diff --git a/core/src/main/java/org/apache/stormcrawler/Constants.java b/core/src/main/java/org/apache/stormcrawler/Constants.java index 882f2d965..f925e79d9 100644 --- a/core/src/main/java/org/apache/stormcrawler/Constants.java +++ b/core/src/main/java/org/apache/stormcrawler/Constants.java @@ -60,4 +60,13 @@ public class Constants { public static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; private Constants() {} + + /** Hard deadline in seconds for a single fetch, -1 to disable. */ + public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout"; + + /** + * Maximum number of helper threads on which fetches with a timeout are run for protocols that + * can not enforce the timeout themselves. Defaults to twice fetcher.threads.number. + */ + public static final String FETCH_TIMEOUT_HELPERS_PARAM_KEY = "fetcher.thread.timeout.helpers"; } diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java new file mode 100644 index 000000000..bf549c640 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.bolt; + +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.storm.task.TopologyContext; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.metrics.CrawlerMetrics; +import org.apache.stormcrawler.protocol.Protocol; +import org.apache.stormcrawler.util.ConfUtils; + +/** + * Enforces {@code fetcher.thread.timeout} on protocol calls (robots.txt lookup and fetch) for the + * fetcher bolts. + * + *

When the timeout is off, or the protocol enforces it itself (see {@link + * Protocol#supportsFetchTimeout()}, true for the default okhttp protocol), the call runs on the + * calling thread. Otherwise it runs on a helper thread and is abandoned there when the deadline + * passes: the helper stays busy until the protocol gives up on its own, so the pool is bounded + * ({@code fetcher.thread.timeout.helpers}) and a call that finds every helper busy fails at once + * with {@link SaturatedException}. Helper threads are created on demand and released after a minute + * of inactivity: with the default protocol none is ever created. + */ +final class FetchTimeoutHelpers { + + /** Thrown when a call run on a helper thread did not complete within the timeout. */ + static final class TimeoutException extends Exception { + TimeoutException(String url, long timeoutSecs) { + super("Fetch timed out after " + timeoutSecs + "s fetching " + url); + } + } + + /** Thrown when no helper thread is available to run a call with a timeout. */ + static final class SaturatedException extends Exception { + SaturatedException(String url) { + super("No fetch helper available for " + url); + } + } + + private final long timeoutSecs; + private final ThreadPoolExecutor helpers; + + /** + * @param conf the bolt configuration + * @param defaultMaxHelpers pool bound used unless {@code fetcher.thread.timeout.helpers} is set + * @param threadNamePrefix prefix of the helper thread names + */ + FetchTimeoutHelpers(Map conf, int defaultMaxHelpers, String threadNamePrefix) { + this.timeoutSecs = ConfUtils.getLong(conf, Constants.FETCH_TIMEOUT_PARAM_KEY, -1); + final int maxHelpers = + ConfUtils.getInt( + conf, Constants.FETCH_TIMEOUT_HELPERS_PARAM_KEY, defaultMaxHelpers); + final AtomicInteger helperNum = new AtomicInteger(); + this.helpers = + new ThreadPoolExecutor( + 0, + Math.max(1, maxHelpers), + 60L, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + r -> { + Thread t = + new Thread(r, threadNamePrefix + helperNum.incrementAndGet()); + t.setDaemon(true); + return t; + }); + } + + /** Registers the {@code fetchhelpers} gauge: number of helper threads busy with a call. */ + void registerMetrics(TopologyContext context, Map conf, int bucketSecs) { + CrawlerMetrics.registerGauge( + context, conf, "fetchhelpers", helpers::getActiveCount, bucketSecs); + } + + /** Whether a timeout is configured at all. */ + boolean enabled() { + return timeoutSecs > 0; + } + + /** Timeout in seconds, -1 when disabled. */ + long timeoutSecs() { + return timeoutSecs; + } + + /** Pool bound. */ + int maxHelpers() { + return helpers.getMaximumPoolSize(); + } + + /** Largest number of helper threads ever alive. */ + int largestPoolSize() { + return helpers.getLargestPoolSize(); + } + + /** + * Runs a protocol call under the timeout. + * + * @throws SaturatedException when every helper is busy + * @throws TimeoutException when the deadline passed + * @throws Exception the protocol's own exception + */ + T call(Callable call, Protocol protocol, String url) throws Exception { + if (timeoutSecs <= 0 || protocol.supportsFetchTimeout()) { + return call.call(); + } + final Future future; + try { + future = helpers.submit(call); + } catch (RejectedExecutionException e) { + throw new SaturatedException(url); + } + try { + return future.get(timeoutSecs, TimeUnit.SECONDS); + } catch (java.util.concurrent.TimeoutException e) { + future.cancel(true); + throw new TimeoutException(url, timeoutSecs); + } catch (CancellationException e) { + throw new Exception("Fetch cancelled for " + url); + } catch (ExecutionException e) { + // unwrap the real cause so the bolts' classification sees it + Throwable cause = e.getCause(); + if (cause instanceof Exception) { + throw (Exception) cause; + } + throw new Exception(cause); + } + } + + void shutdown() { + helpers.shutdownNow(); + } +} diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java index ac141a4ad..2180d24e7 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -34,14 +34,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.BlockingDeque; -import java.util.concurrent.CancellationException; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingDeque; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; @@ -86,14 +79,6 @@ public class FetcherBolt extends StatusEmitterBolt { */ public static final String QUEUED_TIMEOUT_PARAM_KEY = "fetcher.timeout.queue"; - /** - * Hard timeout in seconds for a single call to {@link Protocol#getProtocolOutput}. If a fetch - * exceeds this duration the thread is interrupted, the URL is marked as FETCH_ERROR, and the - * thread moves on to the next item. A value of {@code -1} (the default) disables the bolt-level - * timeout, relying solely on the protocol-level socket timeouts. - */ - public static final String FETCH_TIMEOUT_PARAM_KEY = "fetcher.thread.timeout"; - /** Key name of the custom crawl delay for a queue that may be present in the metadata. */ private static final String CRAWL_DELAY_KEY_NAME = "crawl.delay"; @@ -129,6 +114,14 @@ public class FetcherBolt extends StatusEmitterBolt { private String[] beingFetched; + /** Runs protocol calls under fetcher.thread.timeout, see {@link FetchTimeoutHelpers}. */ + private FetchTimeoutHelpers fetchHelpers; + + /** Largest number of helper threads ever alive; for tests. */ + int helperPoolSize() { + return fetchHelpers == null ? 0 : fetchHelpers.largestPoolSize(); + } + @Override public Map getComponentConfiguration() { Config conf = new Config(); @@ -498,15 +491,6 @@ private class FetcherThread extends Thread { private long timeoutInQueues = -1; - /** Hard timeout in seconds for a single protocol fetch. -1 means disabled. */ - private long fetchTimeout = -1; - - /** - * Single-thread executor used to run the protocol call so that it can be interrupted via - * {@link Future#cancel(boolean)} when the bolt-level timeout fires. - */ - private final ExecutorService fetchExecutor; - // by default remains as is-pre 1.17 private String protocolMetadataPrefix = ""; @@ -520,24 +504,11 @@ public FetcherThread(Config conf, int num) { this.crawlDelayForce = ConfUtils.getBoolean(conf, "fetcher.server.delay.force", false); this.threadNum = num; timeoutInQueues = ConfUtils.getLong(conf, QUEUED_TIMEOUT_PARAM_KEY, timeoutInQueues); - fetchTimeout = ConfUtils.getLong(conf, FETCH_TIMEOUT_PARAM_KEY, fetchTimeout); protocolMetadataPrefix = ConfUtils.getString( conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, protocolMetadataPrefix); - - if (fetchTimeout > 0) { - fetchExecutor = - Executors.newSingleThreadExecutor( - r -> { - Thread t = new Thread(r, "FetcherTimeout #" + num); - t.setDaemon(true); - return t; - }); - } else { - fetchExecutor = null; - } } @Override @@ -598,7 +569,9 @@ public void run() { "No protocol implementation found for " + fit.url); } - BaseRobotRules rules = protocol.getRobotRules(fit.url); + BaseRobotRules rules = + fetchHelpers.call( + () -> protocol.getRobotRules(fit.url), protocol, fit.url); boolean fromCache = false; if (rules instanceof RobotRules && ((RobotRules) rules).getContentLengthFetched().length == 0) { @@ -733,33 +706,11 @@ public void run() { final Metadata fetchMetadata = metadata; ProtocolResponse response; - if (fetchExecutor != null) { - Future future = - fetchExecutor.submit( - () -> protocol.getProtocolOutput(fit.url, fetchMetadata)); - try { - response = future.get(fetchTimeout, TimeUnit.SECONDS); - } catch (TimeoutException e) { - future.cancel(true); - throw new Exception( - "Fetch timed out after " - + fetchTimeout - + "s fetching " - + fit.url, - e); - } catch (CancellationException e) { - throw new Exception("Fetch cancelled for " + fit.url); - } catch (ExecutionException e) { - // unwrap the real cause so existing catch logic handles it - Throwable cause = e.getCause(); - if (cause instanceof Exception) { - throw (Exception) cause; - } - throw new Exception(cause); - } - } else { - response = protocol.getProtocolOutput(fit.url, metadata); - } + response = + fetchHelpers.call( + () -> protocol.getProtocolOutput(fit.url, fetchMetadata), + protocol, + fit.url); long timeFetching = System.currentTimeMillis() - start; @@ -871,10 +822,19 @@ public void run() { } // common exceptions for which we log only a short message - if (exece.getCause() instanceof java.util.concurrent.TimeoutException + if (exece instanceof FetchTimeoutHelpers.TimeoutException + || exece instanceof java.io.InterruptedIOException || message.contains(" timed out")) { LOG.info("Socket timeout fetching {}", fit.url); message = "Socket timeout fetching"; + eventCounter.scope("fetch.timeout").incrBy(1); + } else if (exece instanceof FetchTimeoutHelpers.SaturatedException) { + eventCounter.scope("fetch.helper.rejected").incrBy(1); + LOG.warn( + "{}: all {} fetch helpers are busy", + message, + fetchHelpers.maxHelpers()); + message = "No fetch helper available"; } else if (exece.getCause() instanceof java.net.UnknownHostException || exece instanceof java.net.UnknownHostException) { LOG.info("Unknown host {}", fit.url); @@ -984,6 +944,11 @@ public void prepare( int threadCount = ConfUtils.getInt(conf, "fetcher.threads.number", 10); int startDelay = ConfUtils.getInt(conf, "fetcher.threads.start.delay", 10); + // helpers for protocols which can not cancel a fetch themselves; no thread until needed + fetchHelpers = + new FetchTimeoutHelpers(conf, Math.max(1, threadCount * 2), "FetcherTimeout-"); + fetchHelpers.registerMetrics(context, stormConf, metricsTimeBucketSecs); + for (int i = 0; i < threadCount; i++) { if (startDelay > 0 && i > 0) { // short delay to avoid that DNS or other resources are temporarily @@ -1030,6 +995,9 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { @Override public void cleanup() { super.cleanup(); + if (fetchHelpers != null) { + fetchHelpers.shutdown(); + } protocolFactory.cleanup(); } diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java index 6c2dae6e2..e1524f570 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java @@ -28,13 +28,7 @@ import java.text.SimpleDateFormat; import java.util.Locale; import java.util.Map; -import java.util.concurrent.CancellationException; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.lang3.StringUtils; import org.apache.http.HttpHeaders; @@ -126,10 +120,8 @@ public class SimpleFetcherBolt extends StatusEmitterBolt { // by default remains as is-pre 1.17 private String protocolMetadataPrefix = ""; - /** Hard timeout in seconds for a single protocol fetch. -1 means disabled. */ - private long fetchTimeout = -1; - - private ExecutorService fetchExecutor; + /** Runs protocol calls under fetcher.thread.timeout, see {@link FetchTimeoutHelpers}. */ + private FetchTimeoutHelpers fetchHelpers; private void checkConfiguration() { @@ -222,17 +214,8 @@ public void prepare( ConfUtils.getString( conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, protocolMetadataPrefix); - this.fetchTimeout = - ConfUtils.getLong(conf, FetcherBolt.FETCH_TIMEOUT_PARAM_KEY, fetchTimeout); - if (fetchTimeout > 0) { - fetchExecutor = - Executors.newSingleThreadExecutor( - r -> { - Thread t = new Thread(r, "SimpleFetcherTimeout #" + taskId); - t.setDaemon(true); - return t; - }); - } + fetchHelpers = new FetchTimeoutHelpers(conf, 2, "SimpleFetcherTimeout-" + taskId + "-"); + fetchHelpers.registerMetrics(context, conf, metricsTimeBucketSecs); } @Override @@ -246,8 +229,8 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { public void cleanup() { super.cleanup(); protocolFactory.cleanup(); - if (fetchExecutor != null) { - fetchExecutor.shutdownNow(); + if (fetchHelpers != null) { + fetchHelpers.shutdown(); } } @@ -301,7 +284,8 @@ public void execute(Tuple input) { Protocol protocol = protocolFactory.getProtocol(url); - BaseRobotRules rules = protocol.getRobotRules(urlString); + BaseRobotRules rules = + fetchHelpers.call(() -> protocol.getRobotRules(urlString), protocol, urlString); boolean fromCache = false; if (rules instanceof RobotRules && ((RobotRules) rules).getContentLengthFetched().length == 0) { @@ -446,28 +430,11 @@ public void execute(Tuple input) { final String fetchUrl = urlString; final Metadata fetchMetadata = metadata; ProtocolResponse response; - if (fetchExecutor != null) { - Future future = - fetchExecutor.submit( - () -> protocol.getProtocolOutput(fetchUrl, fetchMetadata)); - try { - response = future.get(fetchTimeout, TimeUnit.SECONDS); - } catch (TimeoutException e) { - future.cancel(true); - throw new Exception( - "Fetch timed out after " + fetchTimeout + "s fetching " + urlString, e); - } catch (CancellationException e) { - throw new Exception("Fetch cancelled for " + urlString); - } catch (ExecutionException e) { - Throwable cause = e.getCause(); - if (cause instanceof Exception) { - throw (Exception) cause; - } - throw new Exception(cause); - } - } else { - response = protocol.getProtocolOutput(urlString, metadata); - } + response = + fetchHelpers.call( + () -> protocol.getProtocolOutput(fetchUrl, fetchMetadata), + protocol, + urlString); long timeFetching = System.currentTimeMillis() - start; final int byteLength = response.getContent().length; @@ -575,10 +542,16 @@ public void execute(Tuple input) { } // common exceptions for which we log only a short message - if (exece.getCause() instanceof java.util.concurrent.TimeoutException + if (exece instanceof FetchTimeoutHelpers.TimeoutException + || exece instanceof java.io.InterruptedIOException || message.contains(" timed out")) { LOG.error("Socket timeout fetching {}", urlString); message = "Socket timeout fetching"; + eventCounter.scope("fetch.timeout").incrBy(1); + } else if (exece instanceof FetchTimeoutHelpers.SaturatedException) { + eventCounter.scope("fetch.helper.rejected").incrBy(1); + LOG.warn("{}: all fetch helpers are busy", message); + message = "No fetch helper available"; } else if (exece.getCause() instanceof java.net.UnknownHostException || exece instanceof java.net.UnknownHostException) { LOG.error("Unknown host {}", urlString); diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java index b19760b93..6a411e856 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/DelegatorProtocol.java @@ -157,6 +157,10 @@ public ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws return protoInstance.getProtocolOutput(url, metadata); } + public boolean supportsFetchTimeout() { + return protoInstance.supportsFetchTimeout(); + } + public BaseRobotRules getRobotRules(String url) { return protoInstance.getRobotRules(url); } @@ -320,6 +324,20 @@ final FilteredProtocol getProtocolFor(String url, Metadata metadata) { return proto.getProtocolOutput(url, metadata); } + /** + * True only when every delegate enforces the fetch timeout itself: the bolt does not know in + * advance which delegate a URL will be routed to. + */ + @Override + public boolean supportsFetchTimeout() { + for (FilteredProtocol p : protocols) { + if (!p.supportsFetchTimeout()) { + return false; + } + } + return true; + } + @Override public void cleanup() { for (FilteredProtocol p : protocols) { diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java index 242395235..20948edf9 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/Protocol.java @@ -53,6 +53,16 @@ public interface Protocol { */ ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws Exception; + /** + * Whether this protocol enforces {@code fetcher.thread.timeout} itself by cancelling the + * request when the deadline passes. When true the fetcher bolts call {@link + * #getProtocolOutput(String, Metadata)} directly instead of running it on a helper thread that + * they abandon on timeout. Defaults to false. + */ + default boolean supportsFetchTimeout() { + return false; + } + BaseRobotRules getRobotRules(String url); void cleanup(); diff --git a/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java b/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java index 90cb4d742..18a8f7c69 100644 --- a/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java +++ b/core/src/main/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocol.java @@ -97,6 +97,9 @@ public class HttpProtocol extends AbstractHttpProtocol { private int completionTimeout = -1; + /** Per-call deadline in seconds from fetcher.thread.timeout, -1 when disabled. */ + private long fetchTimeout = -1; + /** Accept partially fetched content as trimmed content */ private boolean partialContentAsTrimmed = false; @@ -155,6 +158,19 @@ public void configure(Config conf) { this.completionTimeout = ConfUtils.getInt(conf, "topology.message.timeout.secs", completionTimeout); + this.fetchTimeout = + ConfUtils.getLong(conf, Constants.FETCH_TIMEOUT_PARAM_KEY, fetchTimeout); + if (fetchTimeout > 0 && completionTimeout > 0 && fetchTimeout > completionTimeout) { + // the per-call deadline replaces the client-level callTimeout derived from the + // message timeout: never loosen it, Storm would fail the tuple first anyway + LOG.warn( + "{} ({}s) is larger than topology.message.timeout.secs ({}s): using the latter", + Constants.FETCH_TIMEOUT_PARAM_KEY, + fetchTimeout, + completionTimeout); + fetchTimeout = completionTimeout; + } + this.partialContentAsTrimmed = ConfUtils.getBoolean(conf, "http.content.partial.as.trimmed", false); @@ -348,6 +364,11 @@ protected void addHeadersToRequest(Builder rb, Metadata md) { } } + @Override + public boolean supportsFetchTimeout() { + return fetchTimeout > 0; + } + @Override public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) throws Exception { @@ -468,6 +489,12 @@ public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) final Call call = localClient.newCall(request); + if (fetchTimeout > 0) { + // hard deadline for the whole call, enforced by okio's watchdog: on expiry the call + // is cancelled, the socket closed and execute()/the body read throw immediately + call.timeout().timeout(fetchTimeout, TimeUnit.SECONDS); + } + try (Response response = call.execute()) { final Metadata responsemetadata = new Metadata(); diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml index 6945b2c4c..62d95b9ad 100644 --- a/core/src/main/resources/crawler-default.yaml +++ b/core/src/main/resources/crawler-default.yaml @@ -29,9 +29,17 @@ config: fetcher.max.urls.in.queues: -1 fetcher.max.queue.size: -1 fetcher.timeout.queue: -1 - # hard timeout in seconds for a single protocol fetch at the bolt level; - # -1 disables (relies on protocol-level socket timeouts only) + # hard timeout in seconds for a single fetch (robots.txt lookup included), + # independent of the protocol's socket timeouts; -1 disables it. Never + # exceeds topology.message.timeout.secs. With okhttp the HTTP call itself + # is cancelled when the deadline passes (socket closed, URL reported as + # FETCH_ERROR); other protocols run on a helper thread and are abandoned + # there on timeout fetcher.thread.timeout: -1 + # max helper threads per bolt for the abandoned fetches above; default is + # 2 x fetcher.threads.number (2 for SimpleFetcherBolt). When all are busy + # the URL is reported as FETCH_ERROR "No fetch helper available" + # fetcher.thread.timeout.helpers: 100 # max. crawl-delay accepted in robots.txt (in seconds) fetcher.max.crawl.delay: 30 # behavior of fetcher when the crawl-delay in the robots.txt diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java index 35cd7b5f5..8dbb7f3e5 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java @@ -43,6 +43,7 @@ import org.apache.stormcrawler.TestUtil; import org.apache.stormcrawler.persistence.Status; import org.apache.stormcrawler.protocol.ProtocolFactory; +import org.apache.stormcrawler.protocol.StuckProtocol; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -53,8 +54,10 @@ abstract class AbstractFetcherBoltTest { BaseRichBolt bolt; @AfterEach - void cleanupParserBolt() { + void cleanupParserBolt() throws ReflectiveOperationException { bolt.cleanup(); + // the factory is a singleton configured once: never leak a protocol into the next test + resetProtocolFactory(); } @Test @@ -153,6 +156,122 @@ void testThreadTimeout(WireMockRuntimeInfo wmRuntimeInfo) { Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); } + /** + * A fetch that hits the bolt-level timeout must not hold up the fetches that follow it: with + * one fetcher thread, a stuck fetch followed by two fast ones must yield two pages and one + * FETCH_ERROR within a few seconds, not one FETCH_ERROR per URL. + */ + @Test + void stuckFetchDoesNotBlockTheFollowingFetches(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlMatching("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(10_000))); + stubFor(get(urlMatching("/fast.*")).willReturn(aResponse().withStatus(200).withBody("ok"))); + + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + config.put("http.timeout", 30_000); + // same host: the second and third URL wait for the first one to release the queue + config.put("fetcher.server.delay", 0.0f); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + String base = "http://localhost:" + wmRuntimeInfo.getHttpPort(); + for (String path : new String[] {"/slow", "/fast1", "/fast2"}) { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn(base + path); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + } + + await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 3); + + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, statusTuples.size(), "only the slow URL should fail"); + Assertions.assertEquals(base + "/slow", statusTuples.get(0).get(0)); + Assertions.assertEquals(Status.FETCH_ERROR, statusTuples.get(0).get(2)); + Assertions.assertEquals(2, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); + } + + /** + * With a protocol that cannot be cancelled, timed-out fetches are abandoned on helper threads + * from a bounded pool shared by the bolt: every fetch actually starts until the pool is full, + * and the next one is rejected right away instead of queueing behind a stuck helper. + */ + @Test + void abandonedFetchesUseABoundedSharedPool() throws ReflectiveOperationException { + StuckProtocol.STARTED.set(0); + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.protocol.implementation", StuckProtocol.class.getName()); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + config.put("fetcher.thread.timeout.helpers", 2); + config.put("fetcher.server.delay", 0.0f); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + for (String path : new String[] {"/1", "/2", "/3"}) { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn("http://stuck.example.com" + path); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + } + + await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 3); + + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(3, statusTuples.size()); + for (List t : statusTuples) { + Assertions.assertEquals(Status.FETCH_ERROR, t.get(2)); + } + // pool of 2 (twice the fetcher threads): the first two fetches really started + Assertions.assertEquals(2, StuckProtocol.STARTED.get(), "fetches actually started"); + // the third found no free helper and was rejected with an explicit cause + long rejected = + statusTuples.stream() + .map(t -> ((Metadata) t.get(1)).getFirstValue("fetch.exception")) + .filter(m -> m != null && m.contains("helper")) + .count(); + Assertions.assertEquals(1, rejected, "one fetch rejected by the saturated pool"); + } + + /** The robots.txt lookup is part of the fetch: it must be covered by the timeout too. */ + @Test + void hangingRobotsLookupIsReportedAtTheTimeout() throws ReflectiveOperationException { + StuckProtocol.HANG_ROBOTS = true; + try { + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.protocol.implementation", StuckProtocol.class.getName()); + config.put("fetcher.threads.number", 1); + config.put("fetcher.thread.timeout", 1L); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")).thenReturn("http://stuck.example.com/robots"); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + + await().atMost(6, TimeUnit.SECONDS).until(() -> output.getAckedTuples().size() == 1); + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, statusTuples.size()); + Assertions.assertEquals(Status.FETCH_ERROR, statusTuples.get(0).get(2)); + } finally { + StuckProtocol.HANG_ROBOTS = false; + } + } + @Test void invalidProxyMetadataEmitsFetchError(WireMockRuntimeInfo wmRuntimeInfo) throws ReflectiveOperationException { diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java new file mode 100644 index 000000000..9fb24adbe --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.bolt; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.protocol.DummyProtocol; +import org.apache.stormcrawler.protocol.Protocol; +import org.apache.stormcrawler.protocol.StuckProtocol; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(value = 20, unit = TimeUnit.SECONDS) +class FetchTimeoutHelpersTest { + + /** A protocol which claims to enforce the timeout itself. */ + private static final Protocol SELF_TIMING = + new DummyProtocol() { + @Override + public boolean supportsFetchTimeout() { + return true; + } + }; + + private static final Protocol PLAIN = new DummyProtocol(); + + private FetchTimeoutHelpers helpers; + + private static Map conf(long timeoutSecs, Integer maxHelpers) { + Map conf = new HashMap<>(); + conf.put(Constants.FETCH_TIMEOUT_PARAM_KEY, timeoutSecs); + if (maxHelpers != null) { + conf.put(Constants.FETCH_TIMEOUT_HELPERS_PARAM_KEY, maxHelpers); + } + return conf; + } + + private FetchTimeoutHelpers helpers(long timeoutSecs, Integer maxHelpers) { + helpers = new FetchTimeoutHelpers(conf(timeoutSecs, maxHelpers), 4, "test-helper-"); + return helpers; + } + + @AfterEach + void shutdown() { + if (helpers != null) { + helpers.shutdown(); + } + } + + @Test + void disabledRunsOnTheCallingThread() throws Exception { + FetchTimeoutHelpers h = helpers(-1, null); + Assertions.assertFalse(h.enabled()); + Thread caller = Thread.currentThread(); + Thread ran = h.call(Thread::currentThread, PLAIN, "http://a.net/"); + Assertions.assertSame(caller, ran); + Assertions.assertEquals(0, h.largestPoolSize()); + } + + @Test + void protocolEnforcingTheTimeoutRunsOnTheCallingThread() throws Exception { + FetchTimeoutHelpers h = helpers(1, null); + Assertions.assertTrue(h.enabled()); + Assertions.assertEquals(1, h.timeoutSecs()); + Thread ran = h.call(Thread::currentThread, SELF_TIMING, "http://a.net/"); + Assertions.assertSame(Thread.currentThread(), ran); + Assertions.assertEquals(0, h.largestPoolSize()); + } + + @Test + void otherProtocolsRunOnAHelperAndReturnTheResult() throws Exception { + FetchTimeoutHelpers h = helpers(5, null); + Thread ran = h.call(Thread::currentThread, PLAIN, "http://a.net/"); + Assertions.assertNotSame(Thread.currentThread(), ran); + Assertions.assertTrue(ran.getName().startsWith("test-helper-"), ran.getName()); + Assertions.assertTrue(ran.isDaemon()); + Assertions.assertEquals(1, h.largestPoolSize()); + } + + @Test + void protocolExceptionsPropagateUnwrapped() { + FetchTimeoutHelpers h = helpers(5, null); + IOException thrown = + Assertions.assertThrows( + IOException.class, + () -> + h.call( + () -> { + throw new IOException("boom"); + }, + PLAIN, + "http://a.net/")); + Assertions.assertEquals("boom", thrown.getMessage()); + } + + @Test + void errorsAreWrappedInAnException() { + FetchTimeoutHelpers h = helpers(5, null); + Exception thrown = + Assertions.assertThrows( + Exception.class, + () -> + h.call( + () -> { + throw new AssertionError("not an exception"); + }, + PLAIN, + "http://a.net/")); + Assertions.assertInstanceOf(AssertionError.class, thrown.getCause()); + } + + @Test + void deadlineThrowsTypedTimeoutAndInterruptsTheHelper() throws Exception { + FetchTimeoutHelpers h = helpers(1, null); + AtomicBoolean interrupted = new AtomicBoolean(); + CountDownLatch done = new CountDownLatch(1); + long start = System.currentTimeMillis(); + FetchTimeoutHelpers.TimeoutException thrown = + Assertions.assertThrows( + FetchTimeoutHelpers.TimeoutException.class, + () -> + h.call( + () -> { + try { + Thread.sleep(10_000); + } catch (InterruptedException e) { + interrupted.set(true); + } + done.countDown(); + return null; + }, + PLAIN, + "http://a.net/page")); + Assertions.assertTrue(System.currentTimeMillis() - start < 3_000); + Assertions.assertTrue(thrown.getMessage().contains("http://a.net/page")); + // the helper is asked to stop, even though not every protocol honours it + Assertions.assertTrue(done.await(5, TimeUnit.SECONDS)); + Assertions.assertTrue(interrupted.get()); + } + + @Test + void saturationFailsFastWithTypedException() throws Exception { + FetchTimeoutHelpers h = helpers(1, 2); + Assertions.assertEquals(2, h.maxHelpers()); + StuckProtocol stuck = new StuckProtocol(); + // fill the two helpers with fetches that never return + for (int i = 0; i < 2; i++) { + Assertions.assertThrows( + FetchTimeoutHelpers.TimeoutException.class, + () -> h.call(() -> stuck.getProtocolOutput("http://a.net/", null), stuck, "u")); + } + long start = System.currentTimeMillis(); + Assertions.assertThrows( + FetchTimeoutHelpers.SaturatedException.class, + () -> h.call(() -> "never", stuck, "http://a.net/3")); + Assertions.assertTrue(System.currentTimeMillis() - start < 500, "rejected immediately"); + Assertions.assertEquals(2, h.largestPoolSize()); + } + + @Test + void poolBoundComesFromConfigOrDefault() { + Assertions.assertEquals(4, helpers(1, null).maxHelpers()); + helpers.shutdown(); + Assertions.assertEquals(7, helpers(1, 7).maxHelpers()); + helpers.shutdown(); + // never less than one helper + Assertions.assertEquals(1, helpers(1, 0).maxHelpers()); + } + + @Test + void afterShutdownCallsAreRejected() { + FetchTimeoutHelpers h = helpers(1, null); + h.shutdown(); + Assertions.assertThrows( + FetchTimeoutHelpers.SaturatedException.class, + () -> h.call(() -> "x", PLAIN, "http://a.net/")); + } +} diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java index e7f592e14..9d8ddd719 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java @@ -31,6 +31,7 @@ import org.apache.stormcrawler.Constants; import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.persistence.Status; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -138,4 +139,39 @@ void unforcedLongCrawlDelayStillEmitsCrawlDelayErrorWithoutMetadata( assertEquals("crawl_delay", md.getFirstValue(Constants.STATUS_ERROR_CAUSE)); assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); } + + @Test + void noHelperThreadsWithOkhttpAndFetchTimeout(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.thread.timeout", 5L); + fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertEquals( + 0, + ((FetcherBolt) bolt).helperPoolSize(), + "okhttp cancels the call itself: no helper threads expected"); + } + + /** With okhttp the robots.txt fetch goes through the same call deadline as the page. */ + @Test + void slowRobotsTxtIsBoundedByTheFetchTimeoutWithOkhttp(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn(aResponse().withStatus(200).withFixedDelay(10_000))); + stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("http.timeout", 30_000); + config.put("fetcher.thread.timeout", 1L); + long start = System.currentTimeMillis(); + // the robots.txt lookup fails at the deadline; the parser then allows the fetch + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + Assertions.assertNotNull(md); + Assertions.assertTrue( + System.currentTimeMillis() - start < 6_000, "robots.txt lookup was not bounded"); + assertEquals(0, ((FetcherBolt) bolt).helperPoolSize()); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java index 9b6ba8bce..98a6226d3 100644 --- a/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java +++ b/core/src/test/java/org/apache/stormcrawler/protocol/DelegationProtocolTest.java @@ -18,6 +18,10 @@ package org.apache.stormcrawler.protocol; import java.io.FileNotFoundException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import org.apache.storm.Config; import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.protocol.DelegatorProtocol.FilteredProtocol; @@ -73,4 +77,33 @@ void getProtocolTest() throws FileNotFoundException { pf = superProto.getProtocolFor("https://www.example-two.com/large.doc", meta); Assertions.assertEquals("fourth", pf.id); } + + private static DelegatorProtocol delegator(String... classNames) { + Config conf = new Config(); + conf.put("http.agent.name", "this_is_only_a_test"); + conf.put("fetcher.thread.timeout", 1L); + List> entries = new ArrayList<>(); + for (int i = 0; i < classNames.length; i++) { + Map entry = new HashMap<>(); + entry.put("className", classNames[i]); + entry.put("id", "p" + i); + if (i < classNames.length - 1) { + entry.put("filters", Map.of("key" + i, "value")); + } + entries.add(entry); + } + conf.put("protocol.delegator.config", entries); + DelegatorProtocol delegator = new DelegatorProtocol(); + delegator.configure(conf); + return delegator; + } + + @Test + void supportsFetchTimeoutOnlyWhenEveryDelegateDoes() { + String okhttp = org.apache.stormcrawler.protocol.okhttp.HttpProtocol.class.getName(); + String dummy = DummyProtocol.class.getName(); + Assertions.assertTrue(delegator(okhttp, okhttp).supportsFetchTimeout()); + Assertions.assertFalse(delegator(dummy, okhttp).supportsFetchTimeout()); + Assertions.assertFalse(delegator(okhttp, dummy).supportsFetchTimeout()); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java new file mode 100644 index 000000000..30a40ac4e --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.protocol; + +import crawlercommons.robots.BaseRobotRules; +import crawlercommons.robots.SimpleRobotRules; +import crawlercommons.robots.SimpleRobotRules.RobotRulesMode; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.storm.Config; +import org.apache.stormcrawler.Metadata; + +/** + * A protocol whose fetches hang for a long time and ignore interruption, like a client blocked on a + * socket. Counts the fetches that actually started. + */ +public class StuckProtocol implements Protocol { + + public static final AtomicInteger STARTED = new AtomicInteger(); + + public static final long HANG_MILLIS = 10_000; + + /** When true, robots.txt lookups hang as well. */ + public static volatile boolean HANG_ROBOTS = false; + + @Override + public void configure(Config conf) {} + + @Override + public ProtocolResponse getProtocolOutput(String url, Metadata metadata) throws Exception { + STARTED.incrementAndGet(); + hang(); + return new ProtocolResponse("late".getBytes(StandardCharsets.UTF_8), 200, new Metadata()); + } + + @Override + public BaseRobotRules getRobotRules(String url) { + if (HANG_ROBOTS) { + hang(); + } + return new SimpleRobotRules(RobotRulesMode.ALLOW_ALL); + } + + private static void hang() { + long until = System.currentTimeMillis() + HANG_MILLIS; + while (System.currentTimeMillis() < until) { + try { + Thread.sleep(until - System.currentTimeMillis()); + } catch (InterruptedException e) { + // ignored on purpose: an interrupt does not free a blocked socket read either + } + } + } + + @Override + public void cleanup() {} +} diff --git a/core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java b/core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java new file mode 100644 index 000000000..6da59e994 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.protocol.okhttp; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.get; +import static com.github.tomakehurst.wiremock.client.WireMock.stubFor; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; + +import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; +import com.github.tomakehurst.wiremock.junit5.WireMockTest; +import java.io.InterruptedIOException; +import org.apache.storm.Config; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.protocol.ProtocolResponse; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +@WireMockTest +class HttpProtocolFetchTimeoutTest { + + private static HttpProtocol protocol(long fetchTimeoutSecs) { + return protocol(fetchTimeoutSecs, new Config()); + } + + private static HttpProtocol protocol(long fetchTimeoutSecs, Config conf) { + conf.put("http.agent.name", "this_is_only_a_test"); + // socket timeouts far above the fetch timeout so that only the latter can fire + conf.put("http.timeout", 30_000); + if (fetchTimeoutSecs > 0) { + conf.put(Constants.FETCH_TIMEOUT_PARAM_KEY, fetchTimeoutSecs); + } + HttpProtocol protocol = new HttpProtocol(); + protocol.configure(conf); + return protocol; + } + + @Test + void supportsFetchTimeoutOnlyWhenConfigured() { + Assertions.assertTrue(protocol(1).supportsFetchTimeout()); + Assertions.assertFalse(protocol(-1).supportsFetchTimeout()); + } + + @Test + void slowResponseIsCancelledAtTheFetchTimeout(WireMockRuntimeInfo wm) { + stubFor( + get(urlEqualTo("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(5_000))); + HttpProtocol protocol = protocol(1); + long start = System.currentTimeMillis(); + Exception thrown = + Assertions.assertThrows( + Exception.class, + () -> + protocol.getProtocolOutput( + wm.getHttpBaseUrl() + "/slow", new Metadata())); + long elapsed = System.currentTimeMillis() - start; + Assertions.assertTrue( + elapsed < 3_000, "fetch was not cancelled at the timeout, took " + elapsed + " ms"); + Assertions.assertInstanceOf(InterruptedIOException.class, thrown); + } + + /** The fetch timeout must not loosen the deadline derived from the message timeout. */ + @Test + void fetchTimeoutIsClampedToTheMessageTimeout(WireMockRuntimeInfo wm) { + stubFor( + get(urlEqualTo("/slow")) + .willReturn(aResponse().withStatus(200).withFixedDelay(5_000))); + Config conf = new Config(); + conf.put("topology.message.timeout.secs", 1); + HttpProtocol protocol = protocol(60, conf); + long start = System.currentTimeMillis(); + Assertions.assertThrows( + Exception.class, + () -> protocol.getProtocolOutput(wm.getHttpBaseUrl() + "/slow", new Metadata())); + long elapsed = System.currentTimeMillis() - start; + Assertions.assertTrue(elapsed < 3_000, "message timeout was loosened, took " + elapsed); + } + + /** + * With http.content.partial.as.trimmed the content received before the deadline is kept and + * flagged as trimmed for "time"; without it the deadline is a plain failure. + */ + @Test + void deadlineDuringBodyHonoursPartialContentAsTrimmed(WireMockRuntimeInfo wm) throws Exception { + byte[] body = new byte[20_000]; + stubFor( + get(urlEqualTo("/dribble")) + .willReturn( + aResponse() + .withStatus(200) + // uncompressed, so that bytes reach the buffer as they + // arrive + .withHeader("Content-Encoding", "identity") + .withBody(body) + .withChunkedDribbleDelay(40, 8_000))); + Config keep = new Config(); + keep.put("http.content.partial.as.trimmed", true); + ProtocolResponse response = + protocol(1, keep) + .getProtocolOutput(wm.getHttpBaseUrl() + "/dribble", new Metadata()); + Assertions.assertEquals(200, response.getStatusCode()); + Assertions.assertTrue(response.getContent().length < body.length, "content was cut"); + Assertions.assertEquals( + "true", + response.getMetadata().getFirstValue(ProtocolResponse.TRIMMED_RESPONSE_KEY)); + Assertions.assertEquals( + "time", + response.getMetadata().getFirstValue(ProtocolResponse.TRIMMED_RESPONSE_REASON_KEY)); + + Assertions.assertThrows( + Exception.class, + () -> + protocol(1) + .getProtocolOutput( + wm.getHttpBaseUrl() + "/dribble", new Metadata())); + } +} diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index d5996a6fc..bf67c8724 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -192,6 +192,8 @@ is defined. | fetcher.threads.per.queue | 1 | Default number of threads per queue. Can be overridden. | fetcher.threads.start.delay | 10 | Delay (milliseconds) between starting next fetcher thread. Avoids overloading DNS or network resources during fetcher startup when all threads simultaneously start requesting pages. | fetcher.timeout.queue | -1 | Maximum wait time (seconds) for items in the queue. -1 disables timeout. +| fetcher.thread.timeout | -1 | Hard deadline (seconds) for a single fetch, robots.txt lookup included, independent of the protocol's own socket timeouts. -1 disables it. With the default okhttp protocol the deadline is applied to the HTTP call itself: on expiry the call is cancelled, the socket closed and the URL reported as `FETCH_ERROR` with `fetch.exception` "Socket timeout fetching"; if `http.content.partial.as.trimmed` is on, content already received is kept instead and flagged as trimmed for "time". The deadline never exceeds `topology.message.timeout.secs`. For protocols which do not enforce it themselves (see `Protocol.supportsFetchTimeout()`, false for `DelegatorProtocol` unless every delegate supports it) the fetch runs on a helper thread and is abandoned there on timeout. +| fetcher.thread.timeout.helpers | 2 x fetcher.threads.number (2 for SimpleFetcherBolt) | Maximum number of helper threads per bolt instance for the abandoned fetches above. Threads are created on demand and released after a minute idle. When every helper is busy the URL is reported as `FETCH_ERROR` with `fetch.exception` "No fetch helper available": helpers are shared by all hosts, so a host that never answers can make fetches of other hosts fail this way until its helpers time out. The `fetchhelpers` gauge and the `fetch.timeout` and `fetch.helper.rejected` counters make this visible. | fetcherbolt.queue.debug.filepath | "" | Path to a debug log (e.g. /tmp/fetcher-dump-{port}). | http.agent.description | - | Description for the User-Agent header. | http.agent.email | - | Email address in User-Agent header.