Skip to content
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

Add Buffer Latency Metric #4237

Merged
merged 4 commits into from
Mar 7, 2024
Merged

Add Buffer Latency Metric #4237

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9dd30ec
Add Buffer Latency Metric
Mar 5, 2024
5948bcd
Addressed review comments
Mar 6, 2024
f954bef
Fixed failing code verification test by adding new test case
Mar 6, 2024
6debe01
Added null check before calling updateLatency
Mar 6, 2024
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: 2 additions & 0 deletions data-prepper-api/src/main/java/org/opensearch/dataprepper/metrics/MetricNames.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,4 +86,6 @@ private MetricNames() {}
* Delimiter used to separate path components in metric names.
*/
public static final String DELIMITER = ".";

public static final String LATENCY = "latency";
Copy link
Member

Choose a reason for hiding this comment

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

Should we use a different name here? The term latency is ambiguous as there are numerous places where latency exists.

It appears that this is the time before events are read from the buffer. Some name ideas:

  • bufferReadLatency
  • readLatency
  • timeEventIdle
  • atRestTime (to put it in opposition to in-flight)

}
16 changes: 16 additions & 0 deletions data-prepper-api/src/main/java/org/opensearch/dataprepper/model/buffer/AbstractBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,11 @@
import org.opensearch.dataprepper.model.CheckpointState;
import org.opensearch.dataprepper.model.configuration.PluginSetting;
import org.opensearch.dataprepper.model.record.Record;
import org.opensearch.dataprepper.model.event.Event;
import org.opensearch.dataprepper.model.event.DefaultEventHandle;

import java.time.Instant;
import java.time.Duration;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.TimeUnit;
Expand All @@ -32,6 +36,7 @@ public abstract class AbstractBuffer<T extends Record<?>> implements Buffer<T> {
private final Counter writeTimeoutCounter;
private final Counter recordsWriteFailed;
private final Timer writeTimer;
private final Timer latencyTimer;
private final Timer readTimer;
private final Timer checkpointTimer;

Expand All @@ -54,6 +59,7 @@ private AbstractBuffer(final PluginMetrics pluginMetrics, final String pipelineN
this.writeTimeoutCounter = pluginMetrics.counter(MetricNames.WRITE_TIMEOUTS);
this.writeTimer = pluginMetrics.timer(MetricNames.WRITE_TIME_ELAPSED);
this.readTimer = pluginMetrics.timer(MetricNames.READ_TIME_ELAPSED);
this.latencyTimer = pluginMetrics.timer(MetricNames.LATENCY);
this.checkpointTimer = pluginMetrics.timer(MetricNames.CHECKPOINT_TIME_ELAPSED);
}

Expand Down Expand Up @@ -142,6 +148,16 @@ public Map.Entry<Collection<T>, CheckpointState> read(int timeoutInMillis) {
return readResult;
}

protected void updateLatency(Collection<T> records) {
for (T rec : records) {
if (rec instanceof Record) {
Record<Event> record = (Record<Event>)rec;
Copy link
Member

Choose a reason for hiding this comment

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

Not all buffers hold records at the moment.

We should change these lines to support non-events. Also, I think you don't need to cast to DefaultEventHandle.

Object data = rec.getData();
if(data instanceof Event) {
  Event event = (Event) data;
  Instant receivedTime = event.getEventHandle().getInternalOriginationTime();
  latencyTimer.record(Duration.between(receivedTime, Instant.now()));
}

Instant receivedTime = ((DefaultEventHandle)(((Event)record.getData()).getEventHandle())).getInternalOriginationTime();
latencyTimer.record(Duration.between(receivedTime, Instant.now()));
}
}
}

@Override
public void checkpoint(final CheckpointState checkpointState) {
checkpointTimer.record(() -> doCheckpoint(checkpointState));
Expand Down
1 change: 1 addition & 0 deletions ...re/src/main/java/org/opensearch/dataprepper/peerforwarder/PeerForwarderReceiveBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@ public Map.Entry<Collection<T>, CheckpointState> doRead(final int timeoutInMilli
}
}

updateLatency((Collection<T>)records);
final CheckpointState checkpointState = new CheckpointState(recordsRead);
recordsInFlight += checkpointState.getNumRecordsToBeChecked();
return new AbstractMap.SimpleEntry<>(records, checkpointState);
Expand Down
1 change: 1 addition & 0 deletions ...rc/main/java/org/opensearch/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,7 @@ public Map.Entry<Collection<T>, CheckpointState> doRead(int timeoutInMillis) {
}
}

updateLatency((Collection<T>)records);
final CheckpointState checkpointState = new CheckpointState(recordsRead);
return new AbstractMap.SimpleEntry<>(records, checkpointState);
}
Expand Down
6 changes: 4 additions & 2 deletions ...ka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/buffer/KafkaBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ public KafkaBuffer(final PluginSetting pluginSetting, final KafkaBufferConfig ka
final AcknowledgementSetManager acknowledgementSetManager,
final ByteDecoder byteDecoder, final AwsCredentialsSupplier awsCredentialsSupplier,
final CircuitBreaker circuitBreaker) {
super(kafkaBufferConfig.getCustomMetricPrefix().orElse(pluginSetting.getName()), pluginSetting.getPipelineName());
super(kafkaBufferConfig.getCustomMetricPrefix().orElse(pluginSetting.getName()+"buffer"), pluginSetting.getPipelineName());
Copy link
Member

Choose a reason for hiding this comment

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

What will this name look like?

kafkabuffer

Is there a better name we can use here?

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

Yeah. Right now the name looks like "kafka". I am changing it to "kafkabuffer"

Copy link
Member

Choose a reason for hiding this comment

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

I'm wondering if we can have buffer in the base name to start with.

So for example:

my-pipeline.kafka.buffer.readLatency

Perhaps by changing

public static final String READ_LATENCY = "readLatency";

to

public static final String READ_LATENCY = "buffer.readLatency";

Thoughts?

final SerializationFactory serializationFactory = new BufferSerializationFactory(new CommonSerializationFactory());
final KafkaCustomProducerFactory kafkaCustomProducerFactory = new KafkaCustomProducerFactory(serializationFactory, awsCredentialsSupplier, new TopicServiceFactory());
this.byteDecoder = byteDecoder;
Expand Down Expand Up @@ -141,7 +141,9 @@ public void doWriteAll(Collection<Record<Event>> records, int timeoutInMillis) t
public Map.Entry<Collection<Record<Event>>, CheckpointState> doRead(int timeoutInMillis) {
try {
setMdc();
return innerBuffer.read(timeoutInMillis);
Map.Entry<Collection<Record<Event>>, CheckpointState> result = innerBuffer.read(timeoutInMillis);
updateLatency(result.getKey());
return result;
} finally {
resetMdc();
}
Expand Down
Loading