strimzi · mstruk · Jul 7, 2023 · Mar 21, 2023 · Jul 4, 2023 · Jul 4, 2023
diff --git a/README.md b/README.md
@@ -1225,21 +1225,38 @@ Configuring the metrics
 
 By default, the gathering and exporting of metrics is disabled. Metrics are available to get an insight into the performance and failures during token validation, authorization operations and client authentication to the authorization server. You can also monitor the authorization server requests by background services such as refreshing of JWKS keys and refreshing of grants when `KeycloakAuthorizer` is used.
 
-You can enable metrics for token validation on the Kafka broker or for client authentication on the client by setting the following JAAS option to `true`:
+You can enable metrics for token validation, and keycloak authorization on the Kafka broker or for client authentication on the client by setting the following JAAS option to `true`:
 - `oauth.enable.metrics` (e.g.: "true")
 
-You can enable metrics for `KeycloakAuthorizer` by setting an analogous option in Kafka broker's `server.properties` file:
+You can also enable metrics only for keycloak authorization by setting an analogous option in Kafka broker's `server.properties` file:
 - `strimzi.authorization.enable.metrics` (e.g.: "true")
 
-If `OAUTH_ENABLE_METRICS` env variable is set or if `oauth.enable.metrics` system property is set, that will both also enable the metrics for `KeycloakAuthorizer`.
+If `OAUTH_ENABLE_METRICS` env variable is set or if `oauth.enable.metrics` system property is set, that will also enable the metrics for keycloak authorization (as well as for token validation, and client authentication).
 
-If `oauth.config.id` is specified in JAAS configuration of the listener or the client, it will be available in MBean / metric name as `contextId` attribute. If not specified, it will be calculated from JAAS configuration for the validator or default to `client` in client JAAS config, or `keycloak-authorizer` for KeycloakAuthorizer metrics.
+The OAuth metrics ignores the Kafka `metric.reporters` option in order to prevent automatically instantiating double instances of reporters. Most reporters may expect that they are singleton object and may not function properly in multiple copies.
+Instead, there is `strimzi.metric.reporters` option where the reporters that support multiple copies can be specified for the purpose of metrics integration:
+- `strimzi.metric.reporters` (e.g.: "org.apache.kafka.common.metrics.JmxReporter", use ',' as a separator to enable multiple reporters)
+
+The configuration option `strimzi.metric.reporters` has to be configured as an env variable or a system property. Using it on the Kafka broker inside `server.properties` does not work reliably due to multiple pluggability mechanisms that can be used (authorizer, authentication callback handler, client).
+Some of these mechanisms get the `server.properties` filtered so only configuration recognised by Kafka makes it through. However, this is a global OAuth Metrics configuration, and it is initialized on first use by any of the components, using the configuration provided to that component.
+Specifically, the inter-broker client using OAUTHBEARER might be the first to trigger OAuth Metrics initialisation on the broker, and does not see this config option.
+
+In order to reliably configure `strimzi.metric.reporters` one of the following options should be used when starting a Kafka broker:
+- `STRIMZI_METRIC_REPORTERS` env variable
+- `strimzi.metric.reporters` env variable or system property
+
+When OAuth metrics are enabled the OAuth layer can not reuse existing metric reporter instances, and have to create its own copies of metric reporters. Noteably, it does not create its copy of `org.apache.kafka.common.metrics.JmxReporter` automatically. Instead, it has to be explicitly specified in `strimzi.metric.reporters` configuration as item in the list of reporters to instantiate.
+At the moment there is no way to integrate with the existing Kafka metrics / reporters objects already instantiated in different Kafka runtimes (producer, consumer, broker, ...).
 
-Metrics are exposed through JMX managed beans. They can also be exposed as Prometheus metrics by using the Prometheus JMX Exporter agent, and mapping the JMX metrics names to prometheus metrics names.
+NOTE: This breaks compatibility with OAuth versions preceding 0.13.0 where `metric.reporters` was used to configure reporters which were consequently automatically instantiated twice.
 
