This is an automated email from the ASF dual-hosted git repository. jnioche pushed a commit to branch prometheus-reporter in repository https://gitbox.apache.org/repos/asf/storm.git
commit 329d3f0b73b2b82f09e8efe63c67185eb576f39e Author: Julien Nioche <[email protected]> AuthorDate: Sun Oct 11 17:53:24 2026 +0100 Add a Prometheus reporter for the V2 topology metrics Storm ships Console, CSV, JMX and Graphite reporters for the V2 metrics, and storm-metrics-prometheus only pushes the cluster metrics to a Pushgateway. With JMX or Graphite the task dimensions end up flattened in the metric names and have to be mapped back to labels with regular expressions. PrometheusStormReporter is a StormReporter which serves the metrics of each worker over HTTP for Prometheus to scrape. It reads the metrics of each task from the MetricRegistryProvider, so the dimensions become the labels topology, host, port, component, task and stream, and 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 port (9400 by default). The JVM metrics are exposed as well unless jvm.metrics is false. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- DEPENDENCY-LICENSES | 2 + docs/metrics_v2.md | 2 + external/storm-metrics-prometheus/README.md | 70 +++- external/storm-metrics-prometheus/pom.xml | 15 + .../prometheus/PrometheusStormReporter.java | 173 ++++++++++ .../metrics/prometheus/StormMetricsCollector.java | 367 +++++++++++++++++++++ .../prometheus/PrometheusStormReporterTest.java | 144 ++++++++ .../prometheus/StormMetricsCollectorTest.java | 87 +++++ 8 files changed, 858 insertions(+), 2 deletions(-) diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES index ac07efc58..644450b37 100644 --- a/DEPENDENCY-LICENSES +++ b/DEPENDENCY-LICENSES @@ -237,9 +237,11 @@ List of third-party dependencies grouped by their license type. * Prometheus Metrics Config (io.prometheus:prometheus-metrics-config:1.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-config) * Prometheus Metrics Core (io.prometheus:prometheus-metrics-core:1.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-core) * Prometheus Metrics Exporter - Common (io.prometheus:prometheus-metrics-exporter-common:1.9.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.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-httpserver) * Prometheus Metrics Exporter - Pushgateway (io.prometheus:prometheus-metrics-exporter-pushgateway:1.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-pushgateway) * Prometheus Metrics Exposition Formats (io.prometheus:prometheus-metrics-exposition-formats:1.9.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.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-textformats) + * Prometheus Metrics Instrumentation - JVM (io.prometheus:prometheus-metrics-instrumentation-jvm:1.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-instrumentation-jvm) * Prometheus Metrics Model (io.prometheus:prometheus-metrics-model:1.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-model) * Prometheus Metrics Tracer Common (io.prometheus:prometheus-metrics-tracer-common:1.9.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.9.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-initializer) diff --git a/docs/metrics_v2.md b/docs/metrics_v2.md index ae5527eb0..cd87a00d2 100644 --- a/docs/metrics_v2.md +++ b/docs/metrics_v2.md @@ -82,6 +82,8 @@ The following metric reporters are supported Reports metrics to a [Graphite](https://graphiteapp.org) server. * JMX Reporter (`org.apache.storm.metrics2.reporters.JmxStormReporter`): Exposes metrics via JMX. + * Prometheus Reporter (`org.apache.storm.metrics.prometheus.PrometheusStormReporter`, in the `storm-metrics-prometheus` module): + Serves the metrics of each worker over HTTP for [Prometheus](https://prometheus.io) to scrape, with the task dimensions as labels. Custom metrics reporters can be created by implementing `org.apache.storm.metrics2.reporters.StormReporter` interface or extending `org.apache.storm.metrics2.reporters.ScheduledStormReporter` class. diff --git a/external/storm-metrics-prometheus/README.md b/external/storm-metrics-prometheus/README.md index 351742c1a..3f5bee368 100644 --- a/external/storm-metrics-prometheus/README.md +++ b/external/storm-metrics-prometheus/README.md @@ -1,9 +1,75 @@ # Storm Metrics Prometheus -This module contains a reporter to push [Cluster Metrics](https://storm.apache.org/releases/current/ClusterMetrics.html) to +This module contains two reporters: + +* `PrometheusStormReporter`, a [Metrics V2](https://storm.apache.org/releases/current/metrics_v2.html) reporter which serves + the metrics of each worker over HTTP for Prometheus to scrape. +* `PrometheusPreparableReporter`, which pushes the cluster metrics to a Prometheus Pushgateway. + +## Topology metrics + +Unlike the generic JMX or Graphite reporters, `PrometheusStormReporter` 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 +``` + +### Usage + +Add the jar of this module and its dependencies to the topology or to the `lib` directory of your Storm installation, then set +the following in `storm.yaml` or in the topology configuration: + +```yaml +topology.metrics.reporters: + - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter" + # 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` | | 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 become one family named after the whole metric (`status.count` becomes `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 `rate()` gives their rate per second. +* Histograms and timers are exposed as summaries plus a `_max` and a `_mean` gauge. Timers are in seconds. Dropwizard + histograms do not track a sum, so the summaries have no `_sum`; use the `_mean` gauge instead. + +### Prometheus + +Scrape every worker port of every supervisor, e.g. with 4 slots per supervisor: + +```yaml +scrape_configs: + - job_name: storm + static_configs: + - targets: ["supervisor1:7700", "supervisor1:7701", "supervisor1:7702", "supervisor1:7703"] +``` + +A target is only up while a worker runs on the corresponding slot. + +## Cluster metrics + +`PrometheusPreparableReporter` pushes [Cluster Metrics](https://storm.apache.org/releases/current/ClusterMetrics.html) to a [Prometheus Pushgateway](https://github.com/prometheus/pushgateway), where it can be scraped by a Prometheus instance. -## Usage +### Usage To use, edit your `storm.yaml` config file: diff --git a/external/storm-metrics-prometheus/pom.xml b/external/storm-metrics-prometheus/pom.xml index df83b3653..61f70cd96 100644 --- a/external/storm-metrics-prometheus/pom.xml +++ b/external/storm-metrics-prometheus/pom.xml @@ -52,6 +52,16 @@ <artifactId>prometheus-metrics-exporter-pushgateway</artifactId> <version>${prometheus.client.version}</version> </dependency> + <dependency> + <groupId>io.prometheus</groupId> + <artifactId>prometheus-metrics-exporter-httpserver</artifactId> + <version>${prometheus.client.version}</version> + </dependency> + <dependency> + <groupId>io.prometheus</groupId> + <artifactId>prometheus-metrics-instrumentation-jvm</artifactId> + <version>${prometheus.client.version}</version> + </dependency> <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter</artifactId> @@ -67,6 +77,11 @@ <artifactId>junit-jupiter-engine</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-core</artifactId> + <scope>test</scope> + </dependency> <dependency> <groupId>org.testcontainers</groupId> <artifactId>testcontainers</artifactId> diff --git a/external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java b/external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java new file mode 100644 index 000000000..a0716ea60 --- /dev/null +++ b/external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java @@ -0,0 +1,173 @@ +/* + * 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.storm.metrics.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; + +/** + * Metrics V2 reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * + * <pre> + * topology.metrics.reporters: + * - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter" + * port.offset: 1000 + * </pre> + * + * <p>Each worker listens on its own port, which is the worker port plus <code>port.offset</code>, + * e.g. 7700 for a worker on 6700. When the worker port is unknown, as in local mode, the endpoint + * listens on <code>port</code>, which can also be set explicitly. The workers of a local cluster + * share the same endpoint. + */ +public class PrometheusStormReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusStormReporter.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<Integer, Endpoint> ENDPOINTS = new HashMap<>(); + + private StormMetricsCollector.Source source; + private int port; + private String host; + private boolean jvmMetrics; + + @Override + public void prepare(MetricRegistryProvider metricRegistryProvider, Map<String, Object> topoConf, + Map<String, Object> 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<String, Object> topoConf, + Map<String, Object> reporterConf) { + throw new UnsupportedOperationException( + "PrometheusStormReporter needs a MetricRegistryProvider to get the task " + + "dimensions"); + } + + static int resolvePort(Map<String, Object> 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/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java b/external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java new file mode 100644 index 000000000..eae69da2f --- /dev/null +++ b/external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java @@ -0,0 +1,367 @@ +/* + * 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.storm.metrics.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 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: + * + * <ul> + * <li>a scoped metric such as {@code fetcher_counter.status_200} becomes {@code + * fetcher_counter} with the label {@code scope="status_200"};</li> + * <li>a gauge whose value is a map, such as {@code status.count}, becomes {@code status_count} + * with a {@code key} label per entry;</li> + * <li>a Storm built-in metric such as {@code __ack-count-spout:default} becomes {@code + * storm_ack_count} with the labels {@code source_component="spout"} and {@code + * stream="default"}.</li> + * </ul> + * + * <p>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<Source> sources = new CopyOnWriteArrayList<>(); + + /** Families reported with conflicting types, so that the conflict is logged only once. */ + private final Set<String> 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<TaskMetricDimensions, TaskMetricRepo> task + : source.provider.getTaskMetrics().entrySet()) { + Map<String, String> 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<String, String> 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<String, Gauge> gauges, SortedMap<String, Counter> counters, + SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, + SortedMap<String, Timer> 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<Quantile> 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<String, Family> byName = new TreeMap<>(); + private final Set<String> conflicts; + + Families(Set<String> 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<Labels, DataPointSnapshot> points = new LinkedHashMap<>(); + + Family(Type type) { + this.type = type; + } + + @SuppressWarnings("unchecked") + MetricSnapshot snapshot(String name) { + MetricMetadata metadata = new MetricMetadata(name); + Collection<? extends DataPointSnapshot> values = points.values(); + if (type == Type.COUNTER) { + return new CounterSnapshot(metadata, (Collection<CounterDataPointSnapshot>) values); + } + if (type == Type.GAUGE) { + return new GaugeSnapshot(metadata, (Collection<GaugeDataPointSnapshot>) values); + } + return new SummarySnapshot(metadata, (Collection<SummaryDataPointSnapshot>) values); + } + } +} diff --git a/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/PrometheusStormReporterTest.java b/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/PrometheusStormReporterTest.java new file mode 100644 index 000000000..b31552451 --- /dev/null +++ b/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/PrometheusStormReporterTest.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.storm.metrics.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 PrometheusStormReporterTest { + + private static final String TOPOLOGY_ID = "wordcount-1-1700000000"; + private static final int WORKER_PORT = 6700; + + private StormMetricRegistry registry; + private PrometheusStormReporter reporter; + + @BeforeEach + void setUp() { + Map<String, Object> topoConf = new HashMap<>(); + topoConf.put(Config.STORM_ID, TOPOLOGY_ID); + topoConf.put(Config.TOPOLOGY_NAME, "wordcount"); + registry = new StormMetricRegistry(); + registry.start(topoConf, WORKER_PORT); + + reporter = new PrometheusStormReporter(); + 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<Integer>) () -> 7, "fetcher", 3); + registry.gauge("status.count", + (Gauge<Map<String, Long>>) () -> 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<String, String> fetcher = + Map.of("topology", "wordcount", "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(PrometheusStormReporter.DEFAULT_PORT, + PrometheusStormReporter.resolvePort(Map.of(), null)); + assertEquals(7700, PrometheusStormReporter.resolvePort(Map.of(), "6700")); + assertEquals(6800, PrometheusStormReporter.resolvePort(Map.of("port.offset", 100), "6700")); + assertEquals(9999, PrometheusStormReporter.resolvePort(Map.of("port", 9999), "6700")); + } + + private String scrape() throws Exception { + HttpRequest request = HttpRequest.newBuilder( + URI.create("http://localhost:"; + reporter.getBoundPort() + "/metrics")).build(); + HttpResponse<String> 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<String, String> labels, + String extraLabel, String extraValue, double expected) { + Map<String, String> 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/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/StormMetricsCollectorTest.java b/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/StormMetricsCollectorTest.java new file mode 100644 index 000000000..7d3479c63 --- /dev/null +++ b/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/StormMetricsCollectorTest.java @@ -0,0 +1,87 @@ +/* + * 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.storm.metrics.prometheus; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.prometheus.metrics.model.snapshots.Labels; +import java.util.Map; +import org.apache.storm.metrics.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<String, String> dimensions = Map.of( + "topologyId", "wordcount-1-1700000000", + "hostname", "worker1", + "port", "6700", + "componentId", "fetcher", + "taskid", "3"); + assertEquals(Labels.of( + "component", "fetcher", + "host", "worker1", + "port", "6700", + "task", "3", + "topology", "wordcount"), + StormMetricsCollector.taskLabels("wordcount", dimensions)); + } +}
