Feature/update monitoring (#15)

* refactor monitoring metrics

---------

Co-authored-by: Cui Heng, INI-NET-VNC-HCS <Heng.Cui@swisscom.com>

Author: cuihengch

candyfloss/build.gradle

@@ -38,8 +38,9 @@ dependencies {
     implementation libs.slf4j
     implementation libs.vavr
     implementation libs.json.path
-    implementation libs.micrometer
+    implementation libs.micrometer.prom
     implementation libs.logback
+    implementation libs.prometheus.httpserver
     implementation project(":json-transformation")
     implementation project(":protobuf-decoder")
     implementation project(':test-utils')
@@ -67,11 +68,6 @@ jacocoTestReport {
     dependsOn test // tests are required to run before generating the report
 }
 
-task downloadPrometheusExporter(type: Download) {
-    src 'https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.16.1/jmx_prometheus_javaagent-0.16.1.jar'
-    dest new File("${project.projectDir}/src/main/jib", 'jmx_prometheus_javaagent.jar')
-}
-
 
 jar {
     manifest {
@@ -101,15 +97,7 @@ jib {
         jvmFlags = [
                 '-Xms2048m',
                 '-Dconfig.override_with_env_vars=true',
-                '-Dcom.sun.management.jmxremote.port=5555',
-                '-Dcom.sun.management.jmxremote.local.only=true',
-                '-Dcom.sun.management.jmxremote.authenticate=false',
-                '-Dcom.sun.management.jmxremote.ssl=false',
-                '-javaagent:/jmx_prometheus_javaagent.jar=8090:/jmx_prometheus_exporter.yaml',
         ]
         mainClass = 'com.swisscom.daisy.cosmos.candyfloss.CandyflossKStreamsApplication'
     }
 }
-
-tasks.jibDockerBuild.dependsOn tasks.downloadPrometheusExporter
-tasks.jib.dependsOn tasks.downloadPrometheusExporter

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java

@@ -1,6 +1,7 @@
 package com.swisscom.daisy.cosmos.candyfloss;
 
 import com.swisscom.daisy.cosmos.candyfloss.config.JsonKStreamApplicationConfig;
+import com.swisscom.daisy.cosmos.candyfloss.config.MetricRegistryConfig;
 import com.swisscom.daisy.cosmos.candyfloss.config.exceptions.InvalidConfigurations;
 import com.swisscom.daisy.cosmos.candyfloss.messages.ValueErrorMessage;
 import com.swisscom.daisy.cosmos.candyfloss.processors.*;
@@ -10,10 +11,6 @@
 import com.typesafe.config.ConfigFactory;
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde;
-import io.micrometer.core.instrument.Clock;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.jmx.JmxConfig;
-import io.micrometer.jmx.JmxMeterRegistry;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.time.Duration;
@@ -45,8 +42,8 @@ public CandyflossKStreamsApplication(JsonKStreamApplicationConfig config) {
   public static void main(String[] args)
       throws IOException, InvalidConfigurations, InvalidMatchConfiguration {
     // Setting up the metrics registry
-    var jmxMeterRegistry = new JmxMeterRegistry(JmxConfig.DEFAULT, Clock.SYSTEM);
-    Metrics.addRegistry(jmxMeterRegistry);
+    MetricRegistryConfig prometheusRegistry = new MetricRegistryConfig();
+    prometheusRegistry.bindJvm();
 
     logger.info("Loading configurations");
     Config conf = ConfigFactory.load();
@@ -65,12 +62,14 @@ public static void main(String[] args)
               @Override
               public void run() {
                 logger.info("Received shutdown signal, terminating Candyfloss");
+                prometheusRegistry.close();
                 kafkaStreams.close();
                 latch.countDown();
               }
             });
 
     try {
+      prometheusRegistry.start();
       kafkaStreams.start();
       latch.await();
       logger.info("Cosmos Candyfloss gracefully shutdown");

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/CFMetricReporter.java

@@ -0,0 +1,70 @@
+package com.swisscom.daisy.cosmos.candyfloss.config;
+
+import io.micrometer.core.instrument.*;
+import java.util.List;
+import java.util.Map;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.apache.kafka.common.metrics.MetricsReporter;
+
+public class CFMetricReporter implements MetricsReporter {
+  @Override
+  public void init(List<KafkaMetric> metricList) {
+    metricList.forEach(this::metricChange);
+  }
+
+  @Override
+  public void metricChange(KafkaMetric metric) {
+    MetricName metricName = metric.metricName();
+
+    if (metric.metricValue() instanceof Number) {
+      Tags tags = tagsConvert(metricName);
+
+      if (isCounter(metricName)) {
+        FunctionCounter.builder(
+                cleanCounterName(metricName), metric, m -> ((Number) m.metricValue()).doubleValue())
+            .tags(tags)
+            .register(Metrics.globalRegistry);
+      } else {
+        Gauge.builder(metricName.name(), metric, m -> ((Number) m.metricValue()).doubleValue())
+            .description(metricName.description())
+            .tags(tags)
+            .register(Metrics.globalRegistry);
+      }
+    }
+  }
+
+  @Override
+  public void metricRemoval(KafkaMetric metric) {
+    MetricName metricName = metric.metricName();
+    Tags tags = tagsConvert(metricName);
+
+    Metrics.globalRegistry
+        .find(metricName.name())
+        .tags(tags)
+        .meters()
+        .forEach(Metrics.globalRegistry::remove);
+  }
+
+  @Override
+  public void close() {}
+
+  @Override
+  public void configure(Map<String, ?> config) {}
+
+  private boolean isCounter(MetricName metricName) {
+    String name = metricName.name();
+    return name.endsWith("-total") || name.endsWith("-count");
+  }
+
+  private Tags tagsConvert(MetricName metricName) {
+    return Tags.of(
+        metricName.tags().entrySet().stream()
+            .map(entry -> Tag.of(entry.getKey(), entry.getValue()))
+            .toList());
+  }
+
+  private String cleanCounterName(MetricName metricName) {
+    return metricName.name().replace("-total", "").replace("-count", "");
+  }
+}

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java

@@ -11,6 +11,7 @@
 import java.util.Properties;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import org.apache.kafka.streams.StreamsConfig;
 
 /***
  * The main class to parse and represent the application configs read from `application.$env.conf`
@@ -45,6 +46,8 @@ public static JsonKStreamApplicationConfig fromConfig(Config config)
     for (var entry : kafkaConfig.entrySet()) {
       kafka.put(entry.getKey(), kafkaConfig.getString(entry.getKey()));
     }
+    kafka.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, CFMetricReporter.class.getName());
+
     final String inputTopicName = config.getConfig("kstream").getString("input.topic.name");
     final InputType inputType;
     if (config.hasPath("kstream.input.type")) {

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/MetricRegistryConfig.java

@@ -0,0 +1,54 @@
+package com.swisscom.daisy.cosmos.candyfloss.config;
+
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics;
+import io.micrometer.core.instrument.binder.jvm.JvmThreadMetrics;
+import io.micrometer.core.instrument.binder.system.ProcessorMetrics;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
+import io.prometheus.client.exporter.HTTPServer;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetricRegistryConfig {
+  private final PrometheusMeterRegistry prometheusRegistry;
+
+  private JvmGcMetrics jvmGcMetrics;
+
+  private HTTPServer server;
+
+  private final Logger logger = LoggerFactory.getLogger(MetricRegistryConfig.class);
+
+  public MetricRegistryConfig() {
+    prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+  }
+
+  public void bindJvm() {
+    if (jvmGcMetrics == null) {
+      jvmGcMetrics = new JvmGcMetrics();
+    }
+    jvmGcMetrics.bindTo(prometheusRegistry);
+    new ProcessorMetrics().bindTo(prometheusRegistry);
+    new JvmThreadMetrics().bindTo(prometheusRegistry);
+    new JvmMemoryMetrics().bindTo(prometheusRegistry);
+  }
+
+  public void start() {
+    try {
+      server =
+          new HTTPServer(
+              new InetSocketAddress(8090), prometheusRegistry.getPrometheusRegistry(), true);
+      Metrics.addRegistry(prometheusRegistry);
+    } catch (IOException e) {
+      logger.error("Registry service is not properly started", e);
+    }
+  }
+
+  public void close() {
+    jvmGcMetrics.close();
+    server.close();
+  }
+}

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java

@@ -9,10 +9,8 @@
 import com.swisscom.daisy.cosmos.candyfloss.messages.ValueErrorMessage;
 import com.swisscom.daisy.cosmos.candyfloss.processors.exceptions.InvalidCounterKeysConfigurations;
 import com.swisscom.daisy.cosmos.candyfloss.processors.exceptions.InvalidCounterValue;
-import io.micrometer.core.instrument.Counter;
-import io.micrometer.core.instrument.Gauge;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.*;
+import io.micrometer.core.instrument.Timer.Sample;
 import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
 import java.time.*;
@@ -32,6 +30,9 @@
 
 public class CounterNormalizationProcessor
     implements Processor<String, FlattenedMessage, String, ValueErrorMessage<FlattenedMessage>> {
+  private static final String metricTag = "pipeline";
+
+  private static final String timerMetric = "latency_normalization";
 
   private final Logger logger = LoggerFactory.getLogger(CounterNormalizationProcessor.class);
 
@@ -56,15 +57,6 @@ public class CounterNormalizationProcessor
           .register(Metrics.globalRegistry);
   private final AtomicInteger numKeysTotal = new AtomicInteger(0);
   private final AtomicInteger numKeysCleaned = new AtomicInteger(0);
-  private final Gauge numKeysGauge =
-      Gauge.builder("json_streams_old_counter_total_keys", () -> numKeysTotal)
-          .description("Number of counter keys saved in the state store")
-          .register(Metrics.globalRegistry);
-
-  private final Gauge numKeysGaugeCleaned =
-      Gauge.builder("json_streams_old_counter_cleaned_keys", () -> numKeysCleaned)
-          .description("Number of old counter keys deleted from the state store")
-          .register(Metrics.globalRegistry);
 
   private final Duration maxCounterAge;
   private final Duration scanFrequency;
@@ -395,12 +387,18 @@ private Bytes getCounterKey(DocumentContext context, NormalizeCounterConfig coun
 
   @Override
   public void process(Record<String, FlattenedMessage> record) {
+    Sample timer = Timer.start(Metrics.globalRegistry);
+
     var key = record.key();
     var value = record.value();
     var ts = record.timestamp();
 
     KeyValue<String, ValueErrorMessage<FlattenedMessage>> kv = handleRecord(key, value, ts);
 
     context.forward(new Record<>(kv.key, kv.value, record.timestamp()));
+
+    timer.stop(
+        Metrics.globalRegistry.timer(
+            timerMetric, metricTag, value.getTag() == null ? "" : value.getTag()));
   }
 }

candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/FlattenProcessor.java

@@ -6,6 +6,8 @@
 import com.swisscom.daisy.cosmos.candyfloss.transformations.TransformedMessage;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.Timer.Sample;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.kafka.streams.KeyValue;
@@ -15,17 +17,14 @@
 
 public class FlattenProcessor
     implements Processor<String, TransformedMessage, String, ValueErrorMessage<FlattenedMessage>> {
+  private static final String metricTag = "pipeline";
+  private static final String timerMetric = "latency_flatten";
+  private static final String CounterOutMetric = "json_streams_flatten_out";
+
   private final Counter counterIn =
       Counter.builder("json_streams_flatten_in")
           .description("Number of message incoming to the Json Flatten step")
           .register(Metrics.globalRegistry);
-  private final Counter counterOut =
-      Counter.builder("json_streams_flatten_out")
-          .description(
-              "Number of output messages after the flatten step. "
-                  + "Note: json_streams_flatten_out/(json_streams_flatten_in-json_streams_flatten_error)"
-                  + " is flatten message ration")
-          .register(Metrics.globalRegistry);
   private final Counter counterError =
       Counter.builder("json_streams_flatten_error")
           .description("Number of error messages that are discarded to dlq topic")
@@ -40,17 +39,23 @@ public void init(ProcessorContext<String, ValueErrorMessage<FlattenedMessage>> c
 
   @Override
   public void process(Record<String, TransformedMessage> record) {
+    Sample timer = Timer.start(Metrics.globalRegistry);
+    counterIn.increment();
+
     var key = record.key();
     var value = record.value();
     var ts = record.timestamp();
 
     handleRecord(key, value, ts).forEach(kv -> context.forward(new Record<>(kv.key, kv.value, ts)));
+
+    timer.stop(
+        Metrics.globalRegistry.timer(
+            timerMetric, metricTag, value.getTag() == null ? "" : value.getTag()));
   }
 
   public Iterable<KeyValue<String, ValueErrorMessage<FlattenedMessage>>> handleRecord(
       String key, TransformedMessage value, long ts) {
     try {
-      counterIn.increment();
       return flattenMessage(key, value);
     } catch (Exception e) {
       counterError.increment();
@@ -69,7 +74,14 @@ private List<KeyValue<String, ValueErrorMessage<FlattenedMessage>>> flattenMessa
             x ->
                 KeyValue.pair(
                     key, new ValueErrorMessage<>(new FlattenedMessage(x, value.getTag()))))
-        .peek(x -> counterOut.increment())
+        .peek(
+            x -> {
+              FlattenedMessage flatMsg = x.value.getValue();
+              Counter.builder(CounterOutMetric)
+                  .tag(metricTag, flatMsg.getTag() == null ? "" : flatMsg.getTag())
+                  .register(Metrics.globalRegistry)
+                  .increment();
+            })
         .collect(Collectors.toList());
   }
 }