-The OAuth metrics also honor the `metric.reporters`, `metrics.num.samples`, `metrics.recording.level` and `metrics.sample.window.ms` configurations of Kafka runtimes. When OAuth metrics are enabled the OAuth layer creates duplicate instances of `JmxReporter` and the configured `MetricReporter`s since at the moment there is no other way to integrate with the existing metrics system of Kafka runtimes. 
 
-When OAuth metrics are enabled, managed beans are registered on demand, containing the attributes that are easily translated into Prometheus metrics.
+The Kafka options that control sampling are honored: `metrics.num.samples`, `metrics.recording.level` and `metrics.sample.window.ms`.  
+
+A common use-case is for metrics to be exposed through JMX managed beans. They can then also be exposed as Prometheus metrics by using the Prometheus JMX Exporter agent, and mapping the JMX metrics names to prometheus metrics names.
+If `oauth.config.id` is specified in JAAS configuration of the listener or the client, it will be available in MBean / metric name as `contextId` attribute. If not specified, it will be calculated from JAAS configuration for the validator or default to `client` in client JAAS config, or `keycloak-authorizer` for KeycloakAuthorizer metrics.
+
+When `JmxReporter` is enabled, managed beans are registered on demand, containing the attributes that are easily translated into Prometheus metrics.
 
 Each registered MBean contains two counter variables - `count`, and `totalTimeMs`.
 It also contains three gauge variables - `minTimeMs`, `maxTimeMs` and `avgTimeMs`. These are measured within the configured sample time window.

diff --git a/examples/kubernetes/kafka-oauth-authz-metrics-client.yaml b/examples/kubernetes/kafka-oauth-authz-metrics-client.yaml
@@ -46,6 +46,8 @@ spec:
       env:
         - name: OAUTH_ENABLE_METRICS
           value: "true"
+        - name: STRIMZI_METRIC_REPORTERS
+          value: org.apache.kafka.common.metrics.JmxReporter
         - name: SECRET
           valueFrom:
             secretKeyRef:

diff --git a/examples/kubernetes/kafka-oauth-single-authz-metrics.yaml b/examples/kubernetes/kafka-oauth-single-authz-metrics.yaml
@@ -5,7 +5,7 @@ metadata:
   name: my-cluster
 spec:
   kafka:
-    version: 3.3.2
+    version: 3.4.0
     replicas: 1
     listeners:
       - name: plain
@@ -68,12 +68,14 @@ spec:
         env:
           - name: OAUTH_ENABLE_METRICS
             value: "true"
-    #      - name: KAFKA_DEBUG
-    #        value: "y"
-    #      - name: DEBUG_SUSPEND_FLAG
-    #        value: "y"
-    #      - name: JAVA_DEBUG_PORT
-    #        value: "5005"
+          - name: STRIMZI_METRIC_REPORTERS
+            value: org.apache.kafka.common.metrics.JmxReporter
+          #- name: KAFKA_DEBUG
+          #  value: "y"
+          #- name: DEBUG_SUSPEND_FLAG
+          #  value: "n"
+          #- name: JAVA_DEBUG_PORT
+          #  value: "5005"
     jmxOptions: {}
 
   zookeeper:
@@ -94,6 +96,7 @@ spec:
   entityOperator:
     topicOperator: {}
     userOperator: {}
+
 ---
 kind: ConfigMap
 apiVersion: v1

diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/metrics/GlobalConfig.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/metrics/GlobalConfig.java
@@ -0,0 +1,17 @@
+/*
+ * Copyright 2017-2023, Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.oauth.metrics;
+
+import io.strimzi.kafka.oauth.common.Config;
+
+/**
+ * Configuration that can be specified as ENV vars, System properties or in <code>server.properties</code> configuration file,
+ * but not as part of the JAAS configuration.
+ */
+public class GlobalConfig extends Config {
+
+    /** The name of the 'strimzi.metric.reporters' config option */
+    public static final String STRIMZI_METRIC_REPORTERS = "strimzi.metric.reporters";
+}
diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/services/OAuthMetrics.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/services/OAuthMetrics.java
@@ -10,7 +10,6 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.KafkaMetricsContext;
 import org.apache.kafka.common.metrics.MetricConfig;
 import org.apache.kafka.common.metrics.Metrics;
@@ -24,23 +23,40 @@
 import org.apache.kafka.common.utils.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
