Skip to content

MINOR: Use thread name and task for sensor name #5111

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
guozhangwang merged 3 commits into from
Jun 1, 2018
Merged

MINOR: Use thread name and task for sensor name #5111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fbb0c1c
MINOR: Use thread name and task for sensor name
bbejeck May 31, 2018
9ae123a
MINOR: updated to use task name as sensor name with variable
bbejeck Jun 1, 2018
d857d79
MINOR: updates per comments
bbejeck Jun 1, 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
74 changes: 43 additions & 31 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 @@ -229,16 +229,33 @@ public Sensor addLatencyAndThroughputSensor(final String scopeName,
final String operationName,
final Sensor.RecordingLevel recordingLevel,
final String... tags) {

return addLatencyAndThroughputSensor(null,
scopeName,
entityName,
operationName,
recordingLevel,
tags);

}

public Sensor addLatencyAndThroughputSensor(final String taskName,
final String scopeName,
final String entityName,
final String operationName,
final Sensor.RecordingLevel recordingLevel,
final String... tags) {

final Map<String, String> tagMap = constructTags(scopeName, entityName, tags);
final Map<String, String> allTagMap = constructTags(scopeName, "all", tags);

// first add the global operation metrics if not yet, with the global tags only
final Sensor parent = metrics.sensor(sensorName(operationName, null), recordingLevel);
final Sensor parent = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), null), recordingLevel);
addLatencyMetrics(scopeName, parent, operationName, allTagMap);
addThroughputMetrics(scopeName, parent, operationName, allTagMap);

// add the operation metrics with additional tags
final Sensor sensor = metrics.sensor(sensorName(operationName, entityName), recordingLevel, parent);
final Sensor sensor = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), entityName), recordingLevel, parent);
addLatencyMetrics(scopeName, sensor, operationName, tagMap);
addThroughputMetrics(scopeName, sensor, operationName, tagMap);

Expand All @@ -247,20 +264,6 @@ 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 @@ -270,35 +273,44 @@ public Sensor addThroughputSensor(final String scopeName,
final String operationName,
final Sensor.RecordingLevel recordingLevel,
final String... tags) {

return addThroughputSensor(null,
scopeName,
entityName,
operationName,
recordingLevel,
tags);

}

public Sensor addThroughputSensor(final String taskName,
final String scopeName,
final String entityName,
final String operationName,
final Sensor.RecordingLevel recordingLevel,
final String... tags) {

final Map<String, String> tagMap = constructTags(scopeName, entityName, tags);
final Map<String, String> allTagMap = constructTags(scopeName, "all", tags);

// first add the global operation metrics if not yet, with the global tags only
final Sensor parent = metrics.sensor(sensorName(operationName, null), recordingLevel);
final Sensor parent = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), null), recordingLevel);
addThroughputMetrics(scopeName, parent, operationName, allTagMap);

// add the operation metrics with additional tags
final Sensor sensor = metrics.sensor(sensorName(operationName, entityName), recordingLevel, parent);
final Sensor sensor = metrics.sensor(sensorName(buildUniqueSensorName(operationName, taskName), entityName), recordingLevel, parent);
addThroughputMetrics(scopeName, sensor, operationName, tagMap);

parentSensors.put(sensor, parent);

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 String buildUniqueSensorName(String operationName, String taskName) {
String task = taskName == null ? "" : taskName + ".";
return threadName + "." + task + operationName;
}

