feat: Improved instrumentation around core functions

plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer-v2.ts

@@ -310,7 +310,7 @@ export class SessionRecordingIngesterV2 {
     public async handleEachBatch(messages: Message[]): Promise<void> {
         await runInstrumentedFunction({
             statsKey: `recordingingester.handleEachBatch`,
-            timeout: 60 * 1000,
+            logToConsole: true,
             func: async () => {
                 const transaction = Sentry.startTransaction({ name: `blobIngestion_handleEachBatch` }, {})
                 histogramKafkaBatchSize.observe(messages.length)
@@ -567,12 +567,19 @@ export class SessionRecordingIngesterV2 {
         // - have some sort of timeout so we don't get stuck here forever
         if (this.serverConfig.SESSION_RECORDING_PARTITION_REVOKE_OPTIMIZATION) {
             status.info('🔁', `blob_ingester_consumer - flushing ${sessionsToDrop.length} sessions on revoke...`)
-            await Promise.allSettled(
-                sessionsToDrop
-                    .map(([_, x]) => x)
-                    .sort((x) => x.buffer.oldestKafkaTimestamp ?? Infinity)
-                    .map((x) => x.flush('partition_shutdown'))
-            )
+
+            await runInstrumentedFunction({
+                statsKey: `recordingingester.onRevokePartitions.flushSessions`,
+                logToConsole: true,
+                func: async () => {
+                    await Promise.allSettled(
+                        sessionsToDrop
+                            .map(([_, x]) => x)
+                            .sort((x) => x.buffer.oldestKafkaTimestamp ?? Infinity)
+                            .map((x) => x.flush('partition_shutdown'))
+                    )
+                },
+            })
         }
 
         topicPartitions.forEach((topicPartition: TopicPartition) => {

plugin-server/src/main/utils.ts

@@ -11,6 +11,13 @@ interface FunctionInstrumentation<T> {
     timeoutMessage?: string
     timeoutContext?: () => Record<string, any>
     teamId?: number
+    logToConsole?: boolean
+}
+
+const logTime = (startTime: number, statsKey: string, error?: any) => {
+    status.info('⏱️', `${statsKey} took ${Math.round(performance.now() - startTime)}ms`, {
+        error,
+    })
 }
 
 export async function runInstrumentedFunction<T>({
@@ -20,18 +27,27 @@ export async function runInstrumentedFunction<T>({
     func,
     statsKey,
     teamId,
+    logToConsole = false,
 }: FunctionInstrumentation<T>): Promise<T> {
     const t = timeoutGuard(timeoutMessage ?? `Timeout warning for '${statsKey}'!`, timeoutContext, timeout)
+    const startTime = performance.now()
     const end = instrumentedFunctionDuration.startTimer({
         function: statsKey,
     })
+
     try {
         const result = await func()
         end({ success: 'true' })
+        if (logToConsole) {
+            logTime(startTime, statsKey)
+        }
         return result
     } catch (error) {
         end({ success: 'false' })
         status.info('🔔', error)
+        if (logToConsole) {
+            logTime(startTime, statsKey, error)
+        }
         Sentry.captureException(error, { tags: { team_id: teamId } })
         throw error
     } finally {