Add retry count metric (#178)

* Add retry count metric * Rename metric * Update metric before sending request
line · Dec 16, 2022 · f7febfe · f7febfe
1 parent c9d786b
commit f7febfe
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 1 deletion.
diff --git a/processor/src/main/java/com/linecorp/decaton/processor/metrics/Metrics.java b/processor/src/main/java/com/linecorp/decaton/processor/metrics/Metrics.java
@@ -30,6 +30,7 @@
 import com.linecorp.decaton.processor.metrics.internal.AvailableTags;
 
 import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.Meter;
 import io.micrometer.core.instrument.Meter.Id;
@@ -261,6 +262,12 @@ public class RetryMetrics extends AbstractMetrics {
                                    .description("The number of tasks failed to enqueue in retry topic")
                                    .tags(availableTags.subscriptionScope())
                                    .register(registry));
+
+        public final DistributionSummary retryTaskRetries =
+                meter(() -> DistributionSummary.builder("retry.task.retries")
+                                               .description("The number of times a task was retried")
+                                               .tags(availableTags.subscriptionScope())
+                                               .register(registry));
     }
 
     public static Metrics withTags(String... keyValues) {

diff --git a/...va/com/linecorp/decaton/processor/runtime/internal/DecatonTaskRetryQueueingProcessor.java b/...va/com/linecorp/decaton/processor/runtime/internal/DecatonTaskRetryQueueingProcessor.java
@@ -52,17 +52,19 @@ public DecatonTaskRetryQueueingProcessor(SubscriptionScope scope, DecatonTaskPro
     public void process(ProcessingContext<byte[]> context, byte[] serializedTask)
             throws InterruptedException {
         TaskMetadata originalMeta = context.metadata();
+        long nextRetryCount = originalMeta.retryCount() + 1;
         long nextTryTimeMillis = System.currentTimeMillis() + backoff.toMillis();
         TaskMetadataProto taskMetadata =
                 TaskMetadataProto.newBuilder(originalMeta.toProto())
-                                 .setRetryCount(originalMeta.retryCount() + 1)
+                                 .setRetryCount(nextRetryCount)
                                  .setScheduledTimeMillis(nextTryTimeMillis)
                                  .build();
         DecatonTaskRequest request =
                 DecatonTaskRequest.newBuilder()
                                   .setMetadata(taskMetadata)
                                   .setSerializedTask(ByteString.copyFrom(serializedTask))
                                   .build();
+        metrics.retryTaskRetries.record(nextRetryCount);
 
         CompletableFuture<PutTaskResult> future = producer.sendRequest(context.key(), request);
         future.whenComplete((r, e) -> {