Add a Prometheus reporter for the Storm V2 metrics

THIRD-PARTY.txt

@@ -270,6 +270,18 @@ List of third-party dependencies grouped by their license type.
         * Playwright - Driver (com.microsoft.playwright:driver:1.63.0 - https://github.com/microsoft/playwright-java/driver)
         * Playwright - Main Library (com.microsoft.playwright:playwright:1.63.0 - https://github.com/microsoft/playwright-java/playwright)
         * Playwright - Node.js For All Platforms (com.microsoft.playwright:driver-bundle:1.63.0 - https://github.com/microsoft/playwright-java/driver-bundle)
+        * Prometheus Metrics Config (io.prometheus:prometheus-metrics-config:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-config)
+        * Prometheus Metrics Core (io.prometheus:prometheus-metrics-core:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-core)
+        * Prometheus Metrics Exporter - Common (io.prometheus:prometheus-metrics-exporter-common:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-common)
+        * Prometheus Metrics Exporter - HTTP Server (io.prometheus:prometheus-metrics-exporter-httpserver:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-httpserver)
+        * Prometheus Metrics Exposition Formats (io.prometheus:prometheus-metrics-exposition-formats:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-formats)
+        * Prometheus Metrics Exposition Text Formats (io.prometheus:prometheus-metrics-exposition-textformats:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-textformats)
+        * Prometheus Metrics Instrumentation - JVM (io.prometheus:prometheus-metrics-instrumentation-jvm:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-instrumentation-jvm)
+        * Prometheus Metrics Model (io.prometheus:prometheus-metrics-model:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-model)
+        * Prometheus Metrics Tracer Common (io.prometheus:prometheus-metrics-tracer-common:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-common)
+        * Prometheus Metrics Tracer Initializer (io.prometheus:prometheus-metrics-tracer-initializer:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-initializer)
+        * Prometheus Metrics Tracer OpenTelemetry (io.prometheus:prometheus-metrics-tracer-otel:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel)
+        * Prometheus Metrics Tracer OpenTelemetry Agent (io.prometheus:prometheus-metrics-tracer-otel-agent:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel-agent)
         * proto-google-common-protos (com.google.api.grpc:proto-google-common-protos:2.59.2 - https://github.com/googleapis/sdk-platform-java)
         * rank-eval (org.opensearch.plugin:rank-eval-client:2.19.6 - https://github.com/opensearch-project/OpenSearch.git)
         * rest (org.opensearch.client:opensearch-rest-client:2.19.6 - https://github.com/opensearch-project/OpenSearch.git)

external/prometheus/README.md

@@ -0,0 +1,56 @@
+# stormcrawler-prometheus
+
+A [Storm V2 metrics](https://storm.apache.org/releases/current/metrics_v2.html) reporter which serves the metrics of each worker over HTTP for [Prometheus](https://prometheus.io/) to scrape.
+
+Unlike the generic JMX or Graphite reporters, it uses the dimensions Storm keeps for each task as labels, so a metric reported by several tasks ends up in one family:
+
+```
+fetcher_counter_total{topology="crawl",host="worker1",port="6700",component="fetcher",task="3",scope="status_200"} 1234.0
+status_count{topology="crawl",host="worker1",port="6700",component="status",task="4",key="FETCHED"} 56789.0
+storm_ack_count_total{topology="crawl",host="worker1",port="6700",component="status",task="4",source_component="spout",stream="default"} 42.0
+```
+
+## Configuration
+
+Add the dependency to your topology and set the following in the configuration:
+
+```yaml
+  # StormCrawler's metrics must be registered with the V2 API
+  stormcrawler.metrics.version: "v2"
+
+  topology.metrics.reporters:
+    - class: "org.apache.stormcrawler.prometheus.PrometheusReporter"
+      # each worker listens on its port + offset, e.g. 7700 for 6700
+      port.offset: 1000
+```
+
+| Key | Default | Description |
+|---|---|---|
+| `port` | | Port to listen on, takes precedence over `port.offset`. Used as is in local mode, where it defaults to 9400. |
+| `port.offset` | 1000 | Added to the worker port to get the port to listen on. |
+| `host` | all interfaces | Address to bind to. |
+| `jvm.metrics` | true | Also expose the JVM metrics (memory, GC, threads, `process_start_time_seconds`). |
+| `filter` | | Storm metrics filter, as for the other V2 reporters. |
+
+The workers of a local cluster share a single endpoint.
+
+## Naming
+
+* Scoped metrics (`name.scope`) become the family `name` with a `scope` label. Only the first dot separates the scope, so `fetcher_counter.robots.fetched` has `scope="robots.fetched"`.
+* Gauges whose value is a map, such as `status.count`, become one family named after the whole metric (`status_count`) with a `key` label per entry.
+* Storm's built-in metrics (`__ack-count-spout:default`) become `storm_ack_count` with `source_component` and `stream` labels. The `.m1_rate` gauges Storm registers next to its counters are skipped, use `rate()` on the counter instead.
+* Counters and meters are exposed as counters, so `fetcher_average_persec.bytes_fetched_perSec` becomes `fetcher_average_persec_total{scope="bytes_fetched_perSec"}` and `rate()` gives the bytes per second.
+* Histograms and timers are exposed as summaries plus a `_max` and a `_mean` gauge. Timers are in seconds.
+
+## Prometheus
+
+Scrape every worker port of every supervisor, e.g. with 4 slots per supervisor:
+
+```yaml
+scrape_configs:
+  - job_name: stormcrawler
+    static_configs:
+      - targets: ["supervisor1:7700", "supervisor1:7701", "supervisor1:7702", "supervisor1:7703"]
+```
+
+A target is only up while a worker runs on the corresponding slot.

external/prometheus/pom.xml

@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+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.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.stormcrawler</groupId>
+        <artifactId>stormcrawler-external</artifactId>
+        <version>4.0.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>stormcrawler-prometheus</artifactId>
+    <packaging>jar</packaging>
+
+    <name>stormcrawler-prometheus</name>
+    <url>https://github.com/apache/stormcrawler/tree/master/external/prometheus</url>
+    <description>Prometheus reporter for the Storm V2 metrics</description>
+
+    <properties>
+        <prometheus.version>1.8.0</prometheus.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>prometheus-metrics-core</artifactId>
+            <version>${prometheus.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>prometheus-metrics-exporter-httpserver</artifactId>
+            <version>${prometheus.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>prometheus-metrics-instrumentation-jvm</artifactId>
+            <version>${prometheus.version}</version>
+        </dependency>
+    </dependencies>
+</project>

external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java

@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.stormcrawler.prometheus;
+
+import com.codahale.metrics.MetricRegistry;
+import io.prometheus.metrics.exporter.httpserver.HTTPServer;
+import io.prometheus.metrics.instrumentation.jvm.JvmMetrics;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.metrics2.MetricRegistryProvider;
+import org.apache.storm.metrics2.reporters.ScheduledStormReporter;
+import org.apache.storm.metrics2.reporters.StormReporter;
+import org.apache.storm.utils.ObjectReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Storm V2 metrics reporter serving the metrics of a worker over HTTP for Prometheus to scrape.
+ * Requires <code>stormcrawler.metrics.version</code> to be set to "v2" or "both" for the
+ * StormCrawler metrics to be included.
+ *
+ * <pre>
+ * topology.metrics.reporters:
+ *   - class: "org.apache.stormcrawler.prometheus.PrometheusReporter"
+ *     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 PrometheusReporter implements StormReporter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PrometheusReporter.class);
+
+    /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */
+    public static final String PORT_KEY = "port";
+
+    /** Added to the worker port to get the port to listen on, 1000 by default. */
+    public static final String PORT_OFFSET_KEY = "port.offset";
+
+    /** Address to bind to, all interfaces by default. */
+    public static final String HOST_KEY = "host";
+
+    /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */
+    public static final String JVM_METRICS_KEY = "jvm.metrics";
+
+    static final int DEFAULT_PORT_OFFSET = 1000;
+
+    /** Port used when the worker port is unknown, e.g. in local mode. */
+    static final int DEFAULT_PORT = 9400;
+
+    /** System property set by the supervisor when launching a worker. */
+    private static final String WORKER_PORT_PROPERTY = "worker.port";
+
+    /** Endpoints by port, shared by the reporters within a JVM. */
+    private static final Map<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(
+                "PrometheusReporter 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();
+        }
+    }
+}