Skip to content

KAFKA-7223: Suppression Buffer Metrics #5795

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 5 commits into from
Nov 27, 2018
Merged

KAFKA-7223: Suppression Buffer Metrics #5795

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f9c6c8e
KAFKA-7223: Suppression Buffer Metrics
vvcephei Oct 4, 2018
526dc74
remove suppression rate metric
vvcephei Nov 16, 2018
4e392a5
update buffer metrics on init and close
vvcephei Nov 16, 2018
fa4e858
remove unused field
vvcephei Nov 16, 2018
bd8806c
cache processor node metric strings as constants
vvcephei Nov 27, 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
45 changes: 44 additions & 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 @@ -20,12 +20,19 @@
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.Max;
import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.metrics.stats.Sum;
import org.apache.kafka.common.metrics.stats.Total;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;

import java.util.Map;
import java.util.concurrent.TimeUnit;

public class Sensors {

private static final String PROCESSOR_NODE_METRICS_GROUP = "stream-processor-node-metrics";

@vvcephei vvcephei Oct 12, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

might as well consolidate this repeated string.

@mjsax mjsax Nov 6, 2018

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.

I am always confused by metric. What it the group name used for (ie, semantic meaning)?

@vvcephei vvcephei Nov 14, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

AFAICT, I don't think it's "used for" anything in particular. It's part of the metric name, but so is the description.

I assume it's intended to be like a namespace.

@guozhangwang guozhangwang Nov 15, 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.

group name is used as a first "tag" of the metric name in JMX reporter: xxx-metrics:type=[group-name],[tag1]=[value1],...; for other reporters they can use the group name however they like.

@mjsax mjsax Nov 20, 2018 •
edited
Loading

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.

Sorry for my confusion. Let me rephrase my current understanding:
A "metric" is a single value that we track and report as key-value pair. A Sensor groups multiple metric together for ease of use -- each metric gets globally unique name that is put together with many parts.

