This is an automated email from the ASF dual-hosted git repository. tallison pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/tika.git
commit 68a4272e4de38f6eb02a93e5bcce76b72dcc5eca Author: tballison <[email protected]> AuthorDate: Thu Sep 2 14:39:02 2021 -0400 TIKA-3543 -- improve robustness and feature set of HttpFetcher --- CHANGES.txt | 2 + .../tika/pipes/emitter/fs/FileSystemEmitter.java | 7 + tika-pipes/tika-fetchers/tika-fetcher-http/pom.xml | 7 + .../tika/pipes/fetcher/http/HttpFetcher.java | 257 +++++++++++++++++++-- .../tika/pipes/fetcher/http/HttpFetcherTest.java | 19 +- .../src/test/resources/tika-config-http.xml | 5 + 6 files changed, 279 insertions(+), 18 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 9adb889..37990f1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,7 @@ Release 2.1.1 - ??? + * Improve robustness and features of the httpfetcher (TIKA-3543) + * Add optional fetch ranges to FetchEmitTuple to allow range fetching from, e.g. http or s3 (TIKA-3542). diff --git a/tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java b/tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java index cf67509..47c5177 100644 --- a/tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java +++ b/tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java @@ -108,6 +108,13 @@ public class FileSystemEmitter extends AbstractEmitter implements StreamEmitter this.fileExtension = fileExtension; } + /** + * What to do if the target file already exists. NOTE: if more than one + * thread is trying write to the same file and {@link ON_EXISTS#REPLACE} is chosen, + * you still might get a {@link FileAlreadyExistsException}. + * + * @param onExists + */ @Field public void setOnExists(String onExists) { switch (onExists) { diff --git a/tika-pipes/tika-fetchers/tika-fetcher-http/pom.xml b/tika-pipes/tika-fetchers/tika-fetcher-http/pom.xml index 34af101..85d2e06 100644 --- a/tika-pipes/tika-fetchers/tika-fetcher-http/pom.xml +++ b/tika-pipes/tika-fetchers/tika-fetcher-http/pom.xml @@ -45,6 +45,13 @@ <artifactId>tika-httpclient-commons</artifactId> <version>${project.version}</version> </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>tika-core</artifactId> + <version>${project.version}</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> </dependencies> <build> diff --git a/tika-pipes/tika-fetchers/tika-fetcher-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java b/tika-pipes/tika-fetchers/tika-fetcher-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java index 55d9f18..ff0ab36 100644 --- a/tika-pipes/tika-fetchers/tika-fetcher-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java +++ b/tika-pipes/tika-fetchers/tika-fetcher-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java @@ -17,16 +17,37 @@ package org.apache.tika.pipes.fetcher.http; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.MalformedURLException; +import java.net.URI; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.io.IOUtils; +import org.apache.http.Header; +import org.apache.http.HttpConnection; +import org.apache.http.HttpEntity; +import org.apache.http.HttpInetConnection; import org.apache.http.HttpResponse; import org.apache.http.client.HttpClient; +import org.apache.http.client.config.RequestConfig; import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.protocol.HttpClientContext; +import org.apache.http.util.EntityUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,8 +58,11 @@ import org.apache.tika.config.InitializableProblemHandler; import org.apache.tika.config.Param; import org.apache.tika.exception.TikaConfigException; import org.apache.tika.exception.TikaException; +import org.apache.tika.exception.TikaTimeoutException; +import org.apache.tika.io.TemporaryResources; import org.apache.tika.io.TikaInputStream; import org.apache.tika.metadata.Metadata; +import org.apache.tika.metadata.Property; import org.apache.tika.pipes.fetcher.AbstractFetcher; import org.apache.tika.pipes.fetcher.RangeFetcher; @@ -47,10 +71,57 @@ import org.apache.tika.pipes.fetcher.RangeFetcher; */ public class HttpFetcher extends AbstractFetcher implements Initializable, RangeFetcher { + public static String HTTP_HEADER_PREFIX = "http-header:"; + + public static String HTTP_FETCH_PREFIX = "http-connection:"; + + /** + * http status code + */ + public static Property HTTP_STATUS_CODE = + Property.externalInteger(HTTP_HEADER_PREFIX + "status-code"); + /** + * Number of redirects + */ + public static Property HTTP_NUM_REDIRECTS = + Property.externalInteger(HTTP_FETCH_PREFIX + "num-redirects"); + + /** + * If there were redirects, this captures the final URL visited + */ + public static Property HTTP_TARGET_URL = + Property.externalText(HTTP_FETCH_PREFIX + "target-url"); + + public static Property HTTP_TARGET_IP_ADDRESS = + Property.externalText(HTTP_FETCH_PREFIX + "target-ip-address"); + + public static Property HTTP_FETCH_TRUNCATED = + Property.externalBoolean(HTTP_FETCH_PREFIX + "fetch-truncated"); + + public static Property HTTP_CONTENT_ENCODING = + Property.externalText(HTTP_HEADER_PREFIX + "content-encoding"); + + public static Property HTTP_CONTENT_TYPE = + Property.externalText(HTTP_HEADER_PREFIX + "content-type"); + + Logger LOG = LoggerFactory.getLogger(HttpFetcher.class); private HttpClientFactory httpClientFactory; private HttpClient httpClient; + private int maxRedirects = 10; + //overall timeout in milliseconds + private long overallTimeout = -1; + + private long maxSpoolSize = -1; + + //max string length to read from a result if the + //status code was not in the 200 range + private int maxErrMsgSize = 10000; + + //httpHeaders to capture in the metadata + private Set<String> httpHeaders = new HashSet<>(); + public HttpFetcher() throws TikaConfigException { httpClientFactory = new HttpClientFactory(); } @@ -58,7 +129,12 @@ public class HttpFetcher extends AbstractFetcher implements Initializable, Range @Override public InputStream fetch(String fetchKey, Metadata metadata) throws IOException, TikaException { HttpGet get = new HttpGet(fetchKey); - return get(get); + RequestConfig requestConfig = + RequestConfig.custom() + .setMaxRedirects(maxRedirects) + .setRedirectsEnabled(true).build(); + get.setConfig(requestConfig); + return get(get, metadata); } @Override @@ -66,36 +142,137 @@ public class HttpFetcher extends AbstractFetcher implements Initializable, Range throws IOException, TikaException { HttpGet get = new HttpGet(fetchKey); get.setHeader("Range", "bytes=" + startRange + "-" + endRange); - return get(get); + return get(get, metadata); } - private InputStream get(HttpGet get) throws IOException, TikaException { - HttpResponse response = httpClient.execute(get); - int code = response.getStatusLine().getStatusCode(); - if (code < 200 || code > 299) { - throw new IOException("bad status code: " + code + " :: " + - responseToString(response.getEntity().getContent())); + private InputStream get(HttpGet get, Metadata metadata) throws IOException, TikaException { + HttpClientContext context = HttpClientContext.create(); + HttpResponse response = null; + final AtomicBoolean timeout = new AtomicBoolean(false); + Timer timer = null; + try { + if (overallTimeout > -1) { + TimerTask task = new TimerTask() { + @Override + public void run() { + timeout.set(true); + if (get != null) { + get.abort(); + } + + } + }; + timer = new Timer(false); + timer.schedule(task, overallTimeout); + } + response = httpClient.execute(get, context); + + updateMetadata(get.getURI().toString(), response, context, metadata); + + int code = response.getStatusLine().getStatusCode(); + if (code < 200 || code > 299) { + throw new IOException("bad status code: " + code + " :: " + + responseToString(response)); + } + try (InputStream is = response.getEntity().getContent()) { + return spool(is, metadata); + } + } catch (IOException e) { + if (timeout.get() == true) { + throw new TikaTimeoutException("Overall timeout after " + overallTimeout + "ms"); + } else { + throw e; + } + } finally { + if (timer != null) { + timer.cancel(); + timer.purge(); + } + if (response != null && response instanceof CloseableHttpResponse) { + ((CloseableHttpResponse) response).close(); + } } + } - //spool to local + private InputStream spool(InputStream content, Metadata metadata) throws IOException { long start = System.currentTimeMillis(); - TikaInputStream tis = TikaInputStream.get(response.getEntity().getContent()); - tis.getPath(); - if (response instanceof CloseableHttpResponse) { - ((CloseableHttpResponse) response).close(); + TemporaryResources tmp = new TemporaryResources(); + Path tmpFile = tmp.createTempFile(); + if (maxSpoolSize < 0) { + Files.copy(content, tmpFile, StandardCopyOption.REPLACE_EXISTING); + } else { + try (OutputStream os = Files.newOutputStream(tmpFile)) { + long totalRead = IOUtils.copyLarge(content, os, 0, maxSpoolSize); + if (totalRead == maxSpoolSize && content.read() != -1) { + metadata.set(HTTP_FETCH_TRUNCATED, "true"); + } + } } long elapsed = System.currentTimeMillis() - start; LOG.debug("took {} ms to copy to local tmp file", elapsed); - return tis; + return TikaInputStream.get(tmpFile, metadata, tmp); } - private String responseToString(InputStream is) { - try { - return IOUtils.toString(is, StandardCharsets.UTF_8); + private void updateMetadata(String url, HttpResponse response, HttpClientContext context, + Metadata metadata) { + + metadata.set(HTTP_STATUS_CODE, response.getStatusLine().getStatusCode()); + HttpEntity entity = response.getEntity(); + if (entity.getContentEncoding() != null) { + metadata.set(HTTP_CONTENT_ENCODING, entity.getContentEncoding().getValue()); + } + if (entity.getContentType() != null) { + metadata.set(HTTP_CONTENT_TYPE, entity.getContentType().getValue()); + } + + //load headers + for (String h : httpHeaders) { + Header[] headers = response.getHeaders(h); + if (headers != null && headers.length > 0) { + String name = HTTP_HEADER_PREFIX + h; + for (Header header : headers) { + metadata.add(name, header.getValue()); + } + } + } + List<URI> uriList = context.getRedirectLocations(); + if (uriList == null) { + metadata.set(HTTP_NUM_REDIRECTS, 0); + metadata.set(HTTP_TARGET_URL, url); + } else { + metadata.set(HTTP_NUM_REDIRECTS, uriList.size()); + try { + metadata.set(HTTP_TARGET_URL, uriList.get(uriList.size() - 1).toURL().toString()); + } catch (MalformedURLException e) { + //swallow + } + } + HttpConnection connection = context.getConnection(); + if (connection instanceof HttpInetConnection) { + InetAddress inetAddress = ((HttpInetConnection)connection).getRemoteAddress(); + if (inetAddress != null) { + metadata.set(HTTP_TARGET_IP_ADDRESS, inetAddress.getHostAddress()); + } + } + + } + + private String responseToString(HttpResponse response) { + try (InputStream is = response.getEntity().getContent()) { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + IOUtils.copyLarge(is, bos, 0, maxErrMsgSize); + return new String(bos.toByteArray(), StandardCharsets.UTF_8); } catch (IOException e) { LOG.warn("IOexception trying to read error message", e); return ""; + } finally { + try { + EntityUtils.consume(response.getEntity()); + } catch (IOException e) { + //swallow + } } + } @Field @@ -143,6 +320,52 @@ public class HttpFetcher extends AbstractFetcher implements Initializable, Range httpClientFactory.setSocketTimeout(socketTimeout); } + /** + * Set the maximum number of bytes to spool to a temp file. + * If this value is <code>-1</code>, the full stream will be spooled to a temp file + * + * Default size is -1. + * + * @param maxSpoolSize + */ + @Field + public void setMaxSpoolSize(long maxSpoolSize) { + this.maxSpoolSize = maxSpoolSize; + } + + @Field + public void setMaxRedirects(int maxRedirects) { + this.maxRedirects = maxRedirects; + } + + /** + * Which http headers should we capture in the metadata. + * Keys will be prepended with {@link HttpFetcher#HTTP_HEADER_PREFIX} + * + * @param headers + */ + @Field + public void setHttpHeaders(List<String> headers) { + this.httpHeaders.clear(); + this.httpHeaders.addAll(headers); + } + + /** + * This sets an overall timeout on the request. If a server is super slow + * or the file is very long, the other timeouts might not be triggered. + * + * @param overallTimeout + */ + @Field + public void setOverallTimeout(long overallTimeout) { + this.overallTimeout = overallTimeout; + } + + @Field + public void setMaxErrMsgSize(int maxErrMsgSize) { + this.maxErrMsgSize = maxErrMsgSize; + } + @Override public void initialize(Map<String, Param> params) throws TikaConfigException { httpClient = httpClientFactory.build(); diff --git a/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java b/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java index 718dde3..aacebb2 100644 --- a/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java +++ b/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java @@ -18,6 +18,7 @@ package org.apache.tika.pipes.fetcher.http; import static org.junit.jupiter.api.Assertions.assertEquals; +import java.io.ByteArrayOutputStream; import java.io.InputStream; import java.nio.file.Files; import java.nio.file.Path; @@ -25,15 +26,31 @@ import java.nio.file.Paths; import java.nio.file.StandardCopyOption; import java.util.zip.GZIPInputStream; +import org.apache.commons.io.IOUtils; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +import org.apache.tika.TikaTest; import org.apache.tika.io.TemporaryResources; import org.apache.tika.metadata.Metadata; import org.apache.tika.pipes.fetcher.FetcherManager; @Disabled("requires network connectivity") -public class HttpFetcherTest { +public class HttpFetcherTest extends TikaTest { + + @Test + public void testRedirect() throws Exception { + String url = "https://t.co/cvfkWAEIxw?amp=1";; + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + Metadata metadata = new Metadata(); + HttpFetcher httpFetcher = + (HttpFetcher) getFetcherManager("tika-config-http.xml") + .getFetcher("http"); + try (InputStream is = httpFetcher.fetch(url, metadata)) { + IOUtils.copy(is, bos); + } + //debug(metadata); + } @Test public void testRange() throws Exception { diff --git a/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/resources/tika-config-http.xml b/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/resources/tika-config-http.xml index a4f058a..2336899 100644 --- a/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/resources/tika-config-http.xml +++ b/tika-pipes/tika-fetchers/tika-fetcher-http/src/test/resources/tika-config-http.xml @@ -20,6 +20,11 @@ <fetcher class="org.apache.tika.pipes.fetcher.http.HttpFetcher"> <params> <name>http</name> + <httpHeaders> + <header>Connection</header> + <header>Expires</header> + <header>Content-Length</header> + </httpHeaders> </params> </fetcher> </fetchers>
