Add config for custom metrics tags

docs/changelog.rst

@@ -18,6 +18,9 @@ KSQL 5.4.0 includes new features, including:
   Avro record or JSON object, depending on the format in use, or as an anonymous value.
   For more information, refer to :ref:`ksql_single_field_wrapping`.
 
+* A new config `ksql.metrics.tags.custom` for adding custom tags to emitted JMX metrics.
+  See :ref:`ksql-metrics-tags-custom` for usage.
+
 KSQL 5.4.0 includes the following misc. changes:
 
 * Require either the value for a ``@UdfParameter`` or for the UDF JAR to be compiled with

docs/installation/server-config/config-reference.rst

@@ -371,6 +371,16 @@ The corresponding environment variable in the
 `KSQL Server image <https://hub.docker.com/r/confluentinc/cp-ksql-server/>`__ is
 ``KSQL_LISTENERS``.
 
+.. _ksql-metrics-tags-custom:
+
+------------------------
+ksql.metrics.tags.custom
+------------------------
+
+A list of tags to be included with emitted :ref:`JMX metrics <ksql-monitoring-and-metrics>`,
+formatted as a string of ``key:value`` pairs separated by commas.
+For example, ``key1:value1,key2:value2``.
+
 .. _ksql-c3-settings:
 
 |c3| Settings

ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.util;
 
