dpol1 commented on code in PR #2052: URL: https://github.com/apache/stormcrawler/pull/2052#discussion_r4285638255
########## core/src/main/java/org/apache/stormcrawler/filtering/adaptive/CanonicalRules.java: ########## @@ -0,0 +1,629 @@ +/* + * 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.filtering.adaptive; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import crawlercommons.domains.EffectiveTldFinder; +import java.net.MalformedURLException; +import java.net.URL; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.commons.lang3.StringUtils; +import org.apache.stormcrawler.util.ConfUtils; +import org.apache.stormcrawler.util.URLUtil; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Evidence gathered from the canonical tags of a site about which of its query parameters can be + * removed without changing the content. + * + * <p>Fed by {@link org.apache.stormcrawler.parse.filter.CanonicalParamLearner} and read by {@link + * AdaptiveURLNormalizer}: since these live in different components of the same bolt, they share the + * single instance of this JVM, obtained with {@link #getInstance(Map)} and handed back with {@link + * #release()}. Configured from the Storm configuration, see the <code>adaptive.normalizer.*</code> + * options, so that both sides cannot disagree. Safe to use from several threads. + * + * <p>Rules are learnt per host or domain, but a canonical which keeps a parameter on a given path + * vetoes the rule on that path: a parameter dropped on product pages and kept on listings is + * removed from the former only. Kept on more than {@value #MAX_KEPT_PATHS} paths, it is not removed + * anywhere. + */ +public class CanonicalRules { + + private static final Logger LOG = LoggerFactory.getLogger(CanonicalRules.class); + + public static final String CANONICAL_KEY_PARAM = "adaptive.normalizer.canonical.key"; + + public static final String SCOPE_PARAM = "adaptive.normalizer.scope"; + + public static final String MIN_OBSERVATIONS_PARAM = "adaptive.normalizer.min.observations"; + + public static final String MIN_DISTINCT_PATHS_PARAM = "adaptive.normalizer.min.distinct.paths"; + + public static final String CONFIDENCE_PARAM = "adaptive.normalizer.confidence"; + + public static final String MAX_SCOPES_PARAM = "adaptive.normalizer.max.scopes"; + + public static final String MAX_PARAMS_PARAM = "adaptive.normalizer.max.params"; + + public static final String MAX_CACHED_SOURCES_PARAM = "adaptive.normalizer.max.cached.sources"; + + public static final String PROTECTED_PARAMS_PARAM = "adaptive.normalizer.protected.params"; + + /** Longer names are tokens rather than parameters and would only take up room. */ + public static final int MAX_PARAM_NAME_LENGTH = 64; + + /** Distinct paths on which a canonical may keep a parameter before it is never removed. */ + public static final int MAX_KEPT_PATHS = 8; + + /** + * Never removed, however consistently the canonical tags drop them, matched regardless of case: + * a listing serving <code>/list?page=2..N</code> with a canonical of <code>/list</code> would + * otherwise have its paginated content normalised away and never fetched. + */ + private static final List<String> DEFAULT_PROTECTED_PARAMS = + Arrays.asList( + "page", + "p", + "pg", + "pgno", + "pg_no", + "pageno", + "page_no", + "pagenum", + "page_num", + "pagenumber", + "page_number", + "pageindex", + "page_index", + "paged", + "pagina", + "seite", + "offset", + "start", + "from", + "limit", + "per_page", + "perpage", + "q", + "query", + "s", + "search", + "keyword", + "keywords", + "sort", + "order", + "orderby", + "order_by", + "dir", + "lang", + "language", + "hl", + "locale", + "id", + "category", + "cat", + "tag", + "year", + "month", + "day", + "view", + "format", + "type"); + + /** Guarded by the class lock, as is the number of components using it. */ + private static CanonicalRules instance; + + private static int users; + + /** + * The instance shared by the components of this JVM, created from the Storm configuration by + * the first of them. Every caller must {@link #release()} it once done, so that the next + * topology of a long-lived JVM starts afresh. A caller whose settings differ from the existing + * instance's gets that instance nonetheless, with a warning. + */ + public static synchronized CanonicalRules getInstance(@NotNull Map<String, Object> stormConf) { + final Settings settings = Settings.from(stormConf); + if (instance == null) { + instance = new CanonicalRules(settings); + } else if (!instance.settings.equals(settings)) { + LOG.warn("Already configured differently in this JVM, keeping the first settings"); + } + users++; + return instance; + } + + /** Hands the instance back; once no component uses it any more it is discarded. */ + public void release() { + synchronized (CanonicalRules.class) { + if (instance != this) { + return; + } + if (--users <= 0) { + instance = null; + users = 0; + } + } + } + + /** What is read from the Storm configuration, with the defaults applied. */ + private record Settings( + String canonicalKey, + boolean scopeByDomain, + int minObservations, + int minDistinctPaths, + double confidence, + int maxScopes, + int maxParams, + int maxCachedSources, + Set<String> protectedParams) { + + static Settings from(Map<String, Object> stormConf) { + double confidence = ConfUtils.getFloat(stormConf, CONFIDENCE_PARAM, 0.9f); + if (confidence <= 0d || confidence > 1d) { + LOG.warn("Ignoring invalid value for {}: {}", CONFIDENCE_PARAM, confidence); + confidence = 0.9d; + } + + final List<String> configuredProtected = + stormConf.containsKey(PROTECTED_PARAMS_PARAM) + ? ConfUtils.loadListFromConf(PROTECTED_PARAMS_PARAM, stormConf) + : DEFAULT_PROTECTED_PARAMS; + final Set<String> protectedParams = new HashSet<>(); + for (String param : configuredProtected) { + protectedParams.add(param.toLowerCase(Locale.ROOT)); + } + + return new Settings( + ConfUtils.getString(stormConf, CANONICAL_KEY_PARAM, "canonical"), + "domain".equalsIgnoreCase(ConfUtils.getString(stormConf, SCOPE_PARAM, "host")), + Math.max(1, ConfUtils.getInt(stormConf, MIN_OBSERVATIONS_PARAM, 5)), + Math.max(1, ConfUtils.getInt(stormConf, MIN_DISTINCT_PATHS_PARAM, 3)), + confidence, + Math.max(1, ConfUtils.getInt(stormConf, MAX_SCOPES_PARAM, 10_000)), + Math.max(1, ConfUtils.getInt(stormConf, MAX_PARAMS_PARAM, 20)), + Math.max(1, ConfUtils.getInt(stormConf, MAX_CACHED_SOURCES_PARAM, 50_000)), + Collections.unmodifiableSet(protectedParams)); + } + } + + private final Settings settings; + + /** Evidence per host or domain. */ + private final Cache<String, ConcurrentHashMap<String, ParamStats>> scopes; + + /** + * Fingerprints of the pages already learnt from, so that each counts as a single observation. A + * collision merely skips a page. + */ + private final Cache<Long, Boolean> knownSources; + + private CanonicalRules(Settings settings) { + this.settings = settings; + scopes = Caffeine.newBuilder().maximumSize(settings.maxScopes()).build(); + knownSources = Caffeine.newBuilder().maximumSize(settings.maxCachedSources()).build(); + } + + /** Metadata key holding the value of the canonical tag. */ + public String getCanonicalKey() { + return settings.canonicalKey(); + } + + /** Number of hosts or domains tracked. Pending evictions are performed first. */ + public long getTrackedScopes() { + scopes.cleanUp(); + return scopes.estimatedSize(); + } + + /** Number of parameters tracked for that host or domain, <code>max.params</code> at most. */ + public int getTrackedParams(@NotNull String scopeKey) { + final ConcurrentHashMap<String, ParamStats> scopeStats = scopes.getIfPresent(scopeKey); + return scopeStats == null ? 0 : scopeStats.size(); + } + + /** + * Records what the canonical tag of a page says about its query parameters: the ones it dropped + * are evidence that they do not affect the content, the ones it kept are evidence of the + * opposite. + * + * @param sourceUrl the URL of the page which was parsed + * @param canonicalValue the value of its canonical tag, absolute or relative + */ + public void learn(@Nullable URL sourceUrl, @Nullable String canonicalValue) { + if (sourceUrl == null || StringUtils.isBlank(canonicalValue)) { + return; + } + + // checked before the cache of known sources so that the many URLs without a + // query string, which teach us nothing, do not take up room in it + final Set<String> sourceParams = parameterNames(sourceUrl.getQuery()); + if (sourceParams.isEmpty()) { + return; + } + + // a page is a single observation, whether it has one outlink or a thousand + final String sourceForm = sourceUrl.toExternalForm(); + if (knownSources.asMap().putIfAbsent(fingerprint(sourceForm), Boolean.TRUE) != null) { Review Comment: While its fingerprint is in `knownSources`, a refetched page returns here, so a canonical that now keeps the parameter is ignored. Could a known source still record vetoes, without counting the page again? ########## core/src/main/java/org/apache/stormcrawler/bolt/JSoupParserBolt.java: ########## @@ -646,5 +646,8 @@ public void cleanup() { if (parseFilters != null) { parseFilters.cleanup(); } + if (jsoupFilters != null) { + jsoupFilters.cleanup(); Review Comment: okhttp `HttpProtocol` needs the same since #2128. When `http.allow.redirects` is on it builds its own `URLFilters` for the redirect hops, and the `AdaptiveURLNormalizer` in it never releases `CanonicalRules`. -- 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]
