KAFKA-8578: Add basic functionality to expose RocksDB metrics by cadonna · Pull Request #6979 · apache/kafka

cadonna · 2019-06-21T15:50:29Z

Adds RocksDBMetrics class that provides methods to get
sensors from the Kafka metrics registry and to setup the
sensors to record RocksDB metrics
Extends StreamsMetricsImpl with functionality to add the
required metrics to the sensors.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

- Adds `RocksDBMetrics` class that provides methods to get sensors from the Kafka metrics registry and to setup the sensors to record RocksDB metrics - Extends `StreamsMetricsImpl` with functionality to add the required metrics to the sensors.

cadonna · 2019-06-21T15:53:35Z

Call for review @pkleindl @ableegoldman @bbejeck @vvcephei @guozhangwang @mjsax @abbccdda

mjsax · 2019-06-21T17:47:21Z

+        return tagMap;
+    }
+
+    public Map<String, String> storeLevelTagMap(final String taskName, final String storeType, final String storeName) {


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

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

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

mjsax · 2019-06-21T17:49:28Z

    }

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


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

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

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?

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.

mjsax · 2019-06-21T17:49:59Z

+    }
+
+    public Map<String, String> storeLevelTagMap(final String taskName, final String storeType, final String storeName) {
+        final Map<String, String> tagMap = taskLevelTagMap(taskName);


mjsax · 2019-06-21T17:56:08Z

+                                                      final String storeName) {
+        final Sensor sensor = streamsMetrics
+            .storeLevelSensor(taskName, storeName, BYTES_WRITTEN_TO_DB, RecordingLevel.DEBUG);
+        addAmountRateAndTotalMetricsToSensor(sensor,


nit: for method calls, we usually format like this:

addAmountRateAndTotalMetricsToSensor( sensor, ...); Breaking the first parameter already reduced line length and seems preferable.

mjsax · 2019-06-21T17:57:09Z

+                                             streamsMetrics.storeLevelTagMap(taskName,
+                                                                             STORE_TYPE_PREFIX + storeType, storeName),
+                                             BYTES_WRITTEN_TO_DB,
+                                             "Average per-second number of bytes written to the RocksDB state store",


Nit: Average number of bytes written per-second to the RocksDB state store ?

mjsax · 2019-06-21T17:57:42Z

+                                                     final String storeName) {
+        final Sensor sensor = streamsMetrics
+            .storeLevelSensor(taskName, storeName, BYTES_READ_FROM_DB, RecordingLevel.DEBUG);
+        addAmountRateAndTotalMetricsToSensor(sensor,


mjsax · 2019-06-21T17:57:51Z

+            .storeLevelSensor(taskName, storeName, BYTES_WRITTEN_TO_DB, RecordingLevel.DEBUG);
+        addAmountRateAndTotalMetricsToSensor(sensor,
+                                             STATE_LEVEL_GROUP,
+                                             streamsMetrics.storeLevelTagMap(taskName,


mjsax · 2019-06-21T17:57:59Z

+                                             streamsMetrics.storeLevelTagMap(taskName,
+                                                                             STORE_TYPE_PREFIX + storeType, storeName),
+                                             BYTES_READ_FROM_DB,
+                                             "Average per-second number of bytes read from the RocksDB state store",


mjsax · 2019-06-21T18:11:22Z

+    public void shouldAddAmountRateAndSum() {
+        final MockTime time = new MockTime(0);
+
+        addAmountRateAndTotalMetricsToSensor(sensor, group, tags, metricNamePrefix, description1, description2);


Nit: I think it is confusing to import methods under test (makes it unclear if this is a "setup" method -- might be better to not use static imports) ?

mjsax · 2019-06-21T18:17:23Z

+        final double value2 = 18.0;
+        sensor.record(value2, time.milliseconds());
+        assertThat(metric.measurable().measure(new MetricConfig(), time.milliseconds()),
+                   equalTo(initialValue + value1 + value2));


Where does initialValue come from? For example: shouldAddAvgAndTotalMetricsToSensor add the sensor and calls verifySumMetric -- why is the initial sum not zero?

The initial value is not zero, because in shouldAddAvgAndTotalMetricsToSensor two verifications are called, one verifies the Avg metric and the other the Sum metric. The first verification records two values and the second verification also records two values. Inside the metric the values are computed over samples. All four values added in the verifications are added to the samples over which the metric is computed. That means that the two values added for the first verification are already in the metric sample when the second verification is called. Hence, the initial value in the second verification is the sum of the two values added to the metric in the first verification.

I changed the test to get rid of the initial value. Now, I advance mock time as long as it needs to get empty samples.

abbccdda · 2019-06-24T15:47:24Z


    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";


should it be called store-id then?

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

abbccdda · 2019-06-24T15:54:04Z

+        return sensor;
+    }
+
+    public static Sensor memtableBytesFlushedSensor(final StreamsMetricsImpl streamsMetrics,


One thing I would suggest we do is to create an intermediate struct to store all the parameters, in case later we need to add more fields for sensor creation.

Good idea! Done!

abbccdda · 2019-06-24T15:56:30Z

+    private static final String MEMTABLE_BYTES_FLUSHED = "memtable-bytes-flushed";
+    private static final String MEMTABLE_HIT_RATIO = "memtable-hit" + RATIO_SUFFIX;
+    private static final String MEMTABLE_FLUSH_TIME = "memtable-flush-time";
+    private static final String WRITE_STALL_DURATION = "write-stall-duration";


I would suggest we co-locate the description with metrics and refactor out the sensor creation part, this may help reduce code redundancy.

I am afraid, I cannot follow. Could you elaborate?

Maybe @abbccdda was saying to store the description strings in static final fields as well? I'd agree with this, just to be sure that we don't waste memory on duplicate description strings.

Does this make any difference since all description strings are literals in static methods?

Yes, each time the method is invoked (I.e., once per store), a separate copy of the string is placed on the heap. I previously didn't think this would be a big factor, but someone in the community profiled the memory usage of a long-running topology and found that these strings tend to accumulate over time.

There's no need to worry about this for short-scoped strings like exception messages, but metrics are long-lived objects, and we benefit from making them static constants.

Oh, I see makes sense.

abbccdda · 2019-06-24T15:57:25Z

+
+    @Test
+    public void shouldAddAmountRateAndSum() {
+        final MockTime time = new MockTime(0);


How about making time a private field and do a setUp call to reinitialize it every time?

abbccdda · 2019-06-24T15:58:59Z

+        assertThat(metrics.metrics().size(), equalTo(2 + 1)); // one metric is added automatically in the constructor of Metrics
+    }
+
+    private void verifySumMetric(final Metrics metrics,


It's more reusable if we make value1 and value2 both parameters. Maybe we could pass a List along with initialValue in this case?

I would like to leave this method as it is because it suffices to use two values to verify that the statistical function set in the metric is a sum. IMO a test should be as simple as possible to avoid errors. Additionally, the expected sum would not only depend on the list of values but also on the instance of MetricConfig. This dependency would unnecessarily complicate the logic of this verification.

abbccdda · 2019-06-24T15:59:37Z

+        sensor.record(value1, time.milliseconds());
+        final double value2 = 18.0;
+        sensor.record(value2, time.milliseconds());
+        time.sleep(30000);


This sleep is a bit too long for a unit test, could we consider avoiding it?

This is NOT an actual sleep. This call advances the mocked time by 30 seconds. It does not wait at all.

abbccdda · 2019-06-24T16:01:31Z

+        replayAndVerify(sensorCreator);
+    }
+
+    private void verifyRateSensor(final String metricNamePrefix,


nit: I slightly doubt whether these exact match tests are necessary

Could you elaborate on your doubts? How should I otherwise test, if the methods in RocksDBMetrics create the sensors with the correct name, tags and metrics? Please let me know, if you have a better idea.

bbejeck

@cadonna I made a pass overall looks good I only have one minor comment in addition to the other comments.

bbejeck · 2019-06-24T20:12:25Z

+        sensor.record(value1, time.milliseconds());
+        final double value2 = 18.0;
+        sensor.record(value2, time.milliseconds());
+        time.sleep(30000);


here it might work better to use TestUtils.waitForCondition instead of an arbitrary sleep call

This is NOT an actual sleep. This call advances the mocked time by 30 seconds. It does not wait at all.

bbejeck · 2019-06-24T20:26:53Z

Java 8 and Java 11 failed, test results already cleaned up.

Retest this please

guozhangwang · 2019-06-24T23:09:43Z


    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";


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

guozhangwang · 2019-06-24T23:20:46Z

                                  "The average per-second number of " + operation);
    }

+    public static void addAmountRateAndTotalMetricsToSensor(final Sensor sensor,


This is not part of the PR but I realized that CumulativeCount is in streams.processor.internals.metrics but not in common.metrics.stats. Is this intentional? @vvcephei

IIRC, it was done to avoid a KIP.

Of course, that may or may not have been the right thing to do, but I believe it was the idea... Do you think we should move it?

Ok, I took a closer look and had a bit of a flash-back to how confusing these metrics are... Maybe #7057 will help.

Thank you @vvcephei !

guozhangwang · 2019-06-24T23:28:13Z

+
+    public Map<String, String> storeLevelTagMap(final String taskName, final String storeType, final String storeName) {
+        final Map<String, String> tagMap = taskLevelTagMap(taskName);
+        tagMap.put(storeType + STORE_ID_TAG, storeName);


Are the storeType supposed to have the dash already? If not this should be storeType + "-" + STORE_ID_TAG.

Good point! I think I am going to change this to be consistent with the other labels.

guozhangwang · 2019-06-24T23:34:41Z

+                                                 final String storeName) {
+        final Sensor sensor = streamsMetrics
+            .storeLevelSensor(taskName, storeName, COMPACTION_TIME + MIN_SUFFIX, RecordingLevel.DEBUG);
+        addValueMetricToSensor(sensor,


I'm a bit confused here for max / min we are creating a sensor with type Value still, but since I did not see how the record() is going to be invoked maybe I'm just wrong.. otherwise, could we just use a similar function like addAvgAndMax?

For the RocksDB statistics that come in the form of histograms, we already get the min, max, and average values from RocksDB. We still need to decide if we should reset the histogram after each recording to have
kind of a moving window semantic or if we are fine with the histograms from the start of the application. WDYT?

Just chiming in, I think we're just defining the metrics in this PR and then actually recording them later. We might change the actual metric type later on when we look at how it's going to actually get recorded, so maybe we shouldn't worry too much about the selected metric implementation right now.

…rics-basics

vvcephei · 2019-07-09T15:48:59Z

    }

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


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?

vvcephei · 2019-07-09T15:50:55Z

+        return tagMap;
+    }
+
+    public Map<String, String> storeLevelTagMap(final String taskName, final String storeType, final String storeName) {


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

vvcephei · 2019-07-09T15:54:42Z

+    private static final String MEMTABLE_BYTES_FLUSHED = "memtable-bytes-flushed";
+    private static final String MEMTABLE_HIT_RATIO = "memtable-hit" + RATIO_SUFFIX;
+    private static final String MEMTABLE_FLUSH_TIME = "memtable-flush-time";
+    private static final String WRITE_STALL_DURATION = "write-stall-duration";


Maybe @abbccdda was saying to store the description strings in static final fields as well? I'd agree with this, just to be sure that we don't waste memory on duplicate description strings.

vvcephei · 2019-07-09T17:15:02Z

                                  "The average per-second number of " + operation);
    }

+    public static void addAmountRateAndTotalMetricsToSensor(final Sensor sensor,


Ok, I took a closer look and had a bit of a flash-back to how confusing these metrics are... Maybe #7057 will help.

vvcephei · 2019-07-09T17:17:41Z

+        sensor.add(new MetricName(name, group, description, tags), new Value());
+    }
+
+    public static void addAvgAndTotalMetricsToSensor(final Sensor sensor,


Suggested change

public static void addAvgAndTotalMetricsToSensor(final Sensor sensor,

public static void addAvgAndSumMetricsToSensor(final Sensor sensor,

vvcephei · 2019-07-09T17:20:11Z

+                                                 final String storeName) {
+        final Sensor sensor = streamsMetrics
+            .storeLevelSensor(taskName, storeName, COMPACTION_TIME + MIN_SUFFIX, RecordingLevel.DEBUG);
+        addValueMetricToSensor(sensor,


Just chiming in, I think we're just defining the metrics in this PR and then actually recording them later. We might change the actual metric type later on when we look at how it's going to actually get recorded, so maybe we shouldn't worry too much about the selected metric implementation right now.

vvcephei · 2019-07-09T17:29:35Z

+        verifyAll();
+        verify(StreamsMetricsImpl.class);
+
+        assertThat(sensor, is(this.sensor));


Hey, I'm sorry, but can you explain what's going on here? Sensor doesn't override equals, so I'm not seeing how this assertion works.

Here I am just verifying that the methods in RocksDBMetrics pass through the sensor they get from the methods in the StreamsMetricImpl mock. It is an equality check on the reference. Before verifying the sensor, I also verify through the mock whether the metrics with the correct names and tags are added to the sensor.

Oh, right, I forgot that the metrics registry returns the same copy of the sensor when all the name, description, and tags are the same... Thanks.

…rics-basics

cadonna · 2019-07-10T10:16:41Z

Failures unrelated
Retest this, please

…rics-basics

cadonna · 2019-07-12T14:38:03Z

Retest this, please

guozhangwang · 2019-07-19T16:51:40Z

                                  "The average per-second number of " + operation);
    }

+    public static void addAmountRateAndTotalMetricsToSensor(final Sensor sensor,


Thank you @vvcephei !

guozhangwang · 2019-07-19T16:58:37Z

+        addTotalMetricToSensor(sensor, group, tags, operation, descriptionOfTotal);
+    }
+
+    public static void addAmountRateMetricToSensor(final Sensor sensor,


This is not for this PR, but: since in #7057 we are cleaning up the confusing names shall we do the same for function names as well as a follow-up? Also the existing function names do not have ..ToSensor suffix, could we make it consistent (I'm fine with either way, just want to do the same for new and existing functions)?

To make the function name consistent with the Stat name, should this be addRateOfSum?

This is not for this PR, but: since in #7057 we are cleaning up the confusing names shall we do the same for function names as well as a follow-up? Also the existing function names do not have ..ToSensor suffix, could we make it consistent (I'm fine with either way, just want to do the same for new and existing functions)?

I agree with you that we should be consistent. I just did not want to make this PR bigger than needed since it is already quite big. I will open a refactoring PR for this.

To make the function name consistent with the Stat name, should this be addRateOfSum?

Done

guozhangwang · 2019-07-19T17:00:31Z

+                   new Rate(TimeUnit.SECONDS, new Sum()));
+    }
+
+    public static void addTotalMetricToSensor(final Sensor sensor,


Similarly: addSum?

guozhangwang · 2019-07-19T17:04:51Z

        }
+    }
+
+    @Test


This is a meta comment: for SampledStat their recorded values can vary based on the windowing itself, so in unit test it may not always has the exact value, i.e. asserting equalTo(value) may make this flaky, whereas just testing it is non-zero is sufficient.

IIUC, since SampledStat metrics do not maintain any clock internally and I use MockTime in the unit tests, the values recorded in SampledStat should be deterministic. Please, correct me if I am missing anything.

If we are using MockTime then it should be fine.

guozhangwang · 2019-07-19T17:08:18Z

+        "Total number of bytes read from the RocksDB state store";
+    private static final String MEMTABLE_BYTES_FLUSHED_RATE_DESCRIPTION =
+        "Average number of bytes flushed per second from the memtable to disk";
+    private static final String MEMTABLE_BYTES_FLUSHED_AVG_DESCRIPTION =


The description Total number ... seems incorrect for AVG?

Oh... Thank you! Actually, it is the other way around. The description is correct, but the variable name is wrong.

guozhangwang · 2019-07-19T17:09:22Z

+        "Minimum time spent on flushing the memtable to disk in ms";
+    private static final String MEMTABLE_FLUSH_TIME_MAX_DESCRIPTION =
+        "Maximum time spent on flushing the memtable to disk in ms";
+    private static final String WRITE_STALL_DURATION_AVG_DESCRIPTION = "Moving average duration of write stalls in ms";


Moving average duration may be a bit confusing to readers, maybe just Average duration of ...

I will change it to Average duration to be consistent with the other average metrics. However, I think we risk wrong expectations on user side if we do not improve the descriptions of the metrics. Especially, with having compaction-time-avg which is not a moving average, but it has the following description Average time spent on compaction in ms.

…rics-basics

cadonna · 2019-07-31T20:22:24Z

Retest this, please

cadonna · 2019-08-01T08:24:59Z

Retest this, please

guozhangwang · 2019-08-02T16:51:15Z

Merged to trunk, thanks @cadonna !!

Add methods to record RocksDB metrics

3013214

- Adds `RocksDBMetrics` class that provides methods to get sensors from the Kafka metrics registry and to setup the sensors to record RocksDB metrics - Extends `StreamsMetricsImpl` with functionality to add the required metrics to the sensors.

mjsax added the streams label Jun 21, 2019

mjsax reviewed Jun 21, 2019

View reviewed changes

abbccdda reviewed Jun 24, 2019

View reviewed changes

bbejeck reviewed Jun 24, 2019

View reviewed changes

guozhangwang reviewed Jun 24, 2019

View reviewed changes

cadonna added 2 commits July 5, 2019 11:03

Incorporate feedback

5f4624f

Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…

2f3f161

…rics-basics

vvcephei mentioned this pull request Jul 9, 2019

KAFKA-8696: clean up Sum/Count/Total metrics #7057

Merged

3 tasks

vvcephei reviewed Jul 9, 2019

View reviewed changes

cadonna added 2 commits July 9, 2019 21:46

Incorporate feedback

bacdd7c

Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…

1faae95

…rics-basics

cadonna added 2 commits July 11, 2019 17:29

Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…

269ca2a

…rics-basics

Reduce memory footprint for descriptions of RocksDB metrics

b4ccf2d

vvcephei approved these changes Jul 12, 2019

View reviewed changes

guozhangwang reviewed Jul 19, 2019

View reviewed changes

cadonna added 2 commits July 31, 2019 11:29

Merge remote-tracking branch 'upstream/trunk' into AK8578-RocksDB-met…

ba1cbbc

…rics-basics

Incorporate feedback

e531069

guozhangwang merged commit a7d0fdd into apache:trunk Aug 2, 2019

cadonna deleted the AK8578-RocksDB-metrics-basics branch October 21, 2019 11:40

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

	public static void addAvgAndTotalMetricsToSensor(final Sensor sensor,
	public static void addAvgAndSumMetricsToSensor(final Sensor sensor,

                       }
+                  }
+                  @Test

Uh oh!

Conversation

cadonna commented Jun 21, 2019

Committer Checklist (excluded from commit message)

Uh oh!

cadonna commented Jun 21, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment