Skip to content

KAFKA-8578: Add basic functionality to expose RocksDB metrics #6979

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 9 commits into from
Aug 2, 2019
Merged

KAFKA-8578: Add basic functionality to expose RocksDB metrics #6979

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
3013214
Add methods to record RocksDB metrics
cadonna Jun 21, 2019
5f4624f
Incorporate feedback
cadonna Jul 3, 2019
2f3f161
Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…
cadonna Jul 5, 2019
bacdd7c
Incorporate feedback
cadonna Jul 9, 2019
1faae95
Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…
cadonna Jul 9, 2019
269ca2a
Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…
cadonna Jul 11, 2019
b4ccf2d
Reduce memory footprint for descriptions of RocksDB metrics
cadonna Jul 11, 2019
ba1cbbc
Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…
cadonna Jul 31, 2019
e531069
Incorporate feedback
cadonna Jul 31, 2019
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
83 changes: 75 additions & 8 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 @@ -23,9 +23,12 @@
import org.apache.kafka.common.metrics.Sensor.RecordingLevel;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.CumulativeCount;
import org.apache.kafka.common.metrics.stats.CumulativeSum;
import org.apache.kafka.common.metrics.stats.Max;
import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.metrics.stats.Value;
import org.apache.kafka.common.metrics.stats.WindowedCount;
import org.apache.kafka.common.metrics.stats.WindowedSum;
import org.apache.kafka.streams.StreamsMetrics;

import java.util.Arrays;
Expand Down Expand Up @@ -54,17 +57,21 @@ public class StreamsMetricsImpl implements StreamsMetrics {

public static final String THREAD_ID_TAG = "client-id";
public static final String TASK_ID_TAG = "task-id";
public static final String STORE_ID_TAG = "state-id";

@abbccdda abbccdda Jun 24, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

should it be called store-id then?

@guozhangwang guozhangwang Jun 24, 2019

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.

This is to be consistent with existing state store-level ids.


public static final String ALL_TASKS = "all";

public static final String LATENCY_SUFFIX = "-latency";
public static final String AVG_SUFFIX = "-avg";
public static final String MAX_SUFFIX = "-max";
public static final String MIN_SUFFIX = "-min";
public static final String RATE_SUFFIX = "-rate";
public static final String TOTAL_SUFFIX = "-total";
public static final String RATIO_SUFFIX = "-ratio";

public static final String THREAD_LEVEL_GROUP = "stream-metrics";
public static final String TASK_LEVEL_GROUP = "stream-task-metrics";
public static final String STATE_LEVEL_GROUP = "stream-state-metrics";

public static final String PROCESSOR_NODE_METRICS_GROUP = "stream-processor-node-metrics";
public static final String PROCESSOR_NODE_ID_TAG = "processor-node-id";
Expand Down Expand Up @@ -123,6 +130,18 @@ public final void removeAllThreadLevelSensors() {
}
}

public Map<String, String> taskLevelTagMap(final String taskName) {
final Map<String, String> tagMap = threadLevelTagMap();

@mjsax mjsax Jun 21, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

In threadLevelTagMap(String...) there is a check:

        if (tags != null) {
            if ((tags.length % 2) != 0) {
                throw new IllegalArgumentException("Tags needs to be specified in key-value pairs");
            }

Should we do the same here

@cadonna cadonna Jul 2, 2019

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.

I am not sure I can follow. We do not have a parameter String... tags here. What should we check?

@vvcephei vvcephei Jul 9, 2019

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.

Actually, I'm not sure what's going on with threadLevelTagMap(String...).

It has exactly one usage, which is streamsMetrics.threadLevelTagMap(TASK_ID_TAG, ALL_TASKS) which seems to be doing exactly the same thing this method is doing.

Maybe we can just get rid of threadLevelTagMap(String...) and migrate the usage to this method?

@cadonna cadonna Jul 9, 2019

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.

streamsMetrics.threadLevelTagMap(TASK_ID_TAG, ALL_TASKS) is used for a parent sensor that will be removed as part of KIP-444. When the that sensor will be removed also streamsMetrics.threadLevelTagMap(TASK_ID_TAG, ALL_TASKS) and the corresponding method will disappear. I would like to leave it for the future PR, since that code is not touched in this PR.

tagMap.put(TASK_ID_TAG, taskName);
return tagMap;
}

public Map<String, String> storeLevelTagMap(final String taskName, final String storeType, final String storeName) {

@mjsax mjsax Jun 21, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit: line to long (break each parameter in it's own line)

@cadonna cadonna Jul 2, 2019

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.

The line is exactly 120 characters long as allowed by the coding guidelines. ;-)

@vvcephei vvcephei Jul 9, 2019

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.

Can we just add https://checkstyle.sourceforge.io/config_sizes.html#LineLength so we never have to have this discussion again? :)

final Map<String, String> tagMap = taskLevelTagMap(taskName);

@mjsax mjsax Jun 21, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

As above

@cadonna cadonna Jul 2, 2019

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.

As above

tagMap.put(storeType + "-" + STORE_ID_TAG, storeName);
return tagMap;
}

public final Sensor taskLevelSensor(final String taskName,
final String sensorName,
final RecordingLevel recordingLevel,
Expand Down Expand Up @@ -237,9 +256,7 @@ public final Sensor storeLevelSensor(final String taskName,
if (!storeLevelSensors.containsKey(key)) {
storeLevelSensors.put(key, new LinkedList<>());
}

final String fullSensorName = key + SENSOR_NAME_DELIMITER + sensorName;

final Sensor sensor = metrics.sensor(fullSensorName, recordingLevel, parents);

storeLevelSensors.get(key).push(fullSensorName);
Expand Down Expand Up @@ -454,12 +471,62 @@ public static void addInvocationRateAndCount(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation) {
addInvocationRateAndCount(sensor,
group,
tags,
operation,
"The total number of " + operation,
"The average per-second number of " + operation);
addInvocationRateAndCount(
sensor,
group,
tags,
operation,
"The total number of " + operation,
"The average per-second number of " + operation
);
}

public static void addRateOfSumAndSumMetricsToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation,
final String descriptionOfRate,
final String descriptionOfTotal) {
addRateOfSumMetricToSensor(sensor, group, tags, operation, descriptionOfRate);
addSumMetricToSensor(sensor, group, tags, operation, descriptionOfTotal);
}

public static void addRateOfSumMetricToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation,
final String description) {
sensor.add(new MetricName(operation + RATE_SUFFIX, group, description, tags),
new Rate(TimeUnit.SECONDS, new WindowedSum()));
}

public static void addSumMetricToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String operation,
final String description) {
sensor.add(new MetricName(operation + TOTAL_SUFFIX, group, description, tags), new CumulativeSum());
}

public static void addValueMetricToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String name,
final String description) {
sensor.add(new MetricName(name, group, description, tags), new Value());
}

public static void addAvgAndSumMetricsToSensor(final Sensor sensor,
final String group,
final Map<String, String> tags,
final String metricNamePrefix,
final String descriptionOfAvg,
final String descriptionOfTotal) {
sensor.add(new MetricName(metricNamePrefix + AVG_SUFFIX, group, descriptionOfAvg, tags), new Avg());
sensor.add(
new MetricName(metricNamePrefix + TOTAL_SUFFIX, group, descriptionOfTotal, tags),
new CumulativeSum()
);
}

/**
Expand Down
Loading