jnioche commented on code in PR #1984: URL: https://github.com/apache/stormcrawler/pull/1984#discussion_r3568253957
########## external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/HostBackoff.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.Expiry; +import com.github.benmanes.caffeine.cache.Ticker; +import java.util.Collection; +import java.util.HashSet; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.protocol.ProtocolResponse; +import org.apache.stormcrawler.util.ConfUtils; +import org.apache.stormcrawler.util.RetryAfterParser; + +/** + * Per-host back-off state and decision logic for the {@link QueueRegulatorBolt}: given a + * queue-stream tuple, decides whether and until when the host's frontier queue should be blocked. + * + * <p>Two kinds of signal are recognised, both derived from the tuple's metadata: + * + * <ul> + * <li><b>explicit</b> — a rate-limit status code with a usable Retry-After header: the requested + * delay is honoured, capped by {@code urlfrontier.max.retry.after}; + * <li><b>pressure</b> — a rate-limit status code without a usable header (or, when {@code + * urlfrontier.backoff.on.exceptions} is set, a fetch exception): each pressure event blocks + * the host for a growing duration, {@code base × factor^n} capped at {@code max}, the same + * curve Nutch applies since 1.19 (NUTCH-2946). A host that stays quiet for {@code decay} + * seconds <b>after its block has expired</b> is forgiven and restarts from {@code base} — a + * blocked host is silent because it is blocked, so the quiet clock only starts once it may be + * fetched again. + * </ul> + * + * <p>Two guards keep the escalation honest: + * + * <ul> + * <li><b>one event per window</b> — pressure signals arriving while the host's block is still + * active never escalate: URLs of that host already in flight when the block was raised fail + * one after the other, and counting each failure would let a single incident saturate the cap + * (multiple losses inside one RTT are one congestion event, RFC 5681); + * <li><b>only-extend</b> — {@code blockQueueUntil} overwrites, and signals race: a block is only + * ever replaced by a later one, never shrunk. + * </ul> + * + * <p>Signals suppressed by either guard still return the <b>stored</b> block end, so the caller + * re-asserts the same block at the frontier: the RPC is fire-and-forget, and a lost call would + * otherwise leave the frontier unblocked while this state believes the host is covered. The + * repeated call is idempotent and self-limiting — once the frontier applies the block, the host + * stops being served and the signals stop. + * + * <p>A random fraction ({@code urlfrontier.backoff.jitter}) is added to every computed duration so + * that hosts blocked on the same schedule do not all expire together. + * + * <p>All per-host state lives in a single cache entry that expires {@code decay} seconds after the + * block end, which bounds the memory to the hosts blocked recently. State is soft: it restarts + * empty, escalation resumes from {@code base} on the next signal. Not thread-safe; a bolt task + * calls {@link #blockUntilFor} from a single thread. + */ +final class HostBackoff { + + /** Metadata key set by the FetcherBolt with the HTTP status code of the fetch. */ + static final String STATUS_CODE_KEY = "fetch.statusCode"; + + /** + * Metadata key set by the FetcherBolt with the reason of a fetch failure. The FetcherBolt + * removes it before every fetch, so its presence means the last cycle failed — unlike {@link + * #STATUS_CODE_KEY}, which lingers from the previous response and is stale on such tuples. + */ + static final String EXCEPTION_KEY = "fetch.exception"; + + /** Name of the Retry-After HTTP header, lower-cased as stored by the protocol layer. */ + static final String RETRY_AFTER_HEADER = "retry-after"; + + /** + * Queue key used by the status updaters when no partition key can be derived from a URL. The + * queue is shared by unrelated URLs, so it is never blocked. + */ + static final String DEFAULT_QUEUE_KEY = "_DEFAULT_"; + + private static final Set<String> DEFAULT_STATUS_CODES = Set.of("429", "503"); + + /** + * Back-off state of one host: the escalated duration reached so far, the end of the current (or + * last) block, and the entry's lifetime computed when it was written — block remainder plus the + * decay window, so the escalation is forgotten {@code decay} seconds after the block ends, and + * the block guard can never be evicted while the block runs. + */ + private record HostState(long backoffSecs, long blockUntilSecs, long ttlSecs) {} + + private final Cache<String, HostState> hosts; + + /** Metadata key holding the Retry-After header, including the protocol prefix. */ + private final String retryAfterKey; + + /** Upper bound in ms for the honoured Retry-After delay; -1 means no cap. */ + private final long maxRetryAfterMs; + + private final Set<String> statusCodes; + private final long baseSecs; + private final float factor; + private final long maxSecs; + private final long decaySecs; + private final boolean onExceptions; + private final float jitter; + private final Random jitterSource; + + HostBackoff(Map<String, Object> conf) { + this(conf, Ticker.systemTicker(), new Random()); + } + + /** Visible for tests: a controllable ticker drives the expiry, a seeded source the jitter. */ + HostBackoff(Map<String, Object> conf, Ticker ticker, Random jitterSource) { + // the protocol layer stores response headers in the metadata with this + // prefix; the FetcherBolt merges them into the status metadata + this.retryAfterKey = + ConfUtils.getString(conf, ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, "") + + RETRY_AFTER_HEADER; + long maxRetryAfterSecs = + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_MAX_RETRY_AFTER_KEY, + Constants.URLFRONTIER_MAX_RETRY_AFTER_DEFAULT); + this.maxRetryAfterMs = maxRetryAfterSecs < 0 ? -1L : maxRetryAfterSecs * 1000L; + this.statusCodes = statusCodesFromConf(conf); + this.baseSecs = + Math.max( + 0L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_BASE_KEY, + Constants.URLFRONTIER_BACKOFF_BASE_DEFAULT)); + // a factor below 1 would de-escalate, negative jitter would shrink + // blocks, a negative decay would evict a running block: clamp all + // to avoid letting a typo break the invariants silently + this.factor = + Math.max( + 1f, + ConfUtils.getFloat( + conf, + Constants.URLFRONTIER_BACKOFF_FACTOR_KEY, + Constants.URLFRONTIER_BACKOFF_FACTOR_DEFAULT)); + this.maxSecs = + Math.max( + this.baseSecs, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_MAX_KEY, + Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT)); + this.decaySecs = + Math.max( + 0L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_DECAY_KEY, + Constants.URLFRONTIER_BACKOFF_DECAY_DEFAULT)); + this.onExceptions = + ConfUtils.getBoolean( + conf, + Constants.URLFRONTIER_BACKOFF_ON_EXCEPTIONS_KEY, + Constants.URLFRONTIER_BACKOFF_ON_EXCEPTIONS_DEFAULT); + this.jitter = + Math.max( + 0f, + ConfUtils.getFloat( + conf, + Constants.URLFRONTIER_BACKOFF_JITTER_KEY, + Constants.URLFRONTIER_BACKOFF_JITTER_DEFAULT)); + this.jitterSource = jitterSource; + this.hosts = + Caffeine.newBuilder() Review Comment: Does not this re-implements Caffeine's built-in Expiry.writing(...) factory? `Expiry.writing((k, s) -> Duration.ofSeconds(s.ttlSecs()))` should be semantically identical: the function applies on create and update, reads keep the current duration. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
