diff --git a/THIRD-PARTY.txt b/THIRD-PARTY.txt index 0078ae2ed..37e8e32a7 100644 --- a/THIRD-PARTY.txt +++ b/THIRD-PARTY.txt @@ -270,6 +270,18 @@ List of third-party dependencies grouped by their license type. * Playwright - Driver (com.microsoft.playwright:driver:1.63.0 - https://github.com/microsoft/playwright-java/driver) * Playwright - Main Library (com.microsoft.playwright:playwright:1.63.0 - https://github.com/microsoft/playwright-java/playwright) * Playwright - Node.js For All Platforms (com.microsoft.playwright:driver-bundle:1.63.0 - https://github.com/microsoft/playwright-java/driver-bundle) + * Prometheus Metrics Config (io.prometheus:prometheus-metrics-config:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-config) + * Prometheus Metrics Core (io.prometheus:prometheus-metrics-core:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-core) + * Prometheus Metrics Exporter - Common (io.prometheus:prometheus-metrics-exporter-common:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-common) + * Prometheus Metrics Exporter - HTTP Server (io.prometheus:prometheus-metrics-exporter-httpserver:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exporter-httpserver) + * Prometheus Metrics Exposition Formats (io.prometheus:prometheus-metrics-exposition-formats:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-formats) + * Prometheus Metrics Exposition Text Formats (io.prometheus:prometheus-metrics-exposition-textformats:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-exposition-textformats) + * Prometheus Metrics Instrumentation - JVM (io.prometheus:prometheus-metrics-instrumentation-jvm:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-instrumentation-jvm) + * Prometheus Metrics Model (io.prometheus:prometheus-metrics-model:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-model) + * Prometheus Metrics Tracer Common (io.prometheus:prometheus-metrics-tracer-common:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-common) + * Prometheus Metrics Tracer Initializer (io.prometheus:prometheus-metrics-tracer-initializer:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-initializer) + * Prometheus Metrics Tracer OpenTelemetry (io.prometheus:prometheus-metrics-tracer-otel:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel) + * Prometheus Metrics Tracer OpenTelemetry Agent (io.prometheus:prometheus-metrics-tracer-otel-agent:1.8.0 - http://github.com/prometheus/client_java/client_java/prometheus-metrics-tracer/prometheus-metrics-tracer-otel-agent) * proto-google-common-protos (com.google.api.grpc:proto-google-common-protos:2.59.2 - https://github.com/googleapis/sdk-platform-java) * rank-eval (org.opensearch.plugin:rank-eval-client:2.19.6 - https://github.com/opensearch-project/OpenSearch.git) * rest (org.opensearch.client:opensearch-rest-client:2.19.6 - https://github.com/opensearch-project/OpenSearch.git) diff --git a/external/prometheus/README.md b/external/prometheus/README.md new file mode 100644 index 000000000..0304b430d --- /dev/null +++ b/external/prometheus/README.md @@ -0,0 +1,56 @@ +# stormcrawler-prometheus + +A [Storm V2 metrics](https://storm.apache.org/releases/current/metrics_v2.html) reporter which serves the metrics of each worker over HTTP for [Prometheus](https://prometheus.io/) to scrape. + +Unlike the generic JMX or Graphite reporters, it uses the dimensions Storm keeps for each task as labels, so a metric reported by several tasks ends up in one family: + +``` +fetcher_counter_total{topology="crawl",host="worker1",port="6700",component="fetcher",task="3",scope="status_200"} 1234.0 +status_count{topology="crawl",host="worker1",port="6700",component="status",task="4",key="FETCHED"} 56789.0 +storm_ack_count_total{topology="crawl",host="worker1",port="6700",component="status",task="4",source_component="spout",stream="default"} 42.0 +``` + +## Configuration + +Add the dependency to your topology and set the following in the configuration: + +```yaml + # StormCrawler's metrics must be registered with the V2 API + stormcrawler.metrics.version: "v2" + + topology.metrics.reporters: + - class: "org.apache.stormcrawler.prometheus.PrometheusReporter" + # each worker listens on its port + offset, e.g. 7700 for 6700 + port.offset: 1000 +``` + +| Key | Default | Description | +|---|---|---| +| `port` | | Port to listen on, takes precedence over `port.offset`. Used as is in local mode, where it defaults to 9400. | +| `port.offset` | 1000 | Added to the worker port to get the port to listen on. | +| `host` | all interfaces | Address to bind to. | +| `jvm.metrics` | true | Also expose the JVM metrics (memory, GC, threads, `process_start_time_seconds`). | +| `filter` | | Storm metrics filter, as for the other V2 reporters. | + +The workers of a local cluster share a single endpoint. + +## Naming + +* Scoped metrics (`name.scope`) become the family `name` with a `scope` label. Only the first dot separates the scope, so `fetcher_counter.robots.fetched` has `scope="robots.fetched"`. +* Gauges whose value is a map, such as `status.count`, become one family named after the whole metric (`status_count`) with a `key` label per entry. +* Storm's built-in metrics (`__ack-count-spout:default`) become `storm_ack_count` with `source_component` and `stream` labels. The `.m1_rate` gauges Storm registers next to its counters are skipped, use `rate()` on the counter instead. +* Counters and meters are exposed as counters, so `fetcher_average_persec.bytes_fetched_perSec` becomes `fetcher_average_persec_total{scope="bytes_fetched_perSec"}` and `rate()` gives the bytes per second. +* Histograms and timers are exposed as summaries plus a `_max` and a `_mean` gauge. Timers are in seconds. + +## Prometheus + +Scrape every worker port of every supervisor, e.g. with 4 slots per supervisor: + +```yaml +scrape_configs: + - job_name: stormcrawler + static_configs: + - targets: ["supervisor1:7700", "supervisor1:7701", "supervisor1:7702", "supervisor1:7703"] +``` + +A target is only up while a worker runs on the corresponding slot. diff --git a/external/prometheus/pom.xml b/external/prometheus/pom.xml new file mode 100644 index 000000000..d6698cb20 --- /dev/null +++ b/external/prometheus/pom.xml @@ -0,0 +1,62 @@ + + + + + + 4.0.0 + + + org.apache.stormcrawler + stormcrawler-external + 4.0.0-SNAPSHOT + ../pom.xml + + + stormcrawler-prometheus + jar + + stormcrawler-prometheus + https://github.com/apache/stormcrawler/tree/master/external/prometheus + Prometheus reporter for the Storm V2 metrics + + + 1.8.0 + + + + + io.prometheus + prometheus-metrics-core + ${prometheus.version} + + + + io.prometheus + prometheus-metrics-exporter-httpserver + ${prometheus.version} + + + + io.prometheus + prometheus-metrics-instrumentation-jvm + ${prometheus.version} + + + diff --git a/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java new file mode 100644 index 000000000..ae300a6be --- /dev/null +++ b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/PrometheusReporter.java @@ -0,0 +1,188 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.prometheus; + +import com.codahale.metrics.MetricRegistry; +import io.prometheus.metrics.exporter.httpserver.HTTPServer; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.metrics2.MetricRegistryProvider; +import org.apache.storm.metrics2.reporters.ScheduledStormReporter; +import org.apache.storm.metrics2.reporters.StormReporter; +import org.apache.storm.utils.ObjectReader; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Storm V2 metrics reporter serving the metrics of a worker over HTTP for Prometheus to scrape. + * Requires stormcrawler.metrics.version to be set to "v2" or "both" for the + * StormCrawler metrics to be included. + * + * 
+ * topology.metrics.reporters:
+ *   - class: "org.apache.stormcrawler.prometheus.PrometheusReporter"
+ *     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 PrometheusReporter implements StormReporter { + + private static final Logger LOG = LoggerFactory.getLogger(PrometheusReporter.class); + + /** Port to listen on, takes precedence over {@link #PORT_OFFSET_KEY}. */ + public static final String PORT_KEY = "port"; + + /** Added to the worker port to get the port to listen on, 1000 by default. */ + public static final String PORT_OFFSET_KEY = "port.offset"; + + /** Address to bind to, all interfaces by default. */ + public static final String HOST_KEY = "host"; + + /** Whether to expose the JVM metrics (memory, GC, threads, start time), true by default. */ + public static final String JVM_METRICS_KEY = "jvm.metrics"; + + static final int DEFAULT_PORT_OFFSET = 1000; + + /** Port used when the worker port is unknown, e.g. in local mode. */ + static final int DEFAULT_PORT = 9400; + + /** System property set by the supervisor when launching a worker. */ + private static final String WORKER_PORT_PROPERTY = "worker.port"; + + /** Endpoints by port, shared by the reporters within a JVM. */ + private static final Map 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( + "PrometheusReporter needs a MetricRegistryProvider to get the task dimensions"); + } + + static int resolvePort(Map reporterConf, String workerPort) { + Integer port = ObjectReader.getInt(reporterConf.get(PORT_KEY), null); + if (port != null) { + return port; + } + if (workerPort == null) { + return DEFAULT_PORT; + } + int offset = ObjectReader.getInt(reporterConf.get(PORT_OFFSET_KEY), DEFAULT_PORT_OFFSET); + return Integer.parseInt(workerPort) + offset; + } + + @Override + public void start() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + try { + endpoint = new Endpoint(port, host, jvmMetrics); + } catch (IOException e) { + LOG.error("Could not serve the metrics on port {}", port, e); + return; + } + ENDPOINTS.put(port, endpoint); + LOG.info( + "Serving the metrics for Prometheus on port {}", endpoint.server.getPort()); + } + endpoint.collector.add(source); + } + } + + @Override + public void stop() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + if (endpoint == null) { + return; + } + endpoint.collector.remove(source); + if (endpoint.collector.isEmpty()) { + endpoint.server.close(); + ENDPOINTS.remove(port); + } + } + } + + @Override + public void close() { + stop(); + } + + /** Port the endpoint of this reporter is bound to, -1 if it is not running. */ + int getBoundPort() { + synchronized (ENDPOINTS) { + Endpoint endpoint = ENDPOINTS.get(port); + return endpoint == null ? -1 : endpoint.server.getPort(); + } + } + + private static final class Endpoint { + + private final StormMetricsCollector collector = new StormMetricsCollector(); + private final HTTPServer server; + + Endpoint(int port, String host, boolean jvmMetrics) throws IOException { + PrometheusRegistry registry = new PrometheusRegistry(); + registry.register(collector); + if (jvmMetrics) { + JvmMetrics.builder().register(registry); + } + HTTPServer.Builder builder = HTTPServer.builder().port(port).registry(registry); + if (host != null) { + builder.hostname(host); + } + server = builder.buildAndStart(); + } + } +} diff --git a/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java new file mode 100644 index 000000000..22cac7ff7 --- /dev/null +++ b/external/prometheus/src/main/java/org/apache/stormcrawler/prometheus/StormMetricsCollector.java @@ -0,0 +1,398 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.prometheus; + +import com.codahale.metrics.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 Storm 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 = "__"; + + /** + * Storm registers a gauge with this suffix next to each of its counters. Prometheus computes + * rates from the counter itself. + */ + private static final String RATE_SUFFIX = ".m1_rate"; + + private static final Pattern INVALID_CHARS = Pattern.compile("[^a-zA-Z0-9_]"); + + /** The metrics of a worker. */ + static final class Source { + private final MetricRegistryProvider provider; + private final String topology; + private final MetricFilter filter; + + Source(MetricRegistryProvider provider, String topology, MetricFilter filter) { + this.provider = provider; + this.topology = topology; + this.filter = filter; + } + } + + /** Family name and labels derived from the name of a Storm metric. */ + record Name(String family, Labels labels) {} + + private enum Type { + COUNTER, + GAUGE, + SUMMARY + } + + private final List sources = new CopyOnWriteArrayList<>(); + + /** Families reported with conflicting types, so that the conflict is logged only once. */ + private final Set conflicts = ConcurrentHashMap.newKeySet(); + + /** {@link TaskMetricRepo} only hands its metrics over to a {@link ScheduledReporter}. */ + private final Capture capture = new Capture(); + + void add(Source source) { + sources.add(source); + } + + void remove(Source source) { + sources.remove(source); + } + + boolean isEmpty() { + return sources.isEmpty(); + } + + @Override + public synchronized MetricSnapshots collect() { + Families families = new Families(conflicts); + for (Source source : sources) { + for (Map.Entry task : + source.provider.getTaskMetrics().entrySet()) { + Map dimensions = task.getKey().getDimensions(); + capture.families = families; + capture.labels = taskLabels(source.topology, dimensions); + capture.stream = dimensions.get("streamId"); + task.getValue().report(capture, source.filter); + } + } + capture.families = null; + return families.build(); + } + + static Labels taskLabels(String topology, Map dimensions) { + Labels.Builder builder = Labels.builder(); + addLabel(builder, "topology", topology); + addLabel(builder, "host", dimensions.get("hostname")); + addLabel(builder, "port", dimensions.get("port")); + addLabel(builder, "component", dimensions.get("componentId")); + addLabel(builder, "task", dimensions.get("taskid")); + addLabel(builder, "stream", dimensions.get("streamId")); + return builder.build(); + } + + private static void addLabel(Labels.Builder builder, String name, String value) { + if (value != null) { + builder.label(name, value); + } + } + + /** + * Maps the name of a Storm metric to a family name and labels. + * + * @param metric the name of the metric as registered with Storm + * @param stream the stream the metric was registered for, if any + */ + static Name parse(String metric, String stream) { + if (metric.startsWith(STORM_PREFIX)) { + String base = metric.substring(STORM_PREFIX.length()); + Labels labels = Labels.EMPTY; + if (stream != null && base.endsWith("-" + stream)) { + // e.g. __emit-count-default + base = base.substring(0, base.length() - stream.length() - 1); + } else if (stream != null && base.endsWith(":" + stream)) { + // e.g. __ack-count-spout:default, the names of Storm's + // metrics are made of two words + base = base.substring(0, base.length() - stream.length() - 1); + int dash = base.indexOf('-', base.indexOf('-') + 1); + if (dash > 0) { + labels = Labels.of("source_component", base.substring(dash + 1)); + base = base.substring(0, dash); + } + } + return new Name("storm_" + sanitize(base), labels); + } + int dot = metric.indexOf('.'); + if (dot > 0 && dot < metric.length() - 1) { + return new Name( + sanitize(metric.substring(0, dot)), + Labels.of("scope", metric.substring(dot + 1))); + } + return new Name(sanitize(metric), Labels.EMPTY); + } + + static String sanitize(String name) { + String sanitized = INVALID_CHARS.matcher(name).replaceAll("_"); + if (sanitized.isEmpty() || Character.isDigit(sanitized.charAt(0))) { + sanitized = "_" + sanitized; + } + // removes suffixes such as _total which are reserved + return PrometheusNaming.sanitizeMetricName(sanitized); + } + + private static Double toDouble(Object value) { + if (value instanceof Number number) { + return number.doubleValue(); + } + if (value instanceof Boolean bool) { + return bool ? 1d : 0d; + } + return null; + } + + private static GaugeDataPointSnapshot gaugePoint(Labels labels, double value) { + return GaugeDataPointSnapshot.builder().labels(labels).value(value).build(); + } + + /** Converts the metrics of a task. */ + private static final class Capture extends ScheduledReporter { + + private Families families; + private Labels labels; + private String stream; + + Capture() { + super( + new MetricRegistry(), + "prometheus", + MetricFilter.ALL, + TimeUnit.SECONDS, + TimeUnit.SECONDS); + } + + @Override + @SuppressWarnings("rawtypes") + public void report( + SortedMap gauges, + SortedMap counters, + SortedMap histograms, + SortedMap meters, + SortedMap timers) { + gauges.forEach( + (name, gauge) -> { + // gauges call arbitrary code, one failing must not break the scrape + try { + gauge(name, gauge.getValue()); + } catch (RuntimeException e) { + LOG.debug("Could not read gauge {}", name, e); + } + }); + counters.forEach((name, counter) -> counter(name, counter.getCount())); + meters.forEach((name, meter) -> counter(name, meter.getCount())); + histograms.forEach( + (name, histogram) -> + summary(name, "", histogram.getCount(), histogram.getSnapshot(), 1)); + timers.forEach( + (name, timer) -> + summary( + name, + "_seconds", + timer.getCount(), + timer.getSnapshot(), + NANOS_PER_SECOND)); + } + + private void gauge(String metric, Object value) { + if (metric.startsWith(STORM_PREFIX) && metric.endsWith(RATE_SUFFIX)) { + return; + } + if (value instanceof Map map) { + String family = sanitize(metric); + map.forEach( + (key, entry) -> { + Double number = toDouble(entry); + if (number != null) { + Labels keyLabels = labels.add("key", String.valueOf(key)); + families.add(family, Type.GAUGE, gaugePoint(keyLabels, number)); + } + }); + return; + } + Double number = toDouble(value); + if (number == null) { + return; + } + Name name = parse(metric, stream); + families.add( + name.family(), Type.GAUGE, gaugePoint(labels.merge(name.labels()), number)); + } + + private void counter(String metric, long count) { + Name name = parse(metric, stream); + families.add( + name.family(), + Type.COUNTER, + CounterDataPointSnapshot.builder() + .labels(labels.merge(name.labels())) + .value(count) + .build()); + } + + private void summary( + String metric, String unit, long count, Snapshot snapshot, double scale) { + Name name = parse(metric, stream); + String family = name.family() + unit; + Labels pointLabels = labels.merge(name.labels()); + List quantiles = new ArrayList<>(QUANTILES.length); + for (double quantile : QUANTILES) { + quantiles.add(new Quantile(quantile, snapshot.getValue(quantile) / scale)); + } + families.add( + family, + Type.SUMMARY, + SummaryDataPointSnapshot.builder() + .labels(pointLabels) + .count(count) + .quantiles(Quantiles.of(quantiles)) + .build()); + families.add( + family + "_max", + Type.GAUGE, + gaugePoint(pointLabels, snapshot.getMax() / scale)); + families.add( + family + "_mean", + Type.GAUGE, + gaugePoint(pointLabels, snapshot.getMean() / scale)); + } + } + + /** Data points grouped by family. */ + private static final class Families { + + private final Map byName = new TreeMap<>(); + private final Set conflicts; + + Families(Set conflicts) { + this.conflicts = conflicts; + } + + void add(String name, Type type, DataPointSnapshot point) { + Family family = byName.computeIfAbsent(name, n -> new Family(type)); + if (family.type != type) { + if (conflicts.add(name)) { + LOG.warn( + "Metric family {} is reported both as a {} and a {}, ignoring the latter", + name, + family.type, + type); + } + return; + } + family.points.putIfAbsent(point.getLabels(), point); + } + + MetricSnapshots build() { + MetricSnapshots.Builder builder = MetricSnapshots.builder(); + byName.forEach((name, family) -> builder.metricSnapshot(family.snapshot(name))); + return builder.build(); + } + } + + private static final class Family { + + private final Type type; + private final Map points = new LinkedHashMap<>(); + + Family(Type type) { + this.type = type; + } + + @SuppressWarnings("unchecked") + MetricSnapshot snapshot(String name) { + MetricMetadata metadata = new MetricMetadata(name); + Collection values = points.values(); + if (type == Type.COUNTER) { + return new CounterSnapshot(metadata, (Collection) values); + } + if (type == Type.GAUGE) { + return new GaugeSnapshot(metadata, (Collection) values); + } + return new SummarySnapshot(metadata, (Collection) values); + } + } +} diff --git a/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java new file mode 100644 index 000000000..682ae41ec --- /dev/null +++ b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/PrometheusReporterTest.java @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.prometheus; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.storm.Config; +import org.apache.storm.metrics2.StormMetricRegistry; +import org.apache.storm.task.TopologyContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class PrometheusReporterTest { + + private static final String TOPOLOGY_ID = "crawl-1-1700000000"; + private static final int WORKER_PORT = 6700; + + private StormMetricRegistry registry; + private PrometheusReporter reporter; + + @BeforeEach + void setUp() { + Map topoConf = new HashMap<>(); + topoConf.put(Config.STORM_ID, TOPOLOGY_ID); + topoConf.put(Config.TOPOLOGY_NAME, "crawl"); + registry = new StormMetricRegistry(); + registry.start(topoConf, WORKER_PORT); + + reporter = new PrometheusReporter(); + 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", "crawl", "component", "fetcher", "task", "3", "port", "6700"); + assertSample(body, "fetcher_counter_total", fetcher, "scope", "status_200", 5); + assertSample(body, "activethreads", fetcher, null, null, 7); + assertSample(body, "status_count", Map.of("component", "status"), "key", "FETCHED", 12); + assertSample(body, "status_count", Map.of("component", "status"), "key", "DISCOVERED", 30); + assertSample( + body, + "storm_ack_count_total", + Map.of("component", "status", "stream", "default"), + "source_component", + "spout", + 2); + assertFalse(body.contains("m1_rate"), body); + assertSample(body, "fetcher_average_perdoc_count", fetcher, "scope", "time_in_queues", 2); + assertSample(body, "fetcher_average_perdoc_max", fetcher, "scope", "time_in_queues", 300); + assertSample(body, "fetcher_average_perdoc_mean", fetcher, "scope", "time_in_queues", 200); + assertSample( + body, + "fetcher_average_persec_total", + fetcher, + "scope", + "bytes_fetched_perSec", + 1000); + assertSample(body, "query_time_seconds_max", fetcher, null, null, 2); + } + + @Test + void resolvePort() { + assertEquals( + PrometheusReporter.DEFAULT_PORT, PrometheusReporter.resolvePort(Map.of(), null)); + assertEquals(7700, PrometheusReporter.resolvePort(Map.of(), "6700")); + assertEquals(6800, PrometheusReporter.resolvePort(Map.of("port.offset", 100), "6700")); + assertEquals(9999, PrometheusReporter.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/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java new file mode 100644 index 000000000..321c973d4 --- /dev/null +++ b/external/prometheus/src/test/java/org/apache/stormcrawler/prometheus/StormMetricsCollectorTest.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.prometheus; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.prometheus.metrics.model.snapshots.Labels; +import java.util.Map; +import org.apache.stormcrawler.prometheus.StormMetricsCollector.Name; +import org.junit.jupiter.api.Test; + +class StormMetricsCollectorTest { + + @Test + void scopedMetric() { + assertEquals( + new Name("fetcher_counter", Labels.of("scope", "status_200")), + StormMetricsCollector.parse("fetcher_counter.status_200", null)); + // only the first dot separates the scope + assertEquals( + new Name("fetcher_counter", Labels.of("scope", "robots.fetched")), + StormMetricsCollector.parse("fetcher_counter.robots.fetched", null)); + } + + @Test + void plainMetric() { + assertEquals( + new Name("activethreads", Labels.EMPTY), + StormMetricsCollector.parse("activethreads", null)); + assertEquals( + new Name("trailing_", Labels.EMPTY), + StormMetricsCollector.parse("trailing.", null)); + } + + @Test + void stormMetricWithStream() { + assertEquals( + new Name("storm_emit_count", Labels.EMPTY), + StormMetricsCollector.parse("__emit-count-default", "default")); + assertEquals( + new Name("storm_capacity", Labels.EMPTY), + StormMetricsCollector.parse("__capacity", "default")); + } + + @Test + void stormMetricWithSourceComponent() { + assertEquals( + new Name("storm_ack_count", Labels.of("source_component", "spout")), + StormMetricsCollector.parse("__ack-count-spout:default", "default")); + assertEquals( + new Name("storm_process_latency", Labels.of("source_component", "url-partitioner")), + StormMetricsCollector.parse("__process-latency-url-partitioner:status", "status")); + } + + @Test + void stormMetricWithoutStream() { + assertEquals( + new Name("storm_receive_population", Labels.EMPTY), + StormMetricsCollector.parse("__receive.population", null)); + } + + @Test + void sanitize() { + assertEquals( + "bytes_fetched_perSec", StormMetricsCollector.sanitize("bytes_fetched_perSec")); + assertEquals( + "GC_G1_Young_Generation", StormMetricsCollector.sanitize("GC.G1-Young-Generation")); + assertEquals("_2xx", StormMetricsCollector.sanitize("2xx")); + assertEquals("pages", StormMetricsCollector.sanitize("pages_total")); + } + + @Test + void taskLabels() { + Map dimensions = + Map.of( + "topologyId", "crawl-1-1700000000", + "hostname", "worker1", + "port", "6700", + "componentId", "fetcher", + "taskid", "3"); + assertEquals( + Labels.of( + "component", "fetcher", + "host", "worker1", + "port", "6700", + "task", "3", + "topology", "crawl"), + StormMetricsCollector.taskLabels("crawl", dimensions)); + } +} diff --git a/pom.xml b/pom.xml index aff3816aa..ced956c61 100644 --- a/pom.xml +++ b/pom.xml @@ -749,6 +749,7 @@ under the License. external/opensearch external/opensearch-java external/playwright + external/prometheus external/solr external/sql external/tika