From be6bc5a0167612d01f869e30d4aafe43a1b82c89 Mon Sep 17 00:00:00 2001 From: Akash Manna Date: Sun, 16 Aug 2026 17:26:24 +0530 Subject: [PATCH 1/2] Adaptative URL filter to normalize URLs based on canonical tag --- .../adaptive/AdaptiveURLNormalizer.java | 416 +++++++++++++++ .../filtering/AdaptiveURLNormalizerTest.java | 472 ++++++++++++++++++ docs/src/main/asciidoc/internals.adoc | 32 ++ 3 files changed, 920 insertions(+) create mode 100644 core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java create mode 100644 core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java 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..216930351 --- /dev/null +++ b/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java @@ -0,0 +1,416 @@ +/* + * 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 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.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import org.apache.commons.lang3.StringUtils; +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; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Normalizes URLs by removing the query parameters which a site has shown to be irrelevant, based + * on the canonical tags found in its pages. + * + *

Whenever a page is parsed, this filter compares the URL of the page with the value of its + * canonical tag (as extracted into the metadata, see canonicalMetadataKey). When both + * point at the same resource - 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, subsequent URLs for that site + * get the parameter removed, 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. + * + *

The evidence is kept in memory only and is therefore lost when the topology is restarted; it + * is not shared between the instances of the bolt either. Both the number of sites and the number + * of parameters tracked per site are bounded, see maxScopes and maxParams + * . + * + *

Configuration, all parameters are optional: + * + * 

{@code
+ * {
+ *   "class": "org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer",
+ *   "name": "AdaptiveURLNormalizer",
+ *   "params": {
+ *     "canonicalMetadataKey": "canonical",
+ *     "scope": "host",
+ *     "minObservations": 5,
+ *     "confidenceThreshold": 0.9,
+ *     "maxScopes": 10000,
+ *     "maxParams": 100
+ *   }
+ * }
+ * }
+ * + * @see STORMCRAWLER-315 + */ +public class AdaptiveURLNormalizer extends URLFilter { + + private static final Logger LOG = LoggerFactory.getLogger(AdaptiveURLNormalizer.class); + + /** Metadata key under which the canonical tag is stored by default. */ + private static final String DEFAULT_CANONICAL_KEY = "canonical"; + + private static final int DEFAULT_MIN_OBSERVATIONS = 5; + + private static final double DEFAULT_CONFIDENCE_THRESHOLD = 0.9d; + + private static final int DEFAULT_MAX_SCOPES = 10_000; + + private static final int DEFAULT_MAX_PARAMS = 100; + + private String canonicalMetadataKey = DEFAULT_CANONICAL_KEY; + + private int minObservations = DEFAULT_MIN_OBSERVATIONS; + + private double confidenceThreshold = DEFAULT_CONFIDENCE_THRESHOLD; + + private int maxParams = DEFAULT_MAX_PARAMS; + + /** Whether the rules are learnt per domain instead of per host. */ + private boolean scopeByDomain = false; + + /** Evidence gathered so far, keyed by host or domain. */ + private Cache> stats = buildCache(DEFAULT_MAX_SCOPES); + + /** + * URL of the page the last observation was made from. The filter is called once per outlink, + * i.e. many times in a row with the same source, but a page must only count once. + */ + private String lastLearnedSource; + + private static Cache> buildCache(int maxScopes) { + return Caffeine.newBuilder().maximumSize(maxScopes).build(); + } + + @Override + public void configure(@NotNull Map stormConf, @NotNull JsonNode paramNode) { + int maxScopes = DEFAULT_MAX_SCOPES; + + JsonNode node = paramNode.get("canonicalMetadataKey"); + if (node != null) { + final String key = node.asText(); + if (StringUtils.isBlank(key)) { + LOG.warn("Ignoring blank value for canonicalMetadataKey"); + } else { + canonicalMetadataKey = key; + } + } + + node = paramNode.get("scope"); + if (node != null) { + final String scope = node.asText(); + if ("domain".equalsIgnoreCase(scope)) { + scopeByDomain = true; + } else if ("host".equalsIgnoreCase(scope)) { + scopeByDomain = false; + } else { + LOG.warn("Unknown value for scope: {}, using host", scope); + } + } + + node = paramNode.get("minObservations"); + if (node != null) { + final int value = node.asInt(DEFAULT_MIN_OBSERVATIONS); + if (value < 1) { + LOG.warn("Ignoring invalid value for minObservations: {}", value); + } else { + minObservations = value; + } + } + + node = paramNode.get("confidenceThreshold"); + if (node != null) { + final double value = node.asDouble(DEFAULT_CONFIDENCE_THRESHOLD); + if (value <= 0d || value > 1d) { + LOG.warn("Ignoring invalid value for confidenceThreshold: {}", value); + } else { + confidenceThreshold = value; + } + } + + node = paramNode.get("maxScopes"); + if (node != null) { + final int value = node.asInt(DEFAULT_MAX_SCOPES); + if (value < 1) { + LOG.warn("Ignoring invalid value for maxScopes: {}", value); + } else { + maxScopes = value; + } + } + + node = paramNode.get("maxParams"); + if (node != null) { + final int value = node.asInt(DEFAULT_MAX_PARAMS); + if (value < 1) { + LOG.warn("Ignoring invalid value for maxParams: {}", value); + } else { + maxParams = value; + } + } + + stats = buildCache(maxScopes); + } + + @Override + public @Nullable String filter( + @Nullable URL sourceUrl, + @Nullable Metadata sourceMetadata, + @NotNull String urlToFilter) { + learn(sourceUrl, sourceMetadata); + return removeIrrelevantParams(urlToFilter); + } + + /** + * Compares the URL of the page being parsed with its canonical tag and records which of its + * query parameters the site considers irrelevant. + */ + private void learn(@Nullable URL sourceUrl, @Nullable Metadata sourceMetadata) { + if (sourceUrl == null || sourceMetadata == null) { + return; + } + + final String canonicalValue = sourceMetadata.getFirstValue(canonicalMetadataKey); + if (StringUtils.isBlank(canonicalValue)) { + return; + } + + // the filter is called once per outlink: a page must only be counted once + final String sourceForm = sourceUrl.toExternalForm(); + if (sourceForm.equals(lastLearnedSource)) { + return; + } + lastLearnedSource = sourceForm; + + // nothing to learn from a URL without a query string + final Set sourceParams = parameterNames(sourceUrl.getQuery()); + if (sourceParams.isEmpty()) { + 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 Map scopeStats = stats.get(scopeKey, k -> new HashMap<>()); + + for (String param : sourceParams) { + ParamStats paramStats = scopeStats.get(param); + if (paramStats == null) { + if (scopeStats.size() >= maxParams) { + LOG.debug( + "Not tracking parameter {} for {}: limit of {} reached", + param, + scopeKey, + maxParams); + continue; + } + paramStats = new ParamStats(); + scopeStats.put(param, paramStats); + } + final boolean wasRemovable = isRemovable(paramStats); + if (canonicalParams.contains(param)) { + paramStats.kept++; + } else { + paramStats.dropped++; + } + if (!wasRemovable && isRemovable(paramStats)) { + LOG.info( + "Removing param {} from the URLs of {}: dropped by {} of {} canonicals", + param, + scopeKey, + paramStats.dropped, + paramStats.total()); + } + } + } + + /** Removes from the URL the parameters which have been found to be irrelevant for its site. */ + private String removeIrrelevantParams(@NotNull String 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 query = url.getQuery(); + if (StringUtils.isEmpty(query)) { + return urlToFilter; + } + + final String scopeKey = scopeKey(url); + if (scopeKey == null) { + return urlToFilter; + } + + final Map scopeStats = stats.getIfPresent(scopeKey); + if (scopeStats == null) { + return urlToFilter; + } + + final StringBuilder newQuery = new StringBuilder(query.length()); + boolean removedSomething = false; + // the parameters are kept verbatim so that their encoding is left untouched + for (String param : query.split("&", -1)) { + final ParamStats paramStats = scopeStats.get(parameterName(param)); + if (paramStats != null && isRemovable(paramStats)) { + 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(url.getProtocol()).append(':'); + final String authority = url.getAuthority(); + if (StringUtils.isNotEmpty(authority)) { + normalized.append("//").append(authority); + } + normalized.append(url.getPath()); + if (newQuery.length() > 0) { + normalized.append('?').append(newQuery); + } + final String ref = url.getRef(); + if (ref != null) { + normalized.append('#').append(ref); + } + return normalized.toString(); + } + + private boolean isRemovable(ParamStats paramStats) { + final int total = paramStats.total(); + return total >= minObservations + && (double) paramStats.dropped / total >= confidenceThreshold; + } + + /** Key under which the evidence is gathered, i.e. the host or the domain of the URL. */ + private @Nullable String scopeKey(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; + } + + /** 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 source.getPath().equals(canonical.getPath()); + } + + private static int port(URL url) { + return url.getPort() == -1 ? url.getDefaultPort() : url.getPort(); + } + + /** Names of the parameters found in a query string, in their decoded form. */ + private 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. */ + private 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; + } + } + + /** Number of times a given parameter was dropped or kept by a canonical tag. */ + private static final class ParamStats { + + private int dropped; + + private int kept; + + private int total() { + return dropped + kept; + } + } +} 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..6b7c2a495 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java @@ -0,0 +1,472 @@ +/* + * 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 com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import java.net.MalformedURLException; +import java.util.HashMap; +import java.util.Map; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer; +import org.apache.stormcrawler.util.URLUtil; +import org.junit.jupiter.api.Test; + +/** + * Tests the learning of the query parameters which can be removed from the URLs of a site, based on + * the canonical tags found in its pages. + */ +class AdaptiveURLNormalizerTest { + + private static final String CANONICAL = "canonical"; + + private AdaptiveURLNormalizer createFilter() { + return createFilter(new ObjectNode(JsonNodeFactory.instance)); + } + + private AdaptiveURLNormalizer createFilter(ObjectNode filterParams) { + AdaptiveURLNormalizer filter = new AdaptiveURLNormalizer(); + Map conf = new HashMap<>(); + filter.configure(conf, filterParams); + return filter; + } + + private static ObjectNode params() { + return new ObjectNode(JsonNodeFactory.instance); + } + + /** 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 the sid parameter but keeps the id one. + */ + private void observeSessionParam(AdaptiveURLNormalizer filter, int observations) + throws MalformedURLException { + for (int i = 0; i < observations; i++) { + observe( + filter, + "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?id=" + i); + } + } + + @Test + void testUnchangedWithoutEvidence() { + AdaptiveURLNormalizer filter = createFilter(); + assertEquals( + "http://example.com/page?id=1&sid=abc", + apply(filter, "http://example.com/page?id=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 + "?id=" + i + "&sid=abc" + i, null); + } + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testRemovesIrrelevantParameter() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testNotAppliedBelowMinObservations() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 4); + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testMinObservationsIsConfigurable() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("minObservations", 2); + AdaptiveURLNormalizer filter = createFilter(filterParams); + observeSessionParam(filter, 2); + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testParameterKeptByCanonicalIsNeverRemoved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 20); + assertEquals( + "http://example.com/other?id=9", apply(filter, "http://example.com/other?id=9")); + } + + @Test + void testCanonicalIdenticalToSourceKeepsEverything() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 10; i++) { + String page = "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i; + observe(filter, page, page); + } + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testConfidenceThreshold() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("minObservations", 4); + filterParams.put("confidenceThreshold", 0.75d); + AdaptiveURLNormalizer filter = createFilter(filterParams); + + // sort: dropped 3 times out of 4 -> 0.75, at the threshold + // page: dropped 2 times out of 4 -> 0.5, below the threshold + observe(filter, "http://example.com/a?sort=x&page=1", "http://example.com/a"); + observe(filter, "http://example.com/b?sort=x&page=1", "http://example.com/b"); + observe(filter, "http://example.com/c?sort=x&page=1", "http://example.com/c?page=1"); + observe(filter, "http://example.com/d?sort=x&page=1", "http://example.com/d?sort=x&page=1"); + + assertEquals( + "http://example.com/e?page=3", apply(filter, "http://example.com/e?sort=x&page=3")); + } + + @Test + void testEvidenceIsScopedToTheHost() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://another.com/other?id=9&sid=zzz", + apply(filter, "http://another.com/other?id=9&sid=zzz")); + assertEquals( + "http://sub.example.com/other?id=9&sid=zzz", + apply(filter, "http://sub.example.com/other?id=9&sid=zzz")); + } + + @Test + void testEvidenceCanBeScopedToTheDomain() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("scope", "domain"); + AdaptiveURLNormalizer filter = createFilter(filterParams); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://www.example.com/page" + i + "?id=" + i + "&sid=abc" + i, + "http://www.example.com/page" + i + "?id=" + i); + } + assertEquals( + "http://shop.example.com/other?id=9", + apply(filter, "http://shop.example.com/other?id=9&sid=zzz")); + assertEquals( + "http://another.com/other?id=9&sid=zzz", + apply(filter, "http://another.com/other?id=9&sid=zzz")); + } + + @Test + void testHostIsCaseInsensitive() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + observeSessionParam(filter, 5); + assertEquals( + "http://EXAMPLE.com/other?id=9", + apply(filter, "http://EXAMPLE.com/other?id=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 + "?id=" + i + "&sid=abc" + i, + "http://example.com/canonical" + i); + } + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=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 + "?id=" + i + "&sid=abc" + i, + "http://mirror.example.com/page" + i + "?id=" + i); + } + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=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 + "?id=" + i + "&sid=abc" + i, + "/page" + i + "?id=" + i); + } + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=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 + "?id=" + i + "&sid=abc" + i, + "?id=" + i); + } + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=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 + "?id=" + i + "&sid=abc" + i, ":::"); + } + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=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?id=a%20b#top", + apply(filter, "http://example.com/other?sid=zzz&id=a%20b#top")); + } + + @Test + void testNonDefaultPortIsPreserved() throws MalformedURLException { + AdaptiveURLNormalizer filter = createFilter(); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com:8080/page" + i + "?id=" + i + "&sid=abc" + i, + "http://example.com:8080/page" + i + "?id=" + i); + } + assertEquals( + "http://example.com:8080/other?id=9", + apply(filter, "http://example.com:8080/other?id=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 + "?id=" + i + "&sid=abc" + i, + "http://example.com:8080/page" + i + "?id=" + i); + } + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=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 + "?id=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?id=" + i); + } + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=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?id=1&sid=abc"; + String canonical = "http://example.com/page?id=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?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=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"; + assertEquals(malformed, apply(filter, malformed)); + } + + @Test + void testCustomCanonicalMetadataKey() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("canonicalMetadataKey", "parse.canonical"); + AdaptiveURLNormalizer filter = createFilter(filterParams); + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, + "parse.canonical", + "http://example.com/page" + i + "?id=" + i); + } + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testNumberOfTrackedParametersIsBounded() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("maxParams", 1); + AdaptiveURLNormalizer filter = createFilter(filterParams); + + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/first" + i + "?sid=abc" + i, + "http://example.com/first" + i); + } + for (int i = 0; i < 5; i++) { + observe( + filter, + "http://example.com/second" + i + "?ref=abc" + i, + "http://example.com/second" + i); + } + + // only the first parameter seen is tracked + assertEquals( + "http://example.com/other?ref=zzz", + apply(filter, "http://example.com/other?sid=1&ref=zzz")); + } + + @Test + void testNumberOfTrackedSitesIsBounded() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("maxScopes", 1); + AdaptiveURLNormalizer filter = createFilter(filterParams); + + // which site gets evicted once the limit is reached is left to the cache, + // the rules of the only one tracked here must still be applied + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } + + @Test + void testInvalidConfigurationValuesFallBackToDefaults() throws MalformedURLException { + ObjectNode filterParams = params(); + filterParams.put("minObservations", 0); + filterParams.put("confidenceThreshold", 1.5d); + filterParams.put("maxScopes", 0); + filterParams.put("maxParams", -1); + filterParams.put("scope", "unknown"); + filterParams.put("canonicalMetadataKey", " "); + AdaptiveURLNormalizer filter = createFilter(filterParams); + + observeSessionParam(filter, 4); + assertEquals( + "http://example.com/other?id=9&sid=zzz", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + + observeSessionParam(filter, 5); + assertEquals( + "http://example.com/other?id=9", + apply(filter, "http://example.com/other?id=9&sid=zzz")); + } +} diff --git a/docs/src/main/asciidoc/internals.adoc b/docs/src/main/asciidoc/internals.adoc index ab5cddeb5..5ceef8ae4 100644 --- a/docs/src/main/asciidoc/internals.adoc +++ b/docs/src/main/asciidoc/internals.adoc @@ -249,6 +249,38 @@ The JSON configuration allows loading several instances of the same filtering cl ===== Built-in URL Filters +====== Adaptive +The link:https://github.com/apache/stormcrawler/blob/main/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java[AdaptiveURLNormalizer] learns which query parameters can be removed from the URLs of a site by comparing the URL of each page with the value of its canonical tag. + +When both 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 requires the canonical tag to have been extracted into the metadata, for instance with an XPathFilter configured with `"canonical": "//*[@rel=\"canonical\"]/@href"`, and must therefore be placed in a parsing bolt. The evidence is kept in memory only: it is lost when the topology is restarted and is not shared between the instances of the bolt. + +[source,json] +---- +{ + "class": "org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer", + "name": "AdaptiveURLNormalizer", + "params": { + "canonicalMetadataKey": "canonical", + "scope": "host", + "minObservations": 5, + "confidenceThreshold": 0.9, + "maxScopes": 10000, + "maxParams": 100 + } +} +---- + +All the parameters are optional: + +* `canonicalMetadataKey` - metadata key holding the value of the canonical tag, _canonical_ by default. +* `scope` - whether the rules are learnt per _host_ (default) or per _domain_. +* `minObservations` - number of pages a parameter must have been seen on before a rule can be applied to it, 5 by default. +* `confidenceThreshold` - proportion of the canonical tags which must have dropped the parameter, 0.9 by default. +* `maxScopes` - maximum number of hosts or domains tracked, 10000 by default. +* `maxParams` - maximum number of parameters tracked per host or domain, 100 by default. + ====== 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. From 9930c9927e43928fadb768b66ff9f273d6834154 Mon Sep 17 00:00:00 2001 From: Akash Manna Date: Tue, 18 Aug 2026 12:13:27 +0530 Subject: [PATCH 2/2] Add Adaptive URL Normalization feature - Introduced AdaptiveURLNormalizer and CanonicalParamLearner classes to learn and remove irrelevant query parameters based on canonical tags. - Updated documentation to include configuration options for adaptive URL normalization. - Implemented CanonicalRules to manage evidence gathered from canonical tags regarding query parameters. - Added tests for CanonicalParamLearner to ensure correct functionality and integration with AdaptiveURLNormalizer. --- .../adaptive/AdaptiveURLNormalizer.java | 366 ++------------ .../filtering/adaptive/CanonicalRules.java | 406 +++++++++++++++ .../parse/filter/CanonicalParamLearner.java | 76 +++ core/src/main/resources/crawler-default.yaml | 16 + .../filtering/AdaptiveURLNormalizerTest.java | 473 ++++++++++++------ .../filter/CanonicalParamLearnerTest.java | 145 ++++++ docs/src/main/asciidoc/configuration.adoc | 22 + docs/src/main/asciidoc/internals.adoc | 56 ++- 8 files changed, 1080 insertions(+), 480 deletions(-) create mode 100644 core/src/main/java/org/apache/stormcrawler/filtering/adaptive/CanonicalRules.java create mode 100644 core/src/main/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearner.java create mode 100644 core/src/test/java/org/apache/stormcrawler/parse/filter/CanonicalParamLearnerTest.java 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 index 216930351..47147848a 100644 --- a/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java +++ b/core/src/main/java/org/apache/stormcrawler/filtering/adaptive/AdaptiveURLNormalizer.java @@ -18,173 +18,44 @@ package org.apache.stormcrawler.filtering.adaptive; import com.fasterxml.jackson.databind.JsonNode; -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.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Locale; import java.util.Map; -import java.util.Set; -import org.apache.commons.lang3.StringUtils; 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; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** - * Normalizes URLs by removing the query parameters which a site has shown to be irrelevant, based - * on the canonical tags found in its pages. + * 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. * - *

Whenever a page is parsed, this filter compares the URL of the page with the value of its - * canonical tag (as extracted into the metadata, see canonicalMetadataKey). When both - * point at the same resource - 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. + *

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

Once enough evidence has been gathered for a given parameter, subsequent URLs for that site - * get the parameter removed, 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. - * - *

The evidence is kept in memory only and is therefore lost when the topology is restarted; it - * is not shared between the instances of the bolt either. Both the number of sites and the number - * of parameters tracked per site are bounded, see maxScopes and maxParams - * . - * - *

Configuration, all parameters are optional: - * - * 

{@code
- * {
- *   "class": "org.apache.stormcrawler.filtering.adaptive.AdaptiveURLNormalizer",
- *   "name": "AdaptiveURLNormalizer",
- *   "params": {
- *     "canonicalMetadataKey": "canonical",
- *     "scope": "host",
- *     "minObservations": 5,
- *     "confidenceThreshold": 0.9,
- *     "maxScopes": 10000,
- *     "maxParams": 100
- *   }
- * }
- * }
+ *

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 { - private static final Logger LOG = LoggerFactory.getLogger(AdaptiveURLNormalizer.class); - - /** Metadata key under which the canonical tag is stored by default. */ - private static final String DEFAULT_CANONICAL_KEY = "canonical"; - - private static final int DEFAULT_MIN_OBSERVATIONS = 5; - - private static final double DEFAULT_CONFIDENCE_THRESHOLD = 0.9d; - - private static final int DEFAULT_MAX_SCOPES = 10_000; - - private static final int DEFAULT_MAX_PARAMS = 100; - - private String canonicalMetadataKey = DEFAULT_CANONICAL_KEY; - - private int minObservations = DEFAULT_MIN_OBSERVATIONS; - - private double confidenceThreshold = DEFAULT_CONFIDENCE_THRESHOLD; - - private int maxParams = DEFAULT_MAX_PARAMS; - - /** Whether the rules are learnt per domain instead of per host. */ - private boolean scopeByDomain = false; + static final String DEFAULT_STORE = "default"; - /** Evidence gathered so far, keyed by host or domain. */ - private Cache> stats = buildCache(DEFAULT_MAX_SCOPES); - - /** - * URL of the page the last observation was made from. The filter is called once per outlink, - * i.e. many times in a row with the same source, but a page must only count once. - */ - private String lastLearnedSource; - - private static Cache> buildCache(int maxScopes) { - return Caffeine.newBuilder().maximumSize(maxScopes).build(); - } + private CanonicalRules rules; @Override public void configure(@NotNull Map stormConf, @NotNull JsonNode paramNode) { - int maxScopes = DEFAULT_MAX_SCOPES; - - JsonNode node = paramNode.get("canonicalMetadataKey"); - if (node != null) { - final String key = node.asText(); - if (StringUtils.isBlank(key)) { - LOG.warn("Ignoring blank value for canonicalMetadataKey"); - } else { - canonicalMetadataKey = key; - } - } - - node = paramNode.get("scope"); - if (node != null) { - final String scope = node.asText(); - if ("domain".equalsIgnoreCase(scope)) { - scopeByDomain = true; - } else if ("host".equalsIgnoreCase(scope)) { - scopeByDomain = false; - } else { - LOG.warn("Unknown value for scope: {}, using host", scope); - } - } - - node = paramNode.get("minObservations"); - if (node != null) { - final int value = node.asInt(DEFAULT_MIN_OBSERVATIONS); - if (value < 1) { - LOG.warn("Ignoring invalid value for minObservations: {}", value); - } else { - minObservations = value; - } - } - - node = paramNode.get("confidenceThreshold"); - if (node != null) { - final double value = node.asDouble(DEFAULT_CONFIDENCE_THRESHOLD); - if (value <= 0d || value > 1d) { - LOG.warn("Ignoring invalid value for confidenceThreshold: {}", value); - } else { - confidenceThreshold = value; - } - } - - node = paramNode.get("maxScopes"); - if (node != null) { - final int value = node.asInt(DEFAULT_MAX_SCOPES); - if (value < 1) { - LOG.warn("Ignoring invalid value for maxScopes: {}", value); - } else { - maxScopes = value; - } - } - - node = paramNode.get("maxParams"); - if (node != null) { - final int value = node.asInt(DEFAULT_MAX_PARAMS); - if (value < 1) { - LOG.warn("Ignoring invalid value for maxParams: {}", value); - } else { - maxParams = value; - } - } - - stats = buildCache(maxScopes); + final JsonNode node = paramNode.get("store"); + final String store = node == null ? DEFAULT_STORE : node.asText(DEFAULT_STORE); + rules = CanonicalRules.getInstance(stormConf, store); } @Override @@ -192,91 +63,31 @@ public void configure(@NotNull Map stormConf, @NotNull JsonNode @Nullable URL sourceUrl, @Nullable Metadata sourceMetadata, @NotNull String urlToFilter) { - learn(sourceUrl, sourceMetadata); + // 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); } /** - * Compares the URL of the page being parsed with its canonical tag and records which of its - * query parameters the site considers irrelevant. + * 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 void learn(@Nullable URL sourceUrl, @Nullable Metadata sourceMetadata) { - if (sourceUrl == null || sourceMetadata == null) { - return; - } - - final String canonicalValue = sourceMetadata.getFirstValue(canonicalMetadataKey); - if (StringUtils.isBlank(canonicalValue)) { - return; - } - - // the filter is called once per outlink: a page must only be counted once - final String sourceForm = sourceUrl.toExternalForm(); - if (sourceForm.equals(lastLearnedSource)) { - return; - } - lastLearnedSource = sourceForm; - - // nothing to learn from a URL without a query string - final Set sourceParams = parameterNames(sourceUrl.getQuery()); - if (sourceParams.isEmpty()) { - 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; + 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 Set canonicalParams = parameterNames(canonical.getQuery()); - final Map scopeStats = stats.get(scopeKey, k -> new HashMap<>()); - - for (String param : sourceParams) { - ParamStats paramStats = scopeStats.get(param); - if (paramStats == null) { - if (scopeStats.size() >= maxParams) { - LOG.debug( - "Not tracking parameter {} for {}: limit of {} reached", - param, - scopeKey, - maxParams); - continue; - } - paramStats = new ParamStats(); - scopeStats.put(param, paramStats); - } - final boolean wasRemovable = isRemovable(paramStats); - if (canonicalParams.contains(param)) { - paramStats.kept++; - } else { - paramStats.dropped++; - } - if (!wasRemovable && isRemovable(paramStats)) { - LOG.info( - "Removing param {} from the URLs of {}: dropped by {} of {} canonicals", - param, - scopeKey, - paramStats.dropped, - paramStats.total()); - } + final int queryEnd = fragment == -1 ? urlToFilter.length() : fragment; + final String query = urlToFilter.substring(questionMark + 1, queryEnd); + if (query.isEmpty()) { + return urlToFilter; } - } - /** Removes from the URL the parameters which have been found to be irrelevant for its site. */ - private String removeIrrelevantParams(@NotNull String urlToFilter) { final URL url; try { url = URLUtil.toURL(urlToFilter); @@ -285,27 +96,15 @@ private String removeIrrelevantParams(@NotNull String urlToFilter) { return urlToFilter; } - final String query = url.getQuery(); - if (StringUtils.isEmpty(query)) { - return urlToFilter; - } - - final String scopeKey = scopeKey(url); + final String scopeKey = rules.scopeKey(url); if (scopeKey == null) { return urlToFilter; } - final Map scopeStats = stats.getIfPresent(scopeKey); - if (scopeStats == null) { - return urlToFilter; - } - final StringBuilder newQuery = new StringBuilder(query.length()); boolean removedSomething = false; - // the parameters are kept verbatim so that their encoding is left untouched for (String param : query.split("&", -1)) { - final ParamStats paramStats = scopeStats.get(parameterName(param)); - if (paramStats != null && isRemovable(paramStats)) { + if (rules.isRemovable(scopeKey, CanonicalRules.parameterName(param))) { removedSomething = true; continue; } @@ -320,97 +119,16 @@ private String removeIrrelevantParams(@NotNull String urlToFilter) { } final StringBuilder normalized = new StringBuilder(urlToFilter.length()); - normalized.append(url.getProtocol()).append(':'); - final String authority = url.getAuthority(); - if (StringUtils.isNotEmpty(authority)) { - normalized.append("//").append(authority); + 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('/'); } - normalized.append(url.getPath()); if (newQuery.length() > 0) { normalized.append('?').append(newQuery); } - final String ref = url.getRef(); - if (ref != null) { - normalized.append('#').append(ref); - } + normalized.append(urlToFilter, queryEnd, urlToFilter.length()); return normalized.toString(); } - - private boolean isRemovable(ParamStats paramStats) { - final int total = paramStats.total(); - return total >= minObservations - && (double) paramStats.dropped / total >= confidenceThreshold; - } - - /** Key under which the evidence is gathered, i.e. the host or the domain of the URL. */ - private @Nullable String scopeKey(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; - } - - /** 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 source.getPath().equals(canonical.getPath()); - } - - private static int port(URL url) { - return url.getPort() == -1 ? url.getDefaultPort() : url.getPort(); - } - - /** Names of the parameters found in a query string, in their decoded form. */ - private 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. */ - private 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; - } - } - - /** Number of times a given parameter was dropped or kept by a canonical tag. */ - private static final class ParamStats { - - private int dropped; - - private int kept; - - private int total() { - return dropped + kept; - } - } } 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 index 6b7c2a495..04813e313 100644 --- a/core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java +++ b/core/src/test/java/org/apache/stormcrawler/filtering/AdaptiveURLNormalizerTest.java @@ -18,44 +18,59 @@ 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; /** - * Tests the learning of the query parameters which can be removed from the URLs of a site, based on - * the canonical tags found in its pages. + * 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 String CANONICAL = "canonical"; + private static final AtomicInteger STORE_COUNTER = new AtomicInteger(); - private AdaptiveURLNormalizer createFilter() { - return createFilter(new ObjectNode(JsonNodeFactory.instance)); - } + /** Rules are shared per JVM and per name, so every test needs a store of its own. */ + private CanonicalRules rules; - private AdaptiveURLNormalizer createFilter(ObjectNode filterParams) { + private AdaptiveURLNormalizer createFilter(Map conf) { AdaptiveURLNormalizer filter = new AdaptiveURLNormalizer(); - Map conf = new HashMap<>(); + 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 static ObjectNode params() { - return new ObjectNode(JsonNodeFactory.instance); + 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); + return observe(filter, pageUrl, "canonical", canonicalValue); } private String observe( @@ -73,16 +88,14 @@ private String apply(AdaptiveURLNormalizer filter, String url) { return filter.filter(null, null, url); } - /** - * Observes pages whose canonical drops the sid parameter but keeps the id one. - */ + /** 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 + "?id=" + i + "&sid=abc" + i, - "http://example.com/page" + i + "?id=" + i); + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i); } } @@ -90,19 +103,19 @@ private void observeSessionParam(AdaptiveURLNormalizer filter, int observations) void testUnchangedWithoutEvidence() { AdaptiveURLNormalizer filter = createFilter(); assertEquals( - "http://example.com/page?id=1&sid=abc", - apply(filter, "http://example.com/page?id=1&sid=abc")); + "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 + "?id=" + i + "&sid=abc" + i, null); + observe(filter, "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, null); } assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -110,8 +123,8 @@ void testRemovesIrrelevantParameter() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 5); assertEquals( - "http://example.com/other?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -119,19 +132,20 @@ void testNotAppliedBelowMinObservations() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 4); assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test void testMinObservationsIsConfigurable() throws MalformedURLException { - ObjectNode filterParams = params(); - filterParams.put("minObservations", 2); - AdaptiveURLNormalizer filter = createFilter(filterParams); + 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?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -139,37 +153,38 @@ void testParameterKeptByCanonicalIsNeverRemoved() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 20); assertEquals( - "http://example.com/other?id=9", apply(filter, "http://example.com/other?id=9")); + "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 + "?id=" + i + "&sid=abc" + i; + String page = "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i; observe(filter, page, page); } assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test void testConfidenceThreshold() throws MalformedURLException { - ObjectNode filterParams = params(); - filterParams.put("minObservations", 4); - filterParams.put("confidenceThreshold", 0.75d); - AdaptiveURLNormalizer filter = createFilter(filterParams); - - // sort: dropped 3 times out of 4 -> 0.75, at the threshold - // page: dropped 2 times out of 4 -> 0.5, below the threshold - observe(filter, "http://example.com/a?sort=x&page=1", "http://example.com/a"); - observe(filter, "http://example.com/b?sort=x&page=1", "http://example.com/b"); - observe(filter, "http://example.com/c?sort=x&page=1", "http://example.com/c?page=1"); - observe(filter, "http://example.com/d?sort=x&page=1", "http://example.com/d?sort=x&page=1"); + 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?page=3", apply(filter, "http://example.com/e?sort=x&page=3")); + "http://example.com/e?ref=3", apply(filter, "http://example.com/e?sid=x&ref=3")); } @Test @@ -177,30 +192,30 @@ void testEvidenceIsScopedToTheHost() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 5); assertEquals( - "http://another.com/other?id=9&sid=zzz", - apply(filter, "http://another.com/other?id=9&sid=zzz")); + "http://another.com/other?pid=9&sid=zzz", + apply(filter, "http://another.com/other?pid=9&sid=zzz")); assertEquals( - "http://sub.example.com/other?id=9&sid=zzz", - apply(filter, "http://sub.example.com/other?id=9&sid=zzz")); + "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 { - ObjectNode filterParams = params(); - filterParams.put("scope", "domain"); - AdaptiveURLNormalizer filter = createFilter(filterParams); + 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 + "?id=" + i + "&sid=abc" + i, - "http://www.example.com/page" + i + "?id=" + i); + "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?id=9", - apply(filter, "http://shop.example.com/other?id=9&sid=zzz")); + "http://shop.example.com/other?pid=9", + apply(filter, "http://shop.example.com/other?pid=9&sid=zzz")); assertEquals( - "http://another.com/other?id=9&sid=zzz", - apply(filter, "http://another.com/other?id=9&sid=zzz")); + "http://another.com/other?pid=9&sid=zzz", + apply(filter, "http://another.com/other?pid=9&sid=zzz")); } @Test @@ -208,8 +223,8 @@ void testHostIsCaseInsensitive() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 5); assertEquals( - "http://EXAMPLE.com/other?id=9", - apply(filter, "http://EXAMPLE.com/other?id=9&sid=zzz")); + "http://EXAMPLE.com/other?pid=9", + apply(filter, "http://EXAMPLE.com/other?pid=9&sid=zzz")); } @Test @@ -218,12 +233,12 @@ void testCanonicalOnAnotherPathIsIgnored() throws MalformedURLException { for (int i = 0; i < 10; i++) { observe( filter, - "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, "http://example.com/canonical" + i); } assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -232,12 +247,12 @@ void testCanonicalOnAnotherHostIsIgnored() throws MalformedURLException { for (int i = 0; i < 10; i++) { observe( filter, - "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, - "http://mirror.example.com/page" + i + "?id=" + i); + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://mirror.example.com/page" + i + "?pid=" + i); } assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -246,12 +261,12 @@ void testRelativeCanonicalIsResolved() throws MalformedURLException { for (int i = 0; i < 5; i++) { observe( filter, - "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, - "/page" + i + "?id=" + i); + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "/page" + i + "?pid=" + i); } assertEquals( - "http://example.com/other?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -260,23 +275,23 @@ void testPureQueryCanonicalIsResolved() throws MalformedURLException { for (int i = 0; i < 5; i++) { observe( filter, - "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, - "?id=" + i); + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "?pid=" + i); } assertEquals( - "http://example.com/other?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "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 + "?id=" + i + "&sid=abc" + i, ":::"); + observe(filter, "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, ":::"); } assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -296,8 +311,34 @@ void testFragmentAndEncodingArePreserved() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 5); assertEquals( - "http://example.com/other?id=a%20b#top", - apply(filter, "http://example.com/other?sid=zzz&id=a%20b#top")); + "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 @@ -306,12 +347,12 @@ void testNonDefaultPortIsPreserved() throws MalformedURLException { for (int i = 0; i < 5; i++) { observe( filter, - "http://example.com:8080/page" + i + "?id=" + i + "&sid=abc" + i, - "http://example.com:8080/page" + i + "?id=" + i); + "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?id=9", - apply(filter, "http://example.com:8080/other?id=9&sid=zzz")); + "http://example.com:8080/other?pid=9", + apply(filter, "http://example.com:8080/other?pid=9&sid=zzz")); } @Test @@ -320,12 +361,12 @@ void testCanonicalOnAnotherPortIsIgnored() throws MalformedURLException { for (int i = 0; i < 10; i++) { observe( filter, - "http://example.com/page" + i + "?id=" + i + "&sid=abc" + i, - "http://example.com:8080/page" + i + "?id=" + i); + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com:8080/page" + i + "?pid=" + i); } assertEquals( - "http://example.com/other?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9&sid=zzz", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -334,12 +375,12 @@ void testDefaultPortMatchesImplicitOne() throws MalformedURLException { for (int i = 0; i < 5; i++) { observe( filter, - "http://example.com:80/page" + i + "?id=" + i + "&sid=abc" + i, - "http://example.com/page" + i + "?id=" + i); + "http://example.com:80/page" + i + "?pid=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i); } assertEquals( - "http://example.com/other?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test @@ -359,15 +400,123 @@ void testEncodedParameterNames() throws MalformedURLException { @Test void testAPageIsOnlyCountedOnce() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); - String page = "http://example.com/page?id=1&sid=abc"; - String canonical = "http://example.com/page?id=1"; + 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?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "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 @@ -388,85 +537,129 @@ void testNullSourceIsHandled() throws MalformedURLException { void testMalformedURLIsLeftUntouched() throws MalformedURLException { AdaptiveURLNormalizer filter = createFilter(); observeSessionParam(filter, 5); - String malformed = "this is not a URL"; + 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 { - ObjectNode filterParams = params(); - filterParams.put("canonicalMetadataKey", "parse.canonical"); - AdaptiveURLNormalizer filter = createFilter(filterParams); + 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 + "?id=" + i + "&sid=abc" + i, + "http://example.com/page" + i + "?pid=" + i + "&sid=abc" + i, "parse.canonical", - "http://example.com/page" + i + "?id=" + i); + "http://example.com/page" + i + "?pid=" + i); } assertEquals( - "http://example.com/other?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "http://example.com/other?pid=9", + apply(filter, "http://example.com/other?pid=9&sid=zzz")); } @Test - void testNumberOfTrackedParametersIsBounded() throws MalformedURLException { - ObjectNode filterParams = params(); - filterParams.put("maxParams", 1); - AdaptiveURLNormalizer filter = createFilter(filterParams); + void testWeakestParameterIsDiscardedWhenFull() throws MalformedURLException { + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MAX_PARAMS_PARAM, 2); + AdaptiveURLNormalizer filter = createFilter(conf); - for (int i = 0; i < 5; i++) { + // 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/first" + i + "?sid=abc" + i, - "http://example.com/first" + i); + "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/second" + i + "?ref=abc" + i, - "http://example.com/second" + i); + "http://example.com/page" + i + "?sid=abc" + i, + "http://example.com/page" + i); } - - // only the first parameter seen is tracked - assertEquals( - "http://example.com/other?ref=zzz", - apply(filter, "http://example.com/other?sid=1&ref=zzz")); + assertEquals("http://example.com/other", apply(filter, "http://example.com/other?sid=zzz")); } @Test void testNumberOfTrackedSitesIsBounded() throws MalformedURLException { - ObjectNode filterParams = params(); - filterParams.put("maxScopes", 1); - AdaptiveURLNormalizer filter = createFilter(filterParams); + Map conf = new HashMap<>(); + conf.put(CanonicalRules.MAX_SCOPES_PARAM, 1); + AdaptiveURLNormalizer filter = createFilter(conf); - // which site gets evicted once the limit is reached is left to the cache, - // the rules of the only one tracked here must still be applied observeSessionParam(filter, 5); - assertEquals( - "http://example.com/other?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + 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 testInvalidConfigurationValuesFallBackToDefaults() throws MalformedURLException { - ObjectNode filterParams = params(); - filterParams.put("minObservations", 0); - filterParams.put("confidenceThreshold", 1.5d); - filterParams.put("maxScopes", 0); - filterParams.put("maxParams", -1); - filterParams.put("scope", "unknown"); - filterParams.put("canonicalMetadataKey", " "); - AdaptiveURLNormalizer filter = createFilter(filterParams); + 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?id=9&sid=zzz", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "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?id=9", - apply(filter, "http://example.com/other?id=9&sid=zzz")); + "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 5ceef8ae4..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,37 +250,60 @@ 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] learns which query parameters can be removed from the URLs of a site by comparing the URL of each page with the value of its canonical tag. +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 both 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. +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 requires the canonical tag to have been extracted into the metadata, for instance with an XPathFilter configured with `"canonical": "//*[@rel=\"canonical\"]/@href"`, and must therefore be placed in a parsing bolt. The evidence is kept in memory only: it is lost when the topology is restarted and is not shared between the instances of the bolt. +*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": { - "canonicalMetadataKey": "canonical", - "scope": "host", - "minObservations": 5, - "confidenceThreshold": 0.9, - "maxScopes": 10000, - "maxParams": 100 + "store": "default" } } ---- -All the parameters are optional: +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: -* `canonicalMetadataKey` - metadata key holding the value of the canonical tag, _canonical_ by default. -* `scope` - whether the rules are learnt per _host_ (default) or per _domain_. -* `minObservations` - number of pages a parameter must have been seen on before a rule can be applied to it, 5 by default. -* `confidenceThreshold` - proportion of the canonical tags which must have dropped the parameter, 0.9 by default. -* `maxScopes` - maximum number of hosts or domains tracked, 10000 by default. -* `maxParams` - maximum number of parameters tracked per host or domain, 100 by default. +* 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.