kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-2733: Standardize metric name for Kafka Streams
Date Thu, 10 Dec 2015 06:15:40 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 6e5bd2497 -> ec466d358


KAFKA-2733: Standardize metric name for Kafka Streams

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Yasuhiro Matsuda, Jun Rao

Closes #643 from guozhangwang/K2733


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/ec466d35
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/ec466d35
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/ec466d35

Branch: refs/heads/trunk
Commit: ec466d358dcd5ff9e035a6d5de82ada12d187ae0
Parents: 6e5bd24
Author: Guozhang Wang <wangguoz@gmail.com>
Authored: Wed Dec 9 22:15:34 2015 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Wed Dec 9 22:15:34 2015 -0800

----------------------------------------------------------------------
 .../processor/internals/StreamThread.java       | 16 +++++++-------
 .../streams/state/MeteredKeyValueStore.java     | 22 ++++++++++----------
 2 files changed, 20 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/ec466d35/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 38333a2..37d24bb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -721,23 +721,25 @@ public class StreamThread extends Thread {
             for (int i = 0; i < tags.length; i += 2)
                 tagMap.put(tags[i], tags[i + 1]);
 
+            String metricGroupName = "streaming-" + scopeName + "-metrics";
+
             // first add the global operation metrics if not yet, with the global tags only
-            Sensor parent = metrics.sensor(operationName);
-            addLatencyMetrics(this.metricGrpName, parent, "all", operationName, this.metricTags);
+            Sensor parent = metrics.sensor(scopeName + "-" + operationName);
+            addLatencyMetrics(metricGroupName, parent, "all", operationName, this.metricTags);
 
             // add the store operation metrics with additional tags
-            Sensor sensor = metrics.sensor(entityName + "-" + operationName, parent);
-            addLatencyMetrics("streaming-" + scopeName + "-metrics", sensor, entityName,
operationName, tagMap);
+            Sensor sensor = metrics.sensor(scopeName + "-" + entityName + "-" + operationName,
parent);
+            addLatencyMetrics(metricGroupName, sensor, entityName, operationName, tagMap);
 
             return sensor;
         }
 
         private void addLatencyMetrics(String metricGrpName, Sensor sensor, String entityName,
String opName, Map<String, String> tags) {
-            maybeAddMetric(sensor, metrics.metricName(opName + "-avg-latency-ms", metricGrpName,
+            maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-avg-latency-ms",
metricGrpName,
                 "The average latency in milliseconds of " + entityName + " " + opName + "
operation.", tags), new Avg());
-            maybeAddMetric(sensor, metrics.metricName(opName + "-max-latency-ms", metricGrpName,
+            maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-max-latency-ms",
metricGrpName,
                 "The max latency in milliseconds of " + entityName + " " + opName + " operation.",
tags), new Max());
-            maybeAddMetric(sensor, metrics.metricName(opName + "-qps", metricGrpName,
+            maybeAddMetric(sensor, metrics.metricName(entityName + "-" + opName + "-qps",
metricGrpName,
                 "The average number of occurrence of " + entityName + " " + opName + " operation
per second.", tags), new Rate(new Count()));
         }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/ec466d35/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
index d75e7e6..16f57a0 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/MeteredKeyValueStore.java
@@ -31,7 +31,7 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K,
V> {
 
     protected final KeyValueStore<K, V> inner;
     protected final Serdes<K, V> serialization;
-    protected final String metricGrp;
+    protected final String metricScope;
     protected final Time time;
 
     private Sensor putTime;
@@ -48,10 +48,10 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K,
V> {
     private KeyValueStoreChangeLogger<K, V> changeLogger = null;
 
     // always wrap the store with the metered store
-    public MeteredKeyValueStore(final KeyValueStore<K, V> inner, Serdes<K, V>
serialization, String metricGrp, Time time) {
+    public MeteredKeyValueStore(final KeyValueStore<K, V> inner, Serdes<K, V>
serialization, String metricScope, Time time) {
         this.inner = inner;
         this.serialization = serialization;
-        this.metricGrp = metricGrp;
+        this.metricScope = metricScope;
         this.time = time != null ? time : new SystemTime();
     }
 
@@ -69,14 +69,14 @@ public class MeteredKeyValueStore<K, V> implements KeyValueStore<K,
V> {
     public void init(ProcessorContext context) {
         final String name = name();
         this.metrics = context.metrics();
-        this.putTime = this.metrics.addLatencySensor(metricGrp, name, "put", "store-name",
name);
-        this.getTime = this.metrics.addLatencySensor(metricGrp, name, "get", "store-name",
name);
-        this.deleteTime = this.metrics.addLatencySensor(metricGrp, name, "delete", "store-name",
name);
-        this.putAllTime = this.metrics.addLatencySensor(metricGrp, name, "put-all", "store-name",
name);
-        this.allTime = this.metrics.addLatencySensor(metricGrp, name, "all", "store-name",
name);
-        this.rangeTime = this.metrics.addLatencySensor(metricGrp, name, "range", "store-name",
name);
-        this.flushTime = this.metrics.addLatencySensor(metricGrp, name, "flush", "store-name",
name);
-        this.restoreTime = this.metrics.addLatencySensor(metricGrp, name, "restore", "store-name",
name);
+        this.putTime = this.metrics.addLatencySensor(metricScope, name, "put");
+        this.getTime = this.metrics.addLatencySensor(metricScope, name, "get");
+        this.deleteTime = this.metrics.addLatencySensor(metricScope, name, "delete");
+        this.putAllTime = this.metrics.addLatencySensor(metricScope, name, "put-all");
+        this.allTime = this.metrics.addLatencySensor(metricScope, name, "all");
+        this.rangeTime = this.metrics.addLatencySensor(metricScope, name, "range");
+        this.flushTime = this.metrics.addLatencySensor(metricScope, name, "flush");
+        this.restoreTime = this.metrics.addLatencySensor(metricScope, name, "restore");
 
         serialization.init(context);
         this.changeLogger = this.loggingEnabled ? new KeyValueStoreChangeLogger<>(name,
context, serialization) : null;


Mime
View raw message