diff --git a/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java b/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java new file mode 100644 index 000000000..47147848a --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.filtering.adaptive; + +import com.fasterxml.jackson.databind.JsonNode; +import java.net.MalformedURLException; +import java.net.URL; +import java.util.Map; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.filtering.URLFilter; +import org.apache.stormcrawler.util.URLUtil; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * Removes from URLs the query parameters which a site has shown to be irrelevant, based on the + * canonical tags found in its pages. The aim is similar to the Clean-param extension of the + * robots protocol by Yandex, except that the rules are learnt instead of being declared by the + * site. + * + *

Without {@link org.apache.stormcrawler.parse.filter.CanonicalParamLearner} this filter has + * nothing to apply: the parsing bolts filter the outlinks of a page before running the parse + * filters which extract its canonical tag, so the canonical is never in the metadata given here. The + * learner gathers the evidence into the {@link CanonicalRules} instance named by store, + * which also holds the configuration common to both. + * + *

Rules are learnt per JVM, so a URL discovered before a rule was established keeps the form it + * was stored with and two workers may briefly disagree. Rules are only ever added, never withdrawn, + * so the workers converge as the crawl progresses. + * + * @see STORMCRAWLER-315 + */ +public class AdaptiveURLNormalizer extends URLFilter { + + static final String DEFAULT_STORE = "default"; + + private CanonicalRules rules; + + @Override + public void configure(@NotNull Map stormConf, @NotNull JsonNode paramNode) { + final JsonNode node = paramNode.get("store"); + final String store = node == null ? DEFAULT_STORE : node.asText(DEFAULT_STORE); + rules = CanonicalRules.getInstance(stormConf, store); + } + + @Override + public @Nullable String filter( + @Nullable URL sourceUrl, + @Nullable Metadata sourceMetadata, + @NotNull String urlToFilter) { + // usually a no-op, but the canonical is there when the filters are called from a + // parse filter or when the key is in metadata.persist + if (sourceUrl != null && sourceMetadata != null) { + rules.learn(sourceUrl, sourceMetadata.getFirstValue(rules.getCanonicalKey())); + } + return removeIrrelevantParams(urlToFilter); + } + + /** + * Rebuilds the URL without the parameters established as irrelevant for its site. Only the query + * string is rewritten, everything else is copied verbatim, so that a URL which needed sanitizing + * to be parsed is not silently replaced by its sanitized form. + */ + private String removeIrrelevantParams(@NotNull String urlToFilter) { + final int fragment = urlToFilter.indexOf('#'); + final int questionMark = urlToFilter.indexOf('?'); + if (questionMark == -1 || (fragment != -1 && fragment < questionMark)) { + return urlToFilter; + } + final int queryEnd = fragment == -1 ? urlToFilter.length() : fragment; + final String query = urlToFilter.substring(questionMark + 1, queryEnd); + if (query.isEmpty()) { + return urlToFilter; + } + + final URL url; + try { + url = URLUtil.toURL(urlToFilter); + } catch (MalformedURLException e) { + // leave it to the filters in charge of the validity of the URLs + return urlToFilter; + } + + final String scopeKey = rules.scopeKey(url); + if (scopeKey == null) { + return urlToFilter; + } + + final StringBuilder newQuery = new StringBuilder(query.length()); + boolean removedSomething = false; + for (String param : query.split("&", -1)) { + if (rules.isRemovable(scopeKey, CanonicalRules.parameterName(param))) { + removedSomething = true; + continue; + } + if (newQuery.length() > 0) { + newQuery.append('&'); + } + newQuery.append(param); + } + + if (!removedSomething) { + return urlToFilter; + } + + final StringBuilder normalized = new StringBuilder(urlToFilter.length()); + normalized.append(urlToFilter, 0, questionMark); + // http://example.com?a=b has no path: keep the same key as the other normalizers + final String path = url.getPath(); + if (path == null || path.isEmpty()) { + normalized.append('/'); + } + if (newQuery.length() > 0) { + normalized.append('?').append(newQuery); + } + normalized.append(urlToFilter, queryEnd, urlToFilter.length()); + return normalized.toString(); + } +} diff --git a/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/CanonicalRules.java b/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/CanonicalRules.java new file mode 100644 index 000000000..3bb5d0c61 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/CanonicalRules.java @@ -0,0 +1,406 @@ +/* + * 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.PaidLevelDomain; +import java.net.MalformedURLException; +import java.net.URL; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; +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 java.util.concurrent.atomic.AtomicInteger; +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. + * + *

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 an + * instance obtained per JVM and per name with {@link #getInstance(Map, String)}. Configured from the + * Storm configuration, see the adaptive.normalizer.* options, so that both sides cannot + * disagree. Safe to use from several threads. + */ +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"; + + /** + * Never removed, however consistently the canonical tags drop them: a listing serving + * /list?page=2..N with a canonical of /list would otherwise have its + * paginated content normalised away and never fetched. + */ + private static final List DEFAULT_PROTECTED_PARAMS = + Arrays.asList( + "page", "p", "pg", "paged", "offset", "start", "from", "limit", "per_page", + "q", "query", "s", "search", "keyword", "keywords", "sort", "order", "dir", + "lang", "language", "hl", "locale", "id", "category", "cat", "tag", "year", + "month", "day", "view", "format", "type"); + + private static final ConcurrentHashMap INSTANCES = + new ConcurrentHashMap<>(); + + /** Instance registered under that name for this JVM. The configuration of the first caller wins. */ + public static CanonicalRules getInstance( + @NotNull Map stormConf, @NotNull String name) { + return INSTANCES.computeIfAbsent(name, n -> new CanonicalRules(stormConf)); + } + + private final String canonicalKey; + + private final boolean scopeByDomain; + + private final int minObservations; + + private final int minDistinctPaths; + + private final double confidence; + + private final int maxParams; + + private final Set protectedParams; + + /** Evidence per host or domain. */ + private final Cache> scopes; + + /** Pages already learnt from, so that each counts as a single observation. */ + private final Cache knownSources; + + CanonicalRules(@NotNull Map stormConf) { + canonicalKey = ConfUtils.getString(stormConf, CANONICAL_KEY_PARAM, "canonical"); + scopeByDomain = + "domain".equalsIgnoreCase(ConfUtils.getString(stormConf, SCOPE_PARAM, "host")); + minObservations = Math.max(1, ConfUtils.getInt(stormConf, MIN_OBSERVATIONS_PARAM, 5)); + minDistinctPaths = Math.max(1, ConfUtils.getInt(stormConf, MIN_DISTINCT_PATHS_PARAM, 3)); + + final double configuredConfidence = ConfUtils.getFloat(stormConf, CONFIDENCE_PARAM, 0.9f); + if (configuredConfidence <= 0d || configuredConfidence > 1d) { + LOG.warn("Ignoring invalid value for {}: {}", CONFIDENCE_PARAM, configuredConfidence); + confidence = 0.9d; + } else { + confidence = configuredConfidence; + } + + maxParams = Math.max(1, ConfUtils.getInt(stormConf, MAX_PARAMS_PARAM, 100)); + + final Set configuredProtected = new HashSet<>(); + if (stormConf.containsKey(PROTECTED_PARAMS_PARAM)) { + configuredProtected.addAll( + ConfUtils.loadListFromConf(PROTECTED_PARAMS_PARAM, stormConf)); + } else { + configuredProtected.addAll(DEFAULT_PROTECTED_PARAMS); + } + protectedParams = Collections.unmodifiableSet(configuredProtected); + + scopes = + Caffeine.newBuilder() + .maximumSize(Math.max(1, ConfUtils.getInt(stormConf, MAX_SCOPES_PARAM, 10_000))) + .build(); + knownSources = + Caffeine.newBuilder() + .maximumSize( + Math.max( + 1, + ConfUtils.getInt( + stormConf, MAX_CACHED_SOURCES_PARAM, 50_000))) + .build(); + } + + /** Metadata key holding the value of the canonical tag. */ + public String getCanonicalKey() { + return canonicalKey; + } + + /** Number of hosts or domains tracked. Pending evictions are performed first. */ + public long getTrackedScopes() { + scopes.cleanUp(); + return scopes.estimatedSize(); + } + + /** + * 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 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(sourceForm, Boolean.TRUE) != null) { + return; + } + + final URL canonical; + try { + canonical = URLUtil.resolveUrl(sourceUrl, canonicalValue); + } catch (MalformedURLException e) { + LOG.debug("Invalid canonical value {} found in {}", canonicalValue, sourceForm); + return; + } + + // a canonical pointing at another resource tells us nothing about the parameters + if (!sameResource(sourceUrl, canonical)) { + return; + } + + final String scopeKey = scopeKey(sourceUrl); + if (scopeKey == null) { + return; + } + + final Set canonicalParams = parameterNames(canonical.getQuery()); + final ConcurrentHashMap scopeStats = + scopes.get(scopeKey, k -> new ConcurrentHashMap<>()); + final String sourcePath = path(sourceUrl); + + for (String param : sourceParams) { + if (protectedParams.contains(param)) { + continue; + } + final ParamStats stats = statsFor(scopeStats, param, scopeKey); + if (stats == null) { + continue; + } + if (canonicalParams.contains(param)) { + stats.kept.incrementAndGet(); + } else { + stats.dropped.incrementAndGet(); + if (stats.droppedPaths.size() < minDistinctPaths) { + stats.droppedPaths.add(sourcePath); + } + } + promoteIfEstablished(scopeKey, param, stats); + } + } + + /** Whether the parameter has been established as removable for that host or domain. */ + public boolean isRemovable(@Nullable String scopeKey, @NotNull String param) { + if (scopeKey == null) { + return false; + } + final ConcurrentHashMap scopeStats = scopes.getIfPresent(scopeKey); + if (scopeStats == null) { + return false; + } + final ParamStats stats = scopeStats.get(param); + return stats != null && stats.established; + } + + /** Key under which the evidence of a URL is gathered, i.e. its host or its domain. */ + public @Nullable String scopeKey(@NotNull URL url) { + final String host = url.getHost(); + if (StringUtils.isEmpty(host)) { + return null; + } + final String lowerCasedHost = host.toLowerCase(Locale.ROOT); + if (!scopeByDomain) { + return lowerCasedHost; + } + final String domain = PaidLevelDomain.getPLD(lowerCasedHost); + return domain == null ? lowerCasedHost : domain; + } + + /** + * Statistics of a parameter, created if there is room. Room is made by discarding the weakest + * entry, so that a site using per-page tokens as parameter names cannot fill the slots of a host + * for good. + */ + private @Nullable ParamStats statsFor( + ConcurrentHashMap scopeStats, String param, String scopeKey) { + ParamStats stats = scopeStats.get(param); + if (stats != null) { + return stats; + } + synchronized (scopeStats) { + stats = scopeStats.get(param); + if (stats != null) { + return stats; + } + if (scopeStats.size() >= maxParams && !discardWeakest(scopeStats)) { + LOG.debug("Not tracking parameter {} for {}: no room left", param, scopeKey); + return null; + } + stats = new ParamStats(); + scopeStats.put(param, stats); + return stats; + } + } + + /** Discards the parameter with the least evidence, established ones excepted. */ + private boolean discardWeakest(ConcurrentHashMap scopeStats) { + String weakest = null; + int fewest = Integer.MAX_VALUE; + for (Map.Entry entry : scopeStats.entrySet()) { + final ParamStats stats = entry.getValue(); + if (stats.established) { + continue; + } + final int total = stats.total(); + if (total < fewest) { + fewest = total; + weakest = entry.getKey(); + } + } + if (weakest == null) { + return false; + } + scopeStats.remove(weakest); + return true; + } + + /** + * Promotes a parameter to removable once the evidence is sufficient. Promotions are final: a + * rule which came and went would normalise the same URL differently over time. + */ + private void promoteIfEstablished(String scopeKey, String param, ParamStats stats) { + if (stats.established) { + return; + } + final int dropped = stats.dropped.get(); + final int total = dropped + stats.kept.get(); + if (total < minObservations + || (double) dropped / total < confidence + || stats.droppedPaths.size() < minDistinctPaths) { + return; + } + stats.established = true; + LOG.info( + "Removing param {} from the URLs of {}: dropped by {} of {} pages, {} paths", + param, + scopeKey, + dropped, + total, + stats.droppedPaths.size()); + } + + /** Whether both URLs differ by their query string only. */ + private static boolean sameResource(URL source, URL canonical) { + final String sourceHost = source.getHost(); + final String canonicalHost = canonical.getHost(); + if (StringUtils.isEmpty(sourceHost) || !sourceHost.equalsIgnoreCase(canonicalHost)) { + return false; + } + if (!source.getProtocol().equalsIgnoreCase(canonical.getProtocol())) { + return false; + } + if (port(source) != port(canonical)) { + return false; + } + return path(source).equals(path(canonical)); + } + + private static int port(URL url) { + return url.getPort() == -1 ? url.getDefaultPort() : url.getPort(); + } + + /** Path of a URL, never null: opaque URLs such as mailto: have none. */ + private static String path(URL url) { + final String path = url.getPath(); + return path == null ? "" : path; + } + + /** Names of the parameters found in a query string, in their decoded form. */ + static Set parameterNames(@Nullable String query) { + if (StringUtils.isEmpty(query)) { + return Collections.emptySet(); + } + final Set names = new HashSet<>(); + for (String param : query.split("&")) { + if (!param.isEmpty()) { + names.add(parameterName(param)); + } + } + return names; + } + + /** Name of a single name=value pair, in its decoded form. */ + static String parameterName(String param) { + final int equals = param.indexOf('='); + final String name = equals == -1 ? param : param.substring(0, equals); + try { + return URLDecoder.decode(name, StandardCharsets.UTF_8); + } catch (IllegalArgumentException e) { + // malformed percent encoding: compare the names as they are + return name; + } + } + + /** What the canonical tags said about a given parameter of a given host or domain. */ + private static final class ParamStats { + + private final AtomicInteger dropped = new AtomicInteger(); + + private final AtomicInteger kept = new AtomicInteger(); + + /** Distinct paths whose canonical dropped the parameter, capped to what is needed. */ + private final Set droppedPaths = ConcurrentHashMap.newKeySet(); + + private volatile boolean established; + + private int total() { + return dropped.get() + kept.get(); + } + } +} diff --git a/core/src/main/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearner.java b/core/src/main/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearner.java new file mode 100644 index 000000000..79b064cd6 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearner.java @@ -0,0 +1,76 @@ +/* + * 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.parse.filter; + +import com.fasterxml.jackson.databind.JsonNode; +import java.net.MalformedURLException; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; +import org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer; +import org.apache.stormcrawler.filtering.adaptive.CanonicalRules; +import org.apache.stormcrawler.parse.ParseFilter; +import org.apache.stormcrawler.parse.ParseResult; +import org.apache.stormcrawler.util.URLUtil; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.w3c.dom.DocumentFragment; + +/** + * Compares the URL of a page with the value of its canonical tag and records which of its query + * parameters the site considers irrelevant, so that {@link AdaptiveURLNormalizer} can remove them + * from the URLs of that site. + * + *

Must run after the filter extracting the canonical tag into the metadata, typically an + * XPathFilter configured with "canonical": "//*[@rel=\"canonical\"]/@href" + * . Only reads the parse result, never modifies it. + * + * @see CanonicalRules for the configuration, which is shared with the URL filter + * @see STORMCRAWLER-315 + */ +public class CanonicalParamLearner extends ParseFilter { + + private static final Logger LOG = LoggerFactory.getLogger(CanonicalParamLearner.class); + + private CanonicalRules rules; + + @Override + public void configure(@NotNull Map stormConf, @NotNull JsonNode paramNode) { + final JsonNode node = paramNode.get("store"); + final String store = node == null ? "default" : node.asText("default"); + rules = CanonicalRules.getInstance(stormConf, store); + } + + @Override + public void filter(String url, byte[] content, DocumentFragment doc, ParseResult parse) { + // getValues rather than get(url), which would insert an empty ParseData + final String[] canonicals = parse.getValues(url, rules.getCanonicalKey()); + if (canonicals == null || canonicals.length == 0) { + return; + } + final String canonical = canonicals[0]; + if (StringUtils.isBlank(canonical)) { + return; + } + try { + rules.learn(URLUtil.toURL(url), canonical); + } catch (MalformedURLException e) { + LOG.debug("Unable to parse {} while learning from its canonical tag", url); + } + } +} diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml index 27092814f..c18351747 100644 --- a/core/src/main/resources/crawler-default.yaml +++ b/core/src/main/resources/crawler-default.yaml @@ -91,6 +91,22 @@ config: metadata.track.path: true metadata.track.depth: true + # Shared by the AdaptiveURLNormalizer and its companion CanonicalParamLearner. Both are + # optional and must be declared in urlfilters.json / parsefilters.json to have any effect. + adaptive.normalizer.canonical.key: "canonical" + adaptive.normalizer.scope: "host" + adaptive.normalizer.min.observations: 5 + # distinct paths required: guards against self-referencing canonicals on paginated listings + adaptive.normalizer.min.distinct.paths: 3 + adaptive.normalizer.confidence: 0.9 + adaptive.normalizer.max.scopes: 10000 + adaptive.normalizer.max.params: 100 + adaptive.normalizer.max.cached.sources: 50000 + # overrides the built-in list of parameters which are never removed (page, offset, q, id, ...) + # adaptive.normalizer.protected.params: + # - page + # - q + # Agent name info - given here as an example. Do not be an anonymous coward, use your real information! # The full user agent value sent as part of the HTTP requests # is built from the elements below. Only the agent.name is mandatory, diff --git a/core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java b/core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java new file mode 100644 index 000000000..04813e313 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java @@ -0,0 +1,665 @@ +/* + * 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; + +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.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.net.MalformedURLException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer; +import org.apache.stormcrawler.filtering.adaptive.CanonicalRules; +import org.apache.stormcrawler.util.URLUtil; +import org.junit.jupiter.api.Test; + +/** + * sid plays the parameter a site drops from its canonical tags, pid the one it keeps. + * Neither is in the protected list, unlike id. + */ +class AdaptiveURLNormalizerTest { + + private static final AtomicInteger STORE_COUNTER = new AtomicInteger(); + + /** Rules are shared per JVM and per name, so every test needs a store of its own. */ + private CanonicalRules rules; + + private AdaptiveURLNormalizer createFilter(Map conf) { + AdaptiveURLNormalizer filter = new AdaptiveURLNormalizer(); + ObjectNode filterParams = new ObjectNode(JsonNodeFactory.instance); + String store = "test-" + STORE_COUNTER.incrementAndGet(); + filterParams.put("store", store); + filter.configure(conf, filterParams); + rules = CanonicalRules.getInstance(conf, store); + return filter; + } + + private AdaptiveURLNormalizer createFilter() { + return createFilter(new HashMap<>()); + } + + /** Simulates the parsing of a page having the given canonical tag. */ + private String observe(AdaptiveURLNormalizer filter, String pageUrl, String canonicalValue) + throws MalformedURLException { + return observe(filter, pageUrl, "canonical", canonicalValue); + } + + private String observe( + AdaptiveURLNormalizer filter, String pageUrl, String metadataKey, String canonicalValue) + throws MalformedURLException { + Metadata metadata = new Metadata(); + if (canonicalValue != null) { + metadata.setValue(metadataKey, canonicalValue); + } + return filter.filter(URLUtil.toURL(pageUrl), metadata, "http://example.com/seed"); + } + + /** Applies the rules learnt so far without providing any new evidence. */ + private String apply(AdaptiveURLNormalizer filter, String url) { + return filter.filter(null, null, url); + } + + /** Observes pages whose canonical drops sid but keeps pid. */ + private void observeSessionParam(AdaptiveURLNormalizer filter, int observations) + throws MalformedURLException { + for (int i = 0; i < observations; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i); + } + } + + @Test + void testUnchangedWithoutEvidence() { + AdaptiveURLNormalizer filter = createFilter(); + assertEquals( + "http://example.com/page?pid=1&sid=abc", + apply(filter, "http://example.com/page?pid=1&sid=abc")); + } + + @Test + void testNoEvidenceWithoutCanonical() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 20; i++) { + observe(filter, "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, null); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testRemovesIrrelevantParameter() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testNotAppliedBelowMinObservations() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 4); + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testMinObservationsIsConfigurable() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MIN_OBSERVATIONS_PARAM, 2); + conf.put(CanonicalRules.MIN_DISTINCT_PATHS_PARAM, 2); + AdaptiveURLNormalizer filter = createFilter(conf); + observeSessionParam(filter, 2); + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testParameterKeptByCanonicalIsNeverRemoved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 20); + assertEquals( + "http://example.com/other?pid=9", apply(filter, "http://example.com/other?pid=9")); + } + + @Test + void testCanonicalIdenticalToSourceKeepsEverything() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 10; i++) { + String page = "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i; + observe(filter, page, page); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testConfidenceThreshold() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MIN_OBSERVATIONS_PARAM, 4); + conf.put(CanonicalRules.MIN_DISTINCT_PATHS_PARAM, 3); + conf.put(CanonicalRules.CONFIDENCE_PARAM, 0.75d); + AdaptiveURLNormalizer filter = createFilter(conf); + + // sid: dropped 3 times out of 4 -> 0.75, at the threshold + // ref: dropped 2 times out of 4 -> 0.5, below the threshold + observe(filter, "http://example.com/a?sid=x&ref=1", "http://example.com/a"); + observe(filter, "http://example.com/b?sid=x&ref=1", "http://example.com/b"); + observe(filter, "http://example.com/c?sid=x&ref=1", "http://example.com/c?ref=1"); + observe(filter, "http://example.com/d?sid=x&ref=1", "http://example.com/d?sid=x&ref=1"); + + assertEquals( + "http://example.com/e?ref=3", apply(filter, "http://example.com/e?sid=x&ref=3")); + } + + @Test + void testEvidenceIsScopedToTheHost() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://another.com/other?pid=9&sid=zzz", + apply(filter, "http://another.com/other?pid=9&sid=zzz")); + assertEquals( + "http://sub.example.com/other?pid=9&sid=zzz", + apply(filter, "http://sub.example.com/other?pid=9&sid=zzz")); + } + + @Test + void testEvidenceCanBeScopedToTheDomain() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.SCOPE_PARAM, "domain"); + AdaptiveURLNormalizer filter = createFilter(conf); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://www.example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://www.example.com/page" + i + "?pid=" + i); + } + assertEquals( + "http://shop.example.com/other?pid=9", + apply(filter, "http://shop.example.com/other?pid=9&sid=zzz")); + assertEquals( + "http://another.com/other?pid=9&sid=zzz", + apply(filter, "http://another.com/other?pid=9&sid=zzz")); + } + + @Test + void testHostIsCaseInsensitive() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://EXAMPLE.com/other?pid=9", + apply(filter, "http://EXAMPLE.com/other?pid=9&sid=zzz")); + } + + @Test + void testCanonicalOnAnotherPathIsIgnored() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 10; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/canonical" + i); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testCanonicalOnAnotherHostIsIgnored() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 10; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://mirror.example.com/page" + i + "?pid=" + i); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testRelativeCanonicalIsResolved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "/page" + i + "?pid=" + i); + } + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testPureQueryCanonicalIsResolved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "?pid=" + i); + } + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testInvalidCanonicalIsIgnored() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 10; i++) { + observe(filter, "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, ":::"); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testWholeQueryStringCanBeRemoved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?sid=abc" + i, + "http://example.com/page" + i); + } + assertEquals("http://example.com/other", apply(filter, "http://example.com/other?sid=zzz")); + } + + @Test + void testFragmentAndEncodingArePreserved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?pid=a%20b#top", + apply(filter, "http://example.com/other?sid=zzz&pid=a%20b#top")); + } + + @Test + void testQuestionMarkInsideFragmentIsNotAQueryString() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other#anchor?sid=zzz", + apply(filter, "http://example.com/other#anchor?sid=zzz")); + } + + @Test + void testEmptyPathIsGivenASlash() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + // BasicURLNormalizer would store this URL as http://example.com/?pid=2 + assertEquals("http://example.com/?pid=2", apply(filter, "http://example.com?sid=1&pid=2")); + } + + @Test + void testOnlyTheQueryStringIsRewritten() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + // the space would be escaped by URLUtil.toURL: the rest must be left as it was + assertEquals( + "http://example.com/a b?pid=2", apply(filter, "http://example.com/a b?sid=1&pid=2")); + } + + @Test + void testNonDefaultPortIsPreserved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com:8080/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com:8080/page" + i + "?pid=" + i); + } + assertEquals( + "http://example.com:8080/other?pid=9", + apply(filter, "http://example.com:8080/other?pid=9&sid=zzz")); + } + + @Test + void testCanonicalOnAnotherPortIsIgnored() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 10; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com:8080/page" + i + "?pid=" + i); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testDefaultPortMatchesImplicitOne() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com:80/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i); + } + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testEncodedParameterNames() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?session%20id=abc" + i, + "http://example.com/page" + i); + } + assertEquals( + "http://example.com/other", + apply(filter, "http://example.com/other?session%20id=zzz")); + } + + @Test + void testAPageIsOnlyCountedOnce() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + String page = "http://example.com/page?pid=1&sid=abc"; + String canonical = "http://example.com/page?pid=1"; + // the filter is called once per outlink of the same page + for (int i = 0; i < 20; i++) { + observe(filter, page, canonical); + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testAPageIsOnlyCountedOnceAcrossRefetches() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + // the same two pages seen again and again, as would happen over successive fetch cycles + for (int round = 0; round < 10; round++) { + for (int page = 0; page < 2; page++) { + observe( + filter, + "http://example.com/page" + page + "?pid=1&sid=abc", + "http://example.com/page" + page + "?pid=1"); + } + } + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testPaginationSurvivesSelfReferencingCanonicals() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + // every page of a listing declaring the bare listing as canonical + for (int i = 2; i < 30; i++) { + observe(filter, "http://example.com/list?offset=" + i, "http://example.com/list"); + } + assertEquals( + "http://example.com/list?offset=7", apply(filter, "http://example.com/list?offset=7")); + } + + @Test + void testEvidenceFromASinglePathIsNotEnough() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + // sid is not protected, but all the evidence comes from the same path + for (int i = 0; i < 30; i++) { + observe(filter, "http://example.com/list?sid=" + i, "http://example.com/list"); + } + assertEquals( + "http://example.com/list?sid=7", apply(filter, "http://example.com/list?sid=7")); + } + + @Test + void testDistinctPathsRequirementIsConfigurable() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MIN_DISTINCT_PATHS_PARAM, 1); + AdaptiveURLNormalizer filter = createFilter(conf); + for (int i = 0; i < 5; i++) { + observe(filter, "http://example.com/list?sid=" + i, "http://example.com/list"); + } + assertEquals("http://example.com/list", apply(filter, "http://example.com/list?sid=7")); + } + + @Test + void testProtectedParametersAreNeverRemoved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 20; i++) { + observe( + filter, + "http://example.com/page" + i + "?page=2&sid=abc" + i, + "http://example.com/page" + i); + } + // sid was learnt, page is protected even though the evidence is identical + assertEquals( + "http://example.com/other?page=2", + apply(filter, "http://example.com/other?page=2&sid=zzz")); + } + + @Test + void testProtectedParametersCanBeOverridden() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.PROTECTED_PARAMS_PARAM, Collections.emptyList()); + AdaptiveURLNormalizer filter = createFilter(conf); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?page=2&sid=abc" + i, + "http://example.com/page" + i); + } + assertEquals("http://example.com/other", apply(filter, "http://example.com/other?page=2")); + } + + @Test + void testProtectedParametersCanBeListed() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.PROTECTED_PARAMS_PARAM, Arrays.asList("sid")); + AdaptiveURLNormalizer filter = createFilter(conf); + observeSessionParam(filter, 20); + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testAnEstablishedRuleIsNeverWithdrawn() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + + // enough contrary evidence to drop the ratio below the threshold: the rule must stand + for (int i = 100; i < 200; i++) { + String page = "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i; + observe(filter, page, page); + } + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testNullSourceIsHandled() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + assertEquals( + "http://example.com/other?sid=zzz", + filter.filter(null, new Metadata(), "http://example.com/other?sid=zzz")); + assertEquals( + "http://example.com/other?sid=zzz", + filter.filter( + URLUtil.toURL("http://example.com/page?sid=1"), + null, + "http://example.com/other?sid=zzz")); + } + + @Test + void testMalformedURLIsLeftUntouched() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + String malformed = "this is not a URL?sid=1"; + assertEquals(malformed, apply(filter, malformed)); + assertEquals("mailto:someone@example.com", apply(filter, "mailto:someone@example.com")); + } + + @Test + void testCustomCanonicalMetadataKey() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.CANONICAL_KEY_PARAM, "parse.canonical"); + AdaptiveURLNormalizer filter = createFilter(conf); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "parse.canonical", + "http://example.com/page" + i + "?pid=" + i); + } + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testWeakestParameterIsDiscardedWhenFull() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MAX_PARAMS_PARAM, 2); + AdaptiveURLNormalizer filter = createFilter(conf); + + // per-page tokens as parameter names would otherwise fill the slots for good + for (int i = 0; i < 20; i++) { + observe( + filter, + "http://example.com/junk" + i + "?token" + i + "=x&nonce" + i + "=y", + "http://example.com/junk" + i); + } + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?sid=abc" + i, + "http://example.com/page" + i); + } + assertEquals("http://example.com/other", apply(filter, "http://example.com/other?sid=zzz")); + } + + @Test + void testNumberOfTrackedSitesIsBounded() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MAX_SCOPES_PARAM, 1); + AdaptiveURLNormalizer filter = createFilter(conf); + + observeSessionParam(filter, 5); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://another.com/page" + i + "?sid=abc" + i, + "http://another.com/page" + i); + } + + assertEquals(1L, rules.getTrackedScopes(), "the second host should have evicted the first"); + + boolean firstStillKnown = + "http://example.com/other?pid=9" + .equals(apply(filter, "http://example.com/other?pid=9&sid=zzz")); + boolean secondKnown = + "http://another.com/other".equals(apply(filter, "http://another.com/other?sid=zzz")); + assertFalse( + firstStillKnown && secondKnown, "only one host should be retained with maxScopes 1"); + assertTrue(firstStillKnown || secondKnown, "the surviving host should still have its rule"); + } + + @Test + void testInvalidConfidenceFallsBackToTheDefault() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.CONFIDENCE_PARAM, 1.5d); + AdaptiveURLNormalizer filter = createFilter(conf); + + observeSessionParam(filter, 4); + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testSharedByConcurrentThreads() throws Exception { + // the URL filters of a FetcherBolt are shared by all of its fetcher threads + AdaptiveURLNormalizer filter = createFilter(); + final int threads = 4; + final int pagesPerThread = 100; + final ExecutorService pool = Executors.newFixedThreadPool(threads); + final CountDownLatch start = new CountDownLatch(1); + final AtomicReference failure = new AtomicReference<>(); + final List tasks = new ArrayList<>(); + + for (int t = 0; t < threads; t++) { + final int thread = t; + tasks.add( + () -> { + try { + start.await(); + for (int i = 0; i < pagesPerThread; i++) { + String page = + "http://example.com/t" + thread + "p" + i + "?sid=abc" + i; + observe(filter, page, "http://example.com/t" + thread + "p" + i); + apply(filter, "http://example.com/other?sid=zzz"); + } + } catch (Throwable e) { + failure.compareAndSet(null, e); + } + }); + } + + tasks.forEach(pool::execute); + start.countDown(); + pool.shutdown(); + assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS), "the threads should have finished"); + if (failure.get() != null) { + throw new AssertionError("a thread failed", failure.get()); + } + assertEquals("http://example.com/other", apply(filter, "http://example.com/other?sid=zzz")); + } +} diff --git a/core/src/test/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearnerTest.java b/core/src/test/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearnerTest.java new file mode 100644 index 000000000..8d78b5d1d --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearnerTest.java @@ -0,0 +1,145 @@ +/* + * 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.parse.filter; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer; +import org.apache.stormcrawler.filtering.adaptive.CanonicalRules; +import org.apache.stormcrawler.parse.ParseResult; +import org.junit.jupiter.api.Test; + +/** Tests that the learner feeds the rules which the {@link AdaptiveURLNormalizer} applies. */ +class CanonicalParamLearnerTest { + + private static final AtomicInteger STORE_COUNTER = new AtomicInteger(); + + private final Map conf = new HashMap<>(); + + private final String store = "learner-test-" + STORE_COUNTER.incrementAndGet(); + + private CanonicalParamLearner createLearner() { + CanonicalParamLearner learner = new CanonicalParamLearner(); + learner.configure(conf, storeParams()); + return learner; + } + + private AdaptiveURLNormalizer createFilter() { + AdaptiveURLNormalizer filter = new AdaptiveURLNormalizer(); + filter.configure(conf, storeParams()); + return filter; + } + + private ObjectNode storeParams() { + ObjectNode params = new ObjectNode(JsonNodeFactory.instance); + params.put("store", store); + return params; + } + + /** Simulates a page being parsed, its canonical tag already extracted into the metadata. */ + private void parse(CanonicalParamLearner learner, String url, String canonicalValue) { + ParseResult parse = new ParseResult(); + Metadata metadata = new Metadata(); + if (canonicalValue != null) { + metadata.setValue("canonical", canonicalValue); + } + parse.set(url, metadata); + learner.filter(url, new byte[0], null, parse); + } + + @Test + void testLearnerFeedsTheFilter() { + CanonicalParamLearner learner = createLearner(); + AdaptiveURLNormalizer filter = createFilter(); + + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + filter.filter(null, null, "http://example.com/other?pid=9&sid=zzz")); + + for (int i = 0; i < 5; i++) { + parse( + learner, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i); + } + + assertEquals( + "http://example.com/other?pid=9", + filter.filter(null, null, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testPagesWithoutCanonicalAreIgnored() { + CanonicalParamLearner learner = createLearner(); + AdaptiveURLNormalizer filter = createFilter(); + + for (int i = 0; i < 20; i++) { + parse(learner, "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, null); + } + + assertEquals( + "http://example.com/other?pid=9&sid=zzz", + filter.filter(null, null, "http://example.com/other?pid=9&sid=zzz")); + } + + @Test + void testUnknownURLIsIgnored() { + CanonicalParamLearner learner = createLearner(); + ParseResult parse = new ParseResult(); + learner.filter("http://example.com/page?sid=1", new byte[0], null, parse); + // no ParseData must have been created for it + assertTrue(parse.getParseMap().isEmpty()); + } + + @Test + void testMalformedURLIsIgnored() { + CanonicalParamLearner learner = createLearner(); + parse(learner, "this is not a URL", "http://example.com/page"); + } + + @Test + void testTheLearnerDoesNotNeedTheDOM() { + assertEquals(false, createLearner().needsDOM()); + } + + @Test + void testConfigurationIsSharedThroughTheStore() { + conf.put(CanonicalRules.MIN_OBSERVATIONS_PARAM, 2); + conf.put(CanonicalRules.MIN_DISTINCT_PATHS_PARAM, 2); + CanonicalParamLearner learner = createLearner(); + AdaptiveURLNormalizer filter = createFilter(); + + for (int i = 0; i < 2; i++) { + parse( + learner, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i); + } + + assertEquals( + "http://example.com/other?pid=9", + filter.filter(null, null, "http://example.com/other?pid=9&sid=zzz")); + } +} diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index 3fc80f267..b543b87eb 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -311,6 +311,28 @@ NOTE: When a proxy is configured, the connection is established to the proxy and proxy's IP address rather than the target host's resolved address. IP filtering is therefore effectively disabled for proxied fetches. +==== Adaptive URL normalization + +The values below are shared by the xref:internals.adoc#adaptivenormalizer[AdaptiveURLNormalizer] and +its companion `CanonicalParamLearner`. They are read from the Storm configuration rather than from +the JSON filter definitions so that the learning and the filtering sides cannot be configured +differently. They have no effect unless both components are declared. + +[cols="1,1,3", options="header"] +|=== +| key | default value | description + +| adaptive.normalizer.canonical.key | canonical | Metadata key holding the value of the canonical tag, as extracted by the `XPathFilter`. +| adaptive.normalizer.scope | host | Whether the rules are learnt per `host` or per `domain`. +| adaptive.normalizer.min.observations | 5 | Number of distinct pages a parameter must have been seen on before it can be removed. +| adaptive.normalizer.min.distinct.paths | 3 | Number of distinct paths whose canonical must have dropped the parameter. Guards against self-referencing canonicals on paginated listings, which are confined to a single path. +| adaptive.normalizer.confidence | 0.9 | Proportion of the canonical tags which must have dropped the parameter. +| adaptive.normalizer.max.scopes | 10000 | Maximum number of hosts or domains tracked. +| adaptive.normalizer.max.params | 100 | Maximum number of parameters tracked per host or domain. When full, the parameter with the least evidence is discarded. +| adaptive.normalizer.max.cached.sources | 50000 | Maximum number of source URLs remembered, so that a page counts as a single observation however many outlinks it has and however many times it is refetched. +| adaptive.normalizer.protected.params | see description | Parameters which are never removed, however consistently the canonical tags drop them. Defaults to `page`, `p`, `pg`, `paged`, `offset`, `start`, `from`, `limit`, `per_page`, `q`, `query`, `s`, `search`, `keyword`, `keywords`, `sort`, `order`, `dir`, `lang`, `language`, `hl`, `locale`, `id`, `category`, `cat`, `tag`, `year`, `month`, `day`, `view`, `format` and `type`. Set to an empty list to disable. +|=== + ==== Indexing The values below are used by sub-classes of `AbstractIndexerBolt`. diff --git a/docs/src/main/asciidoc/internals.adoc b/docs/src/main/asciidoc/internals.adoc index ab5cddeb5..77e389d56 100644 --- a/docs/src/main/asciidoc/internals.adoc +++ b/docs/src/main/asciidoc/internals.adoc @@ -209,6 +209,7 @@ The archetype includes a default link:https://github.com/apache/stormcrawler/blo } ---- +* **CanonicalParamLearner** – link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearner.java[CanonicalParamLearner] compares the URL of a page with the value of its canonical tag to work out which of its query parameters are irrelevant, for the xref:internals.adoc#adaptivenormalizer[AdaptiveURLNormalizer] to remove. It must be declared after the filter extracting the canonical tag and does not modify the parse result. * **CommaSeparatedToMultivaluedMetadata** – link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/parse/filter/CommaSeparatedToMultivaluedMetadata.java[CommaSeparatedToMultivaluedMetadata] rewrites single metadata values containing comma-separated entries into multiple values for the same key, useful for keyword tags. * **DebugParseFilter** – link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/parse/filter/DebugParseFilter.java[DebugParseFilter] dumps an XML representation of the DOM structure to a temporary file. * **DomainParseFilter** – link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/parse/filter/DomainParseFilter.java[DomainParseFilter] stores the domain or host name in the metadata for later indexing. @@ -249,6 +250,61 @@ The JSON configuration allows loading several instances of the same filtering cl ===== Built-in URL Filters +[[adaptivenormalizer]] +====== Adaptive +The link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java[AdaptiveURLNormalizer] removes from the URLs of a site the query parameters which its canonical tags have shown to be irrelevant. + +When the URL of a page and the value of its canonical tag point at the same resource - i.e. they have the same protocol, host, port and path - and differ only by their query string, the parameters dropped by the canonical are taken as evidence that they do not affect the content, whereas the ones kept by the canonical are evidence of the opposite. Once enough evidence has been gathered for a given parameter, it gets removed from the subsequent URLs of that site, which reduces the amount of duplicates fetched. The aim is similar to the _Clean-param_ extension of the robots protocol by Yandex, except that the rules are learnt instead of being declared by the site. + +*This filter does nothing on its own.* The parsing bolts filter the outlinks of a page _before_ running the filters which extract its canonical tag, so the canonical is never in the metadata passed to the URL filters. The evidence is therefore gathered by a companion parse filter, link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearner.java[CanonicalParamLearner], which runs after the canonical has been extracted and shares its findings with the URL filter through a store named in the configuration of both. Three things are needed: + +. a filter extracting the canonical tag into the metadata. The archetype does this with the JSoup `XPathFilter` in `jsoupfilters.json`, which is always run before the parse filters: ++ +[source,json] +---- +{ + "class": "org.apache.stormcrawler.jsoup.XPathFilter", + "name": "XPathFilter", + "params": { + "canonical": "//*[@rel=\"canonical\"]/@href" + } +} +---- +. the learner, in `parsefilters.json`. If the canonical is extracted by the ParseFilter flavour of `XPathFilter` instead, the learner must be declared *after* it in that same file: ++ +[source,json] +---- +{ + "class": "org.apache.stormcrawler.parse.filter.CanonicalParamLearner", + "name": "CanonicalParamLearner", + "params": { + "store": "default" + } +} +---- +. the URL filter itself, in `urlfilters.json`: ++ +[source,json] +---- +{ + "class": "org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer", + "name": "AdaptiveURLNormalizer", + "params": { + "store": "default" + } +} +---- + +The `store` parameter is optional and defaults to _default_. Everything else is configured with the xref:configuration.adoc[standard configuration mechanism] rather than in the JSON files, so that the two sides cannot disagree. See the `adaptive.normalizer.*` options in the configuration reference. + +By default a parameter must have been dropped by the canonical of at least 5 distinct pages spread over at least 3 distinct paths, and kept by fewer than 10% of them, before it is removed. The requirement on distinct paths guards against self-referencing canonicals, a common misconfiguration where every page of a listing declares the bare listing as canonical: taken at face value, `page` would be found irrelevant and the whole of the paginated content would be normalised away and never fetched. For the same reason, a list of parameter names which are never removed - `page`, `offset`, `q`, `sort`, `id` and similar - is applied on top and can be replaced with `adaptive.normalizer.protected.params`. + +Limitations worth knowing before enabling it: + +* The evidence is held in memory and is lost when the topology is restarted. It is shared by all the components of a worker but not between workers, so two workers may briefly normalise the same URL differently. Rules are only ever added and never withdrawn, so the workers converge as the crawl progresses. +* A URL discovered before a rule was established keeps the form it was stored with; the status store is not rewritten retrospectively. +* Nothing is learnt from a page whose canonical tag is missing, points at another resource, or which has no query string of its own. + ====== Basic The link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/filtering/basic/BasicURLFilter.java[BasicURLFilter] filters based on the length of the URL and the repetition of path elements.