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..f938b14fd 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,17 @@ 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`, and `robots.crawl.delay` in `metadata.persist`. 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 +491,8 @@ 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.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. @@ -490,7 +502,7 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/urlfronti | urlfrontier.max.retry.after | 86400 | Maximum delay in seconds honoured when a server requests a back-off via the Retry-After HTTP response header. -1 disables the cap. | urlfrontier.backoff.base.secs | 60 | First block duration in seconds applied when a host rate-limits without a usable Retry-After header. 0 disables the adaptive back-off. | urlfrontier.backoff.factor | 2.0 | Multiplicative increase applied to the block duration on each new pressure event for the same host. -| urlfrontier.backoff.max.secs | 86400 | Upper bound in seconds for the escalated block duration. +| urlfrontier.backoff.max.secs | 86400 | Upper bound in seconds for the escalated block duration and for a robots crawl-delay forwarded to the frontier. | urlfrontier.backoff.decay.secs | 1800 | Quiet window in seconds after which its escalation is forgotten and the next pressure event starts again from the base duration. | urlfrontier.backoff.on.exceptions | false | Whether fetch exceptions count as pressure events and escalate the host's block. | urlfrontier.backoff.jitter | 0.1 | Random extra fraction added to every computed block duration. diff --git a/docs/src/main/asciidoc/extending.adoc b/docs/src/main/asciidoc/extending.adoc index 7277f88b9..5d1d74794 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`, the frontier forwards a value capped by `urlfrontier.backoff.max.secs` to the host queue ==== Robots.txt Compliance diff --git a/external/urlfrontier/README.md b/external/urlfrontier/README.md index 8647fc040..869ed043f 100644 --- a/external/urlfrontier/README.md +++ b/external/urlfrontier/README.md @@ -24,6 +24,48 @@ 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 +urlfrontier.backoff.max.secs: 86400 + +metadata.persist: + - robots.crawl.delay +``` + +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. `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. A batch size of one bounds each hand-out; it does not recall URLs already +emitted or eliminate concurrent requests while the new delay is reaching the frontier. + 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..976cb7399 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,12 @@ 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; + /** * 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..b33ba93d2 --- /dev/null +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java @@ -0,0 +1,176 @@ +/* + * 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.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +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.backoff.max.secs}. + * + *

The signal is declarative — the last value wins, there is no escalation and no interaction + * with {@link HostBackoff}'s blocks. Values are capped by {@code urlfrontier.backoff.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 latest 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; + + /** At most one unary RPC per host, plus the latest 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 pendingLatest) {} + + 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_BACKOFF_MAX_KEY, + Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT)); + long decaySecs = + Math.max( + 0L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_DECAY_KEY, + Constants.URLFRONTIER_BACKOFF_DECAY_DEFAULT)); + this.lastApplied = + Caffeine.newBuilder() + .expireAfterWrite(decaySecs, TimeUnit.SECONDS) + .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) { + return new Flight(flight.request(), delay); + } + Integer previous = lastApplied.getIfPresent(key); + if (previous != null && previous == delay) { + return null; + } + DelayRequest request = + new DelayRequest(requestTokens.incrementAndGet(), key, delay); + next.set(request); + return new Flight(request, null); + }); + return Optional.ofNullable(next.get()); + } + + /** Completes one RPC and returns the latest 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.put(key, completed.delaySecs()); + } else { + // An error is ambiguous: the server may have applied the request before the + // response was lost. Forget the previous value so a later signal reasserts + // it instead of being incorrectly deduplicated. + lastApplied.invalidate(key); + } + Integer pending = flight.pendingLatest(); + 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..6e9d1418e 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.backoff.max.secs}. The value is + * declarative: the last one seen wins, re-sent at most once per {@code + * urlfrontier.backoff.decay.secs} window. If a site later removes its 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,30 @@ * 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 is safe only with {@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 or eliminate concurrent + * requests during the transition to the new delay. * *

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. * - *

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 latest value; an unchanged delay is re-sent after the + * configured decay window. */ public class QueueRegulatorBolt extends BaseRichBolt { @@ -132,12 +156,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 @@ -191,9 +224,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 +296,87 @@ static Set missingQueueStreamKeys( return missing; } + /** + * Fails fast when robots pacing is enabled but the queue key, hand-out size, delay cap 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; + } + + 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_BACKOFF_MAX_KEY, + Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT); + if (maxDelaySecs <= 0) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires " + + Constants.URLFRONTIER_BACKOFF_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 +384,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..10436c648 --- /dev/null +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java @@ -0,0 +1,193 @@ +/* + * 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 changedValuesAreSerializedAndOnlyTheLatestPendingValueIsSent() { + 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_BACKOFF_DECAY_DEFAULT + 1); + assertEquals(120, request(policy, "120").delaySecs()); + } + + @Test + void valueIsCappedAtBackoffMax() { + CrawlDelayPolicy policy = policy(Constants.URLFRONTIER_BACKOFF_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 failedRpcRetriesTheLatestPendingValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + + CrawlDelayPolicy.DelayRequest second = policy.completed(first, false).orElseThrow(); + assertEquals(60, second.delaySecs()); + assertFalse(policy.completed(second, false).isPresent()); + assertEquals(60, request(policy, "60").delaySecs()); + } + + @Test + void failedChangedRpcInvalidatesThePreviouslyAppliedValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest initial = request(policy, "120"); + assertFalse(policy.completed(initial, true).isPresent()); + + CrawlDelayPolicy.DelayRequest changed = request(policy, "60"); + assertFalse(policy.completed(changed, false).isPresent()); + + // The server may have applied 60 before the response was lost. Do not deduplicate the + // previous 120: reasserting it is the only way to converge safely. + assertEquals(120, request(policy, "120").delaySecs()); + } + + @Test + void repeatedInFlightValueRetriesOnFailureButNotOnSuccess() { + CrawlDelayPolicy failed = policy(); + CrawlDelayPolicy.DelayRequest failedRequest = request(failed, "120"); + assertFalse(failed.requestFor(KEY, md("120")).isPresent()); + assertEquals(120, failed.completed(failedRequest, false).orElseThrow().delaySecs()); + + CrawlDelayPolicy succeeded = policy(); + CrawlDelayPolicy.DelayRequest succeededRequest = request(succeeded, "120"); + assertFalse(succeeded.requestFor(KEY, md("120")).isPresent()); + assertFalse(succeeded.completed(succeededRequest, true).isPresent()); + } + + @Test + void staleCallbackCannotCompleteANewerRequestWithTheSameValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + CrawlDelayPolicy.DelayRequest retry = policy.completed(first, false).orElseThrow(); + + assertFalse(policy.completed(first, true).isPresent()); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + assertEquals(60, policy.completed(retry, true).orElseThrow().delaySecs()); + } + + @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..760bd5e66 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,118 @@ 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_BACKOFF_MAX_KEY, 0); + assertThrows( + IllegalArgumentException.class, + () -> 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..041834422 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,68 @@ 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 + await().pollDelay(Duration.ofSeconds(2)).atMost(3, TimeUnit.SECONDS).until(() -> true); + + // 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; + } }