Add lag metric for system table cleanup cursor (#35994)

To track lag on the cleanup worker.

GitOrigin-RevId: 09511f1d3e3f286e8ae14f347e6a5f9a2273395f

Author: nipunn1313

crates/application/src/system_table_cleanup/metrics.rs

@@ -1,6 +1,9 @@
+use chrono::Utc;
+use common::document::CreationTime;
 use metrics::{
     log_counter,
     log_counter_with_labels,
+    log_distribution_with_labels,
     prometheus::VMHistogram,
     register_convex_counter,
     register_convex_histogram,
@@ -37,3 +40,18 @@ register_convex_counter!(
 pub fn log_exports_s3_cleanup() {
     log_counter(&EXPORT_TABLE_CLEANUP_ROWS_TOTAL, 1)
 }
+
+register_convex_histogram!(
+    SYSTEM_TABLE_CLEANUP_CURSOR_LAG_SECONDS,
+    "Lag between system table cleanup cursor and now",
+    &["table"]
+);
+pub fn log_system_table_cursor_lag(table_name: &TableName, cursor: CreationTime) {
+    let now = Utc::now().timestamp_millis();
+    let delay_ms = (now as f64) - f64::from(cursor);
+    log_distribution_with_labels(
+        &SYSTEM_TABLE_CLEANUP_CURSOR_LAG_SECONDS,
+        delay_ms / 1000.0,
+        vec![StaticMetricLabel::new("table", table_name.to_string())],
+    )
+}

crates/application/src/system_table_cleanup/mod.rs

@@ -55,6 +55,7 @@ use keybroker::Identity;
 use metrics::{
     log_exports_s3_cleanup,
     log_system_table_cleanup_rows,
+    log_system_table_cursor_lag,
     system_table_cleanup_timer,
 };
 use model::{
@@ -334,6 +335,9 @@ impl<RT: Runtime> SystemTableCleanupWorker<RT> {
             .await?;
         tracing::info!("deleted {deleted_count} documents from {table}");
         log_system_table_cleanup_rows(table, deleted_count);
+        if let Some(cursor) = cursor {
+            log_system_table_cursor_lag(table, *cursor);
+        }
         Ok(deleted_count)
     }