From fd7f1dd7a7b8ad8ce213dda23ec2f08a79a728c7 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 6 Sep 2026 23:58:36 +0200 Subject: [PATCH] #2134 fetcher.thread.timeout: cancel the okhttp call, bounded helper pool for other protocols The bolt-level fetch timeout ran the protocol call on a single-thread executor owned by each FetcherThread and abandoned it with future.cancel(true) on expiry. okhttp does not honour the interrupt while connecting or reading, so the helper stayed blocked on the socket and the FetcherThread queued behind it at its next fetch: a host that dribbles bytes could take a thread out of service for as long as http.timeout. It also doubled the thread count of the bolt whenever the option was on, and the robots.txt lookup was not covered by the timeout at all. okhttp: fetcher.thread.timeout is applied as a per-call deadline (Call.timeout()), enforced by okio's shared watchdog: on expiry the call is cancelled, the socket closed and the fetching thread gets an InterruptedIOException at once, classified as "Socket timeout fetching". The deadline is clamped to topology.message.timeout.secs so that it can never loosen the client-level callTimeout. With http.content.partial.as.trimmed the content received before the deadline is kept and flagged as trimmed for "time", as it already was for the call timeout. Protocol gains a default supportsFetchTimeout() (false); okhttp returns true when configured, DelegatorProtocol only when every delegate does. Bolts: the timeout machinery moves to a new package-private FetchTimeoutHelpers owned by FetcherBolt and SimpleFetcherBolt. Its call() runs both the robots.txt lookup and the fetch on the calling thread when the protocol enforces the timeout itself, otherwise on a helper thread from one bounded pool per bolt (fetcher.thread.timeout.helpers, default 2 x fetcher.threads.number, 2 for SimpleFetcherBolt, threads created on demand and released after a minute idle). A deadline throws a typed TimeoutException; a full pool rejects at once with SaturatedException, reported as FETCH_ERROR "No fetch helper available". With the default protocol no helper thread is ever created. New fetchhelpers gauge and fetch.timeout / fetch.helper.rejected counters. The fetcher.thread.timeout key is defined in Constants; the public alias FetcherBolt.FETCH_TIMEOUT_PARAM_KEY (#1861) is removed. Documentation for the parameter, which was missing, is added to configuration.adoc. Tests: HttpProtocolFetchTimeoutTest (cancellation at the deadline, clamp to the message timeout, partial content), FetchTimeoutHelpersTest (every branch of call()), fetcher bolt tests with a protocol that hangs and ignores interruption (stuck fetch not blocking the following ones, bounded pool with rejection, hanging robots.txt reported at the timeout), slow robots.txt bounded with okhttp, no helper threads with okhttp, delegator capability. Fixes #2134. --- .../org/apache/stormcrawler/Constants.java | 9 + .../bolt/FetchTimeoutHelpers.java | 157 ++++++++++++++ .../apache/stormcrawler/bolt/FetcherBolt.java | 100 +++------ .../stormcrawler/bolt/SimpleFetcherBolt.java | 67 ++---- .../protocol/DelegatorProtocol.java | 18 ++ .../stormcrawler/protocol/Protocol.java | 10 + .../protocol/okhttp/HttpProtocol.java | 27 +++ core/src/main/resources/crawler-default.yaml | 12 +- .../bolt/AbstractFetcherBoltTest.java | 121 ++++++++++- .../bolt/FetchTimeoutHelpersTest.java | 200 ++++++++++++++++++ .../stormcrawler/bolt/FetcherBoltTest.java | 36 ++++ .../protocol/DelegationProtocolTest.java | 33 +++ .../stormcrawler/protocol/StuckProtocol.java | 72 +++++++ .../okhttp/HttpProtocolFetchTimeoutTest.java | 134 ++++++++++++ docs/src/main/asciidoc/configuration.adoc | 2 + 15 files changed, 882 insertions(+), 116 deletions(-) create mode 100644 core/src/main/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpers.java create mode 100644 core/src/test/java/org/apache/stormcrawler/bolt/FetchTimeoutHelpersTest.java create mode 100644 core/src/test/java/org/apache/stormcrawler/protocol/StuckProtocol.java create mode 100644 core/src/test/java/org/apache/stormcrawler/protocol/okhttp/HttpProtocolFetchTimeoutTest.java 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.