Skip to content

Minor: Rename methods to add metrics to sensor in StreamsMetricsImpl #7161

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 1 commit into from
Aug 6, 2019
Merged

Minor: Rename methods to add metrics to sensor in StreamsMetricsImpl #7161

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/metrics/Sensors.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ public static Sensor lateRecordDropSensor(final InternalProcessorContext context
LATE_RECORD_DROP,
Sensor.RecordingLevel.INFO
);
StreamsMetricsImpl.addInvocationRateAndCount(
StreamsMetricsImpl.addInvocationRateAndCountToSensor(
sensor,
PROCESSOR_NODE_METRICS_GROUP,
metrics.tagMap("task-id", context.taskId().toString(), PROCESSOR_NODE_ID_TAG, context.currentNode().name()),
Expand Down
13 changes: 7 additions & 6 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 @@ -33,8 +33,7 @@

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.PROCESSOR_NODE_ID_TAG;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.PROCESSOR_NODE_METRICS_GROUP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgMaxLatency;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndMaxLatencyToSensor;

public class ProcessorNode<K, V> {

Expand Down Expand Up @@ -232,12 +231,14 @@ private static Sensor createTaskAndNodeLatencyAndThroughputSensors(final String
final Map<String, String> taskTags,
final Map<String, String> nodeTags) {
final Sensor parent = metrics.taskLevelSensor(taskName, operation, Sensor.RecordingLevel.DEBUG);
addAvgMaxLatency(parent, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);
addInvocationRateAndCount(parent, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);
addAvgAndMaxLatencyToSensor(parent, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);
StreamsMetricsImpl
.addInvocationRateAndCountToSensor(parent, PROCESSOR_NODE_METRICS_GROUP, taskTags, operation);

final Sensor sensor = metrics.nodeLevelSensor(taskName, processorNodeName, operation, Sensor.RecordingLevel.DEBUG, parent);
addAvgMaxLatency(sensor, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);
addInvocationRateAndCount(sensor, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);
addAvgAndMaxLatencyToSensor(sensor, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);
StreamsMetricsImpl
.addInvocationRateAndCountToSensor(sensor, PROCESSOR_NODE_METRICS_GROUP, nodeTags, operation);

return sensor;
}
Expand Down
50 changes: 25 additions & 25 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 @@ -344,13 +344,13 @@ public Sensor addLatencyAndThroughputSensor(final String scopeName,

// first add the global operation metrics if not yet, with the global tags only
final Sensor parent = metrics.sensor(externalParentSensorName(operationName), recordingLevel);
addAvgMaxLatency(parent, group, allTagMap, operationName);
addInvocationRateAndCount(parent, group, allTagMap, operationName);
addAvgAndMaxLatencyToSensor(parent, group, allTagMap, operationName);
addInvocationRateAndCountToSensor(parent, group, allTagMap, operationName);

// add the operation metrics with additional tags
final Sensor sensor = metrics.sensor(externalChildSensorName(operationName, entityName), recordingLevel, parent);
addAvgMaxLatency(sensor, group, tagMap, operationName);
addInvocationRateAndCount(sensor, group, tagMap, operationName);
addAvgAndMaxLatencyToSensor(sensor, group, tagMap, operationName);
addInvocationRateAndCountToSensor(sensor, group, tagMap, operationName);

parentSensors.put(sensor, parent);

Expand All @@ -374,11 +374,11 @@ public Sensor addThroughputSensor(final String scopeName,

// first add the global operation metrics if not yet, with the global tags only
final Sensor parent = metrics.sensor(externalParentSensorName(operationName), recordingLevel);
addInvocationRateAndCount(parent, group, allTagMap, operationName);
addInvocationRateAndCountToSensor(parent, group, allTagMap, operationName);

// add the operation metrics with additional tags
final Sensor sensor = metrics.sensor(externalChildSensorName(operationName, entityName), recordingLevel, parent);
addInvocationRateAndCount(sensor, group, tagMap, operationName);
addInvocationRateAndCountToSensor(sensor, group, tagMap, operationName);

parentSensors.put(sensor, parent);

Expand All @@ -397,10 +397,10 @@ private String externalParentSensorName(final String operationName) {
}


public static void addAvgAndMax(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
public static void addAvgAndMaxToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
sensor.add(
new MetricName(
operation + AVG_SUFFIX,
Expand All @@ -419,10 +419,10 @@ public static void addAvgAndMax(final Sensor sensor,
);
}

public static void addAvgMaxLatency(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
public static void addAvgAndMaxLatencyToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
sensor.add(
new MetricName(
operation + "-latency-avg",
Expand All @@ -441,12 +441,12 @@ public static void addAvgMaxLatency(final Sensor sensor,
);
}

public static void addInvocationRateAndCount(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation,
final String descriptionOfInvocation,
final String descriptionOfRate) {
public static void addInvocationRateAndCountToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation,
final String descriptionOfInvocation,
final String descriptionOfRate) {
sensor.add(
new MetricName(
operation + TOTAL_SUFFIX,
Expand All @@ -467,11 +467,11 @@ public static void addInvocationRateAndCount(final Sensor sensor,
);
}

public static void addInvocationRateAndCount(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
addInvocationRateAndCount(
public static void addInvocationRateAndCountToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
addInvocationRateAndCountToSensor(
sensor,
group,
tags,
Expand Down
30 changes: 15 additions & 15 deletions ...ams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,8 @@
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TASK_ID_TAG;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TASK_LEVEL_GROUP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.THREAD_LEVEL_GROUP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndMax;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndMaxToSensor;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCountToSensor;

public class ThreadMetrics {
private ThreadMetrics() {}
Expand Down Expand Up @@ -74,7 +74,7 @@ private ThreadMetrics() {}

public static Sensor createTaskSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor createTaskSensor = streamsMetrics.threadLevelSensor(CREATE_TASK, RecordingLevel.INFO);
addInvocationRateAndCount(createTaskSensor,
addInvocationRateAndCountToSensor(createTaskSensor,
THREAD_LEVEL_GROUP,
streamsMetrics.threadLevelTagMap(),
CREATE_TASK,
Expand All @@ -85,7 +85,7 @@ public static Sensor createTaskSensor(final StreamsMetricsImpl streamsMetrics) {

public static Sensor closeTaskSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor closeTaskSensor = streamsMetrics.threadLevelSensor(CLOSE_TASK, RecordingLevel.INFO);
addInvocationRateAndCount(closeTaskSensor,
addInvocationRateAndCountToSensor(closeTaskSensor,
THREAD_LEVEL_GROUP,
streamsMetrics.threadLevelTagMap(),
CLOSE_TASK,
Expand All @@ -97,8 +97,8 @@ public static Sensor closeTaskSensor(final StreamsMetricsImpl streamsMetrics) {
public static Sensor commitSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor commitSensor = streamsMetrics.threadLevelSensor(COMMIT, Sensor.RecordingLevel.INFO);
final Map<String, String> tagMap = streamsMetrics.threadLevelTagMap();
addAvgAndMax(commitSensor, THREAD_LEVEL_GROUP, tagMap, COMMIT_LATENCY);
addInvocationRateAndCount(commitSensor,
addAvgAndMaxToSensor(commitSensor, THREAD_LEVEL_GROUP, tagMap, COMMIT_LATENCY);
addInvocationRateAndCountToSensor(commitSensor,
THREAD_LEVEL_GROUP,
tagMap,
COMMIT,
Expand All @@ -110,8 +110,8 @@ public static Sensor commitSensor(final StreamsMetricsImpl streamsMetrics) {
public static Sensor pollSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor pollSensor = streamsMetrics.threadLevelSensor(POLL, Sensor.RecordingLevel.INFO);
final Map<String, String> tagMap = streamsMetrics.threadLevelTagMap();
addAvgAndMax(pollSensor, THREAD_LEVEL_GROUP, tagMap, POLL_LATENCY);
addInvocationRateAndCount(pollSensor,
addAvgAndMaxToSensor(pollSensor, THREAD_LEVEL_GROUP, tagMap, POLL_LATENCY);
addInvocationRateAndCountToSensor(pollSensor,
THREAD_LEVEL_GROUP,
tagMap,
POLL,
Expand All @@ -123,8 +123,8 @@ public static Sensor pollSensor(final StreamsMetricsImpl streamsMetrics) {
public static Sensor processSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor processSensor = streamsMetrics.threadLevelSensor(PROCESS, Sensor.RecordingLevel.INFO);
final Map<String, String> tagMap = streamsMetrics.threadLevelTagMap();
addAvgAndMax(processSensor, THREAD_LEVEL_GROUP, tagMap, PROCESS_LATENCY);
addInvocationRateAndCount(processSensor,
addAvgAndMaxToSensor(processSensor, THREAD_LEVEL_GROUP, tagMap, PROCESS_LATENCY);
addInvocationRateAndCountToSensor(processSensor,
THREAD_LEVEL_GROUP,
tagMap,
PROCESS,
Expand All @@ -137,8 +137,8 @@ public static Sensor processSensor(final StreamsMetricsImpl streamsMetrics) {
public static Sensor punctuateSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor punctuateSensor = streamsMetrics.threadLevelSensor(PUNCTUATE, Sensor.RecordingLevel.INFO);
final Map<String, String> tagMap = streamsMetrics.threadLevelTagMap();
addAvgAndMax(punctuateSensor, THREAD_LEVEL_GROUP, tagMap, PUNCTUATE_LATENCY);
addInvocationRateAndCount(punctuateSensor,
addAvgAndMaxToSensor(punctuateSensor, THREAD_LEVEL_GROUP, tagMap, PUNCTUATE_LATENCY);
addInvocationRateAndCountToSensor(punctuateSensor,
THREAD_LEVEL_GROUP,
tagMap,
PUNCTUATE,
Expand All @@ -150,7 +150,7 @@ public static Sensor punctuateSensor(final StreamsMetricsImpl streamsMetrics) {

public static Sensor skipRecordSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor skippedRecordsSensor = streamsMetrics.threadLevelSensor(SKIP_RECORD, Sensor.RecordingLevel.INFO);
addInvocationRateAndCount(skippedRecordsSensor,
addInvocationRateAndCountToSensor(skippedRecordsSensor,
THREAD_LEVEL_GROUP,
streamsMetrics.threadLevelTagMap(),
SKIP_RECORD,
Expand All @@ -163,11 +163,11 @@ public static Sensor skipRecordSensor(final StreamsMetricsImpl streamsMetrics) {
public static Sensor commitOverTasksSensor(final StreamsMetricsImpl streamsMetrics) {
final Sensor commitOverTasksSensor = streamsMetrics.threadLevelSensor(COMMIT, Sensor.RecordingLevel.DEBUG);
final Map<String, String> tagMap = streamsMetrics.threadLevelTagMap(TASK_ID_TAG, ALL_TASKS);
addAvgAndMax(commitOverTasksSensor,
addAvgAndMaxToSensor(commitOverTasksSensor,
TASK_LEVEL_GROUP,
tagMap,
COMMIT_LATENCY);
addInvocationRateAndCount(commitOverTasksSensor,
addInvocationRateAndCountToSensor(commitOverTasksSensor,
TASK_LEVEL_GROUP,
tagMap,
COMMIT,
Expand Down
4 changes: 2 additions & 2 deletions ...ain/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@
import java.util.Set;

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCountToSensor;

public class AbstractRocksDBSegmentedBytesStore<S extends Segment> implements SegmentedBytesStore {
private static final Logger LOG = LoggerFactory.getLogger(AbstractRocksDBSegmentedBytesStore.class);
Expand Down Expand Up @@ -182,7 +182,7 @@ public void init(final ProcessorContext context,
EXPIRED_WINDOW_RECORD_DROP,
Sensor.RecordingLevel.INFO
);
addInvocationRateAndCount(
addInvocationRateAndCountToSensor(
expiredRecordSensor,
"stream-" + metricScope + "-metrics",
metrics.tagMap("task-id", taskName, metricScope + "-id", name()),
Expand Down
8 changes: 4 additions & 4 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,6 @@
*/
package org.apache.kafka.streams.state.internals;

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;

import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
Expand All @@ -43,6 +40,9 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCountToSensor;

public class InMemorySessionStore implements SessionStore<Bytes, byte[]> {

private static final Logger LOG = LoggerFactory.getLogger(InMemorySessionStore.class);
Expand Down Expand Up @@ -82,7 +82,7 @@ public void init(final ProcessorContext context, final StateStore root) {
EXPIRED_WINDOW_RECORD_DROP,
Sensor.RecordingLevel.INFO
);
addInvocationRateAndCount(
addInvocationRateAndCountToSensor(
expiredRecordSensor,
"stream-" + metricScope + "-metrics",
metrics.tagMap("task-id", taskName, metricScope + "-id", name()),
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@
import java.util.NoSuchElementException;

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.EXPIRED_WINDOW_RECORD_DROP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCountToSensor;
import static org.apache.kafka.streams.state.internals.WindowKeySchema.extractStoreKeyBytes;
import static org.apache.kafka.streams.state.internals.WindowKeySchema.extractStoreTimestamp;

Expand Down Expand Up @@ -98,7 +98,7 @@ public void init(final ProcessorContext context, final StateStore root) {
EXPIRED_WINDOW_RECORD_DROP,
Sensor.RecordingLevel.INFO
);
addInvocationRateAndCount(
addInvocationRateAndCountToSensor(
expiredRecordSensor,
"stream-" + metricScope + "-metrics",
metrics.tagMap("task-id", taskName, metricScope + "-id", name()),
Expand Down
12 changes: 6 additions & 6 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/Sensors.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@

import java.util.Map;

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgMaxLatency;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndMaxLatencyToSensor;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCountToSensor;

public final class Sensors {
private Sensors() {}
Expand All @@ -42,11 +42,11 @@ public static Sensor createTaskAndStoreLatencyAndThroughputSensors(final Sensor.
final Map<String, String> taskTags,
final Map<String, String> storeTags) {
final Sensor taskSensor = metrics.taskLevelSensor(taskName, operation, level);
addAvgMaxLatency(taskSensor, metricsGroup, taskTags, operation);
addInvocationRateAndCount(taskSensor, metricsGroup, taskTags, operation);
addAvgAndMaxLatencyToSensor(taskSensor, metricsGroup, taskTags, operation);
addInvocationRateAndCountToSensor(taskSensor, metricsGroup, taskTags, operation);
final Sensor sensor = metrics.storeLevelSensor(taskName, storeName, operation, level, taskSensor);
addAvgMaxLatency(sensor, metricsGroup, storeTags, operation);
addInvocationRateAndCount(sensor, metricsGroup, storeTags, operation);
addAvgAndMaxLatencyToSensor(sensor, metricsGroup, storeTags, operation);
addInvocationRateAndCountToSensor(sensor, metricsGroup, storeTags, operation);
return sensor;
}

Expand Down
Loading