change from state to updowncounter for wider compatibility

Author: aaaugustine29

instrumentation/jmx-metrics/library/kafka-connect.md

@@ -1,8 +1,8 @@
 # Kafka Connect Metrics
 
 Here is the list of metrics based on MBeans exposed by Apache Kafka Connect. String-valued JMX
-attributes (class/type/version information) are exported as state metrics with value `1` and
-carry the raw string value as metric attributes.
+attributes (class/type/version information) are exported as `UpDownCounter` metrics with value `1`
+and carry the raw string value as metric attributes.
 
 ## Compatibility
 

instrumentation/jmx-metrics/library/src/main/resources/jmx/rules/kafka-connect.yaml

@@ -131,7 +131,7 @@ rules:
       # kafka.connect.worker.rebalance.protocol
       connect-protocol:
         metric: protocol
-        type: state
+        type: updowncounter
         desc: The Connect protocol used by this cluster.
         metricAttribute:
           kafka.connect.protocol.state:
@@ -147,7 +147,7 @@ rules:
       # kafka.connect.worker.rebalance.leader
       leader-name:
         metric: leader
-        type: state
+        type: updowncounter
         desc: The name of the group leader.
         metricAttribute:
           kafka.connect.worker.leader.state:
@@ -169,7 +169,7 @@ rules:
       # kafka.connect.worker.rebalance.active
       rebalancing:
         metric: active
-        type: state
+        type: updowncounter
         unit: "1"
         desc: Whether this worker is currently rebalancing.
         metricAttribute:
@@ -196,15 +196,15 @@ rules:
       # kafka.connect.connector.class
       connector-class:
         metric: class
-        type: state
+        type: updowncounter
         desc: The name of the connector class.
         metricAttribute:
           kafka.connect.connector.class.state:
             configured: "*"
       # kafka.connect.connector.type
       connector-type:
         metric: type
-        type: state
+        type: updowncounter
         desc: The type of the connector. One of 'source' or 'sink'.
         metricAttribute:
           kafka.connect.connector.type:
@@ -214,15 +214,15 @@ rules:
       # kafka.connect.connector.version
       connector-version:
         metric: version
-        type: state
+        type: updowncounter
         desc: The version of the connector class, as reported by the connector.
         metricAttribute:
           kafka.connect.connector.version.state:
             reported: "*"
       # kafka.connect.connector.status
       status:
         metric: status
-        type: state
+        type: updowncounter
         desc: Connector lifecycle state indicator (1 when the state matches the attribute value). Supports Apache and Confluent status values.
         metricAttribute:
           kafka.connect.connector.state:
@@ -247,15 +247,15 @@ rules:
       # kafka.connect.predicate.class
       predicate-class:
         metric: class
-        type: state
+        type: updowncounter
         desc: The class name of the predicate class
         metricAttribute:
           kafka.connect.predicate.class.state:
             configured: "*"
       # kafka.connect.predicate.version
       predicate-version:
         metric: version
-        type: state
+        type: updowncounter
         desc: The version of the predicate class
         metricAttribute:
           kafka.connect.predicate.version.state:
@@ -293,15 +293,15 @@ rules:
       # kafka.connect.task.connector.class
       connector-class:
         metric: connector.class
-        type: state
+        type: updowncounter
         desc: The name of the connector class.
         metricAttribute:
           kafka.connect.task.connector.class.state:
             configured: "*"
       # kafka.connect.task.connector.type
       connector-type:
         metric: connector.type
-        type: state
+        type: updowncounter
         desc: The type of the connector. One of 'source' or 'sink'.
         metricAttribute:
           kafka.connect.task.connector.type:
@@ -311,39 +311,39 @@ rules:
       # kafka.connect.task.connector.version
       connector-version:
         metric: connector.version
-        type: state
+        type: updowncounter
         desc: The version of the connector class, as reported by the connector.
         metricAttribute:
           kafka.connect.task.connector.version.state:
             reported: "*"
       # kafka.connect.task.header.converter.class
       header-converter-class:
         metric: header.converter.class
-        type: state
+        type: updowncounter
         desc: The fully qualified class name from header.converter
         metricAttribute:
           kafka.connect.task.header.converter.class.state:
             configured: "*"
       # kafka.connect.task.header.converter.version
       header-converter-version:
         metric: header.converter.version
-        type: state
+        type: updowncounter
         desc: The version instantiated for header.converter. May be undefined
         metricAttribute:
           kafka.connect.task.header.converter.version.state:
             reported: "*"
       # kafka.connect.task.key.converter.class
       key-converter-class:
         metric: key.converter.class