  1. We prefix metric names (the xxx-metrics: part) with the clientId that is different for each StreamsThread.
  2. What is the purpose of group-name in the metric-name, ie, what metrics should use the same or different group name
  3. Each sensor has a name that is added to all metric names within the sensor (ie, the sensor name groups all it's contains metrics)
  4. we also put additional tags to add more meta information (task-id, processor-id) if appropriate to name each metric uniquely within a sensor.

To clarify: by "grouping" I meant use the same string for a specific part in the metric name. Ie, the prefix groups all metric based on the stream-thread, and the sensor name groups all it's contained metrics (as sub-group within the stream-thread group of metrics). Does this make sense?

@guozhangwang guozhangwang Nov 21, 2018 •
edited by mjsax
Loading

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.

To clarify: there are two entities: the metrics registry which organizes the metrics, and there is metrics reporter which regularly pulls from the registry to report the metric values.

Inside metrics registry there is sensor which as you understand is just a way of grouping metrics into meaningful clusters. The SensorName is just an id for distinguishing sensors in the metrics registry (i.e. you will see logic like if this sensor as already been created in the registry skip this step). A metric name presented in MetricName which contains groupName, tags, etc is just a logical entity in the registry. How to represent the metric names is up to the metrics reporter (different reporters can definitely represent it differently). As for the sensor names, they should never be seen outside the registry as the metrics reporter never exposed them.

@mjsax mjsax Nov 21, 2018

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.

Thanks @guozhangwang

I was just checking, and we also define "stream-processor-node-metrics" in ProcessorNode -- should we unify both, to have one constant only?

Think, we can also simplify ProcessorNode#createTaskAndNodeLatencyAndThroughputSensors and remove the group parameter.

@vvcephei vvcephei Nov 26, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Actually, someone filed a Jira last week reporting that a heap analysis identified this exact group name as responsible for Megabytes of heap space, so perhaps we should consolidate it into one constant!

@mjsax mjsax Nov 26, 2018

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.

If this is an issue, maybe we should do a fix first (in it's own PR, so we can cherry-pick), and rebase/merge this PR later? Thought? Are you planning to do a PR for the reported issue?

@vvcephei vvcephei Nov 27, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

oh. I've already added a commit to this PR.

I am planning a separate PR for a couple of other issues he identified. It's https://issues.apache.org/jira/browse/KAFKA-7660, by the way.


private Sensors() {}

public static Sensor lateRecordDropSensor(final InternalProcessorContext context) {
Expand All @@ -38,7 +45,7 @@ public static Sensor lateRecordDropSensor(final InternalProcessorContext context
);
StreamsMetricsImpl.addInvocationRateAndCount(
sensor,
"stream-processor-node-metrics",
PROCESSOR_NODE_METRICS_GROUP,
metrics.tagMap("task-id", context.taskId().toString(), "processor-node-id", context.currentNode().name()),
"late-record-drop"
);
Expand Down Expand Up @@ -75,4 +82,40 @@ public static Sensor recordLatenessSensor(final InternalProcessorContext context
);
return sensor;
}

public static Sensor suppressionEmitSensor(final InternalProcessorContext context) {

@vvcephei vvcephei Oct 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This sensor is different than that proposed in the kip. During implementation, I noticed a weird asymmetry in which the processor node would measure the suppression, but the buffer would measure the "eviction" aka "emission" aka "forwarding".

I'm proposing to update the KIP to make these two measurements symmetrical.

@bbejeck bbejeck Oct 19, 2018

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.

I didn't think of that before, but now that you mention it, the change makes sense to me.

final StreamsMetricsImpl metrics = context.metrics();

final Sensor sensor = metrics.nodeLevelSensor(
context.taskId().toString(),
context.currentNode().name(),
"suppression-emit",
Sensor.RecordingLevel.DEBUG
);

final Map<String, String> tags = metrics.tagMap(
"task-id", context.taskId().toString(),
"processor-node-id", context.currentNode().name()
);

sensor.add(
new MetricName(

@bbejeck bbejeck Oct 19, 2018

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.

seems like the contents of the two sensor#add methods are the same as above maybe refactor into two methods?

"suppression-emit-rate",
PROCESSOR_NODE_METRICS_GROUP,
"The average number of occurrence of suppression-emit operation per second.",
tags
),
new Rate(TimeUnit.SECONDS, new Sum())
);
sensor.add(
new MetricName(
"suppression-emit-total",
PROCESSOR_NODE_METRICS_GROUP,
"The total number of occurrence of suppression-emit operations.",
tags
),
new Total()
);
return sensor;
}
}
8 changes: 7 additions & 1 deletion ...ain/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,14 @@
*/
package org.apache.kafka.streams.kstream.internals.suppress;

import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.internals.Change;
import org.apache.kafka.streams.kstream.internals.FullChangeSerde;
import org.apache.kafka.streams.kstream.internals.metrics.Sensors;
import org.apache.kafka.streams.kstream.internals.suppress.TimeDefinitions.TimeDefinition;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
Expand All @@ -42,9 +44,10 @@ public class KTableSuppressProcessor<K, V> implements Processor<K, Change<V>> {
private final BufferFullStrategy bufferFullStrategy;
private final boolean shouldSuppressTombstones;
private final String storeName;

private TimeOrderedKeyValueBuffer buffer;
private InternalProcessorContext internalProcessorContext;

private Sensor suppressionEmitSensor;
private Serde<K> keySerde;
private FullChangeSerde<V> valueSerde;

Expand All @@ -68,6 +71,8 @@ public KTableSuppressProcessor(final SuppressedInternal<K> suppress,
@Override
public void init(final ProcessorContext context) {
internalProcessorContext = (InternalProcessorContext) context;
suppressionEmitSensor = Sensors.suppressionEmitSensor(internalProcessorContext);

keySerde = keySerde == null ? (Serde<K>) context.keySerde() : keySerde;
valueSerde = valueSerde == null ? FullChangeSerde.castOrWrap(context.valueSerde()) : valueSerde;
buffer = Objects.requireNonNull((TimeOrderedKeyValueBuffer) context.getStateStore(storeName));
Expand Down Expand Up @@ -123,6 +128,7 @@ private void emit(final KeyValue<Bytes, ContextualRecord> toEmit) {
try {
final K key = keySerde.deserializer().deserialize(null, toEmit.key.get());
internalProcessorContext.forward(key, value);
suppressionEmitSensor.record();
} finally {
internalProcessorContext.setRecordContext(prevRecordContext);
}
Expand Down
26 changes: 25 additions & 1 deletion ...main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,17 +17,20 @@
package org.apache.kafka.streams.state.internals;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
import org.apache.kafka.streams.processor.internals.RecordBatchingStateRestoreCallback;
import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.internals.metrics.Sensors;

import java.nio.ByteBuffer;
import java.util.Collection;
Expand Down Expand Up @@ -57,6 +60,8 @@ public class InMemoryTimeOrderedKeyValueBuffer implements TimeOrderedKeyValueBuf
private long minTimestamp = Long.MAX_VALUE;
private RecordCollector collector;
private String changelogTopic;
private Sensor bufferSizeSensor;

@guozhangwang guozhangwang Nov 15, 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.

In the wiki there is another metric for suppression-mem-buffer-evict, is that not in the scope of this PR?

@vvcephei vvcephei Nov 15, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Ah, now that you mention it, when I originally created this PR, I noticed there were some shortcomings with the metrics I proposed in the KIP, so I wanted to get the reviewers' feedback on the form it has taken in this PR before proposing an update to the KIP.

In particular, I proposed an asymmetric pair of metrics where the processor node would measure the number of incoming events (intermediate-result-suppression), but the buffer would measure the outgoing (emitted==evicted) events. It seemed better to offer a symmetric pair of in/out metrics on either the processor or the buffer. In this PR, I chose to put them both on the node.

In other words, I'm proposing to replace suppression-mem-buffer-evict with suppression-emit.

Of course, in this same review, you have proposed to drop intermediate-result-suppression in favor of the existing process-rate and process-total node-level metrics. This reduces the argument, but I think it still makes sense to make this a node metric instead of a buffer metric, since (IMHO) it more closely matches the intent of using a suppression node to control the emission rate.

@guozhangwang guozhangwang Nov 16, 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.

Thanks for the clarification, that sounds good to me.

private Sensor bufferCountSensor;

private volatile boolean open;

Expand Down Expand Up @@ -174,11 +179,16 @@ public boolean persistent() {

@Override
public void init(final ProcessorContext context, final StateStore root) {
final InternalProcessorContext internalProcessorContext = (InternalProcessorContext) context;
bufferSizeSensor = Sensors.createBufferSizeSensor(this, internalProcessorContext);
bufferCountSensor = Sensors.createBufferCountSensor(this, internalProcessorContext);

context.register(root, (RecordBatchingStateRestoreCallback) this::restoreBatch);
if (loggingEnabled) {
collector = ((RecordCollector.Supplier) context).recordCollector();
changelogTopic = ProcessorStateManager.storeChangelogTopic(context.applicationId(), storeName);
}
updateBufferMetrics();
open = true;
}

Expand All @@ -189,12 +199,13 @@ public boolean isOpen() {

@Override
public void close() {
open = false;
index.clear();
sortedMap.clear();
dirtyKeys.clear();
memBufferSize = 0;
minTimestamp = Long.MAX_VALUE;
open = false;

@mjsax mjsax Nov 21, 2018

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.

Why moving this?

@vvcephei vvcephei Nov 26, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

so that the buffer will report as "closed" immediately after this method is called, rather than only after the close operation has completed.

In retrospect, it seemed safer for :

  • open() to set open=true at when opening has completed
  • close() to set open=false when closing has started

This way, observers who first check that a store is open before querying would never be fooled into querying a partially closed store.

updateBufferMetrics();
}

@Override
Expand Down Expand Up @@ -265,13 +276,15 @@ private void restoreBatch(final Collection<ConsumerRecord<byte[], byte[]>> batch
);
}
}
updateBufferMetrics();
}


@Override
public void evictWhile(final Supplier<Boolean> predicate,
final Consumer<KeyValue<Bytes, ContextualRecord>> callback) {
final Iterator<Map.Entry<BufferKey, ContextualRecord>> delegate = sortedMap.entrySet().iterator();
int evictions = 0;

if (predicate.get()) {
Map.Entry<BufferKey, ContextualRecord> next = null;
Expand All @@ -298,8 +311,13 @@ public void evictWhile(final Supplier<Boolean> predicate,
next = null;
minTimestamp = Long.MAX_VALUE;
}

evictions++;
}
}
if (evictions > 0) {
updateBufferMetrics();
}

@vvcephei vvcephei Oct 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Avoid recording metrics if they haven't changed.

}

@Override
Expand All @@ -308,6 +326,7 @@ public void put(final long time,
final ContextualRecord value) {
cleanPut(time, key, value);
dirtyKeys.add(key);
updateBufferMetrics();
}

private void cleanPut(final long time, final Bytes key, final ContextualRecord value) {
Expand Down Expand Up @@ -355,4 +374,9 @@ private long computeRecordSize(final Bytes key, final ContextualRecord value) {
}
return size;
}

private void updateBufferMetrics() {

@mjsax mjsax Nov 6, 2018

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.

Should we call this in init() and/or close(), too?

@vvcephei vvcephei Nov 15, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

sure; I don't think they're strictly necessary, but it doesn't hurt, and it'll make the system more resilient to future changes.

bufferSizeSensor.record(memBufferSize);
bufferCountSensor.record(index.size());
}
}
69 changes: 68 additions & 1 deletion 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 @@ -16,7 +16,13 @@
*/
package org.apache.kafka.streams.state.internals.metrics;

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.Max;
import org.apache.kafka.common.metrics.stats.Value;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;

import java.util.Map;
Expand All @@ -25,7 +31,6 @@
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addInvocationRateAndCount;

public final class Sensors {

private Sensors() {}

public static Sensor createTaskAndStoreLatencyAndThroughputSensors(final Sensor.RecordingLevel level,
Expand All @@ -44,5 +49,67 @@ public static Sensor createTaskAndStoreLatencyAndThroughputSensors(final Sensor.
addInvocationRateAndCount(sensor, metricsGroup, storeTags, operation);
return sensor;
}

public static Sensor createBufferSizeSensor(final StateStore store,
final InternalProcessorContext context) {
return getBufferSizeOrCountSensor(store, context, "size");
}

public static Sensor createBufferCountSensor(final StateStore store,
final InternalProcessorContext context) {
return getBufferSizeOrCountSensor(store, context, "count");
}

private static Sensor getBufferSizeOrCountSensor(final StateStore store,
final InternalProcessorContext context,
final String property) {
final StreamsMetricsImpl metrics = context.metrics();

final String sensorName = "suppression-buffer-" + property;

final Sensor sensor = metrics.storeLevelSensor(
context.taskId().toString(),
store.name(),
sensorName,
Sensor.RecordingLevel.DEBUG
);

final String metricsGroup = "stream-buffer-metrics";

@vvcephei vvcephei Oct 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

In the KIP, I erroneously proposed to make this a node-level sensor. It should be a store-level sensor instead. I'm proposing to update the KIP if you all agree.

@bbejeck bbejeck Oct 19, 2018

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.

Yeah, that makes sense to me. +1

@guozhangwang guozhangwang Nov 15, 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.

+1

@guozhangwang guozhangwang Nov 15, 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.

This is not newly introduced in this PR but: although we call it stream-buffer-metrics where the buffer-id is the store-name, BUT the store name could be an auto-generated name. Note for the named cache we call it stream-record-cache-metrics but the cache name is the store name, and hence we have the same issue.

But here we use the store.name which would be KTABLE-SUPPRESS-XXXX-store if it is auto generated, maybe it's less confusing to use the processor-name instead which would be KTABLE-SUPPRESS-XXX if it is auto generated. Note even if Suppressed.name() is specified as XYZ with store.name() we still set the buffer-id as XYZ-store instead.

@vvcephei vvcephei Nov 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I think it's good to be aware of this, but it seems like reporting the store name is the right thing to do here.

Note that the buffer in this case is the store for the suppression processor. So, using the processor name instead of the full store name would be more analogous to doing the same thing for a K/V Store.

As written, the metrics that are specific to the buffer (aka store) will be tagged with a name that matches the name of that component when you describe the topology.

Does that seem right, or have I missed your point?


final Map<String, String> tags = metrics.tagMap(
"task-id", context.taskId().toString(),
"buffer-id", store.name()
);

sensor.add(
new MetricName(
sensorName + "-current",
metricsGroup,
"The current " + property + " of buffered records.",
tags),
new Value()
);


sensor.add(
new MetricName(
sensorName + "-avg",
metricsGroup,
"The average " + property + " of buffered records.",
tags),
new Avg()
);

sensor.add(
new MetricName(
sensorName + "-max",
metricsGroup,
"The max " + property + " of buffered records.",
tags),
new Max()
);

return sensor;
}
}

Loading