+import static io.strimzi.kafka.oauth.metrics.GlobalConfig.STRIMZI_METRIC_REPORTERS;
 import static org.apache.kafka.clients.CommonClientConfigs.CLIENT_ID_CONFIG;
 import static org.apache.kafka.clients.CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG;
 import static org.apache.kafka.clients.CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG;
 import static org.apache.kafka.clients.CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG;
 
 /**
  * The singleton for handling a cache of all the Sensors to prevent unnecessary redundant re-registrations.
- * There is a one-to-one mapping between a SensorKey and a Sensor, and one-to-one mapping between a Sensor and an MBean name.
- *
- * MBeans are registered as requested by JmxReporter attached to the Metrics object.
+ * There is a one-to-one mapping between a <code>SensorKey</code> and a <code>Sensor</code>, and one-to-one mapping between a <code>Sensor</code> and an <code>MBean</code> name.
+ * <p>
+ * MBeans are registered as requested by <code>JmxReporter</code> attached to the <code>Metrics</code> object.
+ * The <code>JmxReporter</code> has to be explicitly configured using a config option <code>strimzi.metric.reporters</code>, which is
+ * analogous to the Kafka <code>metric.reporters</code> configuration option. It is not configured by default.
+ * <p>
+ * Since OAuth instantiates its own <code>Metrics</code> object it also has to instantiate reporters to attach them to this <code>Metrics</code> object.
+ * To prevent double instantiation of <code>MetricReporter</code> objects that require to be singleton, all <code>MetricReporter</code> objects
+ * to be integrated with <code>OAuthMetrics</code> have to be separately instantiated.
+ * <p>
+ * Example:
+ * <pre>
+ *    strimzi.metric.reporters=org.apache.kafka.common.metrics.JmxReporter
+ * </pre>
+ * Note: On the Kafka broker it is best to use <code>STRIMZI_METRIC_REPORTERS</code> env variable or <code>strimzi.metric.reporters</code> system property,
+ * rather than a `server.properties` global configuration option.
  */
 public class OAuthMetrics {
 
@@ -57,9 +73,14 @@ public class OAuthMetrics {
      *
      * @param configMap Configuration properties
      */
+    @SuppressWarnings("unchecked")
     OAuthMetrics(Map<String, ?> configMap) {
-        this.configMap = configMap;
         this.config = new Config(configMap);
+
+        // Make sure to add the resolved 'strimzi.metric.reporters' configuration to the config map
+        ((Map<String, Object>) configMap).put(STRIMZI_METRIC_REPORTERS, config.getValue(STRIMZI_METRIC_REPORTERS));
+        this.configMap = configMap;
+
         this.metrics = initKafkaMetrics();
     }
 
@@ -90,26 +111,48 @@ private Metrics initKafkaMetrics() {
 
     private List<MetricsReporter> initReporters() {
         AbstractConfig kafkaConfig = initKafkaConfig();
-        List<MetricsReporter> reporters = kafkaConfig.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
-                MetricsReporter.class);
-
-        JmxReporter reporter = new JmxReporter();
-        reporter.configure(configMap);
-
-        reporters.add(reporter);
+        List<MetricsReporter> reporters = new ArrayList<>(3);
+        if (configMap.get(STRIMZI_METRIC_REPORTERS) != null) {
+            reporters = kafkaConfig.getConfiguredInstances(STRIMZI_METRIC_REPORTERS, MetricsReporter.class);
+        } else {
+            log.warn("Configuration option '{}' is not set, OAuth metrics will not be exported to JMX", STRIMZI_METRIC_REPORTERS);
+        }
         return reporters;
     }
 
     private AbstractConfig initKafkaConfig() {
-        ConfigDef configDef = new ConfigDef()
-                .define(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
-                        ConfigDef.Type.LIST,
-                        Collections.emptyList(),
-                        new ConfigDef.NonNullValidator(),
-                        ConfigDef.Importance.LOW,
-                        CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC);
-
-        return new AbstractConfig(configDef, configMap);
+        ConfigDef configDef = addMetricReporterToConfigDef(new ConfigDef(), STRIMZI_METRIC_REPORTERS);
+        return new AbstractConfig(configDef, toMapOfStringValues(configMap));
+    }
+
+    private ConfigDef addMetricReporterToConfigDef(ConfigDef configDef, String name) {
+        return configDef.define(name,
+                ConfigDef.Type.LIST,
+                Collections.emptyList(),
+                new ConfigDef.NonNullValidator(),
+                ConfigDef.Importance.LOW,
+                CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC);
+    }
+
+    private Map<String, String> toMapOfStringValues(Map<String, ?> configMap) {
+        HashMap<String, String> result = new HashMap<>();
+        for (Map.Entry<String, ?> ent: configMap.entrySet()) {
+            Object val = ent.getValue();
+            if (val == null) {
+                continue;
+            }
+            if (val instanceof Class) {
+                result.put(ent.getKey(), ((Class<?>) val).getCanonicalName());
+            } else if (val instanceof List) {
+                String stringVal = ((List<?>) val).stream().map(String::valueOf).collect(Collectors.joining(","));
+                if (!"".equals(stringVal)) {
+                    result.put(ent.getKey(), stringVal);
+                }
+            } else {
+                result.put(ent.getKey(), String.valueOf(ent.getValue()));
+            }
+        }
+        return result;
     }
 
     private KafkaMetricsContext createKafkaMetricsContext() {

diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/services/Services.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/services/Services.java
@@ -39,6 +39,13 @@ public static synchronized void configure(Map<String, ?> configs) {
         }
     }
 
+    /**
+     * Close any configured Services so they can be reinitialised again
+     */
+    public static synchronized void close() {
+        services = null;
+    }
+
     /**
      * Get a configured singleton instance
      *

diff --git a/testsuite/common/src/main/java/io/strimzi/testsuite/oauth/common/TestUtil.java b/testsuite/common/src/main/java/io/strimzi/testsuite/oauth/common/TestUtil.java
@@ -93,4 +93,32 @@ public static void logStart(String msg) {
         System.out.println("========    "  + msg);
         System.out.println();
     }
+
+    public static int findFirstMatchingInLog(List<String> log, String regex) {
+        int lineNum = 0;
+        Pattern pattern = Pattern.compile(prepareRegex(regex));
+        for (String line: log) {
+            if (pattern.matcher(line).matches()) {
+                return lineNum;
+            }
+            lineNum++;
+        }
+        return -1;
+    }
+
+    public static String prepareRegex(String regex) {
+        String prefix = regex.startsWith("^") ? "" : ".*";
+        String suffix = regex.endsWith("$") ? "" : ".*";
+        return prefix + regex + suffix;
+    }
+
+    public static boolean checkLogForRegex(List<String> log, String regex) {
+        Pattern pattern = Pattern.compile(prepareRegex(regex));
+        for (String line: log) {
+            if (pattern.matcher(line).matches()) {
+                return true;
+            }
+        }
+        return false;
+    }
 }
diff --git a/...uth/auth/metrics/TestMetricsReporter.java → ...h/common/metrics/TestMetricsReporter.java b/...uth/auth/metrics/TestMetricsReporter.java → ...h/common/metrics/TestMetricsReporter.java
@@ -2,7 +2,7 @@
  * Copyright 2017-2022, Strimzi authors.
  * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
-package io.strimzi.testsuite.oauth.auth.metrics;
+package io.strimzi.testsuite.oauth.common.metrics;
 
 import org.apache.kafka.common.metrics.KafkaMetric;
 import org.slf4j.Logger;

diff --git a/testsuite/keycloak-auth-tests/docker-compose.yml b/testsuite/keycloak-auth-tests/docker-compose.yml
@@ -45,7 +45,7 @@ services:
       - "5006:5006"
     volumes:
       - ${PWD}/../docker/target/kafka/libs:/opt/kafka/libs/strimzi
-      - ${PWD}/target/test-classes:/opt/kafka/libs/strimzi/reporters
+      - ${PWD}/../common/target/classes:/opt/kafka/libs/strimzi/reporters
       - ${PWD}/../docker/kafka/config:/opt/kafka/config/strimzi
       - ${PWD}/../docker/kafka/scripts:/opt/kafka/strimzi
     command:
@@ -66,9 +66,6 @@ services:
       - KAFKA_INTER_BROKER_LISTENER_NAME=INTROSPECT
       - KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=OAUTHBEARER
 
-      # Common settings for all the listeners
-      - OAUTH_ENABLE_METRICS=true
-
       # username extraction from JWT token claim
       - OAUTH_USERNAME_CLAIM=preferred_username
       - KAFKA_PRINCIPAL_BUILDER_CLASS=io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder
@@ -129,12 +126,24 @@ services:
       - KAFKA_LISTENER_NAME_FORGE_OAUTHBEARER_SASL_SERVER_CALLBACK_HANDLER_CLASS=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
 
       # Test Metrics Reporters
-      - KAFKA_METRIC_REPORTERS=io.strimzi.testsuite.oauth.auth.metrics.TestMetricsReporter
       - KAFKA_METRICS_CONTEXT_TEST_LABEL=testvalue
       - KAFKA_METRICS_NUM_SAMPLES=3
       - KAFKA_METRICS_RECORDING_LEVEL=DEBUG
       - KAFKA_METRICS_SAMPLE_WINDOW_MS=15000
 
+
+      # OAuth metrics configuration
+
+      - OAUTH_ENABLE_METRICS=true
+      #   When enabling metrics we also have to explicitly configure JmxReporter to have metrics available in JMX
+      #   The following value will be available as env var STRIMZI_METRIC_REPORTERS
+      - STRIMZI_METRIC_REPORTERS=org.apache.kafka.common.metrics.JmxReporter,io.strimzi.testsuite.oauth.common.metrics.TestMetricsReporter
+
+      #   The following value will turn to 'strimzi.metric.reporters=...' in 'strimzi.properties' file
+      #   However, that won't work as the value may be filtered to the component that happens to initialise OAuthMetrics
+      #- KAFKA_STRIMZI_METRIC_REPORTERS=org.apache.kafka.common.metrics.JmxReporter
+
+
       # For start.sh script to know where the keycloak is listening
       - KEYCLOAK_HOST=${KEYCLOAK_HOST:-keycloak}
       - REALM=${REALM:-forge}

diff --git a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java
@@ -62,12 +62,11 @@ public void doTests() throws Exception {
     void oauthMetricsConfigIntegration() {
         System.out.println("    ====    KeycloakAuthenticationTest :: oauthMetricsConfigIntegrationTest");
 
-        // Test MetricReporter config works as expected
+        // Test that MetricReporter config works as expected
         // Get kafka log and make sure the TestMetricReporter was initialised exactly twice
         List<String> lines = getContainerLogsForString(kafkaContainer, "TestMetricsReporter no. ");
-        Assert.assertEquals("Kafka log should contain: \"TestMetricsReporter no. \" exactly twice", 2, lines.size());
+        Assert.assertEquals("Kafka log should contain: \"TestMetricsReporter no. \" exactly once", 1, lines.size());
         Assert.assertTrue("Contains \"TestMetricsReporter no. 1\"", lines.get(0).contains("TestMetricsReporter no. 1 "));
-        Assert.assertTrue("Contains \"TestMetricsReporter no. 2\"", lines.get(1).contains("TestMetricsReporter no. 2 "));
 
         // Ensure the configuration was applied as expected
         lines = getContainerLogsForString(kafkaContainer, "Creating Metrics:");
@@ -82,7 +81,7 @@ void oauthMetricsConfigIntegration() {
         Assert.assertTrue("kafka.broker.id=1", line.contains("kafka.broker.id=1"));
 
         line = lines.get(3);
-        Assert.assertTrue("io.strimzi.testsuite.oauth.auth.metrics.TestMetricsReporter", line.contains("io.strimzi.testsuite.oauth.auth.metrics.TestMetricsReporter"));
+        Assert.assertTrue("io.strimzi.testsuite.oauth.common.metrics.TestMetricsReporter", line.contains("io.strimzi.testsuite.oauth.common.metrics.TestMetricsReporter"));
         Assert.assertTrue("org.apache.kafka.common.metrics.JmxReporter", line.contains("org.apache.kafka.common.metrics.JmxReporter"));
     }