+import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import io.confluent.ksql.config.ConfigItem;
@@ -146,7 +147,7 @@ public class KsqlConfig extends AbstractConfig {
   public static final String KSQL_CUSTOM_METRICS_TAGS = "ksql.metrics.tags.custom";
   private static final String KSQL_CUSTOM_METRICS_TAGS_DOC =
       "A list of tags to be included with emitted JMX metrics, formatted as a string of key:value "
-      + "pairs separated by semicolons. For example, 'key1:value1;key2:value2'.";
+      + "pairs separated by commas. For example, 'key1:value1,key2:value2'.";
 
   public static final String
       defaultSchemaRegistryUrl = "http://localhost:8081";
@@ -718,6 +719,22 @@ public KsqlConfig overrideBreakingConfigsWithOriginalValues(final Map<String, St
     return new KsqlConfig(ConfigGeneration.LEGACY, mergedProperties, mergedStreamConfigProps);
   }
 
+  public Map<String, String> getStringAsMap(final String key) {
+    final String value = getString(key).trim();
+    try {
+      return value.equals("")
+          ? Collections.emptyMap()
+          : Splitter.on(",").trimResults().withKeyValueSeparator(":").split(value);
+    } catch (IllegalArgumentException e) {
+      throw new KsqlException(
+          String.format(
+              "Invalid config value for '%s'. value: %s. reason: %s",
+              key,
+              value,
+              e.getMessage()));
+    }
+  }
+
   private static Set<String> sslConfigNames() {
     final ConfigDef sslConfig = new ConfigDef();
     SslConfigs.addClientSslSupport(sslConfig);

ksql-engine/src/main/java/io/confluent/ksql/KsqlContext.java

@@ -75,7 +75,8 @@ public static KsqlContext create(
     final MutableFunctionRegistry functionRegistry = new InternalFunctionRegistry();
     UdfLoader.newInstance(ksqlConfig, functionRegistry, ".").load();
     final String serviceId = ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG);
-    final String customMetricsTags = ksqlConfig.getString(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS);
+    final Map<String, String> customMetricsTags =
+        ksqlConfig.getStringAsMap(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS);
     final KsqlEngine engine = new KsqlEngine(
         serviceContext,
         processingLogContext,

ksql-engine/src/main/java/io/confluent/ksql/engine/KsqlEngine.java

@@ -39,6 +39,7 @@
 import io.confluent.ksql.util.QueryMetadata;
 import java.io.Closeable;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
@@ -67,7 +68,7 @@ public KsqlEngine(
       final ProcessingLogContext processingLogContext,
       final FunctionRegistry functionRegistry,
       final String serviceId,
-      final String customMetricsTags
+      final Map<String, String> customMetricsTags
   ) {
     this(
         serviceContext,

ksql-engine/src/main/java/io/confluent/ksql/internal/KsqlEngineMetrics.java

@@ -15,12 +15,9 @@
 
 package io.confluent.ksql.internal;
 
-import com.google.common.base.Splitter;
 import io.confluent.ksql.engine.KsqlEngine;
 import io.confluent.ksql.metrics.MetricCollectors;
-import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.KsqlConstants;
-import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.QueryMetadata;
 import java.io.Closeable;
 import java.util.ArrayList;
@@ -65,21 +62,23 @@ public class KsqlEngineMetrics implements Closeable {
   private final KsqlEngine ksqlEngine;
   private final Metrics metrics;
 
-  public KsqlEngineMetrics(final KsqlEngine ksqlEngine, final String customMetricsTags) {
+  public KsqlEngineMetrics(
+      final KsqlEngine ksqlEngine,
+      final Map<String, String> customMetricsTags) {
     this(METRIC_GROUP_PREFIX, ksqlEngine, MetricCollectors.getMetrics(), customMetricsTags);
   }
 
   KsqlEngineMetrics(
       final String metricGroupPrefix,
       final KsqlEngine ksqlEngine,
       final Metrics metrics,
-      final String customMetricsTags) {
+      final Map<String, String> customMetricsTags) {
     this.ksqlEngine = ksqlEngine;
     this.ksqlServiceId = KsqlConstants.KSQL_INTERNAL_TOPIC_PREFIX + ksqlEngine.getServiceId();
     this.sensors = new ArrayList<>();
     this.countMetrics = new ArrayList<>();
     this.metricGroupName = metricGroupPrefix + "-query-stats";
-    this.customMetricsTags = createMetricsTags(customMetricsTags);
+    this.customMetricsTags = customMetricsTags;
 
     this.metrics = metrics;
 
@@ -362,21 +361,6 @@ private void configureNumActiveQueriesForGivenState(
     );
   }
 
-  private static Map<String, String> createMetricsTags(final String tags) {
-    try {
-      return tags.equals("")
-          ? Collections.emptyMap()
-          : Splitter.on(";").trimResults().withKeyValueSeparator(":").split(tags);
-    } catch (IllegalArgumentException e) {
-      throw new KsqlException(
-          String.format(
-              "Invalid config value for '%s'. value: %s. reason: %s",
-              KsqlConfig.KSQL_CUSTOM_METRICS_TAGS,
-              tags,
-              e.getMessage()));
-    }
-  }
-
   private static class CountMetric {
     private final Gauge<Long> count;
     private final MetricName metricName;

ksql-engine/src/test/java/io/confluent/ksql/KsqlContextTestUtil.java

@@ -58,7 +58,7 @@ public static KsqlContext create(
         ProcessingLogContext.create(),
         functionRegistry,
         ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
-        ksqlConfig.getString(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS)
+        ksqlConfig.getStringAsMap(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS)
     );
 
     return new KsqlContext(

ksql-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTestUtil.java

@@ -29,6 +29,8 @@
 import io.confluent.ksql.statement.ConfiguredStatement;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.QueryMetadata;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -48,7 +50,7 @@ public static KsqlEngine createKsqlEngine(
         ProcessingLogContext.create(),
         "test_instance_",
         metaStore,
-        (engine) -> new KsqlEngineMetrics(engine, "")
+        (engine) -> new KsqlEngineMetrics(engine, Collections.emptyMap())
     );
   }
 

ksql-engine/src/test/java/io/confluent/ksql/integration/JsonFormatTest.java

@@ -98,7 +98,7 @@ public void before() throws Exception {
         ProcessingLogContext.create(),
         new InternalFunctionRegistry(),
         ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
-        ksqlConfig.getString(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS));
+        ksqlConfig.getStringAsMap(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS));
 
     topicClient = serviceContext.getTopicClient();
     metaStore = ksqlEngine.getMetaStore();

ksql-engine/src/test/java/io/confluent/ksql/integration/SecureIntegrationTest.java

@@ -271,7 +271,7 @@ private void givenTestSetupWithConfig(final Map<String, Object> ksqlConfigs) {
         ProcessingLogContext.create(),
         new InternalFunctionRegistry(),
         ksqlConfig.getString(KSQL_SERVICE_ID_CONFIG),
-        ksqlConfig.getString(KSQL_CUSTOM_METRICS_TAGS));
+        ksqlConfig.getStringAsMap(KSQL_CUSTOM_METRICS_TAGS));
 
     execInitCreateStreamQueries();
   }

ksql-engine/src/test/java/io/confluent/ksql/internal/KsqlEngineMetricsTest.java

@@ -62,8 +62,7 @@ public class KsqlEngineMetricsTest {
   private KsqlEngineMetrics engineMetrics;
   private static final String KSQL_SERVICE_ID = "test-ksql-service-id";
   private static final String metricNamePrefix = KsqlConstants.KSQL_INTERNAL_TOPIC_PREFIX + KSQL_SERVICE_ID;
-  private static final String CUSTOM_TAGS = "tag1:value1;tag2:value2";
-  private static final Map<String, String> EXPECTED_TAGS = ImmutableMap.of("tag1", "value1", "tag2", "value2");
+  private static final Map<String, String> CUSTOM_TAGS = ImmutableMap.of("tag1", "value1", "tag2", "value2");
 
   @Mock
   private KsqlEngine ksqlEngine;
@@ -265,7 +264,7 @@ private double getMetricValue(final String metricName) {
     return Double.valueOf(
         metrics.metric(
             metrics.metricName(
-                metricName, metricNamePrefix + METRIC_GROUP + "-query-stats", EXPECTED_TAGS)
+                metricName, metricNamePrefix + METRIC_GROUP + "-query-stats", CUSTOM_TAGS)
         ).metricValue().toString()
     );
   }
@@ -275,7 +274,7 @@ private long getLongMetricValue(final String metricName) {
     return Long.parseLong(
         metrics.metric(
             metrics.metricName(
-                metricName, metricNamePrefix + METRIC_GROUP + "-query-stats", EXPECTED_TAGS)
+                metricName, metricNamePrefix + METRIC_GROUP + "-query-stats", CUSTOM_TAGS)
         ).metricValue().toString()
     );
   }

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestApplication.java

@@ -423,7 +423,7 @@ static KsqlRestApplication buildApplication(
         processingLogContext,
         functionRegistry,
         ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
-        ksqlConfig.getString(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS));
+        ksqlConfig.getStringAsMap(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS));
 
     UdfLoader.newInstance(ksqlConfig, functionRegistry, ksqlInstallDir).load();
 

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/StandaloneExecutorFactory.java

@@ -112,7 +112,7 @@ static StandaloneExecutor create(
         processingLogContext,
         functionRegistry,
         ksqlConfig.getString(KsqlConfig.KSQL_SERVICE_ID_CONFIG),
-        ksqlConfig.getString(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS));
+        ksqlConfig.getStringAsMap(KsqlConfig.KSQL_CUSTOM_METRICS_TAGS));
 
     final UdfLoader udfLoader =
         UdfLoader.newInstance(ksqlConfig, functionRegistry, installDir);