diff --git a/core/src/main/java/org/apache/stormcrawler/Constants.java b/core/src/main/java/org/apache/stormcrawler/Constants.java index e06552a6a..882f2d965 100644 --- a/core/src/main/java/org/apache/stormcrawler/Constants.java +++ b/core/src/main/java/org/apache/stormcrawler/Constants.java @@ -29,6 +29,13 @@ public class Constants { public static final String STATUS_ERROR_SOURCE = "error.source"; public static final String STATUS_ERROR_CAUSE = "error.cause"; + /** + * Metadata key holding the robots.txt Crawl-delay in seconds, set by the fetcher bolts when the + * delay exceeds {@code fetcher.max.crawl.delay} and {@code fetcher.max.crawl.delay.force} is + * true, so a frontier-side consumer can enforce it at the source. See #867. + */ + public static final String ROBOTS_CRAWL_DELAY_KEY = "robots.crawl.delay"; + public static final String QUEUE_STREAM_NAME = "queue"; public static final String StatusStreamName = "status"; 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 2439bd76c..ac141a4ad 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -583,6 +583,9 @@ public void run() { // https://github.com/apache/stormcrawler/issues/813 metadata.remove("fetch.exception"); + metadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY); + + String robotsCrawlDelaySecs = null; boolean asap = false; @@ -677,6 +680,12 @@ public void run() { msg); if (force) { fiq.crawlDelay = maxCrawlDelay; + // report the delay the fetcher is not holding, so a frontier-side + // consumer can enforce it at the source (#867) + robotsCrawlDelaySecs = + Long.toString(1L + ((rules.getCrawlDelay() - 1L) / 1000L)); + metadata.setValue( + Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); } else { // pass the info about crawl delay metadata.setValue(Constants.STATUS_ERROR_CAUSE, "crawl_delay"); @@ -792,6 +801,14 @@ public void run() { // metadata persisted or transferred from previous fetches mergedMetadata.putAll(response.getMetadata(), protocolMetadataPrefix); + // Only the locally parsed robots.txt value may populate this control signal. + // A colliding protocol prefix/header must not pace an unrelated queue. + mergedMetadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY); + if (robotsCrawlDelaySecs != null) { + mergedMetadata.setValue( + Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); + } + mergedMetadata.setValue( "fetch.statusCode", Integer.toString(response.getStatusCode())); 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 da40b069a..6c2dae6e2 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java @@ -273,6 +273,9 @@ public void execute(Tuple input) { // https://github.com/apache/stormcrawler/issues/813 metadata.remove("fetch.exception"); + metadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY); + + String robotsCrawlDelaySecs = null; URL url; @@ -401,12 +404,16 @@ public void execute(Tuple input) { // value is negative when not set long robotsDelay = rules.getCrawlDelay(); if (robotsDelay > 0) { - if (robotsDelay > maxCrawlDelay) { + if (robotsDelay > maxCrawlDelay && maxCrawlDelay >= 0) { if (maxCrawlDelayForce) { // cap the value to a maximum // as some sites specify ridiculous values LOG.debug("Delay from robots capped at {} for {}", robotsDelay, url); delay = maxCrawlDelay; + // report the delay the fetcher is not holding, so a frontier-side + // consumer can enforce it at the source (#867) + robotsCrawlDelaySecs = Long.toString(1L + ((robotsDelay - 1L) / 1000L)); + metadata.setValue(Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); } else { LOG.debug( "Skipped URL from queue with overlong crawl-delay ({}): {}", @@ -499,6 +506,13 @@ public void execute(Tuple input) { // persisted or transferred from previous fetches mergedMetadata.putAll(response.getMetadata(), protocolMetadataPrefix); + // Only the locally parsed robots.txt value may populate this control signal. + // A colliding protocol prefix/header must not pace an unrelated queue. + mergedMetadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY); + if (robotsCrawlDelaySecs != null) { + mergedMetadata.setValue(Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); + } + mergedMetadata.setValue("fetch.statusCode", Integer.toString(response.getStatusCode())); mergedMetadata.setValue("fetch.loadingTime", Long.toString(timeFetching)); diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml index 9961320bd..9b176084c 100644 --- a/core/src/main/resources/crawler-default.yaml +++ b/core/src/main/resources/crawler-default.yaml @@ -40,8 +40,11 @@ config: # skip URLs from this queue to avoid that any overlong # crawl-delay throttles the crawler # (if true) - # set the delay to fetcher.max.crawl.delay, - # making fetcher more aggressive than requested + # cap the local delay at fetcher.max.crawl.delay and emit the + # requested value in whole seconds as robots.crawl.delay; + # URLFrontier's QueueRegulatorBolt can forward a capped value when + # its robots crawl-delay integration is explicitly enabled. Without + # that consumer the fetcher remains more aggressive than requested. fetcher.max.crawl.delay.force: false # behavior of fetcher when the crawl-delay in the robots.txt # is smaller (ev. less than one second) than the default delay: 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 4f4aa521b..35cd7b5f5 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java @@ -192,7 +192,59 @@ void invalidProxyMetadataEmitsFetchError(WireMockRuntimeInfo wmRuntimeInfo) Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); } - private void resetProtocolFactory() throws ReflectiveOperationException { + TestOutputCollector fetch( + WireMockRuntimeInfo wmRuntimeInfo, Map config, String path) + throws ReflectiveOperationException { + resetProtocolFactory(); + TestOutputCollector output = new TestOutputCollector(); + bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output)); + + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("source"); + when(tuple.getStringByField("url")) + .thenReturn("http://localhost:" + wmRuntimeInfo.getHttpPort() + path); + when(tuple.getValueByField("metadata")).thenReturn(null); + bolt.execute(tuple); + + await().atMost(30, TimeUnit.SECONDS) + .until( + () -> + output.getEmitted(Utils.DEFAULT_STREAM_ID).size() > 0 + || output.getEmitted(Constants.StatusStreamName).size() + > 0); + return output; + } + + /** + * Fetches a page that is expected to be retrieved successfully (HTTP 200, non-304): the fetcher + * emits it on the default stream (url, content, metadata) for downstream parsing. + */ + Metadata fetchAndGetContentMetadata( + WireMockRuntimeInfo wmRuntimeInfo, Map config, String path) + throws ReflectiveOperationException { + TestOutputCollector output = fetch(wmRuntimeInfo, config, path); + List> contentTuples = output.getEmitted(Utils.DEFAULT_STREAM_ID); + Assertions.assertEquals(1, contentTuples.size()); + Assertions.assertEquals(0, output.getEmitted(Constants.StatusStreamName).size()); + return (Metadata) contentTuples.get(0).get(2); + } + + /** + * Fetches a page that is expected to be rejected before any HTTP request is made (e.g. the + * crawl-delay-too-long guard): the fetcher emits directly on the status stream (url, metadata, + * status). + */ + List fetchAndGetStatusTuple( + WireMockRuntimeInfo wmRuntimeInfo, Map config, String path) + throws ReflectiveOperationException { + TestOutputCollector output = fetch(wmRuntimeInfo, config, path); + List> statusTuples = output.getEmitted(Constants.StatusStreamName); + Assertions.assertEquals(1, statusTuples.size()); + Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size()); + return statusTuples.get(0); + } + + static void resetProtocolFactory() throws ReflectiveOperationException { Field instance = ProtocolFactory.class.getDeclaredField("single_instance"); instance.setAccessible(true); instance.set(null, null); 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 8019bcf34..e7f592e14 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java @@ -17,7 +17,22 @@ package org.apache.stormcrawler.bolt; +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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.persistence.Status; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; public class FetcherBoltTest extends AbstractFetcherBoltTest { @@ -25,4 +40,102 @@ public class FetcherBoltTest extends AbstractFetcherBoltTest { void setUpContext() throws Exception { bolt = new FetcherBolt(); } + + @Test + void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + // robots.txt with a Crawl-delay above fetcher.max.crawl.delay + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 120\n"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "1") + .withBody("hello"))); + + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void fractionalLongCrawlDelayIsRoundedUp(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 30.5\n"))); + 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.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertEquals("31", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 5\n"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "120") + .withBody("hello"))); + + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void unforcedLongCrawlDelayStillEmitsCrawlDelayErrorWithoutMetadata( + WireMockRuntimeInfo wmRuntimeInfo) throws ReflectiveOperationException { + // regression guard: force=false + long delay must keep today's behaviour, i.e. a + // Status.ERROR/crawl_delay emission and no robots.crawl.delay metadata key + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 120\n"))); + 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.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", false); + + List statusTuple = fetchAndGetStatusTuple(wmRuntimeInfo, config, "/page"); + assertEquals(Status.ERROR, statusTuple.get(2)); + Metadata md = (Metadata) statusTuple.get(1); + assertEquals("crawl_delay", md.getFirstValue(Constants.STATUS_ERROR_CAUSE)); + assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } } diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java index da2120228..085ef8056 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java @@ -17,7 +17,20 @@ package org.apache.stormcrawler.bolt; +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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; +import java.util.HashMap; +import java.util.Map; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest { @@ -25,4 +38,97 @@ public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest { void setUpContext() throws Exception { bolt = new SimpleFetcherBolt(); } + + @Test + void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + // robots.txt with a Crawl-delay above fetcher.max.crawl.delay + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 120\n"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "1") + .withBody("hello"))); + + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void fractionalLongCrawlDelayIsRoundedUp(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 30.5\n"))); + 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.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertEquals("31", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 5\n"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "120") + .withBody("hello"))); + + Map config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.max.crawl.delay", 30); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } + + @Test + void negativeMaximumDisablesTheCrawlDelayLimit(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + stubFor( + get(urlEqualTo("/robots.txt")) + .willReturn( + aResponse() + .withStatus(200) + .withBody("User-agent: *\nCrawl-delay: 120\n"))); + 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.max.crawl.delay", -1); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } } diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index fd9e52b61..9f5c0ef7c 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -177,7 +177,7 @@ features you are using in your Apache StormCrawler topology. Some external modul | fetcher.max.crawl.delay | 30 | Maximum accepted Crawl-delay from robots.txt (seconds). If the crawl-delay exceeds this value the behavior depends on the value of fetcher.max.crawl.delay.force. | fetcher.max.crawl.delay.force | false | Configures the behavior of fetcher if the robots.txt crawl-delay exceeds fetcher.max.crawl.delay. If false: the tuple is emitted to the StatusStream -as an ERROR. If true: the queue delay is set to fetcher.max.crawl.delay. +as an ERROR. If true: the fetcher caps its local delay and emits the requested delay in whole seconds as `robots.crawl.delay`; an explicitly enabled URLFrontier `QueueRegulatorBolt` can forward a capped value to the host queue. | fetcher.max.queue.size | -1 | The maximum length of the queue used to store items to be fetched by the FetcherBolt. A setting of -1 sets the length to Integer.MAX_VALUE. | fetcher.max.throttle.sleep | -1 | The maximum amount of time to wait between fetches; if exceeded, the item is sent to the back of the queue. Used in SimpleFetcherBolt. -1 disables it. | fetcher.max.urls.in.queues | -1 | Limits the number of URLs that can be stored in a fetch queue. -1 disables the limit. @@ -472,6 +472,18 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/sql[sql m Integration with link:https://github.com/crawler-commons/url-frontier[URLFrontier], a language-agnostic API for URL management and scheduling. See the link:https://github.com/apache/stormcrawler/tree/main/external/urlfrontier[urlfrontier module] for full details. +`QueueRegulatorBolt` consumes the `queue` stream from the URLFrontier status updater and applies +rate-limit blocks or robots.txt pacing. Robots pacing is enabled when both +`urlfrontier.robots.crawl.delay.enabled` and `fetcher.max.crawl.delay.force` are `true`. It requires `partition.url.mode: byHost`, +`urlfrontier.max.urls.per.bucket: 1`, `robots.crawl.delay` in `metadata.persist`, and a single +URLFrontier endpoint (at most one `urlfrontier.address` entry). The same key +must not match `metadata.transfer`, including wildcards such as `robots.*`, because it is a +per-host control signal. The bolt probes these requirements at startup. A batch size of one +bounds each frontier hand-out but does not recall URLs already emitted or eliminate concurrent +requests while the new delay is reaching the frontier. A custom `metadata.transfer.class` is +responsible for preserving the same persist-only contract for all URLs and values; the startup +probe exercises only representative metadata. + [cols="1,1,3", options="header"] |=== | key | default value | description @@ -480,7 +492,10 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/urlfronti | urlfrontier.port | 7071 | URLFrontier service port. | urlfrontier.address | - | URLFrontier service address (can be used instead of host and port). | urlfrontier.max.buckets | 10 | Number of buckets to request from the frontier. -| urlfrontier.max.urls.per.bucket | 10 | Maximum URLs per bucket. +| urlfrontier.max.urls.per.bucket | 10 | Maximum URLs per bucket. Must be 1 when robots crawl-delay pacing is enabled so one request cannot hand out several URLs from the same host. +| urlfrontier.robots.crawl.delay.enabled | false | Enables forwarding long robots.txt Crawl-delays from the queue stream to URLFrontier. Also requires `fetcher.max.crawl.delay.force: true` and the validated configuration described above. +| urlfrontier.robots.delay.max.secs | 86400 | Upper bound in seconds for a robots crawl-delay forwarded to the frontier. +| urlfrontier.robots.delay.decay.secs | 1800 | Deduplication window in seconds after which an unchanged robots crawl-delay is re-sent to the frontier. | urlfrontier.delay.requestable | - | Time in seconds for which the URLs are locked in URLFrontier. | urlfrontier.cache.expireafter.sec | - | Time in seconds after which the cache expires. | urlfrontier.max.messages.in.flight | - | Maximum number of messages in flight for the StatusUpdater. diff --git a/docs/src/main/asciidoc/extending.adoc b/docs/src/main/asciidoc/extending.adoc index 7277f88b9..59ece5f50 100644 --- a/docs/src/main/asciidoc/extending.adoc +++ b/docs/src/main/asciidoc/extending.adoc @@ -273,7 +273,7 @@ When `robots.txt` specifies a `Crawl-delay`: * If the delay is *smaller* than `fetcher.server.delay` and `fetcher.server.delay.force` is `true`, the larger default delay is used instead * If the delay is *larger* than `fetcher.max.crawl.delay`: ** `fetcher.max.crawl.delay.force: false` (default) — the queue is skipped and the URL gets an ERROR status -** `fetcher.max.crawl.delay.force: true` — the delay is capped at `fetcher.max.crawl.delay` +** `fetcher.max.crawl.delay.force: true` — the fetcher caps its local delay at `fetcher.max.crawl.delay` and reports the requested delay in whole seconds as `robots.crawl.delay`; when `QueueRegulatorBolt` is wired and `urlfrontier.robots.crawl.delay.enabled: true`, a value capped by `urlfrontier.robots.delay.max.secs` is forwarded to the host queue in the frontier ==== Robots.txt Compliance diff --git a/external/urlfrontier/README.md b/external/urlfrontier/README.md index 8647fc040..8731099bc 100644 --- a/external/urlfrontier/README.md +++ b/external/urlfrontier/README.md @@ -24,6 +24,67 @@ urlfrontier.max.buckets: 10 urlfrontier.max.urls.per.bucket:10 ``` +## Robots crawl-delay pacing + +`QueueRegulatorBolt` can pace host queues when a robots.txt Crawl-delay exceeds the fetcher's local +limit. Wire it to the `queue` stream emitted by `StatusUpdaterBolt`: + +```yaml +bolts: + - id: "queue-regulator" + className: "org.apache.stormcrawler.urlfrontier.QueueRegulatorBolt" + parallelism: 1 + +streams: + - from: "status" + to: "queue-regulator" + grouping: + type: FIELDS + args: ["key"] + streamId: "queue" +``` + +Robots pacing is opt-in. It requires host partitioning, one URL per frontier hand-out, a positive +delay cap, and a persist-only control signal: + +```yaml +partition.url.mode: byHost +fetcher.max.crawl.delay.force: true +urlfrontier.robots.crawl.delay.enabled: true +urlfrontier.max.urls.per.bucket: 1 +# cap for the forwarded delay, seconds (default 86400) +urlfrontier.robots.delay.max.secs: 86400 +# dedupe window after which an unchanged delay is re-sent, seconds (default 1800) +urlfrontier.robots.delay.decay.secs: 1800 + +metadata.persist: + - robots.crawl.delay +``` + +The forwarded value is conservative: the maximum observed delay wins, and a lower delay can only +be applied after the previous maximum expires from the decay window. `setDelay(key, 0)` is +not sent if a site later removes its Crawl-delay, so that host can remain slower than necessary but +is not made less polite. + +Do not include `robots.crawl.delay` in `metadata.transfer`, directly or through a wildcard such as +`robots.*`: an outlink must not inherit its parent's host delay. The bolt rejects an unsafe robots +configuration at startup. A custom `metadata.transfer.class` must preserve this contract for every +URL and value; the startup probe can only exercise representative metadata. A batch size of one +bounds each hand-out; it does not recall URLs already emitted. + +Robots pacing also requires a single URLFrontier endpoint — at most one `urlfrontier.address` +entry; the `urlfrontier.host`/`urlfrontier.port` fallback is fine. Keyed `setDelay` calls are +not propagated across URLFrontier nodes +([crawler-commons/url-frontier#146](https://github.com/crawler-commons/url-frontier/issues/146)), +so with several nodes only the queues owned by the connected node would be paced. The bolt fails +fast on multiple configured addresses when pacing is enabled, and only warns otherwise (rate-limit +blocks stay best-effort). A cluster behind a single load-balanced address cannot be detected and +has the same limitation. With several concurrent `getURLs` clients (multiple Spout tasks, or +several topologies on the same frontier) the frontier's per-queue politeness gate is not atomic +([crawler-commons/url-frontier#147](https://github.com/crawler-commons/url-frontier/issues/147)) +and concurrent requests can each be served a URL from the same queue inside the delay window; +prefer a single Spout task per frontier when pacing must be strict. + Your StormCrawler topology requires the following dependency in its pom.xml (just like with any other module) ``` diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java index 8fd7f4deb..79929106e 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java @@ -30,6 +30,7 @@ private Constants() {} // Spout public static final String URLFRONTIER_MAX_URLS_PER_BUCKET_KEY = "urlfrontier.max.urls.per.bucket"; + public static final int URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT = 10; public static final String URLFRONTIER_MAX_BUCKETS_KEY = "urlfrontier.max.buckets"; public static final String URLFRONTIER_DELAY_REQUESTABLE_KEY = "urlfrontier.delay.requestable"; @@ -97,6 +98,27 @@ private Constants() {} public static final boolean URLFRONTIER_BACKOFF_ON_EXCEPTIONS_DEFAULT = false; + /** Enables forwarding long robots.txt Crawl-delays to URLFrontier. Defaults to false. */ + public static final String URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY = + "urlfrontier.robots.crawl.delay.enabled"; + + public static final boolean URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_DEFAULT = false; + + /** Upper bound in seconds for the forwarded robots crawl-delay. Defaults to 86400 (24h). */ + public static final String URLFRONTIER_ROBOTS_DELAY_MAX_KEY = + "urlfrontier.robots.delay.max.secs"; + + public static final int URLFRONTIER_ROBOTS_DELAY_MAX_DEFAULT = 86400; + + /** + * Deduplication window in seconds, after which an unchanged robots crawl-delay is reasserted to + * the frontier. Defaults to 1800 (30m). + */ + public static final String URLFRONTIER_ROBOTS_DELAY_DECAY_KEY = + "urlfrontier.robots.delay.decay.secs"; + + public static final int URLFRONTIER_ROBOTS_DELAY_DECAY_DEFAULT = 1800; + /** * Random extra fraction added to every computed block duration, so that hosts blocked on the * same schedule do not all expire together. Defaults to 0.1; 0 disables jitter. diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java new file mode 100644 index 000000000..75f86c91d --- /dev/null +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java @@ -0,0 +1,216 @@ +/* + * 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.urlfrontier; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Ticker; +import java.time.Duration; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.util.ConfUtils; + +/** + * Decides whether a queue-stream tuple carries a robots.txt crawl-delay worth forwarding to the + * frontier via {@code setDelay}. The fetcher reports the delay (in seconds, {@code + * robots.crawl.delay} metadata) only when it exceeds {@code fetcher.max.crawl.delay} and {@code + * fetcher.max.crawl.delay.force} lets it keep fetching. When frontier pacing is explicitly enabled, + * later hand-outs use the requested interval capped by {@code urlfrontier.robots.delay.max.secs}. + * + *

The signal is conservative — the maximum observed value wins among active signals, preventing + * a shorter delay from lowering the pace of a shared queue. A lower observed delay can be applied + * only after the previously successful maximum expires from the decay window. A failed RPC is + * ambiguous — the server may have applied it before the response was lost — so its value stays as a + * floor for the same window and the next signal reasserts at least that value. There is no + * interaction with {@link HostBackoff}'s blocks. Values are capped by {@code + * urlfrontier.robots.delay.max.secs} (robots can request absurd delays; the parse-time cap is + * disabled in StormCrawler) and deduplicated per host after a successful RPC. Only one request per + * host may be active; changes received meanwhile are coalesced to the maximum value, preventing + * older unary RPCs from overwriting newer values out of order. + * + *

The decision runs on the bolt thread while gRPC callbacks may complete requests concurrently. + * Per-key {@code ConcurrentMap.compute} operations serialize those transitions. + */ +final class CrawlDelayPolicy { + + /** Last delay applied successfully per host; expiry bounds the dedupe window and memory. */ + private final Cache lastApplied; + + /** + * Highest value whose RPC failed ambiguously per host: the server may have applied it before + * the response was lost, so the next signal reasserts at least this floor instead of trusting a + * lower observed value. Expiry mirrors the decay window of {@link #lastApplied}. + */ + private final Cache reassertFloor; + + /** At most one unary RPC per host, plus the maximum value observed while it is active. */ + private final ConcurrentMap active = new ConcurrentHashMap<>(); + + private final AtomicLong requestTokens = new AtomicLong(); + + private final boolean enabled; + + private final long maxSecs; + + record DelayRequest(long token, String key, int delaySecs) {} + + private record Flight(DelayRequest request, Integer pendingMax) {} + + CrawlDelayPolicy(Map conf) { + this(conf, Ticker.systemTicker()); + } + + /** Visible for tests: a controllable ticker drives the dedupe window. */ + CrawlDelayPolicy(Map conf, Ticker ticker) { + this.enabled = + ConfUtils.getBoolean( + conf, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_DEFAULT) + && ConfUtils.getBoolean(conf, "fetcher.max.crawl.delay.force", false); + this.maxSecs = + Math.max( + 1L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_ROBOTS_DELAY_MAX_KEY, + Constants.URLFRONTIER_ROBOTS_DELAY_MAX_DEFAULT)); + long decaySecs = + Math.max( + 0L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_ROBOTS_DELAY_DECAY_KEY, + Constants.URLFRONTIER_ROBOTS_DELAY_DECAY_DEFAULT)); + this.lastApplied = + Caffeine.newBuilder() + .expireAfterWrite(Duration.ofSeconds(decaySecs)) + .ticker(ticker) + .build(); + this.reassertFloor = + Caffeine.newBuilder() + .expireAfterWrite(Duration.ofSeconds(decaySecs)) + .ticker(ticker) + .build(); + } + + /** Returns a request to start now, or empty when it is invalid, deduplicated or coalesced. */ + Optional requestFor(String key, Metadata metadata) { + if (!enabled + || key == null + || metadata == null + || HostBackoff.DEFAULT_QUEUE_KEY.equals(key)) { + return Optional.empty(); + } + final String value = + metadata.getFirstValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY); + if (value == null) { + return Optional.empty(); + } + long secs; + try { + secs = Long.parseLong(value); + } catch (NumberFormatException e) { + return Optional.empty(); + } + if (secs <= 0) { + return Optional.empty(); + } + final int delay = (int) Math.min(secs, Math.min(maxSecs, Integer.MAX_VALUE)); + AtomicReference next = new AtomicReference<>(); + active.compute( + key, + (ignored, flight) -> { + if (flight != null) { + int currentMax = flight.request().delaySecs(); + if (flight.pendingMax() != null) { + currentMax = Math.max(currentMax, flight.pendingMax()); + } + if (delay > currentMax) { + return new Flight(flight.request(), delay); + } + return flight; + } + Integer floor = reassertFloor.getIfPresent(key); + int target = floor == null ? delay : Math.max(delay, floor); + Integer previous = lastApplied.getIfPresent(key); + if (previous != null && previous >= target) { + return null; + } + DelayRequest request = + new DelayRequest(requestTokens.incrementAndGet(), key, target); + next.set(request); + return new Flight(request, null); + }); + return Optional.ofNullable(next.get()); + } + + /** Completes one RPC and returns the maximum coalesced value, if another RPC must follow. */ + Optional completed(DelayRequest completed, boolean success) { + AtomicReference next = new AtomicReference<>(); + active.computeIfPresent( + completed.key(), + (key, flight) -> { + if (flight.request().token() != completed.token()) { + return flight; + } + if (success) { + lastApplied + .asMap() + .compute( + key, + (k, prev) -> { + if (prev == null) { + return completed.delaySecs(); + } + return Math.max(prev, completed.delaySecs()); + }); + reassertFloor + .asMap() + .computeIfPresent( + key, (k, f) -> f <= completed.delaySecs() ? null : f); + } else { + // An error is ambiguous: the server may have applied the request before the + // response was lost. Keep the value as a floor and forget the previous one, + // so the next signal is not deduplicated and reasserts at least this value + // instead of lowering a possibly applied delay before the window expires. + reassertFloor.asMap().merge(key, completed.delaySecs(), Math::max); + lastApplied.invalidate(key); + } + Integer pending = flight.pendingMax(); + if (pending == null || (success && pending <= completed.delaySecs())) { + return null; + } + DelayRequest request = + new DelayRequest(requestTokens.incrementAndGet(), key, pending); + next.set(request); + return new Flight(request, null); + }); + return Optional.ofNullable(next.get()); + } + + /** Drops in-flight bookkeeping during bolt cleanup; terminal callbacks then become no-ops. */ + void clear() { + active.clear(); + } +} diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java index ae403b124..b5bc5b338 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java @@ -22,6 +22,7 @@ import crawlercommons.urlfrontier.URLFrontierGrpc.URLFrontierStub; import crawlercommons.urlfrontier.Urlfrontier.BlockQueueParams; import crawlercommons.urlfrontier.Urlfrontier.Empty; +import crawlercommons.urlfrontier.Urlfrontier.QueueDelayParams; import io.grpc.ManagedChannel; import io.grpc.stub.StreamObserver; import java.util.Collections; @@ -43,8 +44,9 @@ /** * Regulates a URLFrontier queue from the {@code queue} stream emitted by the status updater: blocks - * it via {@code blockQueueUntil} whenever the tuple's metadata reports a rate-limit response. See - * issues #867, #784 and #1106. + * it via {@code blockQueueUntil} when metadata reports a rate-limit response, and paces it via + * {@code setDelay} when the fetcher reports a long robots.txt Crawl-delay. See issues #867, #784, + * #1106 and #1979. * *

Two cases are handled, both gated on the status codes configured with {@code * urlfrontier.backoff.status.codes} (default 429 and 503): @@ -68,6 +70,16 @@ * block so that hosts blocked on the same schedule do not all retry at once, and a block is only * ever extended, never shrunk by a later, shorter signal. * + *

Independently of rate limits, when robots pacing is explicitly enabled and the tuple's + * metadata carries {@code robots.crawl.delay} (written by the fetcher bolts when a robots.txt + * {@code Crawl-delay} exceeds {@code fetcher.max.crawl.delay}), the bolt forwards it to the + * frontier via {@code setDelay}, capped by {@code urlfrontier.robots.delay.max.secs}. The value is + * conservative: the maximum observed value wins, re-sent at most once per {@code + * urlfrontier.robots.delay.decay.secs} window. If a site later removes its Crawl-delay or lowers it + * below {@code fetcher.max.crawl.delay}, no signal arrives and the frontier keeps the old pace: + * throughput loss on that host, never impoliteness; {@code setDelay(key, 0)} is deliberately never + * called as it would erase a server-side default for the queue. + * *

Wire it with a fields grouping on {@code "key"} from the status updater's {@code queue} * stream: the per-host state lives in the bolt task, so all signals for a host must reach the same * task. The {@code "key"} is the frontier queue key derived from {@code partition.url.mode} (the @@ -87,18 +99,42 @@ * when the back-off on exceptions is enabled. A warning is logged at startup when the configuration * would strip them. * - *

Blocking stops new hand-outs; it does not recall URLs of the host already prefetched by the - * spout into the topology, which still fail against the rate-limited server and reschedule via the - * status stream. Keep {@code urlfrontier.max.urls.per.bucket} and {@code - * topology.max.spout.pending} small for a block to bite quickly. + *

Robots pacing is enabled by {@code urlfrontier.robots.crawl.delay.enabled=true} together with + * {@code fetcher.max.crawl.delay.force=true}. It requires a single URLFrontier endpoint — at most + * one {@code urlfrontier.address} entry; the {@code urlfrontier.host}/{@code urlfrontier.port} + * fallback is fine: with several addresses a keyed {@code setDelay} silently does nothing on the + * nodes that do not own the queue (crawler-commons/url-frontier#146), and a cluster behind one + * load-balanced address cannot be detected and has the same problem. It also requires {@code + * partition.url.mode=byHost}, {@code urlfrontier.max.urls.per.bucket=1}, and {@code + * robots.crawl.delay} in {@code metadata.persist} but not in {@code metadata.transfer} + * (including wildcards such as {@code robots.*}). The bolt probes these conditions at startup and + * fails fast rather than silently pacing a shared queue, transferring a host's delay to an outlink, + * or handing out several URLs from that host in one request. A custom {@code + * metadata.transfer.class} remains responsible for satisfying the same persist-only contract for + * all URLs and values. + * + *

Neither blocking nor pacing recalls URLs already prefetched by the spout. For rate-limit + * blocks, keep {@code urlfrontier.max.urls.per.bucket} and {@code topology.max.spout.pending} small + * for the block to bite quickly. Robots pacing requires the stricter per-queue batch size of one; + * this bounds each frontier hand-out but cannot recall earlier hand-outs. + * + *

With several concurrent {@code getURLs} clients — multiple Spout tasks, or several topologies + * on the same frontier — the frontier's per-queue politeness gate is not atomic + * (crawler-commons/url-frontier#147): concurrent requests can each be served a URL from the same + * queue inside the delay window. This is not limited to the transition to a new delay; it can recur + * on every concurrently aligned refill. Prefer a single Spout task per frontier when pacing must be + * strict. * *

Connects via {@code urlfrontier.address}, falling back to {@code urlfrontier.host} / {@code - * urlfrontier.port}. The block is issued with {@code local=false} and propagates to the whole - * cluster, so with several frontier nodes a single connection to any one of them is enough. + * urlfrontier.port}. Keyed {@code blockQueueUntil} and {@code setDelay} calls only take effect on + * the frontier node that owns the queue: in url-frontier 2.5 the {@code local=false} flag is not + * propagated by the distributed service (crawler-commons/url-frontier#146). With several configured + * addresses, blocks apply best-effort on the connected node only; a warning is logged at startup. * - *

The block is fire-and-forget with a short deadline: a failed or expired call is logged but the - * tuple is acked anyway. A missed block heals itself: every further rate-limit signal for the host - * re-asserts the stored block until the frontier stops serving it. + *

The RPCs have a short deadline: a failed or expired call is logged but the tuple is acked + * anyway. A missed block heals itself on further rate-limit signals. Robots delay calls are + * serialized per host and coalesce to the maximum observed value; an unchanged delay is re-sent + * after the configured robots delay decay window. */ public class QueueRegulatorBolt extends BaseRichBolt { @@ -132,12 +168,21 @@ public void onCompleted() {} /** Per-host back-off state and decision logic. */ private HostBackoff backoff; + /** Per-host robots crawl-delay forwarding decision. */ + private CrawlDelayPolicy crawlDelayPolicy; + + /** Prevents terminal callbacks from chaining another RPC after cleanup starts. */ + private volatile boolean closed; + @Override public void prepare(Map conf, TopologyContext context, OutputCollector c) { + this.closed = false; this.collector = c; this.globalCrawlID = ConfUtils.getString(conf, Constants.URLFRONTIER_CRAWL_ID_KEY, CrawlID.DEFAULT); this.backoff = new HostBackoff(conf); + validateRobotsPacingConfiguration(conf); + this.crawlDelayPolicy = new CrawlDelayPolicy(conf); // the status updater emits the queue stream after the metadata.persist // filter: warn upfront when the keys this bolt relies on would be // stripped, as the bolt would otherwise silently never block anything @@ -150,10 +195,18 @@ public void prepare(Map conf, TopologyContext context, OutputCol + " missing entries to metadata.persist.", missing); } - // a single connection is enough even with several frontier nodes, as - // the block is issued with local=false and propagates cluster-wide + // keyed blockQueueUntil/setDelay do not propagate across nodes in + // url-frontier 2.5 (local=false is ignored by the distributed service, + // crawler-commons/url-frontier#146): with several addresses each task + // only regulates the queues owned by the node it connects to List addresses = ConfUtils.loadListFromConf(Constants.URLFRONTIER_ADDRESS_KEY, conf); + if (addresses.size() > 1) { + LOG.warn( + "Multiple configured URLFrontier addresses: blocks and delays only take effect" + + " on the queues owned by the connected node" + + " (crawler-commons/url-frontier#146)"); + } String address; if (addresses.isEmpty()) { address = @@ -191,9 +244,53 @@ public void execute(Tuple t) { frontier.withDeadlineAfter(BLOCK_RPC_DEADLINE_SECS, TimeUnit.SECONDS) .blockQueueUntil(params, NOOP_OBSERVER); } + crawlDelayPolicy.requestFor(key, metadata).ifPresent(this::setDelay); collector.ack(t); } + private void setDelay(CrawlDelayPolicy.DelayRequest request) { + if (closed) { + return; + } + LOG.debug("Setting delay of {}s on queue {}", request.delaySecs(), request.key()); + QueueDelayParams delayParams = + QueueDelayParams.newBuilder() + .setKey(request.key()) + .setCrawlID(globalCrawlID) + .setDelayRequestable(request.delaySecs()) + .setLocal(false) + .build(); + try { + frontier.withDeadlineAfter(BLOCK_RPC_DEADLINE_SECS, TimeUnit.SECONDS) + .setDelay(delayParams, setDelayObserver(request)); + } catch (RuntimeException e) { + LOG.warn("setDelay failed", e); + crawlDelayPolicy.completed(request, false).ifPresent(this::setDelay); + } + } + + private StreamObserver setDelayObserver(CrawlDelayPolicy.DelayRequest request) { + return new StreamObserver<>() { + @Override + public void onNext(Empty value) {} + + @Override + public void onError(Throwable t) { + LOG.warn("setDelay failed", t); + crawlDelayPolicy + .completed(request, false) + .ifPresent(QueueRegulatorBolt.this::setDelay); + } + + @Override + public void onCompleted() { + crawlDelayPolicy + .completed(request, true) + .ifPresent(QueueRegulatorBolt.this::setDelay); + } + }; + } + /** * Returns the queue-stream keys this bolt relies on that would not survive the {@code * metadata.persist} filter applied by the status updater before emitting. The {@code @@ -219,6 +316,97 @@ static Set missingQueueStreamKeys( return missing; } + /** + * Fails fast when robots pacing is enabled but the queue key, hand-out size, delay cap, + * frontier addressing, or metadata routing could apply an unsafe pace or target the wrong host. + */ + static void validateRobotsPacingConfiguration(Map conf) { + if (!ConfUtils.getBoolean( + conf, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_DEFAULT)) { + return; + } + + List addresses = + ConfUtils.loadListFromConf(Constants.URLFRONTIER_ADDRESS_KEY, conf); + if (addresses.size() > 1) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires a single URLFrontier endpoint (at most one" + + " urlfrontier.address entry): keyed setDelay calls are not propagated" + + " across nodes and would silently not pace queues owned by other" + + " nodes (crawler-commons/url-frontier#146)"); + } + + if (!ConfUtils.getBoolean(conf, "fetcher.max.crawl.delay.force", false)) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires fetcher.max.crawl.delay.force=true"); + } + + int maxCrawlDelay = ConfUtils.getInt(conf, "fetcher.max.crawl.delay", 30); + if (maxCrawlDelay < 0) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires fetcher.max.crawl.delay >= 0"); + } + + String partitionMode = + ConfUtils.getString( + conf, + org.apache.stormcrawler.Constants.PARTITION_MODEParamName, + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST); + if (!org.apache.stormcrawler.Constants.PARTITION_MODE_HOST.equals(partitionMode)) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires partition.url.mode=" + + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST + + "; found " + + partitionMode); + } + + int maxURLsPerBucket = + ConfUtils.getInt( + conf, + Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, + Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT); + if (maxURLsPerBucket != 1) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires " + + Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY + + "=1; found " + + maxURLsPerBucket); + } + + long maxDelaySecs = + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_ROBOTS_DELAY_MAX_KEY, + Constants.URLFRONTIER_ROBOTS_DELAY_MAX_DEFAULT); + if (maxDelaySecs <= 0) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires " + + Constants.URLFRONTIER_ROBOTS_DELAY_MAX_KEY + + " > 0"); + } + + final String robotsDelayKey = org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY; + MetadataTransfer transfer = MetadataTransfer.getInstance(conf); + Metadata probe = new Metadata(); + probe.setValue(robotsDelayKey, "1"); + Metadata persisted = transfer.filter(probe); + if (persisted == null || persisted.getFirstValue(robotsDelayKey) == null) { + throw new IllegalArgumentException( + "robots.crawl.delay must be included in metadata.persist"); + } + + Metadata outlink = + transfer.getMetaForOutlink( + "http://target.example/", "http://source.example/", probe); + if (outlink == null || outlink.getFirstValue(robotsDelayKey) != null) { + throw new IllegalArgumentException( + "robots.crawl.delay must not appear in metadata.transfer, including via a" + + " wildcard: it is a per-host control signal"); + } + } + @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { // terminal bolt @@ -226,6 +414,10 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { @Override public void cleanup() { + closed = true; + if (crawlDelayPolicy != null) { + crawlDelayPolicy.clear(); + } if (channel != null) { channel.shutdown(); } diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java index 6aa9ef7f8..9763051ec 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java @@ -24,6 +24,7 @@ import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_DELAY_REQUESTABLE_KEY; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_HOST_KEY; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_MAX_BUCKETS_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_PORT_KEY; @@ -70,7 +71,11 @@ public void open( SpoutOutputCollector collector) { super.open(stormConf, context, collector); - maxURLsPerBucket = ConfUtils.getInt(stormConf, URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 10); + maxURLsPerBucket = + ConfUtils.getInt( + stormConf, + URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, + URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT); maxBucketNum = ConfUtils.getInt(stormConf, URLFRONTIER_MAX_BUCKETS_KEY, 10); diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java new file mode 100644 index 000000000..ad5201fa0 --- /dev/null +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java @@ -0,0 +1,232 @@ +/* + * 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.urlfrontier; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.github.benmanes.caffeine.cache.Ticker; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.stormcrawler.Metadata; +import org.junit.jupiter.api.Test; + +/** + * Unit coverage for {@link CrawlDelayPolicy} - whether a queue-stream tuple carries a robots + * crawl-delay worth forwarding to the frontier, with per-host dedupe bounded by the decay window. + */ +class CrawlDelayPolicyTest { + + private static final String KEY = "example.com"; + + private static final class FakeTicker implements Ticker { + private long nanos; + + @Override + public long read() { + return nanos; + } + + void advanceSecs(long secs) { + nanos += TimeUnit.SECONDS.toNanos(secs); + } + } + + private final FakeTicker ticker = new FakeTicker(); + + private CrawlDelayPolicy policy(Object... keysAndValues) { + Map conf = new HashMap<>(); + conf.put("fetcher.max.crawl.delay.force", true); + conf.put(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, true); + for (int i = 0; i < keysAndValues.length; i += 2) { + conf.put((String) keysAndValues[i], keysAndValues[i + 1]); + } + return new CrawlDelayPolicy(conf, ticker); + } + + private static Metadata md(String delay) { + Metadata md = new Metadata(); + if (delay != null) { + md.setValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY, delay); + } + return md; + } + + @Test + void newValueIsForwarded() { + assertEquals(120, request(policy(), "120").delaySecs()); + } + + @Test + void unchangedValueIsDeduplicatedWithinTheWindow() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + assertFalse(policy.completed(first, true).isPresent()); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + } + + @Test + void changedValuesAreSerializedAndOnlyTheMaximumPendingValueIsSent() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + assertFalse(policy.requestFor(KEY, md("180")).isPresent()); + + CrawlDelayPolicy.DelayRequest second = policy.completed(first, true).orElseThrow(); + assertEquals(180, second.delaySecs()); + assertFalse(policy.completed(second, true).isPresent()); + } + + @Test + void unchangedValueIsResentAfterTheWindow() { + // the dedupe entry expires so a lost RPC converges within one window + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.completed(first, true).isPresent()); + ticker.advanceSecs(Constants.URLFRONTIER_ROBOTS_DELAY_DECAY_DEFAULT + 1); + assertEquals(120, request(policy, "120").delaySecs()); + } + + @Test + void lowerValueIsAppliedOnlyAfterTheMaximumExpiresFromTheWindow() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.completed(first, true).isPresent()); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + + ticker.advanceSecs(Constants.URLFRONTIER_ROBOTS_DELAY_DECAY_DEFAULT + 1); + assertEquals(60, request(policy, "60").delaySecs()); + } + + @Test + void valueIsCappedAtRobotsDelayMax() { + CrawlDelayPolicy policy = policy(Constants.URLFRONTIER_ROBOTS_DELAY_MAX_KEY, 3600); + assertEquals(3600, request(policy, "86400").delaySecs()); + } + + @Test + void signalIsIgnoredWhenFetcherForceIsDisabled() { + CrawlDelayPolicy policy = policy("fetcher.max.crawl.delay.force", false); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + } + + @Test + void signalIsIgnoredWhenRobotsPacingIsDisabled() { + CrawlDelayPolicy policy = + policy(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, false); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + } + + @Test + void failedRpcDoesNotRetryALowerObservedValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + + // the lower value was dropped at observation time: nothing pends, no self-retry + assertFalse(policy.completed(first, false).isPresent()); + // the failure invalidated the dedupe entry: the next signal reasserts the value + assertEquals(120, request(policy, "120").delaySecs()); + } + + @Test + void failedRpcRetriesAHigherPendingValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("180")).isPresent()); + + CrawlDelayPolicy.DelayRequest second = policy.completed(first, false).orElseThrow(); + assertEquals(180, second.delaySecs()); + assertFalse(policy.completed(second, false).isPresent()); + } + + @Test + void failedChangedRpcBecomesTheFloorForTheNextSignal() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest initial = request(policy, "120"); + assertFalse(policy.completed(initial, true).isPresent()); + + CrawlDelayPolicy.DelayRequest changed = request(policy, "180"); + assertFalse(policy.completed(changed, false).isPresent()); + + // The server may have applied 180 before the response was lost. The next signal is not + // deduplicated, and it must reassert the ambiguous maximum, not its own lower value. + assertEquals(180, request(policy, "120").delaySecs()); + } + + @Test + void theAmbiguousFloorExpiresWithTheDecayWindow() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest changed = request(policy, "180"); + assertFalse(policy.completed(changed, false).isPresent()); + + ticker.advanceSecs(Constants.URLFRONTIER_ROBOTS_DELAY_DECAY_DEFAULT + 1); + assertEquals(120, request(policy, "120").delaySecs()); + } + + @Test + void repeatedInFlightValueNeverRetriesAndFailureReassertsOnTheNextSignal() { + CrawlDelayPolicy failed = policy(); + CrawlDelayPolicy.DelayRequest failedRequest = request(failed, "120"); + assertFalse(failed.requestFor(KEY, md("120")).isPresent()); + assertFalse(failed.completed(failedRequest, false).isPresent()); + // the failure invalidated the dedupe entry: the next signal reasserts the value + assertEquals(120, request(failed, "120").delaySecs()); + + CrawlDelayPolicy succeeded = policy(); + CrawlDelayPolicy.DelayRequest succeededRequest = request(succeeded, "120"); + assertFalse(succeeded.requestFor(KEY, md("120")).isPresent()); + assertFalse(succeeded.completed(succeededRequest, true).isPresent()); + // the success recorded the value: an unchanged signal is deduplicated + assertFalse(succeeded.requestFor(KEY, md("120")).isPresent()); + } + + @Test + void staleCallbackCannotCompleteANewerRequest() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("180")).isPresent()); + CrawlDelayPolicy.DelayRequest retry = policy.completed(first, false).orElseThrow(); + assertEquals(180, retry.delaySecs()); + + // the retry replaced the failed request: a late duplicate callback for it is ignored + assertFalse(policy.completed(first, true).isPresent()); + assertFalse(policy.completed(retry, true).isPresent()); + // the retry recorded 180: an unchanged signal is deduplicated + assertFalse(policy.requestFor(KEY, md("180")).isPresent()); + } + + @Test + void malformedAbsentZeroAndDefaultQueueAreIgnored() { + CrawlDelayPolicy policy = policy(); + assertFalse(policy.requestFor(KEY, md("not-a-number")).isPresent()); + assertFalse(policy.requestFor(KEY, md("0")).isPresent()); + assertFalse(policy.requestFor(KEY, md(null)).isPresent()); + assertFalse(policy.requestFor(KEY, null).isPresent()); + assertFalse(policy.requestFor("_DEFAULT_", md("120")).isPresent()); + } + + private static CrawlDelayPolicy.DelayRequest request(CrawlDelayPolicy policy, String delay) { + var request = policy.requestFor(KEY, md(delay)); + assertTrue(request.isPresent()); + return request.orElseThrow(); + } +} diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java index 8f068ca2d..3f8e22e74 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java @@ -17,7 +17,9 @@ package org.apache.stormcrawler.urlfrontier; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.HashMap; @@ -37,15 +39,15 @@ class QueueRegulatorBoltDecisionTest { private static final String RETRY_AFTER_KEY = "protocol.retry-after"; @Test - void missingQueueStreamKeysReportsBothWithDefaultConfig() { - // with the default metadata.persist neither key survives the filter + void missingQueueStreamKeysReportsAllWithDefaultConfig() { + // with the default metadata.persist none of the keys survives the filter Set missing = QueueRegulatorBolt.missingQueueStreamKeys(new HashMap<>(), RETRY_AFTER_KEY, false); assertEquals(Set.of("fetch.statusCode", RETRY_AFTER_KEY), missing); } @Test - void missingQueueStreamKeysEmptyWhenBothPersisted() { + void missingQueueStreamKeysEmptyWhenAllPersisted() { Map conf = new HashMap<>(); conf.put("metadata.persist", List.of("fetch.statusCode", RETRY_AFTER_KEY)); assertTrue( @@ -72,4 +74,140 @@ void exceptionKeyOnlyRequiredWhenBackoffOnExceptionsIsEnabled() { Set.of("fetch.exception"), QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, true)); } + + @Test + void robotsValidationIsSkippedWhenFrontierPacingIsDisabled() { + assertDoesNotThrow( + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(new HashMap<>())); + } + + @Test + void legacyFetcherForceDoesNotEnableFrontierPacing() { + Map conf = new HashMap<>(); + conf.put("fetcher.max.crawl.delay.force", true); + assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void enabledRobotsPacingRequiresFetcherForce() { + Map conf = validRobotsConf(); + conf.put("fetcher.max.crawl.delay.force", false); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void validPersistOnlyRobotsConfigurationPasses() { + assertDoesNotThrow( + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(validRobotsConf())); + } + + @Test + void robotsSignalMustSurvivePersistence() { + Map conf = validRobotsConf(); + conf.remove("metadata.persist"); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsSignalMustNotTransferToOutlinks() { + Map conf = validRobotsConf(); + conf.put("metadata.transfer", List.of("robots.*")); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPersistenceWildcardIsAccepted() { + Map conf = validRobotsConf(); + conf.put("metadata.persist", List.of("robots.*")); + assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPacingRequiresHostPartitioning() { + for (String mode : + List.of( + org.apache.stormcrawler.Constants.PARTITION_MODE_DOMAIN, + org.apache.stormcrawler.Constants.PARTITION_MODE_IP)) { + Map conf = validRobotsConf(); + conf.put(org.apache.stormcrawler.Constants.PARTITION_MODEParamName, mode); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + } + + @Test + void robotsPacingRequiresSingleUrlHandouts() { + Map defaultBatch = validRobotsConf(); + defaultBatch.remove(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(defaultBatch)); + + Map multiUrlBatch = validRobotsConf(); + multiUrlBatch.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 2); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(multiUrlBatch)); + } + + @Test + void robotsPacingRequiresAnEnabledCrawlDelayLimit() { + Map conf = validRobotsConf(); + conf.put("fetcher.max.crawl.delay", -1); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPacingRequiresAPositiveFrontierDelayCap() { + Map conf = validRobotsConf(); + conf.put(Constants.URLFRONTIER_ROBOTS_DELAY_MAX_KEY, 0); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPacingRequiresASingleFrontierAddress() { + // keyed setDelay is not propagated across nodes (crawler-commons/url-frontier#146) + Map conf = validRobotsConf(); + conf.put(Constants.URLFRONTIER_ADDRESS_KEY, List.of("host1:7071", "host2:7071")); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + + Map single = validRobotsConf(); + single.put(Constants.URLFRONTIER_ADDRESS_KEY, List.of("host1:7071")); + assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(single)); + } + + @Test + void multipleFrontierAddressesAreAcceptedWhenRobotsPacingIsDisabled() { + // the #1984 rate-limit blocks stay best-effort on the connected node: warn, do not fail + Map conf = new HashMap<>(); + conf.put(Constants.URLFRONTIER_ADDRESS_KEY, List.of("host1:7071", "host2:7071")); + assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + private static Map validRobotsConf() { + Map conf = new HashMap<>(); + conf.put(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, true); + conf.put("fetcher.max.crawl.delay.force", true); + conf.put( + org.apache.stormcrawler.Constants.PARTITION_MODEParamName, + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST); + conf.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 1); + conf.put( + "metadata.persist", + List.of(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY)); + return conf; + } } diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java index 87a88dcc1..0d55c9c14 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java @@ -18,6 +18,7 @@ package org.apache.stormcrawler.urlfrontier; import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -33,6 +34,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -90,6 +92,11 @@ private Map frontierConfig() { /** Seeds one DISCOVERED url for {@link #HOST} so the queue exists and is fetchable. */ private void seedOneUrl() { + seedUrls("/"); + } + + /** Seeds one DISCOVERED url for {@link #HOST} per path so the queue exists and is fetchable. */ + private void seedUrls(String... paths) { StatusUpdaterBolt seeder = new StatusUpdaterBolt(); TestOutputCollector out = new TestOutputCollector(); Map config = frontierConfig(); @@ -101,18 +108,30 @@ private void seedOneUrl() { config.put("urlfrontier.cache.expireafter.sec", 10); seeder.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(out)); - String url = "http://" + HOST + "/"; - Tuple tuple = mock(Tuple.class); - when(tuple.getValueByField("status")).thenReturn(Status.DISCOVERED); - when(tuple.getStringByField("url")).thenReturn(url); - when(tuple.getValueByField("metadata")).thenReturn(new Metadata()); - seeder.execute(tuple); + Set urls = new HashSet<>(); + for (String path : paths) { + String url = "http://" + HOST + path; + urls.add(url); + Tuple tuple = mock(Tuple.class); + when(tuple.getValueByField("status")).thenReturn(Status.DISCOVERED); + when(tuple.getStringByField("url")).thenReturn(url); + when(tuple.getValueByField("metadata")).thenReturn(new Metadata()); + seeder.execute(tuple); + } await().atMost(20, TimeUnit.SECONDS) .until( () -> - out.getAckedTuples().stream() - .anyMatch(t -> url.equals(t.getStringByField("url")))); + urls.stream() + .allMatch( + url -> + out.getAckedTuples().stream() + .anyMatch( + t -> + url.equals( + t + .getStringByField( + "url"))))); seeder.cleanup(); } @@ -224,4 +243,66 @@ void blocksHostQueueOnHeaderless429() { bolt.cleanup(); } + + @Test + @Timeout(value = 2, unit = TimeUnit.MINUTES) + void pacesHostQueueOnRobotsCrawlDelay() { + seedUrls("/1", "/2"); + + // Model one hand-out before the response carrying the robots delay reaches the queue + // stream. Production uses max.urls.per.bucket=1 to prevent a single request from draining + // a larger host batch; setDelay cannot recall earlier or concurrent hand-outs. + assertEquals(1, countServed(1), "the first hand-out should serve one URL"); + + QueueRegulatorBolt bolt = new QueueRegulatorBolt(); + Map conf = frontierConfig(); + conf.put(ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, "protocol."); + conf.put(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, true); + conf.put("fetcher.max.crawl.delay.force", true); + conf.put( + org.apache.stormcrawler.Constants.PARTITION_MODEParamName, + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST); + conf.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 1); + conf.put( + "metadata.persist", + List.of( + "fetch.statusCode", + "protocol.retry-after", + org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY)); + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), mock(OutputCollector.class)); + Tuple t = mock(Tuple.class); + when(t.getStringByField("key")).thenReturn(HOST); + Metadata md = new Metadata(); + md.setValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY, "300"); + when(t.getValueByField("metadata")).thenReturn(md); + bolt.execute(t); + + // blind delay for the fire-and-forget RPC to land, as in the sibling tests. + // The queue is now not requestable again before 300s: a later call must come back empty + // although /2 was not part of the earlier hand-out and remains queued. + // The pause also pins the unit of setDelayRequestable: were it + // milliseconds, the 300ms window would have expired and /2 would be + // handed out. + await().pollDelay(Duration.ofSeconds(2)).atMost(3, TimeUnit.SECONDS).until(() -> true); + assertEquals(0, countServed(10), "the queue delay should pace the second URL"); + + bolt.cleanup(); + } + + /** Drains one getURLs call and returns how many URLs it handed out. */ + private int countServed(int maxUrlsPerQueue) { + GetParams params = + GetParams.newBuilder() + .setCrawlID(CrawlID.DEFAULT) + .setMaxQueues(10) + .setMaxUrlsPerQueue(maxUrlsPerQueue) + .build(); + int served = 0; + Iterator it = blocking.getURLs(params); + while (it.hasNext()) { + it.next(); + served++; + } + return served; + } }