From d17664ac5c9231a97172f9cec1ab1f78a341b1d7 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Fri, 4 Jun 2021 12:23:06 +0200 Subject: [PATCH 1/2] NMS-12987: use gauges instead of distributions (distributions are not supported by the combination Beam/Flink/Prometheus) --- main/src/main/java/org/opennms/nephron/Pipeline.java | 1 + 1 file changed, 1 insertion(+) diff --git a/main/src/main/java/org/opennms/nephron/Pipeline.java b/main/src/main/java/org/opennms/nephron/Pipeline.java index 1b311124..da2b7913 100644 --- a/main/src/main/java/org/opennms/nephron/Pipeline.java +++ b/main/src/main/java/org/opennms/nephron/Pipeline.java @@ -45,6 +45,7 @@ import org.apache.beam.sdk.io.kafka.KafkaRecord; import org.apache.beam.sdk.io.kafka.TimestampPolicyFactory; import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.Distribution; import org.apache.beam.sdk.metrics.Gauge; import org.apache.beam.sdk.metrics.Metrics; import org.apache.beam.sdk.transforms.Combine; From 5d6022930c0062aa9b790af9ebd0162c9c9296f5 Mon Sep 17 00:00:00 2001 From: Stefan Wachter Date: Fri, 4 Jun 2021 15:11:50 +0200 Subject: [PATCH 2/2] NMS-12987: report currentTimeMillis - flow.lastSwitched as kafka_drift (watermarks are also based on lastSwitched) --- main/src/main/java/org/opennms/nephron/Pipeline.java | 1 + 1 file changed, 1 insertion(+) diff --git a/main/src/main/java/org/opennms/nephron/Pipeline.java b/main/src/main/java/org/opennms/nephron/Pipeline.java index da2b7913..9d5d0cb2 100644 --- a/main/src/main/java/org/opennms/nephron/Pipeline.java +++ b/main/src/main/java/org/opennms/nephron/Pipeline.java @@ -341,6 +341,7 @@ public static class ReadFromKafka extends PTransform kafkaConsumerConfig; + // metric name: flink_taskmanager_job_task_operator_flows_from_kafka_drift private final Counter flowsFromKafka = Metrics.counter("flows", "from_kafka"); // a distribution would be more interesting for from_kafka_drift // -> Unfortunately histograms are not supported Beam/Flink/Prometheus