diff --git a/DEPENDENCY-LICENSES b/DEPENDENCY-LICENSES index ac07efc587..644450b379 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 ae5527eb07..cd87a00d26 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 351742c1a3..f413db28d3 100644 --- a/external/storm-metrics-prometheus/README.md +++ b/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: diff --git a/external/storm-metrics-prometheus/pom.xml b/external/storm-metrics-prometheus/pom.xml index df83b36539..61f70cd967 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 0000000000..4fe1231c4d --- /dev/null +++ b/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. + * + * 
+ * 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) { + 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); + } + } +} 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 0000000000..95b51aae19 --- /dev/null +++ b/external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java @@ -0,0 +1,400 @@ +/* + * 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.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; +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: + * + *

+ * + *

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 = "__"; + + /** 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. + */ + 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); + } + } 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); + } + 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) { + // 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()); + } catch (RuntimeException e) { + LOG.debug("Could not read gauge {}", name, e); + } + }); + 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) { + 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, 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(); + } + } + + 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 0000000000..1b2a73a4a7 --- /dev/null +++ b/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/PrometheusStormReporterTest.java @@ -0,0 +1,199 @@ +/* + * 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.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; + +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 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, + 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 0000000000..4d175410e6 --- /dev/null +++ b/external/storm-metrics-prometheus/src/test/java/org/apache/storm/metrics/prometheus/StormMetricsCollectorTest.java @@ -0,0 +1,97 @@ +/* + * 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 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")); + 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)); + } +}