Adopt asynchronous helpers in Kafka target script (#23)

Author: rmfitzpatrick

contrib/jmx-metrics/docs/target-systems/kafka.md

@@ -10,121 +10,121 @@ These metrics are sourced from Kafka's exposed Yammer metrics for each instance:
 * Name: `kafka.messages.in`
 * Description: Number of messages in per second
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.bytes.in`
 * Description: Bytes in per second from clients
 * Unit: `by`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.bytes.out`
 * Description: Bytes out per second to clients
 * Unit: `by`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.isr.shrinks`
 * Description: In-sync replica shrinks per second
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.isr.expands`
 * Description: In-sync replica expands per second
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.max.lag`
 * Description: Max lag in messages between follower and leader replicas
 * Unit: `1`
-* Instrument Type: LongUpDownCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.controller.active.count`
 * Description: Controller is active on broker
 * Unit: `1`
-* Instrument Type: LongUpDownCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.partitions.offline.count`
 * Description: Number of partitions without an active leader
 * Unit: `1`
-* Instrument Type: LongUpDownCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.partitions.underreplicated.count`
 * Description: Number of under replicated partitions
 * Unit: `1`
-* Instrument Type: LongUpDownCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.leader.election.rate`
 * Description: Leader election rate - non-zero indicates broker failures
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.unclean.election.rate`
 * Description: Unclean leader election rate - non-zero indicates broker failures
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.request.queue`
 * Description: Size of the request queue
 * Unit: `1`
-* Instrument Type: LongUpDownCounter
+* Instrument Type: LongValueObserver
 
 * Name: `kafka.fetch.consumer.total.time.count`
 * Description: Fetch consumer request count
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongSumObserver
 
 * Name: `kafka.fetch.consumer.total.time.median`
 * Description: Fetch consumer request time - 50th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 * Name: `kafka.fetch.consumer.total.time.99p`
 * Description: Fetch consumer request time - 99th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 * Name: `kafka.fetch.follower.total.time.count`
 * Description: Fetch follower request count
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongSumObserver
 
 * Name: `kafka.fetch.follower.total.time.median`
 * Description: Fetch follower request time - 50th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 * Name: `kafka.fetch.follower.total.time.99p`
 * Description: Fetch follower request time - 99th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 * Name: `kafka.produce.total.time.count`
 * Description: Produce request count
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongSumObserver
 
 * Name: `kafka.produce.total.time.median`
 * Description: Produce request time - 50th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 * Name: `kafka.produce.total.time.99p`
 * Description: Produce request time - 99th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 ### Log metrics
 
 * Name: `kafka.logs.flush.time.count`
 * Description: Log flush count
 * Unit: `1`
-* Instrument Type: LongCounter
+* Instrument Type: LongSumObserver
 
 * Name: `kafka.logs.flush.time.median`
 * Description: Log flush time - 50th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver
 
 * Name: `kafka.logs.flush.time.99p`
 * Description: Log flush time - 99th percentile
 * Unit: `ms`
-* Instrument Type: DoubleUpDownCounter
+* Instrument Type: DoubleValueObserver

contrib/jmx-metrics/src/main/resources/target-systems/kafka.groovy

@@ -16,80 +16,80 @@
 
 def messagesInPerSec = otel.mbean("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec")
 otel.instrument(messagesInPerSec, "kafka.messages.in", "number of messages in per second",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longValueObserver)
 
 def bytesInPerSec = otel.mbean("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec")
 otel.instrument(bytesInPerSec, "kafka.bytes.in", "bytes in per second from clients",
-        "by", "Count", otel.&longCounter)
+        "by", "Count", otel.&longValueObserver)
 
 def bytesOutPerSec = otel.mbean("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec")
 otel.instrument(bytesOutPerSec, "kafka.bytes.out", "bytes out per second to clients",
-        "by", "Count", otel.&longCounter)
+        "by", "Count", otel.&longValueObserver)
 
 def isrShrinksPerSec = otel.mbean("kafka.server:type=ReplicaManager,name=IsrShrinksPerSec")
 otel.instrument(isrShrinksPerSec, "kafka.isr.shrinks", "in-sync replica shrinks per second",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longValueObserver)
 
 def isrExpandsPerSec = otel.mbean("kafka.server:type=ReplicaManager,name=IsrExpandsPerSec")
 otel.instrument(isrExpandsPerSec, "kafka.isr.expands", "in-sync replica expands per second",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longValueObserver)
 
 def maxLag = otel.mbean("kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica")
 otel.instrument(maxLag, "kafka.max.lag", "max lag in messages between follower and leader replicas",
-        "1", "Value", otel.&longUpDownCounter)
+        "1", "Value", otel.&longValueObserver)
 
 def activeControllerCount = otel.mbean("kafka.controller:type=KafkaController,name=ActiveControllerCount")
 otel.instrument(activeControllerCount, "kafka.controller.active.count", "controller is active on broker",
-        "1", "Value", otel.&longUpDownCounter)
+        "1", "Value", otel.&longValueObserver)
 
 def offlinePartitionsCount = otel.mbean("kafka.controller:type=KafkaController,name=OfflinePartitionsCount")
 otel.instrument(offlinePartitionsCount, "kafka.partitions.offline.count", "number of partitions without an active leader",
-        "1", "Value", otel.&longUpDownCounter)
+        "1", "Value", otel.&longValueObserver)
 
 def underReplicatedPartitionsCount = otel.mbean("kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions")
 otel.instrument(underReplicatedPartitionsCount, "kafka.partitions.underreplicated.count", "number of under replicated partitions",
-        "1", "Value", otel.&longUpDownCounter)
+        "1", "Value", otel.&longValueObserver)
 
 def leaderElectionRate = otel.mbean("kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs")
 otel.instrument(leaderElectionRate, "kafka.leader.election.rate", "leader election rate - non-zero indicates broker failures",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longValueObserver)
 
 def uncleanLeaderElections = otel.mbean("kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec")
 otel.instrument(uncleanLeaderElections, "kafka.unclean.election.rate", "unclean leader election rate - non-zero indicates broker failures",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longValueObserver)
 
 def requestQueueSize = otel.mbean("kafka.network:type=RequestChannel,name=RequestQueueSize")
 otel.instrument(requestQueueSize, "kafka.request.queue", "size of the request queue",
-        "1", "Value", otel.&longUpDownCounter)
+        "1", "Value", otel.&longValueObserver)
 
 def fetchConsumer = otel.mbean("kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer")
 otel.instrument(fetchConsumer, "kafka.fetch.consumer.total.time.count", "fetch consumer request count",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longSumObserver)
 otel.instrument(fetchConsumer, "kafka.fetch.consumer.total.time.median", "fetch consumer request time - 50th percentile",
-        "ms", "50thPercentile", otel.&doubleUpDownCounter)
+        "ms", "50thPercentile", otel.&doubleValueObserver)
 otel.instrument(fetchConsumer, "kafka.fetch.consumer.total.time.99p", "fetch consumer request time - 99th percentile",
-        "ms", "99thPercentile", otel.&doubleUpDownCounter)
+        "ms", "99thPercentile", otel.&doubleValueObserver)
 
 def fetchFollower = otel.mbean("kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchFollower")
 otel.instrument(fetchFollower, "kafka.fetch.follower.total.time.count", "fetch follower request count",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longSumObserver)
 otel.instrument(fetchFollower, "kafka.fetch.follower.total.time.median", "fetch follower request time - 50th percentile",
-        "ms", "50thPercentile", otel.&doubleUpDownCounter)
+        "ms", "50thPercentile", otel.&doubleValueObserver)
 otel.instrument(fetchFollower, "kafka.fetch.follower.total.time.99p", "fetch follower request time - 99th percentile",
-        "ms", "99thPercentile", otel.&doubleUpDownCounter)
+        "ms", "99thPercentile", otel.&doubleValueObserver)
 
 def produce = otel.mbean("kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce")
 otel.instrument(produce, "kafka.produce.total.time.count", "produce request count",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longSumObserver)
 otel.instrument(produce, "kafka.produce.total.time.median", "produce request time - 50th percentile",
-        "ms", "50thPercentile", otel.&doubleUpDownCounter)
+        "ms", "50thPercentile", otel.&doubleValueObserver)
 otel.instrument(produce, "kafka.produce.total.time.99p", "produce request time - 99th percentile",
-        "ms", "99thPercentile", otel.&doubleUpDownCounter)
+        "ms", "99thPercentile", otel.&doubleValueObserver)
 
 def logFlushRate = otel.mbean("kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs")
 otel.instrument(logFlushRate, "kafka.logs.flush.time.count", "log flush count",
-        "1", "Count", otel.&longCounter)
+        "1", "Count", otel.&longSumObserver)
 otel.instrument(logFlushRate, "kafka.logs.flush.time.median", "log flush time - 50th percentile",
-        "ms", "50thPercentile", otel.&doubleUpDownCounter)
+        "ms", "50thPercentile", otel.&doubleValueObserver)
 otel.instrument(logFlushRate, "kafka.logs.flush.time.99p", "log flush time - 99th percentile",
-        "ms", "99thPercentile", otel.&doubleUpDownCounter)
+        "ms", "99thPercentile", otel.&doubleValueObserver)