Add a gauge which tracks the number of pending activity records

crates/handlers/src/activity_tracker/worker.rs

@@ -10,7 +10,7 @@ use chrono::{DateTime, Utc};
 use mas_storage::{RepositoryAccess, RepositoryError, user::BrowserSessionRepository};
 use opentelemetry::{
     Key, KeyValue,
-    metrics::{Counter, Histogram},
+    metrics::{Counter, Gauge, Histogram},
 };
 use sqlx::PgPool;
 use tokio_util::sync::CancellationToken;
@@ -45,6 +45,7 @@ struct ActivityRecord {
 pub struct Worker {
     pool: PgPool,
     pending_records: HashMap<(SessionKind, Ulid), ActivityRecord>,
+    pending_records_gauge: Gauge<u64>,
     message_counter: Counter<u64>,
     flush_time_histogram: Histogram<u64>,
 }
@@ -80,9 +81,17 @@ impl Worker {
             .with_unit("ms")
             .build();
 
+        let pending_records_gauge = METER
+            .u64_gauge("mas.activity_tracker.pending_records")
+            .with_description("The number of pending activity records")
+            .with_unit("{records}")
+            .build();
+        pending_records_gauge.record(0, &[]);
+
         Self {
             pool,
             pending_records: HashMap::with_capacity(MAX_PENDING_RECORDS),
+            pending_records_gauge,
             message_counter,
             flush_time_histogram,
         }
@@ -165,6 +174,10 @@ impl Worker {
                     let _ = tx.send(());
                 }
             }
+
+            // Update the gauge
+            self.pending_records_gauge
+                .record(self.pending_records.len() as u64, &[]);
         }
 
         // Flush one last time