-        type: state
+        type: updowncounter
         desc: The fully qualified class name from key.converter
         metricAttribute:
           kafka.connect.task.key.converter.class.state:
             configured: "*"
       # kafka.connect.task.key.converter.version
       key-converter-version:
         metric: key.converter.version
-        type: state
+        type: updowncounter
         desc: The version instantiated for key.converter. May be undefined
         metricAttribute:
           kafka.connect.task.key.converter.version.state:
@@ -391,7 +391,7 @@ rules:
       # kafka.connect.task.status
       status:
         metric: status
-        type: state
+        type: updowncounter
         desc: The status of the connector task. Supports Apache (unassigned, running, paused, failed, restarting) and Confluent (unassigned, running, paused, failed, destroyed) values.
         metricAttribute:
           kafka.connect.task.state:
@@ -405,31 +405,31 @@ rules:
       # kafka.connect.task.class
       task-class:
         metric: task.class
-        type: state
+        type: updowncounter
         desc: The class name of the task.
         metricAttribute:
           kafka.connect.task.class.state:
             configured: "*"
       # kafka.connect.task.version
       task-version:
         metric: task.version
-        type: state
+        type: updowncounter
         desc: The version of the task.
         metricAttribute:
           kafka.connect.task.version.state:
             reported: "*"
       # kafka.connect.task.value.converter.class
       value-converter-class:
         metric: value.converter.class
-        type: state
+        type: updowncounter
         desc: The fully qualified class name from value.converter
         metricAttribute:
           kafka.connect.task.value.converter.class.state:
             configured: "*"
       # kafka.connect.task.value.converter.version
       value-converter-version:
         metric: value.converter.version
-        type: state
+        type: updowncounter
         desc: The version instantiated for value.converter. May be undefined
         metricAttribute:
           kafka.connect.task.value.converter.version.state:
@@ -447,15 +447,15 @@ rules:
       # kafka.connect.transform.class
       transform-class:
         metric: class
-        type: state
+        type: updowncounter
         desc: The class name of the transformation class
         metricAttribute:
           kafka.connect.transform.class.state:
             configured: "*"
       # kafka.connect.transform.version
       transform-version:
         metric: version
-        type: state
+        type: updowncounter
         desc: The version of the transformation class
         metricAttribute:
           kafka.connect.transform.version.state:

instrumentation/jmx-metrics/library/src/test/java/io/opentelemetry/instrumentation/jmx/rules/KafkaConnectRuleTest.java

@@ -10,6 +10,7 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.instrumentation.jmx.JmxTelemetry;
+import io.opentelemetry.instrumentation.jmx.engine.MetricInfo;
 import io.opentelemetry.instrumentation.jmx.yaml.JmxConfig;
 import io.opentelemetry.instrumentation.jmx.yaml.JmxRule;
 import io.opentelemetry.instrumentation.jmx.yaml.Metric;
@@ -61,6 +62,21 @@ void kafkaConnectConfigParsesAndBuilds() throws Exception {
     }
   }
 
+  @Test
+  void kafkaConnectRulesUseBasicMetricTypes() throws Exception {
+    JmxConfig config = loadKafkaConnectConfig();
+
+    assertThat(config.getRules())
+        .allSatisfy(
+            rule -> {
+              assertThat(rule.getMetricType()).isNotEqualTo(MetricInfo.Type.STATE);
+              rule
+                  .getMapping()
+                  .values()
+                  .forEach(metric -> assertThat(metric.getMetricType()).isNotEqualTo(MetricInfo.Type.STATE));
+            });
+  }
+
   @Test
   void connectorStatusStateMappingPresent() throws Exception {
     JmxConfig config = loadKafkaConnectConfig();
@@ -69,6 +85,8 @@ void connectorStatusStateMappingPresent() throws Exception {
         getRuleForBean(config, "kafka.connect:type=connector-metrics,connector=*");
 
     StateMapping stateMapping = getMetric(connectorRule, "status").getStateMapping();
+    assertThat(getMetric(connectorRule, "status").getMetricType())
+        .isEqualTo(MetricInfo.Type.UPDOWNCOUNTER);
     assertThat(stateMapping.isEmpty()).isFalse();
     assertThat(stateMapping.getStateKeys())
         .contains(
@@ -95,6 +113,8 @@ void taskStatusStateMappingSuperset() throws Exception {
         getRuleForBean(config, "kafka.connect:type=connector-task-metrics,connector=*,task=*");
 
     StateMapping stateMapping = getMetric(connectorTaskRule, "status").getStateMapping();
+    assertThat(getMetric(connectorTaskRule, "status").getMetricType())
+        .isEqualTo(MetricInfo.Type.UPDOWNCOUNTER);
     assertThat(stateMapping.isEmpty()).isFalse();
     assertThat(stateMapping.getStateKeys())
         .contains(