From 8b1fe690eb96557c6c7d870eb85b645fc8389764 Mon Sep 17 00:00:00 2001 From: lewismc Date: Thu, 11 Dec 2025 18:07:50 -0800 Subject: [PATCH] NUTCH-3134 Add latency metrics with percentile support to Fetcher, Parser, and Indexer --- .../apache/nutch/fetcher/FetcherThread.java | 13 ++ .../nutch/indexer/IndexerMapReduce.java | 21 +++ .../apache/nutch/metrics/LatencyTracker.java | 144 ++++++++++++++++++ .../apache/nutch/metrics/NutchMetrics.java | 22 +++ .../org/apache/nutch/parse/ParseSegment.java | 15 +- 5 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 src/java/org/apache/nutch/metrics/LatencyTracker.java diff --git a/src/java/org/apache/nutch/fetcher/FetcherThread.java b/src/java/org/apache/nutch/fetcher/FetcherThread.java index 8b4e5c95c3..02b7cd3e86 100644 --- a/src/java/org/apache/nutch/fetcher/FetcherThread.java +++ b/src/java/org/apache/nutch/fetcher/FetcherThread.java @@ -38,6 +38,7 @@ import org.apache.nutch.crawl.SignatureFactory; import org.apache.nutch.fetcher.Fetcher.FetcherRun; import org.apache.nutch.fetcher.FetcherThreadEvent.PublishEventType; +import org.apache.nutch.metrics.LatencyTracker; import org.apache.nutch.metrics.NutchMetrics; import org.apache.nutch.metadata.Metadata; import org.apache.nutch.metadata.Nutch; @@ -166,6 +167,9 @@ public class FetcherThread extends Thread { private Counter outlinksDetectedCounter; private Counter outlinksFollowingCounter; + // Latency tracker for fetch timing metrics + private LatencyTracker fetchLatencyTracker; + public FetcherThread(Configuration conf, AtomicInteger activeThreads, FetchItemQueues fetchQueues, QueueFeeder feeder, AtomicInteger spinWaiting, AtomicLong lastRequestStart, FetcherRun.Context context, AtomicInteger errors, String segmentName, boolean parsing, boolean storingContent, @@ -284,6 +288,10 @@ private void initCounters() { NutchMetrics.GROUP_FETCHER_OUTLINKS, NutchMetrics.FETCHER_OUTLINKS_DETECTED_TOTAL); outlinksFollowingCounter = context.getCounter( NutchMetrics.GROUP_FETCHER_OUTLINKS, NutchMetrics.FETCHER_OUTLINKS_FOLLOWING_TOTAL); + + // Initialize latency tracker for fetch timing + fetchLatencyTracker = new LatencyTracker( + NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_LATENCY); } @Override @@ -417,8 +425,11 @@ public void run() { fit.queueID, fiq.crawlDelay, fit.url); } } + // Track fetch latency + long fetchStart = System.currentTimeMillis(); ProtocolOutput output = protocol.getProtocolOutput(fit.url, fit.datum); + fetchLatencyTracker.record(System.currentTimeMillis() - fetchStart); ProtocolStatus status = output.getStatus(); Content content = output.getContent(); ParseStatus pstatus = null; @@ -557,6 +568,8 @@ public void run() { if (fit != null) { fetchQueues.finishFetchItem(fit); } + // Emit fetch latency metrics + fetchLatencyTracker.emitCounters(context); activeThreads.decrementAndGet(); // count threads LOG.info("{} {} -finishing thread {}, activeThreads={}", getName(), Thread.currentThread().getId(), getName(), activeThreads); diff --git a/src/java/org/apache/nutch/indexer/IndexerMapReduce.java b/src/java/org/apache/nutch/indexer/IndexerMapReduce.java index 33f2f244a6..9086a19839 100644 --- a/src/java/org/apache/nutch/indexer/IndexerMapReduce.java +++ b/src/java/org/apache/nutch/indexer/IndexerMapReduce.java @@ -40,6 +40,7 @@ import org.apache.nutch.crawl.Inlinks; import org.apache.nutch.crawl.LinkDb; import org.apache.nutch.crawl.NutchWritable; +import org.apache.nutch.metrics.LatencyTracker; import org.apache.nutch.metrics.NutchMetrics; import org.apache.nutch.metadata.Metadata; import org.apache.nutch.metadata.Nutch; @@ -215,6 +216,9 @@ public static class IndexerReducer extends private URLNormalizers urlNormalizers; private URLFilters urlFilters; + // Latency tracker for indexing timing metrics + private LatencyTracker indexLatencyTracker; + @Override public void setup(Reducer.Context context) { Configuration conf = context.getConfiguration(); @@ -239,6 +243,17 @@ public void setup(Reducer.Context c if (filter) { urlFilters = new URLFilters(conf); } + + // Initialize latency tracker for indexing timing + indexLatencyTracker = new LatencyTracker( + NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_LATENCY); + } + + @Override + public void cleanup(Reducer.Context context) + throws IOException, InterruptedException { + // Emit indexing latency metrics + indexLatencyTracker.emitCounters(context); } @Override @@ -343,6 +358,9 @@ public void reduce(Text key, Iterable values, return; } + // Start timing document indexing + long indexStart = System.currentTimeMillis(); + NutchDocument doc = new NutchDocument(); doc.add("id", key.toString()); @@ -432,6 +450,9 @@ public void reduce(Text key, Iterable values, doc.add("binaryContent", binary); } + // Record indexing latency + indexLatencyTracker.record(System.currentTimeMillis() - indexStart); + context.getCounter(NutchMetrics.GROUP_INDEXER, NutchMetrics.INDEXER_INDEXED_TOTAL).increment(1); diff --git a/src/java/org/apache/nutch/metrics/LatencyTracker.java b/src/java/org/apache/nutch/metrics/LatencyTracker.java new file mode 100644 index 0000000000..3777bb29e3 --- /dev/null +++ b/src/java/org/apache/nutch/metrics/LatencyTracker.java @@ -0,0 +1,144 @@ +/* + * 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.nutch.metrics; + +import org.apache.hadoop.mapreduce.TaskInputOutputContext; + +import com.tdunning.math.stats.TDigest; + +/** + * A utility class for tracking latency metrics using TDigest for percentile + * calculation. + * + *

This class wraps a TDigest data structure to collect latency samples and + * emit Hadoop counters with count, sum, and percentile values (p50, p95, p99). + * + *

Usage: + * 

+ * // In mapper/reducer setup
+ * latencyTracker = new LatencyTracker(NutchMetrics.GROUP_FETCHER, NutchMetrics.FETCHER_LATENCY);
+ * 
+ * // During processing
+ * long start = System.currentTimeMillis();
+ * // ... operation ...
+ * latencyTracker.record(System.currentTimeMillis() - start);
+ * 
+ * // In cleanup
+ * latencyTracker.emitCounters(context);
+ *
+ * + *

Emits the following counters: + *

    + *
  • {prefix}_count_total - total number of samples
    • + *
  • {prefix}_sum_ms - sum of all latencies in milliseconds
    • + *
  • {prefix}_p50_ms - 50th percentile (median) latency
    • + *
  • {prefix}_p95_ms - 95th percentile latency
    • + *
  • {prefix}_p99_ms - 99th percentile latency
    • + *
+ * + * @since 1.22 + */ +public class LatencyTracker { + + /** Default compression factor for TDigest (controls accuracy vs memory). */ + private static final double DEFAULT_COMPRESSION = 100.0; + + private final TDigest digest; + private final String group; + private final String prefix; + private long count = 0; + private long sum = 0; + + /** + * Creates a new LatencyTracker. + * + * @param group the Hadoop counter group name + * @param prefix the prefix for counter names (e.g., "fetch_latency") + */ + public LatencyTracker(String group, String prefix) { + this.digest = TDigest.createDigest(DEFAULT_COMPRESSION); + this.group = group; + this.prefix = prefix; + } + + /** + * Records a latency sample. + * + * @param latencyMs the latency in milliseconds + */ + public void record(long latencyMs) { + digest.add(latencyMs); + count++; + sum += latencyMs; + } + + /** + * Returns the number of recorded samples. + * + * @return the count of recorded latency samples + */ + public long getCount() { + return count; + } + + /** + * Returns the sum of all recorded latencies. + * + * @return the sum of latencies in milliseconds + */ + public long getSum() { + return sum; + } + + /** + * Returns the percentile value for the given quantile. + * + * @param quantile the quantile (0.0 to 1.0) + * @return the percentile value in milliseconds + */ + public long getPercentile(double quantile) { + if (count == 0) { + return 0; + } + return (long) digest.quantile(quantile); + } + + /** + * Emits all latency counters to the Hadoop context. + * + *

Should be called once during cleanup to emit aggregated metrics. + * + * @param context the Hadoop task context + */ + public void emitCounters(TaskInputOutputContext context) { + context.getCounter(group, prefix + "_count_total").setValue(count); + context.getCounter(group, prefix + "_sum_ms").setValue(sum); + + if (count > 0) { + context.getCounter(group, prefix + "_p50_ms").setValue((long) digest.quantile(0.50)); + context.getCounter(group, prefix + "_p95_ms").setValue((long) digest.quantile(0.95)); + context.getCounter(group, prefix + "_p99_ms").setValue((long) digest.quantile(0.99)); + } else { + // Set to 0 if no samples recorded + context.getCounter(group, prefix + "_p50_ms").setValue(0); + context.getCounter(group, prefix + "_p95_ms").setValue(0); + context.getCounter(group, prefix + "_p99_ms").setValue(0); + } + } +} + + diff --git a/src/java/org/apache/nutch/metrics/NutchMetrics.java b/src/java/org/apache/nutch/metrics/NutchMetrics.java index e64a8d6d00..dea34be7f6 100644 --- a/src/java/org/apache/nutch/metrics/NutchMetrics.java +++ b/src/java/org/apache/nutch/metrics/NutchMetrics.java @@ -367,5 +367,27 @@ private NutchMetrics() { /** Empty results in domain statistics. */ public static final String DOMAIN_STATS_EMPTY_RESULT_TOTAL = "empty_result_total"; + + // ========================================================================= + // Latency Metric Prefixes (used with LatencyTracker) + // ========================================================================= + + /** + * Prefix for fetch latency metrics. + * Used with {@link LatencyTracker} to emit fetch timing counters. + */ + public static final String FETCHER_LATENCY = "fetch_latency"; + + /** + * Prefix for parse latency metrics. + * Used with {@link LatencyTracker} to emit parse timing counters. + */ + public static final String PARSER_LATENCY = "parse_latency"; + + /** + * Prefix for indexer latency metrics. + * Used with {@link LatencyTracker} to emit indexing timing counters. + */ + public static final String INDEXER_LATENCY = "index_latency"; } diff --git a/src/java/org/apache/nutch/parse/ParseSegment.java b/src/java/org/apache/nutch/parse/ParseSegment.java index 5ec74ea9fe..a7fbe066ce 100644 --- a/src/java/org/apache/nutch/parse/ParseSegment.java +++ b/src/java/org/apache/nutch/parse/ParseSegment.java @@ -37,6 +37,7 @@ import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; import org.apache.nutch.metadata.Metadata; import org.apache.nutch.metadata.Nutch; +import org.apache.nutch.metrics.LatencyTracker; import org.apache.nutch.metrics.NutchMetrics; import org.apache.nutch.net.protocols.Response; import org.apache.nutch.protocol.Content; @@ -81,12 +82,22 @@ public static class ParseSegmentMapper extends private Text newKey = new Text(); private ScoringFilters scfilters; private boolean skipTruncated; + private LatencyTracker parseLatencyTracker; @Override public void setup(Mapper, Content, Text, ParseImpl>.Context context) { Configuration conf = context.getConfiguration(); scfilters = new ScoringFilters(conf); skipTruncated = conf.getBoolean(SKIP_TRUNCATED, true); + parseLatencyTracker = new LatencyTracker( + NutchMetrics.GROUP_PARSER, NutchMetrics.PARSER_LATENCY); + } + + @Override + public void cleanup(Mapper, Content, Text, ParseImpl>.Context context) + throws IOException, InterruptedException { + // Emit parse latency metrics + parseLatencyTracker.emitCounters(context); } @Override @@ -156,7 +167,9 @@ public void map(WritableComparable key, Content content, } long end = System.currentTimeMillis(); - LOG.info("Parsed ({}ms): {}", (end - start), url); + long parseTime = end - start; + parseLatencyTracker.record(parseTime); + LOG.info("Parsed ({}ms): {}", parseTime, url); context.write( url,