Initial import (#2)

Signed-off-by: Mickael Maison <[email protected]>

.gitignore

@@ -0,0 +1,33 @@
+*.class
+
+# Mobile Tools for Java (J2ME)
+.mtj.tmp/
+
+# Package Files #
+*.jar
+*.war
+*.ear
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+
+# Maven stuff
+**/target/*
+
+# Eclipse stuff
+**/.project
+**/.settings/*
+**/.prefs
+**/.classpath
+/target/
+
+# IntelliJ IDEA specific
+.idea/
+*.iml
+
+# VS Code
+.factorypath
+.vscode
+
+.DS_Store
+

README.md

@@ -6,6 +6,33 @@
 This repository contains the _Prometheus Metrics Reporter for Apache Kafka server and client components_ as proposed in [Strimzi Proposal #64](https://github.com/strimzi/proposals/blob/main/064-prometheus-metrics-reporter.md).
 The implementation is currently still in progress.
 
+## Build
+
+```shell
+mvn package assembly:single
+```
+
+## Run
+
+### Kafka Brokers
+
+Add the following to your broker configuration:
+```properties
+metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
+kafka.metrics.reporters=io.strimzi.kafka.metrics.YammerPrometheusMetricsReporter
+```
+
+### Kafka Clients
+
+Add the following to your client configuration:
+```properties
+metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter
+```
+
+## Access Metrics
+
+Metrics are exposed on `http://localhost:8080/metrics`.
+
 ## Getting help
 
 If you encounter any issues while using Strimzi, you can get help using:

pom.xml

@@ -0,0 +1,143 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>io.strimzi</groupId>
+  <artifactId>metrics-reporter</artifactId>
+  <version>1.0-SNAPSHOT</version>
+
+  <name>metrics-reporter</name>
+  <description>Prometheus Metrics Reporter for Apache Kafka server and client components.</description>
+  <url>https://strimzi.io/</url>
+
+  <scm>
+    <connection>scm:git:git://github.com/strimzi/metrics-reporter.git</connection>
+    <developerConnection>scm:git:ssh://github.com:strimzi/metrics-reporter.git</developerConnection>
+    <url>https://github.com/strimzi/metrics-reporter</url>
+  </scm>
+
+  <issueManagement>
+    <system>GitHub</system>
+    <url>https://github.com/strimzi/metrics-reporter/issues</url>
+  </issueManagement>
+
+  <licenses>
+    <license>
+      <name>The Apache License, Version 2.0</name>
+      <url>https://www.apache.org/licenses/LICENSE-2.0.txt</url>
+    </license>
+  </licenses>
+
+  <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <maven.compiler.source>11</maven.compiler.source>
+    <maven.compiler.target>11</maven.compiler.target>
+    <kafka.version>3.6.1</kafka.version>
+    <prometheus.version>0.16.0</prometheus.version>
+    <yammer.version>2.2.0</yammer.version>
+    <slf4j.version>2.0.6</slf4j.version>
+    <junit.version>5.10.1</junit.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>${kafka.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-server-common</artifactId>
+      <version>${kafka.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka_2.13</artifactId>
+      <version>${kafka.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.yammer.metrics</groupId>
+      <artifactId>metrics-core</artifactId>
+      <version>${yammer.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>${slf4j.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient</artifactId>
+      <version>${prometheus.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient_hotspot</artifactId>
+      <version>${prometheus.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient_httpserver</artifactId>
+      <version>${prometheus.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <version>${junit.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <pluginManagement><!-- lock down plugins versions to avoid using Maven defaults (may be moved to parent pom) -->
+      <plugins>
+        <!-- clean lifecycle, see https://maven.apache.org/ref/current/maven-core/lifecycles.html#clean_Lifecycle -->
+        <plugin>
+          <artifactId>maven-clean-plugin</artifactId>
+          <version>3.1.0</version>
+        </plugin>
+        <!-- default lifecycle, jar packaging: see https://maven.apache.org/ref/current/maven-core/default-bindings.html#Plugin_bindings_for_jar_packaging -->
+        <plugin>
+          <artifactId>maven-resources-plugin</artifactId>
+          <version>3.0.2</version>
+        </plugin>
+        <plugin>
+          <artifactId>maven-compiler-plugin</artifactId>
+          <version>3.8.0</version>
+          <configuration>
+            <source>11</source>
+            <target>11</target>
+          </configuration>
+        </plugin>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-assembly-plugin</artifactId>
+          <version>3.1.1</version>
+          <configuration>
+            <descriptorRefs>
+              <descriptorRef>jar-with-dependencies</descriptorRef>
+            </descriptorRefs>
+          </configuration>
+          <executions>
+            <execution>
+              <id>make-assembly</id>
+              <phase>package</phase>
+              <goals>
+                <goal>single</goal>
+              </goals>
+            </execution>
+          </executions>
+        </plugin>
+      </plugins>
+    </pluginManagement>
+  </build>
+</project>

src/main/java/io/strimzi/kafka/metrics/KafkaMetricsCollector.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.metrics;
+
+import io.prometheus.client.Collector;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * Prometheus Collector to store and export metrics retrieved by the reporters.
+ */
+public class KafkaMetricsCollector extends Collector {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaMetricsCollector.class.getName());
+
+    private final Map<MetricName, KafkaMetric> metrics;
+    private final PrometheusMetricsReporterConfig config;
+    private String prefix;
+
+    public KafkaMetricsCollector(PrometheusMetricsReporterConfig config) {
+        this.config = config;
+        this.metrics = new ConcurrentHashMap<>();
+    }
+
+    public void setPrefix(String prefix) {
+        this.prefix = prefix;
+    }
+
+    @Override
+    public List<MetricFamilySamples> collect() {
+        List<MetricFamilySamples> samples = new ArrayList<>();
+
+        for (Map.Entry<MetricName, KafkaMetric> entry : metrics.entrySet()) {
+            MetricName metricName = entry.getKey();
+            KafkaMetric kafkaMetric = entry.getValue();
+            LOG.trace("Collecting Kafka metric {}", metricName);
+
+            String name = metricName(metricName);
+            // TODO Filtering should take labels into account
+            if (!config.isAllowed(name)) {
+                LOG.info("Kafka metric {} is not allowed", name);
+                continue;
+            }
+            LOG.info("Kafka metric {} is allowed", name);
+            LOG.info("labels " + metricName.tags());
+            MetricFamilySamples sample = convert(name, metricName.description(), kafkaMetric, metricName.tags());
+            if (sample != null) {
+                samples.add(sample);
+            }
+        }
+        return samples;
+    }
+
+    public void addMetric(KafkaMetric metric) {
+        metrics.put(metric.metricName(), metric);
+    }
+
+    public void removeMetric(KafkaMetric metric) {
+        metrics.remove(metric.metricName());
+    }
+
+    String metricName(MetricName metricName) {
+        String prefix = this.prefix
+                .replace('.', '_')
+                .replace('-', '_')
+                .toLowerCase();
+        String group = metricName.group()
+                .replace('.', '_')
+                .replace('-', '_')
+                .toLowerCase();
+        String name = metricName.name()
+                .replace('.', '_')
+                .replace('-', '_')
+                .toLowerCase();
+        return prefix + '_' + group + '_' + name;
+    }
+
+    static MetricFamilySamples convert(String name, String help, KafkaMetric metric, Map<String, String> labels) {
+        Object value = metric.metricValue();
+        if (!(value instanceof Number)) {
+            // Prometheus only accepts numeric metrics.
+            // Kafka gauges can have arbitrary types, so skip them for now
+            // TODO move non-numeric values to labels
+            return null;
+        }
+        Map<String, String> sanitizedLabels = labels.entrySet().stream()
+                .collect(Collectors.toMap(
+                        e -> Collector.sanitizeMetricName(e.getKey()),
+                        Map.Entry::getValue,
+                        (v1, v2) -> { throw new IllegalStateException("Unexpected duplicate key " + v1); },
+                        LinkedHashMap::new));
+        return new MetricFamilySamplesBuilder(Type.GAUGE, help)
+                .addSample(name, ((Number) value).doubleValue(), sanitizedLabels)
+                .build();
+    }
+}