feat: Add batch duration metric for ingestion consumer loop

PostHog · Nov 24, 2023 · 44bbd05 · 44bbd05
1 parent d980d13
commit 44bbd05
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/plugin-server/src/kafka/batch-consumer.ts b/plugin-server/src/kafka/batch-consumer.ts
@@ -221,6 +221,7 @@ export const startBatchConsumer = async ({
                 batchesProcessed += 1
 
                 const processingTimeMs = new Date().valueOf() - startProcessingTimeMs
+                consumedBatchDuration.labels({ topic, groupId }).observe(processingTimeMs)
                 if (processingTimeMs > SLOW_BATCH_PROCESSING_LOG_THRESHOLD_MS) {
                     status.warn(
                         '🕒',
@@ -293,6 +294,12 @@ export const startBatchConsumer = async ({
     return { isHealthy, stop, join, consumer }
 }
 
+export const consumedBatchDuration = new Histogram({
+    name: 'consumed_batch_duration_ms',
+    help: 'Main loop consumer batch processing duration in ms',
+    labelNames: ['topic', 'groupId'],
+})
+
 export const consumerBatchSize = new Histogram({
     name: 'consumed_batch_size',
     help: 'Size of the batch fetched by the consumer',