From b22c47d2d26a0ea2a7408a9c06a1d747f9eabf72 Mon Sep 17 00:00:00 2001 From: Julien Nioche Date: Sat, 10 Oct 2026 21:59:54 +0100 Subject: [PATCH] Add a Prometheus reporter for the Storm V2 metrics New module external/prometheus with PrometheusReporter, a Storm V2 StormReporter which serves the metrics of each worker over HTTP for Prometheus to scrape. Storm only ships Console, CSV, JMX and Graphite reporters for the topology metrics; with JMX or Graphite the dimensions end up flattened in the metric names and have to be mapped back to labels with regular expressions. Storm passes its reporters a MetricRegistryProvider, which gives the metrics of each task with their dimensions. These become the labels topology, host, port, component, task and stream, so a metric reported by several tasks ends up in a single family: - a scoped metric such as fetcher_counter.status_200 becomes fetcher_counter with scope="status_200" - a gauge whose value is a map, such as status.count, becomes status_count with a key label per entry - Storm's own metrics such as __ack-count-spout:default become storm_ack_count with source_component and stream labels; the .m1_rate gauges registered next to Storm's counters are skipped Counters and meters are exposed as counters, histograms and timers as summaries with a _max and a _mean gauge, timers in seconds. Each worker listens on its port plus port.offset (1000 by default), using the worker.port system property set by the supervisor. In local mode the workers share a single endpoint on the port given by port, 9400 by default. The JVM metrics are exposed too unless jvm.metrics is false. Co-Authored-By: Claude Opus 5.5 --- THIRD-PARTY.txt | 12 + external/prometheus/README.md | 56 +++ external/prometheus/pom.xml | 62 +++ .../prometheus/PrometheusReporter.java | 188 +++++++++ .../prometheus/StormMetricsCollector.java | 398 ++++++++++++++++++ .../prometheus/PrometheusReporterTest.java | 179 ++++++++ .../prometheus/StormMetricsCollectorTest.java | 105 +++++ pom.xml | 1 + 8 files changed, 1001 insertions(+) create mode 100644 external/prometheus/README.md create mode 100644 external/prometheus/pom.xml create mode 100644 external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java create mode 100644 external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java create mode 100644 external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java create mode 100644 external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java diff --git a/THIRD-PARTY.txt b/THIRD-PARTY.txt index 0078ae2ed..37e8e32a7 100644 --- a/THIRD-PARTY.txt +++ b/THIRD-PARTY.txt @@ -270,6 +270,18 @@ List of third-party dependencies grouped by their license type. * Playwright - Driver (com.microsoft.playwright:driver:1.63.0 - https://github.com/microsoft/playwright-java/driver) * Playwright - Main Library (com.microsoft.playwright:playwright:1.63.0 - https://github.com/microsoft/playwright-java/playwright) * Playwright - Node.js For All Platforms (com.microsoft.playwright:driver-bundle:1.63.0 - https://github.com/microsoft/playwright-java/driver-bundle) + * Prometheus Metrics Config (io.prometheus:prometheus-metrics-config:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-config) + * Prometheus Metrics Core (io.prometheus:prometheus-metrics-core:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-core) + * Prometheus Metrics Exporter - Common (io.prometheus:prometheus-metrics-exporter-common:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-common) + * Prometheus Metrics Exporter - HTTP Server (io.prometheus:prometheus-metrics-exporter-httpserver:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-httpserver) + * Prometheus Metrics Exposition Formats (io.prometheus:prometheus-metrics-exposition-formats:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-formats) + * Prometheus Metrics Exposition Text Formats (io.prometheus:prometheus-metrics-exposition-textformats:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-textformats) + * Prometheus Metrics Instrumentation - JVM (io.prometheus:prometheus-metrics-instrumentation-jvm:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-instrumentation-jvm) + * Prometheus Metrics Model (io.prometheus:prometheus-metrics-model:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-model) + * Prometheus Metrics Tracer Common (io.prometheus:prometheus-metrics-tracer-common:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-common) + * Prometheus Metrics Tracer Initializer (io.prometheus:prometheus-metrics-tracer-initializer:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-initializer) + * Prometheus Metrics Tracer OpenTelemetry (io.prometheus:prometheus-metrics-tracer-otel:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel) + * Prometheus Metrics Tracer OpenTelemetry Agent (io.prometheus:prometheus-metrics-tracer-otel-agent:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel-agent) * proto-google-common-protos (com.google.api.grpc:proto-google-common-protos:2.59.2 - https://github.com/googleapis/sdk-platform-java) * rank-eval (org.opensearch.plugin:rank-eval-client:2.19.6 - https://github.com/opensearch-project/OpenSearch.git) * rest (org.opensearch.client:opensearch-rest-client:2.19.6 - https://github.com/opensearch-project/OpenSearch.git) diff --git a/external/prometheus/README.md b/external/prometheus/README.md new file mode 100644 index 000000000..0304b430d --- /dev/null +++ b/external/prometheus/README.md @@ -0,0 +1,56 @@ +# stormcrawler-prometheus + +A [Storm V2 metrics](https://storm.apache.org/releases/current/metrics_v2.html) reporter which serves the metrics of each worker over HTTP for [Prometheus](https://prometheus.io/) to scrape. + +Unlike the generic JMX or Graphite reporters, it uses the dimensions Storm keeps for each task as labels, so a metric reported by several tasks ends up in one family: + +``` +fetcher_counter_total{topology="crawl",host="worker1",port="6700",component="fetcher",task="3",scope="status_200"} 1234.0 +status_count{topology="crawl",host="worker1",port="6700",component="status",task="4",key="FETCHED"} 56789.0 +storm_ack_count_total{topology="crawl",host="worker1",port="6700",component="status",task="4",source_component="spout",stream="default"} 42.0 +``` + +## Configuration + +Add the dependency to your topology and set the following in the configuration: + +```yaml + # StormCrawler's metrics must be registered with the V2 API + stormcrawler.metrics.version: "v2" + + topology.metrics.reporters: + - class: "org.apache.stormcrawler.prometheus.PrometheusReporter" + # each worker listens on its port + offset, e.g. 7700 for 6700 + port.offset: 1000 +``` + +| Key | Default | Description | +|---|---|---| +| `port` | | Port to listen on, takes precedence over `port.offset`. Used as is in local mode, where it defaults to 9400. | +| `port.offset` | 1000 | Added to the worker port to get the port to listen on. | +| `host` | all interfaces | Address to bind to. | +| `jvm.metrics` | true | Also expose the JVM metrics (memory, GC, threads, `process_start_time_seconds`). | +| `filter` | | Storm metrics filter, as for the other V2 reporters. | + +The workers of a local cluster share a single endpoint. + +## Naming + +* Scoped metrics (`name.scope`) become the family `name` with a `scope` label. Only the first dot separates the scope, so `fetcher_counter.robots.fetched` has `scope="robots.fetched"`. +* Gauges whose value is a map, such as `status.count`, become one family named after the whole metric (`status_count`) with a `key` label per entry. +* Storm's built-in metrics (`__ack-count-spout:default`) become `storm_ack_count` with `source_component` and `stream` labels. The `.m1_rate` gauges Storm registers next to its counters are skipped, use `rate()` on the counter instead. +* Counters and meters are exposed as counters, so `fetcher_average_persec.bytes_fetched_perSec` becomes `fetcher_average_persec_total{scope="bytes_fetched_perSec"}` and `rate()` gives the bytes per second. +* Histograms and timers are exposed as summaries plus a `_max` and a `_mean` gauge. Timers are in seconds. + +## Prometheus + +Scrape every worker port of every supervisor, e.g. with 4 slots per supervisor: + +```yaml +scrape_configs: + - job_name: stormcrawler + static_configs: + - targets: ["supervisor1:7700", "supervisor1:7701", "supervisor1:7702", "supervisor1:7703"] +``` + +A target is only up while a worker runs on the corresponding slot. diff --git a/external/prometheus/pom.xml b/external/prometheus/pom.xml new file mode 100644 index 000000000..d6698cb20 --- /dev/null +++ b/external/prometheus/pom.xml @@ -0,0 +1,62 @@ + + + + + + 4.0.0 + + + org.apache.stormcrawler + stormcrawler-external + 4.0.0-SNAPSHOT + ../pom.xml + + + stormcrawler-prometheus + jar + + stormcrawler-prometheus + https://github.com/apache/stormcrawler/tree/master/external/prometheus + Prometheus reporter for the Storm V2 metrics + + + 1.8.0 + + + + + io.prometheus + prometheus-metrics-core + ${prometheus.version} + + + + io.prometheus + prometheus-metrics-exporter-httpserver + ${prometheus.version} + + + + io.prometheus + prometheus-metrics-instrumentation-jvm + ${prometheus.version} + + + diff --git a/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java new file mode 100644 index 000000000..ae300a6be --- /dev/null +++ b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java @@ -0,0 +1,188 @@ +/* + * 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.prometheus; + +import com.codahale.metrics.MetricRegistry; +import io.prometheus.metrics.exporter.httpserver.HTTPServer; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.reporters.ScheduledStormReporter; +import org.apache.storm.metrics2.reporters.StormReporter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Storm V2 metrics reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * Requires stormcrawler.metrics.version to be set to "v2" or "both" for the + * StormCrawler metrics to be included. + * + * 
+ * topology.metrics.reporters:
+ *   - class: "org.apache.stormcrawler.prometheus.PrometheusReporter"
+ *     port.offset: 1000
+ *
+ * + *

Each worker listens on its own port, which is the worker port plus port.offset, + * e.g. 7700 for a worker on 6700. When the worker port is unknown, as in local mode, the endpoint + * listens on port, which can also be set explicitly. The workers of a local cluster + * share the same endpoint. + */ +public class PrometheusReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusReporter.class); + + /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */ + public static final String PORT_KEY = "port"; + + /** Added to the worker port to get the port to listen on, 1000 by default. */ + public static final String PORT_OFFSET_KEY = "port.offset"; + + /** Address to bind to, all interfaces by default. */ + public static final String HOST_KEY = "host"; + + /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */ + public static final String JVM_METRICS_KEY = "jvm.metrics"; + + static final int DEFAULT_PORT_OFFSET = 1000; + + /** Port used when the worker port is unknown, e.g. in local mode. */ + static final int DEFAULT_PORT = 9400; + + /** System property set by the supervisor when launching a worker. */ + private static final String WORKER_PORT_PROPERTY = "worker.port"; + + /** Endpoints by port, shared by the reporters within a JVM. */ + private static final Map ENDPOINTS = new HashMap<>(); + + private StormMetricsCollector.Source source; + private int port; + private String host; + private boolean jvmMetrics; + + @Override + public void prepare( + MetricRegistryProvider metricRegistryProvider, + Map topoConf, + Map reporterConf) { + String topology = + ObjectReader.getString( + topoConf.get(Config.TOPOLOGY_NAME), + ObjectReader.getString(topoConf.get(Config.STORM_ID), null)); + source = + new StormMetricsCollector.Source( + metricRegistryProvider, + topology, + ScheduledStormReporter.getMetricsFilter(reporterConf)); + port = resolvePort(reporterConf, System.getProperty(WORKER_PORT_PROPERTY)); + host = ObjectReader.getString(reporterConf.get(HOST_KEY), null); + jvmMetrics = ObjectReader.getBoolean(reporterConf.get(JVM_METRICS_KEY), true); + } + + @Override + public void prepare( + MetricRegistry metricsRegistry, + Map topoConf, + Map reporterConf) { + throw new UnsupportedOperationException( + "PrometheusReporter needs a MetricRegistryProvider to get the task dimensions"); + } + + static int resolvePort(Map reporterConf, String workerPort) { + Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); + if (port != null) { + return port; + } + if (workerPort == null) { + return DEFAULT_PORT; + } + int offset = ObjectReader.getInt(reporterConf.get(PORT_OFFSET_KEY), DEFAULT_PORT_OFFSET); + return Integer.parseInt(workerPort) + offset; + } + + @Override + public void start() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + try { + endpoint = new Endpoint(port, host, jvmMetrics); + } catch (IOException e) { + LOG.error("Could not serve the metrics on port {}", port, e); + return; + } + ENDPOINTS.put(port, endpoint); + LOG.info( + "Serving the metrics for Prometheus on port {}", endpoint.server.getPort()); + } + endpoint.collector.add(source); + } + } + + @Override + public void stop() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + return; + } + endpoint.collector.remove(source); + if (endpoint.collector.isEmpty()) { + endpoint.server.close(); + ENDPOINTS.remove(port); + } + } + } + + @Override + public void close() { + stop(); + } + + /** Port the endpoint of this reporter is bound to, -1 if it is not running. */ + int getBoundPort() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + return endpoint == null ? -1 : endpoint.server.getPort(); + } + } + + private static final class Endpoint { + + private final StormMetricsCollector collector = new StormMetricsCollector(); + private final HTTPServer server; + + Endpoint(int port, String host, boolean jvmMetrics) throws IOException { + PrometheusRegistry registry = new PrometheusRegistry(); + registry.register(collector); + if (jvmMetrics) { + JvmMetrics.builder().register(registry); + } + HTTPServer.Builder builder = HTTPServer.builder().port(port).registry(registry); + if (host != null) { + builder.hostname(host); + } + server = builder.buildAndStart(); + } + } +} diff --git a/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java new file mode 100644 index 000000000..22cac7ff7 --- /dev/null +++ b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java @@ -0,0 +1,398 @@ +/* + * 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.prometheus; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricFilter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.ScheduledReporter; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.Timer; +import io.prometheus.metrics.model.registry.MultiCollector; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.CounterSnapshot.CounterDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.DataPointSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot.GaugeDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import io.prometheus.metrics.model.snapshots.MetricMetadata; +import io.prometheus.metrics.model.snapshots.MetricSnapshot; +import io.prometheus.metrics.model.snapshots.MetricSnapshots; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import io.prometheus.metrics.model.snapshots.Quantile; +import io.prometheus.metrics.model.snapshots.Quantiles; +import io.prometheus.metrics.model.snapshots.SummarySnapshot; +import io.prometheus.metrics.model.snapshots.SummarySnapshot.SummaryDataPointSnapshot; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.TaskMetricDimensions; +import org.apache.storm.metrics2.TaskMetricRepo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns the Storm V2 metrics of one or more workers into Prometheus metric families. The dimensions + * Storm keeps for each task (topology, host, port, component, task and stream) become labels, so + * that a metric reported by several tasks ends up in a single family: + * + *

+ * + *

Counters and meters are exposed as counters, histograms and timers as summaries along with a + * {@code _max} and a {@code _mean} gauge. Timers are converted to seconds. + */ +class StormMetricsCollector implements MultiCollector { + + private static final Logger LOG = LoggerFactory.getLogger(StormMetricsCollector.class); + + private static final double[] QUANTILES = {0.5, 0.75, 0.95, 0.98, 0.99, 0.999}; + + private static final double NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1); + + /** Prefix of the metrics Storm registers for each task. */ + private static final String STORM_PREFIX = "__"; + + /** + * Storm registers a gauge with this suffix next to each of its counters. Prometheus computes + * rates from the counter itself. + */ + private static final String RATE_SUFFIX = ".m1_rate"; + + private static final Pattern INVALID_CHARS = Pattern.compile("[^a-zA-Z0-9_]"); + + /** The metrics of a worker. */ + static final class Source { + private final MetricRegistryProvider provider; + private final String topology; + private final MetricFilter filter; + + Source(MetricRegistryProvider provider, String topology, MetricFilter filter) { + this.provider = provider; + this.topology = topology; + this.filter = filter; + } + } + + /** Family name and labels derived from the name of a Storm metric. */ + record Name(String family, Labels labels) {} + + private enum Type { + COUNTER, + GAUGE, + SUMMARY + } + + private final List sources = new CopyOnWriteArrayList<>(); + + /** Families reported with conflicting types, so that the conflict is logged only once. */ + private final Set conflicts = ConcurrentHashMap.newKeySet(); + + /** {@link TaskMetricRepo} only hands its metrics over to a {@link ScheduledReporter}. */ + private final Capture capture = new Capture(); + + void add(Source source) { + sources.add(source); + } + + void remove(Source source) { + sources.remove(source); + } + + boolean isEmpty() { + return sources.isEmpty(); + } + + @Override + public synchronized MetricSnapshots collect() { + Families families = new Families(conflicts); + for (Source source : sources) { + for (Map.Entry task : + source.provider.getTaskMetrics().entrySet()) { + Map dimensions = task.getKey().getDimensions(); + capture.families = families; + capture.labels = taskLabels(source.topology, dimensions); + capture.stream = dimensions.get("streamId"); + task.getValue().report(capture, source.filter); + } + } + capture.families = null; + return families.build(); + } + + static Labels taskLabels(String topology, Map dimensions) { + Labels.Builder builder = Labels.builder(); + addLabel(builder, "topology", topology); + addLabel(builder, "host", dimensions.get("hostname")); + addLabel(builder, "port", dimensions.get("port")); + addLabel(builder, "component", dimensions.get("componentId")); + addLabel(builder, "task", dimensions.get("taskid")); + addLabel(builder, "stream", dimensions.get("streamId")); + return builder.build(); + } + + private static void addLabel(Labels.Builder builder, String name, String value) { + if (value != null) { + builder.label(name, value); + } + } + + /** + * Maps the name of a Storm metric to a family name and labels. + * + * @param metric the name of the metric as registered with Storm + * @param stream the stream the metric was registered for, if any + */ + static Name parse(String metric, String stream) { + if (metric.startsWith(STORM_PREFIX)) { + String base = metric.substring(STORM_PREFIX.length()); + Labels labels = Labels.EMPTY; + if (stream != null && base.endsWith("-" + stream)) { + // e.g. __emit-count-default + base = base.substring(0, base.length() - stream.length() - 1); + } else if (stream != null && base.endsWith(":" + stream)) { + // e.g. __ack-count-spout:default, the names of Storm's + // metrics are made of two words + base = base.substring(0, base.length() - stream.length() - 1); + int dash = base.indexOf('-', base.indexOf('-') + 1); + if (dash > 0) { + labels = Labels.of("source_component", base.substring(dash + 1)); + base = base.substring(0, dash); + } + } + return new Name("storm_" + sanitize(base), labels); + } + int dot = metric.indexOf('.'); + if (dot > 0 && dot < metric.length() - 1) { + return new Name( + sanitize(metric.substring(0, dot)), + Labels.of("scope", metric.substring(dot + 1))); + } + return new Name(sanitize(metric), Labels.EMPTY); + } + + static String sanitize(String name) { + String sanitized = INVALID_CHARS.matcher(name).replaceAll("_"); + if (sanitized.isEmpty() || Character.isDigit(sanitized.charAt(0))) { + sanitized = "_" + sanitized; + } + // removes suffixes such as _total which are reserved + return PrometheusNaming.sanitizeMetricName(sanitized); + } + + private static Double toDouble(Object value) { + if (value instanceof Number number) { + return number.doubleValue(); + } + if (value instanceof Boolean bool) { + return bool ? 1d : 0d; + } + return null; + } + + private static GaugeDataPointSnapshot gaugePoint(Labels labels, double value) { + return GaugeDataPointSnapshot.builder().labels(labels).value(value).build(); + } + + /** Converts the metrics of a task. */ + private static final class Capture extends ScheduledReporter { + + private Families families; + private Labels labels; + private String stream; + + Capture() { + super( + new MetricRegistry(), + "prometheus", + MetricFilter.ALL, + TimeUnit.SECONDS, + TimeUnit.SECONDS); + } + + @Override + @SuppressWarnings("rawtypes") + public void report( + SortedMap gauges, + SortedMap counters, + SortedMap histograms, + SortedMap meters, + SortedMap timers) { + gauges.forEach( + (name, gauge) -> { + // gauges call arbitrary code, one failing must not break the scrape + try { + gauge(name, gauge.getValue()); + } catch (RuntimeException e) { + LOG.debug("Could not read gauge {}", name, e); + } + }); + counters.forEach((name, counter) -> counter(name, counter.getCount())); + meters.forEach((name, meter) -> counter(name, meter.getCount())); + histograms.forEach( + (name, histogram) -> + summary(name, "", histogram.getCount(), histogram.getSnapshot(), 1)); + timers.forEach( + (name, timer) -> + summary( + name, + "_seconds", + timer.getCount(), + timer.getSnapshot(), + NANOS_PER_SECOND)); + } + + private void gauge(String metric, Object value) { + if (metric.startsWith(STORM_PREFIX) && metric.endsWith(RATE_SUFFIX)) { + return; + } + if (value instanceof Map map) { + String family = sanitize(metric); + map.forEach( + (key, entry) -> { + Double number = toDouble(entry); + if (number != null) { + Labels keyLabels = labels.add("key", String.valueOf(key)); + families.add(family, Type.GAUGE, gaugePoint(keyLabels, number)); + } + }); + return; + } + Double number = toDouble(value); + if (number == null) { + return; + } + Name name = parse(metric, stream); + families.add( + name.family(), Type.GAUGE, gaugePoint(labels.merge(name.labels()), number)); + } + + private void counter(String metric, long count) { + Name name = parse(metric, stream); + families.add( + name.family(), + Type.COUNTER, + CounterDataPointSnapshot.builder() + .labels(labels.merge(name.labels())) + .value(count) + .build()); + } + + private void summary( + String metric, String unit, long count, Snapshot snapshot, double scale) { + Name name = parse(metric, stream); + String family = name.family() + unit; + Labels pointLabels = labels.merge(name.labels()); + List quantiles = new ArrayList<>(QUANTILES.length); + for (double quantile : QUANTILES) { + quantiles.add(new Quantile(quantile, snapshot.getValue(quantile) / scale)); + } + families.add( + family, + Type.SUMMARY, + SummaryDataPointSnapshot.builder() + .labels(pointLabels) + .count(count) + .quantiles(Quantiles.of(quantiles)) + .build()); + families.add( + family + "_max", + Type.GAUGE, + gaugePoint(pointLabels, snapshot.getMax() / scale)); + families.add( + family + "_mean", + Type.GAUGE, + gaugePoint(pointLabels, snapshot.getMean() / scale)); + } + } + + /** Data points grouped by family. */ + private static final class Families { + + private final Map byName = new TreeMap<>(); + private final Set conflicts; + + Families(Set conflicts) { + this.conflicts = conflicts; + } + + void add(String name, Type type, DataPointSnapshot point) { + Family family = byName.computeIfAbsent(name, n -> new Family(type)); + if (family.type != type) { + if (conflicts.add(name)) { + LOG.warn( + "Metric family {} is reported both as a {} and a {}, ignoring the latter", + name, + family.type, + type); + } + return; + } + family.points.putIfAbsent(point.getLabels(), point); + } + + MetricSnapshots build() { + MetricSnapshots.Builder builder = MetricSnapshots.builder(); + byName.forEach((name, family) -> builder.metricSnapshot(family.snapshot(name))); + return builder.build(); + } + } + + private static final class Family { + + private final Type type; + private final Map points = new LinkedHashMap<>(); + + Family(Type type) { + this.type = type; + } + + @SuppressWarnings("unchecked") + MetricSnapshot snapshot(String name) { + MetricMetadata metadata = new MetricMetadata(name); + Collection values = points.values(); + if (type == Type.COUNTER) { + return new CounterSnapshot(metadata, (Collection) values); + } + if (type == Type.GAUGE) { + return new GaugeSnapshot(metadata, (Collection) values); + } + return new SummarySnapshot(metadata, (Collection) values); + } + } +} diff --git a/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java new file mode 100644 index 000000000..682ae41ec --- /dev/null +++ b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java @@ -0,0 +1,179 @@ +/* + * 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.prometheus; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.Config; +import org.apache.storm.metrics2.StormMetricRegistry; +import org.apache.storm.task.TopologyContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class PrometheusReporterTest { + + private static final String TOPOLOGY_ID = "crawl-1-1700000000"; + private static final int WORKER_PORT = 6700; + + private StormMetricRegistry registry; + private PrometheusReporter reporter; + + @BeforeEach + void setUp() { + Map topoConf = new HashMap<>(); + topoConf.put(Config.STORM_ID, TOPOLOGY_ID); + topoConf.put(Config.TOPOLOGY_NAME, "crawl"); + registry = new StormMetricRegistry(); + registry.start(topoConf, WORKER_PORT); + + reporter = new PrometheusReporter(); + reporter.prepare(registry, topoConf, Map.of("port", 0, "jvm.metrics", false)); + reporter.start(); + } + + @AfterEach + void tearDown() { + reporter.stop(); + registry.stop(); + } + + @Test + void servesTaskMetrics() throws Exception { + registry.counter("fetcher_counter.status_200", "fetcher", 3).inc(5); + registry.gauge("activethreads", (Gauge) () -> 7, "fetcher", 3); + registry.gauge( + "status.count", + (Gauge>) () -> Map.of("FETCHED", 12L, "DISCOVERED", 30L), + "status", + 4); + registry.rateCounter( + "__ack-count-spout:default", + TOPOLOGY_ID, + "status", + 4, + WORKER_PORT, + "default") + .inc(2); + + TopologyContext context = mock(TopologyContext.class); + when(context.getStormId()).thenReturn(TOPOLOGY_ID); + when(context.getThisComponentId()).thenReturn("fetcher"); + when(context.getThisTaskId()).thenReturn(3); + when(context.getThisWorkerPort()).thenReturn(WORKER_PORT); + Histogram histogram = registry.histogram("fetcher_average_perdoc.time_in_queues", context); + histogram.update(100); + histogram.update(300); + registry.meter("fetcher_average_persec.bytes_fetched_perSec", context).mark(1000); + registry.timer("query_time", context).update(2, TimeUnit.SECONDS); + + String body = scrape(); + + Map fetcher = + Map.of("topology", "crawl", "component", "fetcher", "task", "3", "port", "6700"); + assertSample(body, "fetcher_counter_total", fetcher, "scope", "status_200", 5); + assertSample(body, "activethreads", fetcher, null, null, 7); + assertSample(body, "status_count", Map.of("component", "status"), "key", "FETCHED", 12); + assertSample(body, "status_count", Map.of("component", "status"), "key", "DISCOVERED", 30); + assertSample( + body, + "storm_ack_count_total", + Map.of("component", "status", "stream", "default"), + "source_component", + "spout", + 2); + assertFalse(body.contains("m1_rate"), body); + assertSample(body, "fetcher_average_perdoc_count", fetcher, "scope", "time_in_queues", 2); + assertSample(body, "fetcher_average_perdoc_max", fetcher, "scope", "time_in_queues", 300); + assertSample(body, "fetcher_average_perdoc_mean", fetcher, "scope", "time_in_queues", 200); + assertSample( + body, + "fetcher_average_persec_total", + fetcher, + "scope", + "bytes_fetched_perSec", + 1000); + assertSample(body, "query_time_seconds_max", fetcher, null, null, 2); + } + + @Test + void resolvePort() { + assertEquals( + PrometheusReporter.DEFAULT_PORT, PrometheusReporter.resolvePort(Map.of(), null)); + assertEquals(7700, PrometheusReporter.resolvePort(Map.of(), "6700")); + assertEquals(6800, PrometheusReporter.resolvePort(Map.of("port.offset", 100), "6700")); + assertEquals(9999, PrometheusReporter.resolvePort(Map.of("port", 9999), "6700")); + } + + private String scrape() throws Exception { + HttpRequest request = + HttpRequest.newBuilder( + URI.create( + "http://localhost:" + reporter.getBoundPort() + "/metrics")) + .build(); + HttpResponse response = + HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString()); + assertEquals(200, response.statusCode()); + return response.body(); + } + + /** Checks that a line of the text format has the given name, labels and value. */ + private static void assertSample( + String body, + String name, + Map labels, + String extraLabel, + String extraValue, + double expected) { + Map wanted = new HashMap<>(labels); + if (extraLabel != null) { + wanted.put(extraLabel, extraValue); + } + for (String line : body.split("\n")) { + if (!line.startsWith(name + "{")) { + continue; + } + String labelPart = line.substring(name.length() + 1, line.lastIndexOf('}')); + boolean matches = + wanted.entrySet().stream() + .allMatch( + e -> + labelPart.contains( + e.getKey() + "=\"" + e.getValue() + "\"")); + if (matches) { + double value = Double.parseDouble(line.substring(line.lastIndexOf(' ') + 1)); + assertEquals(expected, value, 0.0001, line); + return; + } + } + fail("No sample " + name + " with labels " + wanted + " in\n" + body); + } +} diff --git a/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java new file mode 100644 index 000000000..321c973d4 --- /dev/null +++ b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java @@ -0,0 +1,105 @@ +/* + * 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.prometheus; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.prometheus.metrics.model.snapshots.Labels; +import java.util.Map; +import org.apache.stormcrawler.prometheus.StormMetricsCollector.Name; +import org.junit.jupiter.api.Test; + +class StormMetricsCollectorTest { + + @Test + void scopedMetric() { + assertEquals( + new Name("fetcher_counter", Labels.of("scope", "status_200")), + StormMetricsCollector.parse("fetcher_counter.status_200", null)); + // only the first dot separates the scope + assertEquals( + new Name("fetcher_counter", Labels.of("scope", "robots.fetched")), + StormMetricsCollector.parse("fetcher_counter.robots.fetched", null)); + } + + @Test + void plainMetric() { + assertEquals( + new Name("activethreads", Labels.EMPTY), + StormMetricsCollector.parse("activethreads", null)); + assertEquals( + new Name("trailing_", Labels.EMPTY), + StormMetricsCollector.parse("trailing.", null)); + } + + @Test + void stormMetricWithStream() { + assertEquals( + new Name("storm_emit_count", Labels.EMPTY), + StormMetricsCollector.parse("__emit-count-default", "default")); + assertEquals( + new Name("storm_capacity", Labels.EMPTY), + StormMetricsCollector.parse("__capacity", "default")); + } + + @Test + void stormMetricWithSourceComponent() { + assertEquals( + new Name("storm_ack_count", Labels.of("source_component", "spout")), + StormMetricsCollector.parse("__ack-count-spout:default", "default")); + assertEquals( + new Name("storm_process_latency", Labels.of("source_component", "url-partitioner")), + StormMetricsCollector.parse("__process-latency-url-partitioner:status", "status")); + } + + @Test + void stormMetricWithoutStream() { + assertEquals( + new Name("storm_receive_population", Labels.EMPTY), + StormMetricsCollector.parse("__receive.population", null)); + } + + @Test + void sanitize() { + assertEquals( + "bytes_fetched_perSec", StormMetricsCollector.sanitize("bytes_fetched_perSec")); + assertEquals( + "GC_G1_Young_Generation", StormMetricsCollector.sanitize("GC.G1-Young-Generation")); + assertEquals("_2xx", StormMetricsCollector.sanitize("2xx")); + assertEquals("pages", StormMetricsCollector.sanitize("pages_total")); + } + + @Test + void taskLabels() { + Map dimensions = + Map.of( + "topologyId", "crawl-1-1700000000", + "hostname", "worker1", + "port", "6700", + "componentId", "fetcher", + "taskid", "3"); + assertEquals( + Labels.of( + "component", "fetcher", + "host", "worker1", + "port", "6700", + "task", "3", + "topology", "crawl"), + StormMetricsCollector.taskLabels("crawl", dimensions)); + } +} diff --git a/pom.xml b/pom.xml index aff3816aa..ced956c61 100644 --- a/pom.xml +++ b/pom.xml @@ -749,6 +749,7 @@ under the License. external/opensearch external/opensearch-java external/playwright + external/prometheus external/solr external/sql external/tika