private void addLatencyMetrics(final String scopeName, final Sensor sensor, final String opName, final Map<String, String> tags) {
Expand Down
27 changes: 15 additions & 12 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,24 +118,27 @@ public void testMetrics() {
assertNotNull(metrics.getSensor("name-mock.0_0." + throughputOperation));

for (final String opName : latencyOperations) {
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-avg", groupName, metricTags);

@vvcephei vvcephei May 31, 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.

L116 and 118 also have changes that should be reverted.

@bbejeck bbejeck May 31, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

hmm, I'm not sure about that. Those lines are getting the sensor by name which still uses the combination of thread.name() + taskID + operationName or am I missing something?

@vvcephei vvcephei Jun 1, 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.

oh! yeah, you're right. Sorry!

StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-max", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-rate", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-total", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-avg", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-max", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-rate", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-total", groupName, metricTags);
}
assertNotNull(metrics.metrics().get(metrics.metricName("mock.0_0." + throughputOperation + "-rate", groupName,
"The average number of occurrence of " + "mock.0_0." + throughputOperation + " operation per second.", metricTags)));
assertNotNull(metrics.metrics().get(metrics.metricName(throughputOperation + "-rate", groupName,
"The average number of occurrence of " + throughputOperation + " operation per second.",
metricTags)));

// test "all"
metricTags.put("processor-node-id", "all");
for (final String opName : latencyOperations) {
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-avg", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-latency-max", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-rate", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), "mock.0_0." + opName + "-total", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-avg", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-latency-max", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-rate", groupName, metricTags);
StreamsTestUtils.getMetricByNameFilterByTags(metrics.metrics(), opName + "-total", groupName, metricTags);
}
assertNotNull(metrics.metrics().get(metrics.metricName("mock.0_0." + throughputOperation + "-rate", groupName,
"The average number of occurrence of " + "mock.0_0." + throughputOperation + " operation per second.", metricTags)));
assertNotNull(metrics.metrics().get(metrics.metricName(throughputOperation + "-rate",
groupName,
"The average number of occurrence of " + throughputOperation + " operation per second.",
metricTags)));


}
Expand Down
16 changes: 8 additions & 8 deletions ...src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueBytesStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ public void shouldWriteBytesToInnerStoreAndRecordPutMetric() {

metered.put(key, value);

final KafkaMetric metric = metric("test.0_0.put-rate");
final KafkaMetric metric = metric("put-rate");

assertTrue(metric.value() > 0);
EasyMock.verify(inner);
Expand All @@ -114,7 +114,7 @@ public void shouldGetBytesFromInnerStoreAndReturnGetMetric() {

assertThat(metered.get(key), equalTo(value));

final KafkaMetric metric = metric("test.0_0.get-rate");
final KafkaMetric metric = metric("get-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -127,7 +127,7 @@ public void shouldPutIfAbsentAndRecordPutIfAbsentMetric() {

metered.putIfAbsent(key, value);

final KafkaMetric metric = metric("test.0_0.put-if-absent-rate");
final KafkaMetric metric = metric("put-if-absent-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -146,7 +146,7 @@ public void shouldPutAllToInnerStoreAndRecordPutAllMetric() {

metered.putAll(Collections.singletonList(KeyValue.pair(key, value)));

final KafkaMetric metric = metric("test.0_0.put-all-rate");
final KafkaMetric metric = metric("put-all-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -159,7 +159,7 @@ public void shouldDeleteFromInnerStoreAndRecordDeleteMetric() {

metered.delete(key);

final KafkaMetric metric = metric("test.0_0.delete-rate");
final KafkaMetric metric = metric("delete-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -176,7 +176,7 @@ public void shouldGetRangeFromInnerStoreAndRecordRangeMetric() {
assertFalse(iterator.hasNext());
iterator.close();

final KafkaMetric metric = metric("test.0_0.range-rate");
final KafkaMetric metric = metric("range-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -193,7 +193,7 @@ public void shouldGetAllFromInnerStoreAndRecordAllMetric() {
assertFalse(iterator.hasNext());
iterator.close();

final KafkaMetric metric = metric(new MetricName("test.0_0.all-rate", "stream-scope-metrics", "", tags));
final KafkaMetric metric = metric(new MetricName("all-rate", "stream-scope-metrics", "", tags));
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -206,7 +206,7 @@ public void shouldFlushInnerWhenFlushTimeRecords() {

metered.flush();

final KafkaMetric metric = metric("test.0_0.flush-rate");
final KafkaMetric metric = metric("flush-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand Down
14 changes: 7 additions & 7 deletions streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ public void shouldWriteBytesToInnerStoreAndRecordPutMetric() {

metered.put(new Windowed<>(key, new SessionWindow(0, 0)), key);

final KafkaMetric metric = metric("test.0_0.put-rate");
final KafkaMetric metric = metric("put-rate");
assertTrue(((Double) metric.metricValue()) > 0);
EasyMock.verify(inner);
}
Expand All @@ -115,7 +115,7 @@ public void shouldFindSessionsFromStoreAndRecordFetchMetric() {
assertFalse(iterator.hasNext());
iterator.close();

final KafkaMetric metric = metric("test.0_0.fetch-rate");
final KafkaMetric metric = metric("fetch-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -132,7 +132,7 @@ public void shouldFindSessionRangeFromStoreAndRecordFetchMetric() {
assertFalse(iterator.hasNext());
iterator.close();

final KafkaMetric metric = metric("test.0_0.fetch-rate");
final KafkaMetric metric = metric("fetch-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -146,7 +146,7 @@ public void shouldRemoveFromStoreAndRecordRemoveMetric() {

metered.remove(new Windowed<>(key, new SessionWindow(0, 0)));

final KafkaMetric metric = metric("test.0_0.remove-rate");
final KafkaMetric metric = metric("remove-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -163,7 +163,7 @@ public void shouldFetchForKeyAndRecordFetchMetric() {
assertFalse(iterator.hasNext());
iterator.close();

final KafkaMetric metric = metric("test.0_0.fetch-rate");
final KafkaMetric metric = metric("fetch-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}
Expand All @@ -180,15 +180,15 @@ public void shouldFetchRangeFromStoreAndRecordFetchMetric() {
assertFalse(iterator.hasNext());
iterator.close();

final KafkaMetric metric = metric("test.0_0.fetch-rate");
final KafkaMetric metric = metric("fetch-rate");
assertTrue(metric.value() > 0);
EasyMock.verify(inner);
}

@Test
public void shouldRecordRestoreTimeOnInit() {
init();
final KafkaMetric metric = metric("test.0_0.restore-rate");
final KafkaMetric metric = metric("restore-rate");
assertTrue(metric.value() > 0);
}

Expand Down
20 changes: 10 additions & 10 deletions streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,8 @@ public void shouldRecordRestoreLatencyOnInit() {
EasyMock.replay(innerStoreMock);
store.init(context, store);
final Map<MetricName, ? extends Metric> metrics = context.metrics().metrics();
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.restore-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.restore-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "restore-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "restore-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
}

@Test
Expand All @@ -103,8 +103,8 @@ public void shouldRecordPutLatency() {
store.init(context, store);
store.put("a", "a");
final Map<MetricName, ? extends Metric> metrics = context.metrics().metrics();
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.put-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.put-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "put-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "put-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
EasyMock.verify(innerStoreMock);
}

Expand All @@ -116,8 +116,8 @@ public void shouldRecordFetchLatency() {
store.init(context, store);
store.fetch("a", 1, 1).close(); // recorded on close;
final Map<MetricName, ? extends Metric> metrics = context.metrics().metrics();
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
EasyMock.verify(innerStoreMock);
}

Expand All @@ -129,8 +129,8 @@ public void shouldRecordFetchRangeLatency() {
store.init(context, store);
store.fetch("a", "b", 1, 1).close(); // recorded on close;
final Map<MetricName, ? extends Metric> metrics = context.metrics().metrics();
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "fetch-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
EasyMock.verify(innerStoreMock);
}

Expand All @@ -144,8 +144,8 @@ public void shouldRecordFlushLatency() {
store.init(context, store);
store.flush();
final Map<MetricName, ? extends Metric> metrics = context.metrics().metrics();
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.flush-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "test.0_0.flush-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "flush-total", "stream-scope-metrics", singletonMap("scope-id", "all")).metricValue());
assertEquals(1.0, getMetricByNameFilterByTags(metrics, "flush-total", "stream-scope-metrics", singletonMap("scope-id", "mocked-store")).metricValue());
EasyMock.verify(innerStoreMock);
}

Expand Down