Skip to content

MINOR: Ensure sensor names are unique in Kafka Streams #5009

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 2 commits into from
May 12, 2018
Merged

MINOR: Ensure sensor names are unique in Kafka Streams #5009

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
daad05e
MINOR: Ensure sensor names are unique in Kafka Streams
mjsax May 9, 2018
134321d
Github comment
mjsax May 11, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -531,6 +531,7 @@ public synchronized KafkaMetric removeMetric(MetricName metricName) {
log.error("Error when removing metric from " + reporter.getClass().getName(), e);
}
}
log.trace("Removed metric named {}", metricName);
}
return metric;
}
Expand Down Expand Up @@ -564,6 +565,7 @@ synchronized void registerMetric(KafkaMetric metric) {
log.error("Error when registering metric on " + reporter.getClass().getName(), e);
}
}
log.trace("Registered metric named {}", metricName);
}

/**
Expand Down
5 changes: 5 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,34 +186,39 @@ private NodeMetrics(final StreamsMetricsImpl metrics, final String processorNode

// these are all latency metrics
this.nodeProcessTimeSensor = metrics.addLatencyAndThroughputSensor(
context.taskId().toString(),
"processor-node",
processorNodeName,
"process",
Sensor.RecordingLevel.DEBUG,
"task-id", context.taskId().toString()
);
this.nodePunctuateTimeSensor = metrics.addLatencyAndThroughputSensor(
context.taskId().toString(),
"processor-node",
processorNodeName,
"punctuate",
Sensor.RecordingLevel.DEBUG,
"task-id", context.taskId().toString()
);
this.nodeCreationSensor = metrics.addLatencyAndThroughputSensor(
context.taskId().toString(),
"processor-node",
processorNodeName,
"create",
Sensor.RecordingLevel.DEBUG,
"task-id", context.taskId().toString()
);
this.nodeDestructionSensor = metrics.addLatencyAndThroughputSensor(
context.taskId().toString(),
"processor-node",
processorNodeName,
"destroy",
Sensor.RecordingLevel.DEBUG,
"task-id", context.taskId().toString()
);
this.sourceNodeForwardSensor = metrics.addThroughputSensor(
context.taskId().toString(),
"processor-node",
processorNodeName,
"forward",
Expand Down
29 changes: 29 additions & 0 deletions ...rc/main/java/org/apache/kafka/streams/processor/internals/metrics/StreamsMetricsImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -247,6 +247,20 @@ public Sensor addLatencyAndThroughputSensor(final String scopeName,
return sensor;
}

public Sensor addLatencyAndThroughputSensor(final String taskName,
final String scopeName,
final String entityName,
final String operationName,
final Sensor.RecordingLevel recordingLevel,
final String... tags) {
return addLatencyAndThroughputSensor(
scopeName,
entityName,
threadName + "." + taskName + "." + operationName,
recordingLevel,
tags);
}

/**
* @throws IllegalArgumentException if tags is not constructed in key-value pairs
*/
Expand All @@ -272,6 +286,21 @@ public Sensor addThroughputSensor(final String scopeName,
return sensor;
}

public Sensor addThroughputSensor(final String taskName,
final String scopeName,
final String entityName,
final String operationName,
final Sensor.RecordingLevel recordingLevel,
final String... tags) {
return addThroughputSensor(
scopeName,
entityName,
threadName + "." + taskName + "." + operationName,
recordingLevel,
tags
);
}

