Add new metrics for tracking bytes received and processed by KDS #5237
Conversation
sb2k16
requested review from
chenqi0805,
engechas,
graytaylor0,
dinujoh,
kkondaka,
KarstenSchnitter,
dlvenable and
oeyh
as code owners
| @@ -177,6 +191,7 @@ public void processRecords(ProcessRecordsInput processRecordsInput) { | |||
| // Flush buffer at the end | |||
| bufferAccumulator.flush(); | |||
| recordsProcessed.increment(eventCount); | |||
| bytesProcessedSummary.record(recordBytes.stream().mapToLong(Integer::longValue).sum()); | |||
There was a problem hiding this comment.
These two metrics are the same. We should remove the bytesReceived and keep only bytesProcessed.
| @@ -177,6 +191,7 @@ public void processRecords(ProcessRecordsInput processRecordsInput) { | |||
| // Flush buffer at the end | |||
| bufferAccumulator.flush(); | |||
| recordsProcessed.increment(eventCount); | |||
| bytesProcessedSummary.record(recordBytes.stream().mapToLong(Integer::longValue).sum()); | |||
There was a problem hiding this comment.
I think we should record each record independently. This will allow us to know the record size as well.
processRecordsInput.records()
.stream()
.map(kinesisClientRecord.data().remaining())
.map(Integer::longValue)
.forEach(bytesProcessedSummary::record);
| @AllArgsConstructor | ||
| public class KinesisInputOutputRecord { | ||
| private Record<Event> dataPrepperRecord; | ||
| private KinesisClientRecord kinesisClientRecord; |
There was a problem hiding this comment.
Rather than passing this entire record I think you could just provide it as the byte size.
private long recordBytesCount;
| Event event = record.getData(); | ||
| for (KinesisInputOutputRecord kinesisInputOutputRecord: kinesisOutputRecords) { | ||
| Record<Event> dataPrepperRecord = kinesisInputOutputRecord.getDataPrepperRecord(); | ||
| int incomingRecordSizeBytes = kinesisInputOutputRecord.getKinesisClientRecord().data().position(); |
There was a problem hiding this comment.
Can you use ByteBuffer's capacity instead?
https://docs.oracle.com/javase/8/docs/api/java/nio/Buffer.html#capacity--
Per my comment above, you could calculate this elsewhere.
There was a problem hiding this comment.
@dlvenable - It looks like capacity is the total size of the buffer. To get KCL record bytes, it might not be using the entire capacity. However, position returns the offset upto which the data has been read.
Signed-off-by: Souvik Bose <souvbose@amazon.com>
Signed-off-by: Souvik Bose <souvbose@amazon.com>
Signed-off-by: Souvik Bose <souvbose@amazon.com>
sb2k16
force-pushed
the
kinesis-bytes-metrics
branch
from
e7138ac to
d271864
Compare
| acknowledgementSetOpt.ifPresent(acknowledgementSet -> acknowledgementSet.add(event)); | ||
|
|
||
| bufferAccumulator.add(record); | ||
| bufferAccumulator.add(dataPrepperRecord); | ||
| bytesProcessedSummary.record(incomingRecordSizeBytes); |
There was a problem hiding this comment.
you have this line repeated in line number 180 as well? Not sure if this is intentional
Description
This PR is to add two new metrics for KDS source plugin named
bytesReceivedandbytesProcessedto track the size of throughput being ingested from Kinesis Data Streams.Issues Resolved
Resolves #1082
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.