From 329d3f0b73b2b82f09e8efe63c67185eb576f39e Mon Sep 17 00:00:00 2001 From: Julien Nioche Date: Sun, 11 Oct 2026 17:53:24 +0100 Subject: [PATCH 1/2] 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 --- 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 +++++++++ .../prometheus/StormMetricsCollector.java | 367 ++++++++++++++++++ .../PrometheusStormReporterTest.java | 144 +++++++ .../prometheus/StormMetricsCollectorTest.java | 87 +++++ 8 files changed, 858 insertions(+), 2 deletions(-) create mode 100644 external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java create mode 100644 external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java create mode 100644 external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/PrometheusStormReporterTest.java create mode 100644 external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/StormMetricsCollectorTest.java diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES index ac07efc587e..644450b379f 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 ae5527eb073..cd87a00d26b 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 351742c1a32..3f5bee3686d 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 df83b36539d..61f70cd9672 100644 --- a/external/storm-metrics-prometheus/pom.xml +++ b/external/storm-metrics-prometheus/pom.xml @@ -52,6 +52,16 @@ prometheus-metrics-exporter-pushgateway ${prometheus.client.version} + + io.prometheus + prometheus-metrics-exporter-httpserver + ${prometheus.client.version} + + + io.prometheus + prometheus-metrics-instrumentation-jvm + ${prometheus.client.version} + org.junit.jupiter junit-jupiter @@ -67,6 +77,11 @@ junit-jupiter-engine test + + org.mockito + mockito-core + test + org.testcontainers testcontainers 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 00000000000..a0716ea602c --- /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. + * + * 
+ * topology.metrics.reporters:
+ *   - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter"
+ *     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 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 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( + "PrometheusStormReporter 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/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 00000000000..eae69da2ff8 --- /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: + * + *

    + *
  • a scoped metric such as {@code fetcher_counter.status_200} becomes {@code + * fetcher_counter} with the label {@code scope="status_200"};
    • + *
  • a gauge whose value is a map, such as {@code status.count}, becomes {@code status_count} + * with a {@code key} label per entry;
    • + *
  • 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"}.
    • + *
+ * + *

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/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 00000000000..b31552451d3 --- /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 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) () -> 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", "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 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/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 00000000000..7d3479c63bc --- /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 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)); + } +} From b03f38f54c70caa734b997458c693257df9596e2 Mon Sep 17 00:00:00 2001 From: Julien Nioche Date: Mon, 12 Oct 2026 12:06:50 +0100 Subject: [PATCH 2/2] Address review comments on the Prometheus reporter - Convert counters and meters before gauges, and skip the .m1_rate gauge of any RateCounter, not only those in Storm's namespace - Suffix every type when metrics of different types map to the same family, so that nothing is dropped and the result does not depend on the order in which tasks are reported - Turn the remote host and port of the Netty client gauges into labels - Catch IllegalArgumentException when starting the endpoint so that an out of range port does not kill the worker - Warn when an explicit port is used under a supervisor, and when a reporter sharing an endpoint has different host or jvm.metrics settings - Register the JVM metrics once per JVM to avoid leaking a registry and a GC listener every time an endpoint is recreated Co-Authored-By: Claude Opus 5.5 --- external/storm-metrics-prometheus/README.md | 12 +++- .../prometheus/PrometheusStormReporter.java | 29 ++++++++- .../prometheus/StormMetricsCollector.java | 59 +++++++++++++++---- .../PrometheusStormReporterTest.java | 55 +++++++++++++++++ .../prometheus/StormMetricsCollectorTest.java | 10 ++++ 5 files changed, 147 insertions(+), 18 deletions(-) diff --git a/external/storm-metrics-prometheus/README.md b/external/storm-metrics-prometheus/README.md index 3f5bee3686d..f413db28d3b 100644 --- a/external/storm-metrics-prometheus/README.md +++ b/external/storm-metrics-prometheus/README.md @@ -31,13 +31,14 @@ topology.metrics.reporters: | 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` | | Port to listen on, takes precedence over `port.offset`. Meant for local mode, where it defaults to 9400. On a cluster all workers of a host would get the same port and only the first one binds, so use `port.offset` there. | | `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. +The workers of a local cluster share a single endpoint. The `host` and `jvm.metrics` settings of the first reporter to +start apply to it, a warning is logged if a later reporter uses different ones. ### Naming @@ -46,10 +47,15 @@ The workers of a local cluster share a single endpoint. * 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. + The `.m1_rate` gauges registered next to rate counters are skipped, use `rate()` on the counter instead. +* The Netty client gauges (`__send-iconnection-sent-worker2.example.com:6701`) become `storm_send_iconnection_sent`, + `storm_send_iconnection_pending`, `storm_send_iconnection_reconnects` and `storm_send_iconnection_lostOnSend` with + `remote_host` and `remote_port` labels. * 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. +* A family can only have one type. If metrics of different types map to the same family (e.g. a counter `fetcher.pages` + and a gauge `fetcher.queue`), each type gets a suffix instead: `fetcher_counter`, `fetcher_gauge`, `fetcher_summary`. ### Prometheus 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 index a0716ea602c..4fe1231c4d7 100644 --- 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 @@ -15,10 +15,12 @@ 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.MultiCollector; import io.prometheus.metrics.model.registry.PrometheusRegistry; import java.io.IOException; import java.util.HashMap; import java.util.Map; +import java.util.Objects; import org.apache.storm.Config; import org.apache.storm.metrics2.MetricRegistryProvider; import org.apache.storm.metrics2.reporters.ScheduledStormReporter; @@ -96,6 +98,11 @@ public void prepare(MetricRegistry metricsRegistry, Map topoConf static int resolvePort(Map reporterConf, String workerPort) { Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); if (port != null) { + if (workerPort != null) { + LOG.warn("{} is set to {} for worker port {}, workers sharing this host " + + "will fail to bind, use {} instead", + PORT_KEY, port, workerPort, PORT_OFFSET_KEY); + } return port; } if (workerPort == null) { @@ -112,13 +119,17 @@ public void start() { if (endpoint == null) { try { endpoint = new Endpoint(port, host, jvmMetrics); - } catch (IOException e) { + } catch (IOException | IllegalArgumentException 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()); + } else if (!Objects.equals(host, endpoint.host) || jvmMetrics != endpoint.jvmMetrics) { + LOG.warn("Endpoint on port {} already runs with host={} and jvm.metrics={}, " + + "ignoring host={} and jvm.metrics={}", + port, endpoint.host, endpoint.jvmMetrics, host, jvmMetrics); } endpoint.collector.add(source); } @@ -156,12 +167,16 @@ private static final class Endpoint { private final StormMetricsCollector collector = new StormMetricsCollector(); private final HTTPServer server; + private final String host; + private final boolean jvmMetrics; Endpoint(int port, String host, boolean jvmMetrics) throws IOException { + this.host = host; + this.jvmMetrics = jvmMetrics; PrometheusRegistry registry = new PrometheusRegistry(); registry.register(collector); if (jvmMetrics) { - JvmMetrics.builder().register(registry); + registry.register((MultiCollector) JvmRegistry.INSTANCE::scrape); } HTTPServer.Builder builder = HTTPServer.builder().port(port).registry(registry); if (host != null) { @@ -170,4 +185,14 @@ private static final class Endpoint { server = builder.buildAndStart(); } } + + /** JVM metrics registered once per JVM, as {@link JvmMetrics} never releases a registry. */ + private static final class JvmRegistry { + + static final PrometheusRegistry INSTANCE = new PrometheusRegistry(); + + static { + JvmMetrics.builder().register(INSTANCE); + } + } } 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 index eae69da2ff8..95b51aae19a 100644 --- 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 @@ -38,8 +38,10 @@ import io.prometheus.metrics.model.snapshots.SummarySnapshot.SummaryDataPointSnapshot; import java.util.ArrayList; import java.util.Collection; +import java.util.EnumMap; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.SortedMap; @@ -83,6 +85,9 @@ class StormMetricsCollector implements MultiCollector { /** Prefix of the metrics Storm registers for each task. */ private static final String STORM_PREFIX = "__"; + /** Netty client gauges, suffixed with the remote host and port. */ + private static final String ICONNECTION_PREFIX = "send-iconnection-"; + /** * Storm registers a gauge with this suffix next to each of its counters. Prometheus computes * rates from the counter itself. @@ -190,6 +195,15 @@ static Name parse(String metric, String stream) { labels = Labels.of("source_component", base.substring(dash + 1)); base = base.substring(0, dash); } + } else if (stream == null && base.startsWith(ICONNECTION_PREFIX)) { + // e.g. __send-iconnection-sent-worker2.example.com:6701 + int dash = base.indexOf('-', ICONNECTION_PREFIX.length()); + int colon = base.lastIndexOf(':'); + if (dash > 0 && colon > dash + 1) { + labels = Labels.of("remote_host", base.substring(dash + 1, colon), + "remote_port", base.substring(colon + 1)); + base = base.substring(0, dash); + } } return new Name("storm_" + sanitize(base), labels); } @@ -241,7 +255,15 @@ private static final class Capture extends ScheduledReporter { public void report(SortedMap gauges, SortedMap counters, SortedMap histograms, SortedMap meters, SortedMap timers) { + // counters first so that they keep their family when a gauge has the same base name + counters.forEach((name, counter) -> counter(name, counter.getCount())); + meters.forEach((name, meter) -> counter(name, meter.getCount())); gauges.forEach((name, gauge) -> { + // the rate gauge of a RateCounter, the counter itself is exported + if (name.endsWith(RATE_SUFFIX) && counters.containsKey( + name.substring(0, name.length() - RATE_SUFFIX.length()))) { + return; + } // gauges call arbitrary code, one failing must not break the scrape try { gauge(name, gauge.getValue()); @@ -249,8 +271,6 @@ public void report(SortedMap gauges, SortedMap c 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) -> @@ -316,7 +336,7 @@ private void summary(String metric, String unit, long count, Snapshot snapshot, /** Data points grouped by family. */ private static final class Families { - private final Map byName = new TreeMap<>(); + private final Map> byName = new TreeMap<>(); private final Set conflicts; Families(Set conflicts) { @@ -324,20 +344,33 @@ private static final class Families { } 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); + byName.computeIfAbsent(name, n -> new EnumMap<>(Type.class)) + .computeIfAbsent(type, Family::new) + .points.putIfAbsent(point.getLabels(), point); } MetricSnapshots build() { MetricSnapshots.Builder builder = MetricSnapshots.builder(); - byName.forEach((name, family) -> builder.metricSnapshot(family.snapshot(name))); + byName.forEach((name, types) -> { + if (types.size() == 1) { + types.forEach((type, family) -> builder.metricSnapshot(family.snapshot(name))); + return; + } + // a family has a single type in Prometheus, suffix all of them so that the result + // does not depend on the order in which the tasks are reported + if (conflicts.add(name)) { + LOG.warn("Metric family {} is reported as {}, exposing it as {}_", name, + types.keySet(), name); + } + types.forEach((type, family) -> { + String suffixed = name + "_" + type.name().toLowerCase(Locale.ROOT); + if (byName.containsKey(suffixed)) { + LOG.debug("Skipping {} as {} already exists", type, suffixed); + return; + } + builder.metricSnapshot(family.snapshot(suffixed)); + }); + }); return builder.build(); } } 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 index b31552451d3..1b2a73a4a7c 100644 --- 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 @@ -12,8 +12,10 @@ package org.apache.storm.metrics.prometheus; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -101,6 +103,59 @@ void servesTaskMetrics() throws Exception { assertSample(body, "query_time_seconds_max", fetcher, null, null, 2); } + @Test + void counterSharingFamilyWithScopedGauge() throws Exception { + registry.counter("pages", "fetcher", 3).inc(4); + registry.gauge("pages.foo", (Gauge) () -> 9, "fetcher", 3); + // the rate gauge of a RateCounter outside of Storm's namespace + registry.rateCounter("fetched", "fetcher", 3).inc(6); + + String body = scrape(); + + Map fetcher = Map.of("component", "fetcher", "task", "3"); + assertSample(body, "pages_counter_total", fetcher, null, null, 4); + assertSample(body, "pages_gauge", fetcher, "scope", "foo", 9); + assertSample(body, "fetched_total", fetcher, null, null, 6); + assertFalse(body.contains("m1_rate"), body); + } + + @Test + void familyWithDifferentTypesInTwoTasks() throws Exception { + registry.counter("fetcher.pages", "fetcher", 3).inc(5); + registry.gauge("fetcher.queue", (Gauge) () -> 8, "parser", 4); + + String body = scrape(); + + assertSample(body, "fetcher_counter_total", Map.of("component", "fetcher"), "scope", + "pages", 5); + assertSample(body, "fetcher_gauge", Map.of("component", "parser"), "scope", "queue", 8); + } + + @Test + void portOutOfRangeDoesNotFailStart() { + PrometheusStormReporter outOfRange = new PrometheusStormReporter(); + outOfRange.prepare(registry, Map.of(Config.TOPOLOGY_NAME, "wordcount"), + Map.of("port", 70000)); + assertDoesNotThrow(outOfRange::start); + assertEquals(-1, outOfRange.getBoundPort()); + outOfRange.stop(); + } + + @Test + void jvmMetricsAfterEndpointRestart() throws Exception { + // frees the endpoint on port 0 so that the reporters below create their own + reporter.stop(); + for (int i = 0; i < 2; i++) { + reporter = new PrometheusStormReporter(); + reporter.prepare(registry, Map.of(Config.TOPOLOGY_NAME, "wordcount"), + Map.of("port", 0)); + reporter.start(); + String body = scrape(); + assertTrue(body.contains("jvm_memory_used_bytes"), body); + reporter.stop(); + } + } + @Test void resolvePort() { assertEquals(PrometheusStormReporter.DEFAULT_PORT, 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 index 7d3479c63bc..4d175410e68 100644 --- 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 @@ -60,6 +60,16 @@ void stormMetricWithoutStream() { StormMetricsCollector.parse("__receive.population", null)); } + @Test + void stormConnectionMetric() { + assertEquals(new Name("storm_send_iconnection_sent", + Labels.of("remote_host", "worker-2.example.com", "remote_port", "6701")), + StormMetricsCollector.parse("__send-iconnection-sent-worker-2.example.com:6701", null)); + assertEquals(new Name("storm_send_iconnection_lostOnSend", + Labels.of("remote_host", "10.0.0.5", "remote_port", "6700")), + StormMetricsCollector.parse("__send-iconnection-lostOnSend-10.0.0.5:6700", null)); + } + @Test void sanitize() { assertEquals("bytes_fetched_perSec", StormMetricsCollector.sanitize("bytes_fetched_perSec"));