Repository: kafka Updated Branches: refs/heads/trunk 6e5bd2497 -> ec466d358
KAFKA-2733: Standardize metric name for Kafka Streams Author: Guozhang Wang <[email protected]> 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 <[email protected]> Authored: Wed Dec 9 22:15:34 2015 -0800 Committer: Guozhang Wang <[email protected]> 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;
