luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #868 from QAQJ/consumer-level-metrics

Browse source 
Adding a new metric: Application-level MillisBehindLatest
This commit is contained in:
ava huang 2021-12-06 12:22:06 -08:00 committed by GitHub
commit bedae95db9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 20 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ProcessTask.java
View file

@ -45,6 +45,7 @@ import software.amazon.kinesis.schemaregistry.SchemaRegistryDecoder;
@KinesisClientInternalApi @KinesisClientInternalApi
public class ProcessTask implements ConsumerTask { public class ProcessTask implements ConsumerTask {
private static final String PROCESS_TASK_OPERATION = "ProcessTask"; private static final String PROCESS_TASK_OPERATION = "ProcessTask";
private static final String APPLICATION_TRACKER_OPERATION = "ApplicationTracker";
private static final String DATA_BYTES_PROCESSED_METRIC = "DataBytesProcessed"; private static final String DATA_BYTES_PROCESSED_METRIC = "DataBytesProcessed";
private static final String RECORDS_PROCESSED_METRIC = "RecordsProcessed"; private static final String RECORDS_PROCESSED_METRIC = "RecordsProcessed";
private static final String RECORD_PROCESSOR_PROCESS_RECORDS_METRIC = "RecordProcessor.processRecords"; private static final String RECORD_PROCESSOR_PROCESS_RECORDS_METRIC = "RecordProcessor.processRecords";
@ -112,20 +113,28 @@ public class ProcessTask implements ConsumerTask {
*/ */
@Override @Override
public TaskResult call() { public TaskResult call() {
final MetricsScope scope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION); /**
* NOTE: the difference between appScope and shardScope is, appScope doesn't have shardId as a dimension,
* therefore all data added to appScope, although from different shard consumer, will be sent to the same metric,
* which is the app-level MillsBehindLatest metric.
*/
final MetricsScope appScope = MetricsUtil.createMetricsWithOperation(metricsFactory, APPLICATION_TRACKER_OPERATION);
final MetricsScope shardScope = MetricsUtil.createMetricsWithOperation(metricsFactory, PROCESS_TASK_OPERATION);
shardInfo.streamIdentifierSerOpt() shardInfo.streamIdentifierSerOpt()
.ifPresent(streamId -> MetricsUtil.addStreamId(scope, StreamIdentifier.multiStreamInstance(streamId))); .ifPresent(streamId -> MetricsUtil.addStreamId(shardScope, StreamIdentifier.multiStreamInstance(streamId)));
MetricsUtil.addShardId(scope, shardInfo.shardId()); MetricsUtil.addShardId(shardScope, shardInfo.shardId());
long startTimeMillis = System.currentTimeMillis(); long startTimeMillis = System.currentTimeMillis();
boolean success = false; boolean success = false;
try { try {
scope.addData(RECORDS_PROCESSED_METRIC, 0, StandardUnit.COUNT, MetricsLevel.SUMMARY); shardScope.addData(RECORDS_PROCESSED_METRIC, 0, StandardUnit.COUNT, MetricsLevel.SUMMARY);
scope.addData(DATA_BYTES_PROCESSED_METRIC, 0, StandardUnit.BYTES, MetricsLevel.SUMMARY); shardScope.addData(DATA_BYTES_PROCESSED_METRIC, 0, StandardUnit.BYTES, MetricsLevel.SUMMARY);
Exception exception = null; Exception exception = null;
try { try {
if (processRecordsInput.millisBehindLatest() != null) { if (processRecordsInput.millisBehindLatest() != null) {
scope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(), shardScope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(),
StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY);
appScope.addData(MILLIS_BEHIND_LATEST_METRIC, processRecordsInput.millisBehindLatest(),
StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY); StandardUnit.MILLISECONDS, MetricsLevel.SUMMARY);
} }
@ -142,11 +151,11 @@ public class ProcessTask implements ConsumerTask {
} }
if (!records.isEmpty()) { if (!records.isEmpty()) {
scope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY); shardScope.addData(RECORDS_PROCESSED_METRIC, records.size(), StandardUnit.COUNT, MetricsLevel.SUMMARY);
} }
recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber( recordProcessorCheckpointer.largestPermittedCheckpointValue(filterAndGetMaxExtendedSequenceNumber(
scope, records, recordProcessorCheckpointer.lastCheckpointValue(), shardScope, records, recordProcessorCheckpointer.lastCheckpointValue(),
recordProcessorCheckpointer.largestPermittedCheckpointValue())); recordProcessorCheckpointer.largestPermittedCheckpointValue()));
if (shouldCallProcessRecords(records)) { if (shouldCallProcessRecords(records)) {
@ -165,8 +174,9 @@ public class ProcessTask implements ConsumerTask {
} }
return new TaskResult(exception); return new TaskResult(exception);
} finally { } finally {
MetricsUtil.addSuccessAndLatency(scope, success, startTimeMillis, MetricsLevel.SUMMARY); MetricsUtil.addSuccessAndLatency(shardScope, success, startTimeMillis, MetricsLevel.SUMMARY);
MetricsUtil.endScope(scope); MetricsUtil.endScope(shardScope);
MetricsUtil.endScope(appScope);
} }
} }