chore: add two new metrics for blobby commits

PostHog · Jan 17, 2024 · 6a4b712 · 6a4b712
1 parent ed56f24
commit 6a4b712
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/plugin-server/src/main/ingestion-queues/session-recording/services/session-manager.ts b/plugin-server/src/main/ingestion-queues/session-recording/services/session-manager.ts
@@ -189,7 +189,7 @@ export class SessionManager {
             throw error
         }
 
-        // NOTE: This is uncompressed size estimate but thats okay as we currently want to over-flush to see if we can shake out a bug
+        // NOTE: This is uncompressed size estimate but that's okay as we currently want to over-flush to see if we can shake out a bug
         if (this.buffer.sizeEstimate >= this.serverConfig.SESSION_RECORDING_MAX_BUFFER_SIZE_KB * 1024) {
             await this.flush('buffer_size')
         }

diff --git a/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts b/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer.ts
@@ -90,6 +90,18 @@ const counterKafkaMessageReceived = new Counter({
     labelNames: ['partition'],
 })
 
+const counterCommitSkippedDueToPotentiallyBlockingSession = new Counter({
+    name: 'recording_blob_ingestion_commit_skipped_due_to_potentially_blocking_session',
+    help: 'The number of times we skipped committing due to a potentially blocking session',
+})
+
+const histogramActiveSessionsWhenCommitIsBlocked = new Histogram({
+    name: 'recording_blob_ingestion_active_sessions_when_commit_is_blocked',
+    help: 'The number of active sessions on a partition when we skip committing due to a potentially blocking session',
+    buckets: [0, 1, 2, 3, 4, 5, 10, 20, 50, 100, 1000, 10000, Infinity],
+})
+}
+
 type PartitionMetrics = {
     lastMessageTimestamp?: number
     lastMessageOffset?: number
@@ -648,9 +660,11 @@ export class SessionRecordingIngester {
 
                 let potentiallyBlockingSession: SessionManager | undefined
 
+                let activeSessionsOnThisPartition = 0
                 for (const sessionManager of blockingSessions) {
                     if (sessionManager.partition === partition) {
                         const lowestOffset = sessionManager.getLowestOffset()
+                        activeSessionsOnThisPartition++
                         if (
                             lowestOffset !== null &&
                             lowestOffset < (potentiallyBlockingSession?.getLowestOffset() || Infinity)
@@ -684,6 +698,8 @@ export class SessionRecordingIngester {
                             highestOffsetToCommit,
                         }
                     )
+                    counterCommitSkippedDueToPotentiallyBlockingSession.inc()
+                    histogramActiveSessionsWhenCommitIsBlocked.observe(activeSessionsOnThisPartition)
                     return
                 }