Add a Prometheus reporter for the V2 topology metrics

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)

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.

external/storm-metrics-prometheus/README.md

@@ -1,9 +1,81 @@
 # 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`. 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 `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
+
+* 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 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
+
+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:
 

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>

external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java

@@ -0,0 +1,198 @@
+/*
+ * 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.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;
+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) {
+            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) {
+            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 | 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);
+        }
+    }
+
+    @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;
+        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) {
+                registry.register((MultiCollector) JvmRegistry.INSTANCE::scrape);
+            }
+            HTTPServer.Builder builder = HTTPServer.builder().port(port).registry(registry);
+            if (host != null) {
+                builder.hostname(host);
+            }
+            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);
+        }
+    }
+}