Add retry count metric

line · Dec 16, 2022 · bc8a8ce · bc8a8ce
1 parent c9d786b
commit bc8a8ce
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 retryCount =
+                meter(() -> DistributionSummary.builder("retry.count")
+                                               .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,10 +52,11 @@ 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 =
@@ -71,6 +72,7 @@ public void process(ProcessingContext<byte[]> context, byte[] serializedTask)
             } else {
                 metrics.retryQueueingFailed.increment();
             }
+            metrics.retryCount.record(nextRetryCount);
         });
         context.deferCompletion().completeWith(future);
     }