private void addLatencyMetrics(final String scopeName, final Sensor sensor, final String opName, final Map<String, String> tags) {
sensor.add(
metrics.metricName(
Expand Down
61 changes: 31 additions & 30 deletions ...ams/src/main/java/org/apache/kafka/streams/state/internals/InnerMeteredKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;

Expand Down Expand Up @@ -49,7 +49,7 @@ class InnerMeteredKeyValueStore<K, IK, V, IV> extends WrappedStateStore.Abstract
private Sensor allTime;
private Sensor rangeTime;
private Sensor flushTime;
private StreamsMetrics metrics;
private StreamsMetricsImpl metrics;
private ProcessorContext context;
private StateStore root;

Expand Down Expand Up @@ -89,63 +89,64 @@ interface TypeConverter<K, IK, V, IV> {
public void init(ProcessorContext context, StateStore root) {
final String name = name();
final String tagKey = "task-id";
final String tagValue = context.taskId().toString();
final String taskName = context.taskId().toString();
this.context = context;
this.root = root;
this.metrics = context.metrics();
this.putTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
this.metrics = (StreamsMetricsImpl) context.metrics();
this.putTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"put",
Sensor.RecordingLevel.DEBUG,
tagKey, tagValue);
this.putIfAbsentTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.putIfAbsentTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"put-if-absent",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
this.getTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.getTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"get",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
this.deleteTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.deleteTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"delete",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
this.putAllTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.putAllTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"put-all",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
this.allTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.allTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"all",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
this.rangeTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.rangeTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"range",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
this.flushTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
this.flushTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"flush",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(metricScope,
tagKey, taskName);
final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(taskName,
metricScope,
name,
"restore",
Sensor.RecordingLevel.DEBUG,
tagKey,
tagValue);
tagKey, taskName);

@guozhangwang guozhangwang May 11, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: one space.


// register and possibly restore the state from the logs
if (restoreTime.shouldRecord()) {
Expand Down
28 changes: 14 additions & 14 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.StateSerdes;
Expand All @@ -38,7 +38,7 @@ public class MeteredSessionStore<K, V> extends WrappedStateStore.AbstractStateSt
private final Serde<V> valueSerde;
private final Time time;
private StateSerdes<K, V> serdes;
private StreamsMetrics metrics;
private StreamsMetricsImpl metrics;
private Sensor putTime;
private Sensor fetchTime;
private Sensor flushTime;
Expand All @@ -65,19 +65,19 @@ public void init(final ProcessorContext context, final StateStore root) {
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
final String tagKey = "task-id";
final String tagValue = context.taskId().toString();
this.metrics = context.metrics();
this.putTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "put",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
this.fetchTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "fetch",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
this.flushTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "flush",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
this.removeTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "remove",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
final String taskName = context.taskId().toString();
this.metrics = (StreamsMetricsImpl) context.metrics();
this.putTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "put",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
this.fetchTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "fetch",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
this.flushTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "flush",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
this.removeTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "remove",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);

final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "restore",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "restore",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
// register and possibly restore the state from the logs
final long startNs = time.nanoseconds();
try {
Expand Down
24 changes: 12 additions & 12 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredWindowStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.WindowStore;
Expand All @@ -37,7 +37,7 @@ public class MeteredWindowStore<K, V> extends WrappedStateStore.AbstractStateSto
private final Time time;
private final Serde<K> keySerde;
private final Serde<V> valueSerde;
private StreamsMetrics metrics;
private StreamsMetricsImpl metrics;
private Sensor putTime;
private Sensor fetchTime;
private Sensor flushTime;
Expand Down Expand Up @@ -65,16 +65,16 @@ public void init(final ProcessorContext context, final StateStore root) {
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
final String tagKey = "task-id";
final String tagValue = context.taskId().toString();
this.metrics = context.metrics();
this.putTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "put",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
this.fetchTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "fetch",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
this.flushTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "flush",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(metricScope, name(), "restore",
Sensor.RecordingLevel.DEBUG, tagKey, tagValue);
final String taskName = context.taskId().toString();
this.metrics = (StreamsMetricsImpl) context.metrics();
this.putTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "put",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
this.fetchTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "fetch",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
this.flushTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "flush",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
final Sensor restoreTime = this.metrics.addLatencyAndThroughputSensor(taskName, metricScope, name(), "restore",
Sensor.RecordingLevel.DEBUG, tagKey, taskName);
// register and possibly restore the state from the logs
final long startNs = time.nanoseconds();
try {
Expand Down
Loading