rzo1 commented on code in PR #9158: URL: https://github.com/apache/storm/pull/9158#discussion_r4242611026
########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java: ########## @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.MetricRegistry; +import io.prometheus.metrics.exporter.httpserver.HTTPServer; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.reporters.ScheduledStormReporter; +import org.apache.storm.metrics2.reporters.StormReporter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Metrics V2 reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * + * <pre> + * topology.metrics.reporters: + * - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter" + * port.offset: 1000 + * </pre> + * + * <p>Each worker listens on its own port, which is the worker port plus <code>port.offset</code>, + * e.g. 7700 for a worker on 6700. When the worker port is unknown, as in local mode, the endpoint + * listens on <code>port</code>, which can also be set explicitly. The workers of a local cluster + * share the same endpoint. + */ +public class PrometheusStormReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusStormReporter.class); + + /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */ + public static final String PORT_KEY = "port"; + + /** Added to the worker port to get the port to listen on, 1000 by default. */ + public static final String PORT_OFFSET_KEY = "port.offset"; + + /** Address to bind to, all interfaces by default. */ + public static final String HOST_KEY = "host"; + + /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */ + public static final String JVM_METRICS_KEY = "jvm.metrics"; + + static final int DEFAULT_PORT_OFFSET = 1000; + + /** Port used when the worker port is unknown, e.g. in local mode. */ + static final int DEFAULT_PORT = 9400; + + /** System property set by the supervisor when launching a worker. */ + private static final String WORKER_PORT_PROPERTY = "worker.port"; + + /** Endpoints by port, shared by the reporters within a JVM. */ + private static final Map<Integer, Endpoint> ENDPOINTS = new HashMap<>(); + + private StormMetricsCollector.Source source; + private int port; + private String host; + private boolean jvmMetrics; + + @Override + public void prepare(MetricRegistryProvider metricRegistryProvider, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + String topology = ObjectReader.getString(topoConf.get(Config.TOPOLOGY_NAME), + ObjectReader.getString(topoConf.get(Config.STORM_ID), null)); + source = new StormMetricsCollector.Source(metricRegistryProvider, topology, + ScheduledStormReporter.getMetricsFilter(reporterConf)); + port = resolvePort(reporterConf, System.getProperty(WORKER_PORT_PROPERTY)); + host = ObjectReader.getString(reporterConf.get(HOST_KEY), null); + jvmMetrics = ObjectReader.getBoolean(reporterConf.get(JVM_METRICS_KEY), true); + } + + @Override + public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + throw new UnsupportedOperationException( + "PrometheusStormReporter needs a MetricRegistryProvider to get the task " + + "dimensions"); + } + + static int resolvePort(Map<String, Object> reporterConf, String workerPort) { + Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); + if (port != null) { + return port; + } + if (workerPort == null) { + return DEFAULT_PORT; + } + int offset = ObjectReader.getInt(reporterConf.get(PORT_OFFSET_KEY), DEFAULT_PORT_OFFSET); + return Integer.parseInt(workerPort) + offset; + } + + @Override + public void start() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + try { + endpoint = new Endpoint(port, host, jvmMetrics); + } catch (IOException e) { + LOG.error("Could not serve the metrics on port {}", port, e); + return; + } Review Comment: Only `IOException` is caught here. A resolved port above 65535 (e.g. a slot port above 64535 with the default `port.offset` of 1000, or a bad explicit `port`) makes `InetSocketAddress` throw `IllegalArgumentException` from `buildAndStart()`. Nothing in `StormMetricRegistry.startReporter` or `Worker.start` catches it, so the worker dies and the supervisor relaunches it into the same failure, while a bind failure is only logged. Catch `IOException | IllegalArgumentException` here (folded into the suggestion on lines 111-123 below, since they overlap). If you'd rather fail fast, range-check in `resolvePort` and throw with a message naming `port`/`port.offset`, instead of the bare `port out of range:65600`. ########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java: ########## @@ -0,0 +1,367 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricFilter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.ScheduledReporter; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.Timer; +import io.prometheus.metrics.model.registry.MultiCollector; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.CounterSnapshot.CounterDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.DataPointSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot.GaugeDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import io.prometheus.metrics.model.snapshots.MetricMetadata; +import io.prometheus.metrics.model.snapshots.MetricSnapshot; +import io.prometheus.metrics.model.snapshots.MetricSnapshots; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import io.prometheus.metrics.model.snapshots.Quantile; +import io.prometheus.metrics.model.snapshots.Quantiles; +import io.prometheus.metrics.model.snapshots.SummarySnapshot; +import io.prometheus.metrics.model.snapshots.SummarySnapshot.SummaryDataPointSnapshot; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.TaskMetricDimensions; +import org.apache.storm.metrics2.TaskMetricRepo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns the V2 metrics of one or more workers into Prometheus metric families. The dimensions + * Storm keeps for each task (topology, host, port, component, task and stream) become labels, so + * that a metric reported by several tasks ends up in a single family: + * + * <ul> + * <li>a scoped metric such as {@code fetcher_counter.status_200} becomes {@code + * fetcher_counter} with the label {@code scope="status_200"};</li> + * <li>a gauge whose value is a map, such as {@code status.count}, becomes {@code status_count} + * with a {@code key} label per entry;</li> + * <li>a Storm built-in metric such as {@code __ack-count-spout:default} becomes {@code + * storm_ack_count} with the labels {@code source_component="spout"} and {@code + * stream="default"}.</li> + * </ul> + * + * <p>Counters and meters are exposed as counters, histograms and timers as summaries along with a + * {@code _max} and a {@code _mean} gauge. Timers are converted to seconds. + */ +class StormMetricsCollector implements MultiCollector { + + private static final Logger LOG = LoggerFactory.getLogger(StormMetricsCollector.class); + + private static final double[] QUANTILES = {0.5, 0.75, 0.95, 0.98, 0.99, 0.999}; + + private static final double NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1); + + /** Prefix of the metrics Storm registers for each task. */ + private static final String STORM_PREFIX = "__"; + + /** + * Storm registers a gauge with this suffix next to each of its counters. Prometheus computes + * rates from the counter itself. + */ + private static final String RATE_SUFFIX = ".m1_rate"; + + private static final Pattern INVALID_CHARS = Pattern.compile("[^a-zA-Z0-9_]"); + + /** The metrics of a worker. */ + static final class Source { + private final MetricRegistryProvider provider; + private final String topology; + private final MetricFilter filter; + + Source(MetricRegistryProvider provider, String topology, MetricFilter filter) { + this.provider = provider; + this.topology = topology; + this.filter = filter; + } + } + + /** Family name and labels derived from the name of a Storm metric. */ + record Name(String family, Labels labels) { + } + + private enum Type { + COUNTER, + GAUGE, + SUMMARY + } + + private final List<Source> sources = new CopyOnWriteArrayList<>(); + + /** Families reported with conflicting types, so that the conflict is logged only once. */ + private final Set<String> conflicts = ConcurrentHashMap.newKeySet(); + + /** {@link TaskMetricRepo} only hands its metrics over to a {@link ScheduledReporter}. */ + private final Capture capture = new Capture(); + + void add(Source source) { + sources.add(source); + } + + void remove(Source source) { + sources.remove(source); + } + + boolean isEmpty() { + return sources.isEmpty(); + } + + @Override + public synchronized MetricSnapshots collect() { + Families families = new Families(conflicts); + for (Source source : sources) { + for (Map.Entry<TaskMetricDimensions, TaskMetricRepo> task + : source.provider.getTaskMetrics().entrySet()) { + Map<String, String> dimensions = task.getKey().getDimensions(); + capture.families = families; + capture.labels = taskLabels(source.topology, dimensions); + capture.stream = dimensions.get("streamId"); + task.getValue().report(capture, source.filter); + } + } + capture.families = null; + return families.build(); + } + + static Labels taskLabels(String topology, Map<String, String> dimensions) { + Labels.Builder builder = Labels.builder(); + addLabel(builder, "topology", topology); + addLabel(builder, "host", dimensions.get("hostname")); + addLabel(builder, "port", dimensions.get("port")); + addLabel(builder, "component", dimensions.get("componentId")); + addLabel(builder, "task", dimensions.get("taskid")); + addLabel(builder, "stream", dimensions.get("streamId")); + return builder.build(); + } + + private static void addLabel(Labels.Builder builder, String name, String value) { + if (value != null) { + builder.label(name, value); + } + } + + /** + * Maps the name of a Storm metric to a family name and labels. + * + * @param metric the name of the metric as registered with Storm + * @param stream the stream the metric was registered for, if any + */ + static Name parse(String metric, String stream) { + if (metric.startsWith(STORM_PREFIX)) { + String base = metric.substring(STORM_PREFIX.length()); + Labels labels = Labels.EMPTY; + if (stream != null && base.endsWith("-" + stream)) { + // e.g. __emit-count-default + base = base.substring(0, base.length() - stream.length() - 1); + } else if (stream != null && base.endsWith(":" + stream)) { + // e.g. __ack-count-spout:default, the names of Storm's + // metrics are made of two words + base = base.substring(0, base.length() - stream.length() - 1); + int dash = base.indexOf('-', base.indexOf('-') + 1); + if (dash > 0) { + labels = Labels.of("source_component", base.substring(dash + 1)); + base = base.substring(0, dash); + } + } + return new Name("storm_" + sanitize(base), labels); + } + int dot = metric.indexOf('.'); + if (dot > 0 && dot < metric.length() - 1) { + return new Name(sanitize(metric.substring(0, dot)), + Labels.of("scope", metric.substring(dot + 1))); + } + return new Name(sanitize(metric), Labels.EMPTY); + } + + static String sanitize(String name) { + String sanitized = INVALID_CHARS.matcher(name).replaceAll("_"); + if (sanitized.isEmpty() || Character.isDigit(sanitized.charAt(0))) { + sanitized = "_" + sanitized; + } + // removes suffixes such as _total which are reserved + return PrometheusNaming.sanitizeMetricName(sanitized); + } + + private static Double toDouble(Object value) { + if (value instanceof Number number) { + return number.doubleValue(); + } + if (value instanceof Boolean bool) { + return bool ? 1d : 0d; + } + return null; + } + + private static GaugeDataPointSnapshot gaugePoint(Labels labels, double value) { + return GaugeDataPointSnapshot.builder().labels(labels).value(value).build(); + } + + /** Converts the metrics of a task. */ + private static final class Capture extends ScheduledReporter { + + private Families families; + private Labels labels; + private String stream; + + Capture() { + super(new MetricRegistry(), "prometheus", MetricFilter.ALL, TimeUnit.SECONDS, + TimeUnit.SECONDS); + } + + @Override + @SuppressWarnings("rawtypes") + public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters, + SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, + SortedMap<String, Timer> timers) { + gauges.forEach((name, gauge) -> { + // gauges call arbitrary code, one failing must not break the scrape + try { + gauge(name, gauge.getValue()); + } catch (RuntimeException e) { + LOG.debug("Could not read gauge {}", name, e); + } + }); + counters.forEach((name, counter) -> counter(name, counter.getCount())); + meters.forEach((name, meter) -> counter(name, meter.getCount())); Review Comment: Gauges are converted first, so a gauge `X.<scope>` takes family `X` as GAUGE before counter or meter `X` gets there. `Families.add` then drops the counter, and the only trace is one WARN. You can hit this through `TopologyContext` with `registerCounter("pages")` plus `registerGauge("pages.foo", ...)` on the same task. A non-`__` `RateCounter` hits it too, because its `.m1_rate` gauge gets past the skip on line 262. Converting counters and meters first keeps the counter, and skipping any `.m1_rate` gauge that has a matching counter covers `RateCounter`s outside the `__` namespace. ```suggestion 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); } }); ``` A test with a counter `pages` and a gauge `pages.foo` on the same task, asserting that the counter values are exported, would cover this. ########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java: ########## @@ -0,0 +1,367 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricFilter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.ScheduledReporter; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.Timer; +import io.prometheus.metrics.model.registry.MultiCollector; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.CounterSnapshot.CounterDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.DataPointSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot.GaugeDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import io.prometheus.metrics.model.snapshots.MetricMetadata; +import io.prometheus.metrics.model.snapshots.MetricSnapshot; +import io.prometheus.metrics.model.snapshots.MetricSnapshots; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import io.prometheus.metrics.model.snapshots.Quantile; +import io.prometheus.metrics.model.snapshots.Quantiles; +import io.prometheus.metrics.model.snapshots.SummarySnapshot; +import io.prometheus.metrics.model.snapshots.SummarySnapshot.SummaryDataPointSnapshot; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.TaskMetricDimensions; +import org.apache.storm.metrics2.TaskMetricRepo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns the V2 metrics of one or more workers into Prometheus metric families. The dimensions + * Storm keeps for each task (topology, host, port, component, task and stream) become labels, so + * that a metric reported by several tasks ends up in a single family: + * + * <ul> + * <li>a scoped metric such as {@code fetcher_counter.status_200} becomes {@code + * fetcher_counter} with the label {@code scope="status_200"};</li> + * <li>a gauge whose value is a map, such as {@code status.count}, becomes {@code status_count} + * with a {@code key} label per entry;</li> + * <li>a Storm built-in metric such as {@code __ack-count-spout:default} becomes {@code + * storm_ack_count} with the labels {@code source_component="spout"} and {@code + * stream="default"}.</li> + * </ul> + * + * <p>Counters and meters are exposed as counters, histograms and timers as summaries along with a + * {@code _max} and a {@code _mean} gauge. Timers are converted to seconds. + */ +class StormMetricsCollector implements MultiCollector { + + private static final Logger LOG = LoggerFactory.getLogger(StormMetricsCollector.class); + + private static final double[] QUANTILES = {0.5, 0.75, 0.95, 0.98, 0.99, 0.999}; + + private static final double NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1); + + /** Prefix of the metrics Storm registers for each task. */ + private static final String STORM_PREFIX = "__"; + + /** + * Storm registers a gauge with this suffix next to each of its counters. Prometheus computes + * rates from the counter itself. + */ + private static final String RATE_SUFFIX = ".m1_rate"; + + private static final Pattern INVALID_CHARS = Pattern.compile("[^a-zA-Z0-9_]"); + + /** The metrics of a worker. */ + static final class Source { + private final MetricRegistryProvider provider; + private final String topology; + private final MetricFilter filter; + + Source(MetricRegistryProvider provider, String topology, MetricFilter filter) { + this.provider = provider; + this.topology = topology; + this.filter = filter; + } + } + + /** Family name and labels derived from the name of a Storm metric. */ + record Name(String family, Labels labels) { + } + + private enum Type { + COUNTER, + GAUGE, + SUMMARY + } + + private final List<Source> sources = new CopyOnWriteArrayList<>(); + + /** Families reported with conflicting types, so that the conflict is logged only once. */ + private final Set<String> conflicts = ConcurrentHashMap.newKeySet(); + + /** {@link TaskMetricRepo} only hands its metrics over to a {@link ScheduledReporter}. */ + private final Capture capture = new Capture(); + + void add(Source source) { + sources.add(source); + } + + void remove(Source source) { + sources.remove(source); + } + + boolean isEmpty() { + return sources.isEmpty(); + } + + @Override + public synchronized MetricSnapshots collect() { + Families families = new Families(conflicts); + for (Source source : sources) { + for (Map.Entry<TaskMetricDimensions, TaskMetricRepo> task + : source.provider.getTaskMetrics().entrySet()) { + Map<String, String> dimensions = task.getKey().getDimensions(); + capture.families = families; + capture.labels = taskLabels(source.topology, dimensions); + capture.stream = dimensions.get("streamId"); + task.getValue().report(capture, source.filter); + } + } + capture.families = null; + return families.build(); + } + + static Labels taskLabels(String topology, Map<String, String> dimensions) { + Labels.Builder builder = Labels.builder(); + addLabel(builder, "topology", topology); + addLabel(builder, "host", dimensions.get("hostname")); + addLabel(builder, "port", dimensions.get("port")); + addLabel(builder, "component", dimensions.get("componentId")); + addLabel(builder, "task", dimensions.get("taskid")); + addLabel(builder, "stream", dimensions.get("streamId")); + return builder.build(); + } + + private static void addLabel(Labels.Builder builder, String name, String value) { + if (value != null) { + builder.label(name, value); + } + } + + /** + * Maps the name of a Storm metric to a family name and labels. + * + * @param metric the name of the metric as registered with Storm + * @param stream the stream the metric was registered for, if any + */ + static Name parse(String metric, String stream) { + if (metric.startsWith(STORM_PREFIX)) { + String base = metric.substring(STORM_PREFIX.length()); + Labels labels = Labels.EMPTY; + if (stream != null && base.endsWith("-" + stream)) { + // e.g. __emit-count-default + base = base.substring(0, base.length() - stream.length() - 1); + } else if (stream != null && base.endsWith(":" + stream)) { + // e.g. __ack-count-spout:default, the names of Storm's + // metrics are made of two words + base = base.substring(0, base.length() - stream.length() - 1); + int dash = base.indexOf('-', base.indexOf('-') + 1); + if (dash > 0) { + labels = Labels.of("source_component", base.substring(dash + 1)); + base = base.substring(0, dash); + } + } Review Comment: The Netty client gauges from `Client` (`__send-iconnection-{reconnects,sent,pending,lostOnSend}-<host>:<port>`, on by default via `topology.enable.send.iconnection.metrics`) are registered on `__system` with no stream, so they fall through to line 194 and the remote host and port get sanitized into the family name, e.g. `storm_send_iconnection_sent_worker2_example_com_6701`. That makes 4 families per remote connection, which you can't aggregate without regex on `__name__`. Split the trailing `-<host>:<port>` into labels instead: ```suggestion } 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); } } ``` This also needs the `ICONNECTION_PREFIX` constant next to `STORM_PREFIX`, and a test (see diff). The README could list the resulting `storm_send_iconnection_*` families with their `remote_host`/`remote_port` labels. <details><summary>Other changes needed</summary> ```diff --- 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 @@ -83,6 +83,9 @@ /** 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. --- 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 @@ 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")); ``` </details> ########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/StormMetricsCollector.java: ########## @@ -0,0 +1,367 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricFilter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.ScheduledReporter; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.Timer; +import io.prometheus.metrics.model.registry.MultiCollector; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.CounterSnapshot.CounterDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.DataPointSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot; +import io.prometheus.metrics.model.snapshots.GaugeSnapshot.GaugeDataPointSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import io.prometheus.metrics.model.snapshots.MetricMetadata; +import io.prometheus.metrics.model.snapshots.MetricSnapshot; +import io.prometheus.metrics.model.snapshots.MetricSnapshots; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import io.prometheus.metrics.model.snapshots.Quantile; +import io.prometheus.metrics.model.snapshots.Quantiles; +import io.prometheus.metrics.model.snapshots.SummarySnapshot; +import io.prometheus.metrics.model.snapshots.SummarySnapshot.SummaryDataPointSnapshot; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.regex.Pattern; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.TaskMetricDimensions; +import org.apache.storm.metrics2.TaskMetricRepo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns the V2 metrics of one or more workers into Prometheus metric families. The dimensions + * Storm keeps for each task (topology, host, port, component, task and stream) become labels, so + * that a metric reported by several tasks ends up in a single family: + * + * <ul> + * <li>a scoped metric such as {@code fetcher_counter.status_200} becomes {@code + * fetcher_counter} with the label {@code scope="status_200"};</li> + * <li>a gauge whose value is a map, such as {@code status.count}, becomes {@code status_count} + * with a {@code key} label per entry;</li> + * <li>a Storm built-in metric such as {@code __ack-count-spout:default} becomes {@code + * storm_ack_count} with the labels {@code source_component="spout"} and {@code + * stream="default"}.</li> + * </ul> + * + * <p>Counters and meters are exposed as counters, histograms and timers as summaries along with a + * {@code _max} and a {@code _mean} gauge. Timers are converted to seconds. + */ +class StormMetricsCollector implements MultiCollector { + + private static final Logger LOG = LoggerFactory.getLogger(StormMetricsCollector.class); + + private static final double[] QUANTILES = {0.5, 0.75, 0.95, 0.98, 0.99, 0.999}; + + private static final double NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1); + + /** Prefix of the metrics Storm registers for each task. */ + private static final String STORM_PREFIX = "__"; + + /** + * Storm registers a gauge with this suffix next to each of its counters. Prometheus computes + * rates from the counter itself. + */ + private static final String RATE_SUFFIX = ".m1_rate"; + + private static final Pattern INVALID_CHARS = Pattern.compile("[^a-zA-Z0-9_]"); + + /** The metrics of a worker. */ + static final class Source { + private final MetricRegistryProvider provider; + private final String topology; + private final MetricFilter filter; + + Source(MetricRegistryProvider provider, String topology, MetricFilter filter) { + this.provider = provider; + this.topology = topology; + this.filter = filter; + } + } + + /** Family name and labels derived from the name of a Storm metric. */ + record Name(String family, Labels labels) { + } + + private enum Type { + COUNTER, + GAUGE, + SUMMARY + } + + private final List<Source> sources = new CopyOnWriteArrayList<>(); + + /** Families reported with conflicting types, so that the conflict is logged only once. */ + private final Set<String> conflicts = ConcurrentHashMap.newKeySet(); + + /** {@link TaskMetricRepo} only hands its metrics over to a {@link ScheduledReporter}. */ + private final Capture capture = new Capture(); + + void add(Source source) { + sources.add(source); + } + + void remove(Source source) { + sources.remove(source); + } + + boolean isEmpty() { + return sources.isEmpty(); + } + + @Override + public synchronized MetricSnapshots collect() { + Families families = new Families(conflicts); + for (Source source : sources) { + for (Map.Entry<TaskMetricDimensions, TaskMetricRepo> task + : source.provider.getTaskMetrics().entrySet()) { + Map<String, String> dimensions = task.getKey().getDimensions(); + capture.families = families; + capture.labels = taskLabels(source.topology, dimensions); + capture.stream = dimensions.get("streamId"); + task.getValue().report(capture, source.filter); + } + } + capture.families = null; + return families.build(); + } + + static Labels taskLabels(String topology, Map<String, String> dimensions) { + Labels.Builder builder = Labels.builder(); + addLabel(builder, "topology", topology); + addLabel(builder, "host", dimensions.get("hostname")); + addLabel(builder, "port", dimensions.get("port")); + addLabel(builder, "component", dimensions.get("componentId")); + addLabel(builder, "task", dimensions.get("taskid")); + addLabel(builder, "stream", dimensions.get("streamId")); + return builder.build(); + } + + private static void addLabel(Labels.Builder builder, String name, String value) { + if (value != null) { + builder.label(name, value); + } + } + + /** + * Maps the name of a Storm metric to a family name and labels. + * + * @param metric the name of the metric as registered with Storm + * @param stream the stream the metric was registered for, if any + */ + static Name parse(String metric, String stream) { + if (metric.startsWith(STORM_PREFIX)) { + String base = metric.substring(STORM_PREFIX.length()); + Labels labels = Labels.EMPTY; + if (stream != null && base.endsWith("-" + stream)) { + // e.g. __emit-count-default + base = base.substring(0, base.length() - stream.length() - 1); + } else if (stream != null && base.endsWith(":" + stream)) { + // e.g. __ack-count-spout:default, the names of Storm's + // metrics are made of two words + base = base.substring(0, base.length() - stream.length() - 1); + int dash = base.indexOf('-', base.indexOf('-') + 1); + if (dash > 0) { + labels = Labels.of("source_component", base.substring(dash + 1)); + base = base.substring(0, dash); + } + } + return new Name("storm_" + sanitize(base), labels); + } + int dot = metric.indexOf('.'); + if (dot > 0 && dot < metric.length() - 1) { + return new Name(sanitize(metric.substring(0, dot)), + Labels.of("scope", metric.substring(dot + 1))); + } + return new Name(sanitize(metric), Labels.EMPTY); + } + + static String sanitize(String name) { + String sanitized = INVALID_CHARS.matcher(name).replaceAll("_"); + if (sanitized.isEmpty() || Character.isDigit(sanitized.charAt(0))) { + sanitized = "_" + sanitized; + } + // removes suffixes such as _total which are reserved + return PrometheusNaming.sanitizeMetricName(sanitized); + } + + private static Double toDouble(Object value) { + if (value instanceof Number number) { + return number.doubleValue(); + } + if (value instanceof Boolean bool) { + return bool ? 1d : 0d; + } + return null; + } + + private static GaugeDataPointSnapshot gaugePoint(Labels labels, double value) { + return GaugeDataPointSnapshot.builder().labels(labels).value(value).build(); + } + + /** Converts the metrics of a task. */ + private static final class Capture extends ScheduledReporter { + + private Families families; + private Labels labels; + private String stream; + + Capture() { + super(new MetricRegistry(), "prometheus", MetricFilter.ALL, TimeUnit.SECONDS, + TimeUnit.SECONDS); + } + + @Override + @SuppressWarnings("rawtypes") + public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters, + SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, + SortedMap<String, Timer> timers) { + gauges.forEach((name, gauge) -> { + // gauges call arbitrary code, one failing must not break the scrape + try { + gauge(name, gauge.getValue()); + } catch (RuntimeException e) { + LOG.debug("Could not read gauge {}", name, e); + } + }); + counters.forEach((name, counter) -> counter(name, counter.getCount())); + meters.forEach((name, meter) -> counter(name, meter.getCount())); + histograms.forEach((name, histogram) -> + summary(name, "", histogram.getCount(), histogram.getSnapshot(), 1)); + timers.forEach((name, timer) -> + summary(name, "_seconds", timer.getCount(), timer.getSnapshot(), + NANOS_PER_SECOND)); + } + + private void gauge(String metric, Object value) { + if (metric.startsWith(STORM_PREFIX) && metric.endsWith(RATE_SUFFIX)) { + return; + } + if (value instanceof Map<?, ?> map) { + String family = sanitize(metric); + map.forEach((key, entry) -> { + Double number = toDouble(entry); + if (number != null) { + Labels keyLabels = labels.add("key", String.valueOf(key)); + families.add(family, Type.GAUGE, gaugePoint(keyLabels, number)); + } + }); + return; + } + Double number = toDouble(value); + if (number == null) { + return; + } + Name name = parse(metric, stream); + families.add(name.family(), Type.GAUGE, + gaugePoint(labels.merge(name.labels()), number)); + } + + private void counter(String metric, long count) { + Name name = parse(metric, stream); + families.add(name.family(), Type.COUNTER, + CounterDataPointSnapshot.builder() + .labels(labels.merge(name.labels())) + .value(count) + .build()); + } + + private void summary(String metric, String unit, long count, Snapshot snapshot, + double scale) { + Name name = parse(metric, stream); + String family = name.family() + unit; + Labels pointLabels = labels.merge(name.labels()); + List<Quantile> quantiles = new ArrayList<>(QUANTILES.length); + for (double quantile : QUANTILES) { + quantiles.add(new Quantile(quantile, snapshot.getValue(quantile) / scale)); + } + families.add(family, Type.SUMMARY, + SummaryDataPointSnapshot.builder() + .labels(pointLabels) + .count(count) + .quantiles(Quantiles.of(quantiles)) + .build()); + families.add(family + "_max", Type.GAUGE, + gaugePoint(pointLabels, snapshot.getMax() / scale)); + families.add(family + "_mean", Type.GAUGE, + gaugePoint(pointLabels, snapshot.getMean() / scale)); + } + } + + /** Data points grouped by family. */ + private static final class Families { + + private final Map<String, Family> byName = new TreeMap<>(); + private final Set<String> conflicts; + + Families(Set<String> conflicts) { + this.conflicts = conflicts; + } + + void add(String name, Type type, DataPointSnapshot point) { + Family family = byName.computeIfAbsent(name, n -> new Family(type)); + if (family.type != type) { + if (conflicts.add(name)) { + LOG.warn("Metric family {} is reported both as a {} and a {}, " + + "ignoring the latter", name, family.type, type); + } + return; + } + family.points.putIfAbsent(point.getLabels(), point); + } + + MetricSnapshots build() { + MetricSnapshots.Builder builder = MetricSnapshots.builder(); + byName.forEach((name, family) -> builder.metricSnapshot(family.snapshot(name))); + return builder.build(); + } + } Review Comment: The first point seen in a scrape fixes the family type, and every point of another type is dropped. Across tasks that order is `ConcurrentHashMap` hash order. Within a task, gauges always come before counters, meters, histograms and timers. So with counter `fetcher.pages` in one bolt and gauge `fetcher.queue` in another, whichever task is iterated second loses all of its `fetcher` series on every scrape. Because `conflicts` is never cleared, that happens silently after the first warning. Prometheus cannot mix types in one family, but the resolution should not depend on iteration order. Keep the points per type and suffix every type on a collision, so nothing is dropped and the names stay stable: ```suggestion /** Data points grouped by family. */ private static final class Families { private final Map<String, Map<Type, Family>> byName = new TreeMap<>(); private final Set<String> conflicts; Families(Set<String> 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 {}_<type>", 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(); } } ``` This also needs `java.util.EnumMap` and `java.util.Locale` imports, plus a line in the README Naming section (diff below). A `collect()` test with a counter and a gauge under the same prefix in two tasks would cover it. <details><summary>Other changes needed</summary> ```diff --- 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,9 +38,11 @@ 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; --- a/external/storm-metrics-prometheus/README.md +++ b/external/storm-metrics-prometheus/README.md @@ -50,5 +50,7 @@ * 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 ``` </details> ########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java: ########## @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.MetricRegistry; +import io.prometheus.metrics.exporter.httpserver.HTTPServer; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.reporters.ScheduledStormReporter; +import org.apache.storm.metrics2.reporters.StormReporter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Metrics V2 reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * + * <pre> + * topology.metrics.reporters: + * - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter" + * port.offset: 1000 + * </pre> + * + * <p>Each worker listens on its own port, which is the worker port plus <code>port.offset</code>, + * e.g. 7700 for a worker on 6700. When the worker port is unknown, as in local mode, the endpoint + * listens on <code>port</code>, which can also be set explicitly. The workers of a local cluster + * share the same endpoint. + */ +public class PrometheusStormReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusStormReporter.class); + + /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */ + public static final String PORT_KEY = "port"; + + /** Added to the worker port to get the port to listen on, 1000 by default. */ + public static final String PORT_OFFSET_KEY = "port.offset"; + + /** Address to bind to, all interfaces by default. */ + public static final String HOST_KEY = "host"; + + /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */ + public static final String JVM_METRICS_KEY = "jvm.metrics"; + + static final int DEFAULT_PORT_OFFSET = 1000; + + /** Port used when the worker port is unknown, e.g. in local mode. */ + static final int DEFAULT_PORT = 9400; + + /** System property set by the supervisor when launching a worker. */ + private static final String WORKER_PORT_PROPERTY = "worker.port"; + + /** Endpoints by port, shared by the reporters within a JVM. */ + private static final Map<Integer, Endpoint> ENDPOINTS = new HashMap<>(); + + private StormMetricsCollector.Source source; + private int port; + private String host; + private boolean jvmMetrics; + + @Override + public void prepare(MetricRegistryProvider metricRegistryProvider, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + String topology = ObjectReader.getString(topoConf.get(Config.TOPOLOGY_NAME), + ObjectReader.getString(topoConf.get(Config.STORM_ID), null)); + source = new StormMetricsCollector.Source(metricRegistryProvider, topology, + ScheduledStormReporter.getMetricsFilter(reporterConf)); + port = resolvePort(reporterConf, System.getProperty(WORKER_PORT_PROPERTY)); + host = ObjectReader.getString(reporterConf.get(HOST_KEY), null); + jvmMetrics = ObjectReader.getBoolean(reporterConf.get(JVM_METRICS_KEY), true); + } + + @Override + public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + throw new UnsupportedOperationException( + "PrometheusStormReporter needs a MetricRegistryProvider to get the task " + + "dimensions"); + } + + static int resolvePort(Map<String, Object> reporterConf, String workerPort) { + Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); + if (port != null) { + return port; + } + if (workerPort == null) { + return DEFAULT_PORT; + } + int offset = ObjectReader.getInt(reporterConf.get(PORT_OFFSET_KEY), DEFAULT_PORT_OFFSET); + return Integer.parseInt(workerPort) + offset; + } + + @Override + public void start() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + try { + endpoint = new Endpoint(port, host, jvmMetrics); + } catch (IOException e) { + LOG.error("Could not serve the metrics on port {}", port, e); + return; + } + ENDPOINTS.put(port, endpoint); + LOG.info("Serving the metrics for Prometheus on port {}", + endpoint.server.getPort()); + } + endpoint.collector.add(source); + } + } + + @Override + public void stop() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + return; + } + endpoint.collector.remove(source); + if (endpoint.collector.isEmpty()) { + endpoint.server.close(); + ENDPOINTS.remove(port); + } + } + } + + @Override + public void close() { + stop(); + } + + /** Port the endpoint of this reporter is bound to, -1 if it is not running. */ + int getBoundPort() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + return endpoint == null ? -1 : endpoint.server.getPort(); + } + } + + private static final class Endpoint { + + private final StormMetricsCollector collector = new StormMetricsCollector(); + private final HTTPServer server; + + Endpoint(int port, String host, boolean jvmMetrics) throws IOException { + PrometheusRegistry registry = new PrometheusRegistry(); + registry.register(collector); + if (jvmMetrics) { + JvmMetrics.builder().register(registry); + } + HTTPServer.Builder builder = HTTPServer.builder().port(port).registry(registry); + if (host != null) { + builder.hostname(host); + } + server = builder.buildAndStart(); + } + } Review Comment: `JvmMetrics.builder().register(registry)` adds each new `PrometheusRegistry` to the static `JvmMetrics.REGISTERED` set, and nothing ever removes it. `JvmMemoryPoolAllocationMetrics` also adds a GC `NotificationListener` on every register and never removes it. So every endpoint create/close cycle in one JVM, e.g. local-mode submit/kill loops on 9400, leaves behind a registry, the JVM collectors and a GC listener. Register the JVM metrics once on a lazily created static registry and have each endpoint registry delegate to it. Closing an endpoint then leaks nothing. Needs `import io.prometheus.metrics.model.registry.MultiCollector;`. The cast avoids ambiguity between the `register(Collector)` and `register(MultiCollector)` overloads. ```suggestion 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) { 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); } } ``` <details><summary>Other changes needed</summary> ```diff --- 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,6 +15,7 @@ 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; ``` </details> ########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java: ########## @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.MetricRegistry; +import io.prometheus.metrics.exporter.httpserver.HTTPServer; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.reporters.ScheduledStormReporter; +import org.apache.storm.metrics2.reporters.StormReporter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Metrics V2 reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * + * <pre> + * topology.metrics.reporters: + * - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter" + * port.offset: 1000 + * </pre> + * + * <p>Each worker listens on its own port, which is the worker port plus <code>port.offset</code>, + * e.g. 7700 for a worker on 6700. When the worker port is unknown, as in local mode, the endpoint + * listens on <code>port</code>, which can also be set explicitly. The workers of a local cluster + * share the same endpoint. + */ +public class PrometheusStormReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusStormReporter.class); + + /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */ + public static final String PORT_KEY = "port"; + + /** Added to the worker port to get the port to listen on, 1000 by default. */ + public static final String PORT_OFFSET_KEY = "port.offset"; + + /** Address to bind to, all interfaces by default. */ + public static final String HOST_KEY = "host"; + + /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */ + public static final String JVM_METRICS_KEY = "jvm.metrics"; + + static final int DEFAULT_PORT_OFFSET = 1000; + + /** Port used when the worker port is unknown, e.g. in local mode. */ + static final int DEFAULT_PORT = 9400; + + /** System property set by the supervisor when launching a worker. */ + private static final String WORKER_PORT_PROPERTY = "worker.port"; + + /** Endpoints by port, shared by the reporters within a JVM. */ + private static final Map<Integer, Endpoint> ENDPOINTS = new HashMap<>(); + + private StormMetricsCollector.Source source; + private int port; + private String host; + private boolean jvmMetrics; + + @Override + public void prepare(MetricRegistryProvider metricRegistryProvider, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + String topology = ObjectReader.getString(topoConf.get(Config.TOPOLOGY_NAME), + ObjectReader.getString(topoConf.get(Config.STORM_ID), null)); + source = new StormMetricsCollector.Source(metricRegistryProvider, topology, + ScheduledStormReporter.getMetricsFilter(reporterConf)); + port = resolvePort(reporterConf, System.getProperty(WORKER_PORT_PROPERTY)); + host = ObjectReader.getString(reporterConf.get(HOST_KEY), null); + jvmMetrics = ObjectReader.getBoolean(reporterConf.get(JVM_METRICS_KEY), true); + } + + @Override + public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + throw new UnsupportedOperationException( + "PrometheusStormReporter needs a MetricRegistryProvider to get the task " + + "dimensions"); + } + + static int resolvePort(Map<String, Object> reporterConf, String workerPort) { + Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); + if (port != null) { + return port; + } Review Comment: An explicit `port` wins even when `worker.port` is set, so on a real cluster every worker on a host resolves to the same port. Only the first one binds. The others hit the `IOException` branch in `start()`, log an ERROR and keep running with no endpoint, and nothing else reports it, so Prometheus silently loses those workers. The README only says `port` "takes precedence over `port.offset`". At minimum, warn when `port` is used under a supervisor and document that it is meant for local mode or single-slot hosts. Rethrowing from `start()` would be louder but would fail worker startup, so I'd keep the log there. ```suggestion 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; } ``` The README needs an update as well (see extra changes). <details><summary>Other changes needed</summary> ```diff --- a/external/storm-metrics-prometheus/README.md +++ b/external/storm-metrics-prometheus/README.md @@ -34 +34 @@ -| `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. | ``` </details> ########## external/storm-metrics-prometheus/src/main/java/org/apache/storm/metrics/prometheus/PrometheusStormReporter.java: ########## @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.metrics.prometheus; + +import com.codahale.metrics.MetricRegistry; +import io.prometheus.metrics.exporter.httpserver.HTTPServer; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.reporters.ScheduledStormReporter; +import org.apache.storm.metrics2.reporters.StormReporter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Metrics V2 reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * + * <pre> + * topology.metrics.reporters: + * - class: "org.apache.storm.metrics.prometheus.PrometheusStormReporter" + * port.offset: 1000 + * </pre> + * + * <p>Each worker listens on its own port, which is the worker port plus <code>port.offset</code>, + * e.g. 7700 for a worker on 6700. When the worker port is unknown, as in local mode, the endpoint + * listens on <code>port</code>, which can also be set explicitly. The workers of a local cluster + * share the same endpoint. + */ +public class PrometheusStormReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusStormReporter.class); + + /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */ + public static final String PORT_KEY = "port"; + + /** Added to the worker port to get the port to listen on, 1000 by default. */ + public static final String PORT_OFFSET_KEY = "port.offset"; + + /** Address to bind to, all interfaces by default. */ + public static final String HOST_KEY = "host"; + + /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */ + public static final String JVM_METRICS_KEY = "jvm.metrics"; + + static final int DEFAULT_PORT_OFFSET = 1000; + + /** Port used when the worker port is unknown, e.g. in local mode. */ + static final int DEFAULT_PORT = 9400; + + /** System property set by the supervisor when launching a worker. */ + private static final String WORKER_PORT_PROPERTY = "worker.port"; + + /** Endpoints by port, shared by the reporters within a JVM. */ + private static final Map<Integer, Endpoint> ENDPOINTS = new HashMap<>(); + + private StormMetricsCollector.Source source; + private int port; + private String host; + private boolean jvmMetrics; + + @Override + public void prepare(MetricRegistryProvider metricRegistryProvider, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + String topology = ObjectReader.getString(topoConf.get(Config.TOPOLOGY_NAME), + ObjectReader.getString(topoConf.get(Config.STORM_ID), null)); + source = new StormMetricsCollector.Source(metricRegistryProvider, topology, + ScheduledStormReporter.getMetricsFilter(reporterConf)); + port = resolvePort(reporterConf, System.getProperty(WORKER_PORT_PROPERTY)); + host = ObjectReader.getString(reporterConf.get(HOST_KEY), null); + jvmMetrics = ObjectReader.getBoolean(reporterConf.get(JVM_METRICS_KEY), true); + } + + @Override + public void prepare(MetricRegistry metricsRegistry, Map<String, Object> topoConf, + Map<String, Object> reporterConf) { + throw new UnsupportedOperationException( + "PrometheusStormReporter needs a MetricRegistryProvider to get the task " + + "dimensions"); + } + + static int resolvePort(Map<String, Object> reporterConf, String workerPort) { + Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); + if (port != null) { + return port; + } + if (workerPort == null) { + return DEFAULT_PORT; + } + int offset = ObjectReader.getInt(reporterConf.get(PORT_OFFSET_KEY), DEFAULT_PORT_OFFSET); + return Integer.parseInt(workerPort) + offset; + } + + @Override + public void start() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + try { + endpoint = new Endpoint(port, host, jvmMetrics); + } catch (IOException e) { + LOG.error("Could not serve the metrics on port {}", port, e); + return; + } + ENDPOINTS.put(port, endpoint); + LOG.info("Serving the metrics for Prometheus on port {}", + endpoint.server.getPort()); + } + endpoint.collector.add(source); Review Comment: `host` and `jvm.metrics` only take effect for the reporter that creates the endpoint. A later reporter on the same port (two topologies in one `LocalCluster`, which all default to 9400) has its values dropped, and nothing is logged. Either document that the first reporter's settings apply, or warn when they differ. ```suggestion 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); ``` `Endpoint` also has to keep `host` and `jvmMetrics`, and `java.util.Objects` needs an import (see the diff). The catch also includes `IllegalArgumentException`, see the comment on lines 113-118. If the `Endpoint` suggestion further down is applied as well, the two fields go into that version. <details><summary>Other changes needed</summary> ```diff --- 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 @@ -19,6 +19,7 @@ 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; @@ -156,8 +157,12 @@ 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) { ``` </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
