Log more metrics for especially slow subscription processing (#39192)

GitOrigin-RevId: a1b61ebb12dfa6ffe0736aff5049f682fa3d305d

Author: thomasballinger

crates/common/src/knobs.rs

@@ -1358,3 +1358,13 @@ pub static SUBSCRIPTION_INVALIDATION_DELAY_THRESHOLD: LazyLock<usize> =
 /// invalidated to determine the delay before invalidating them.
 pub static SUBSCRIPTION_INVALIDATION_DELAY_MULTIPLIER: LazyLock<u64> =
     LazyLock::new(|| env_config("SUBSCRIPTION_INVALIDATION_DELAY_MULTIPLIER", 5));
+
+/// When processing a single write log entry takes longer than that time, log
+/// extra detail.
+pub static SUBSCRIPTION_PROCESS_LOG_ENTRY_TRACING_THRESHOLD: LazyLock<u64> =
+    LazyLock::new(|| env_config("SUBSCRIPTION_PROCESS_LOG_ENTRY_TRACING_THRESHOLD", 2));
+
+/// When advancing the write log takes longer than this amount, log extra
+/// details.
+pub static SUBSCRIPTION_ADVANCE_LOG_TRACING_THRESHOLD: LazyLock<u64> =
+    LazyLock::new(|| env_config("SUBSCRIPTION_ADVANCE_LOG_TRACING_THRESHOLD", 10));

crates/database/src/metrics.rs

@@ -1082,6 +1082,14 @@ pub fn subscriptions_log_iterate_timer() -> Timer<VMHistogram> {
     Timer::new(&SUBSCRIPTION_LOG_ITERATE_SECONDS)
 }
 
+register_convex_histogram!(
+    SUBSCRIPTION_PROCESS_WRITE_LOG_ENTRY_SECONDS,
+    "Time to process one write log entry when advancing subscriptions",
+);
+pub fn subscription_process_write_log_entry_timer() -> Timer<VMHistogram> {
+    Timer::new(&SUBSCRIPTION_PROCESS_WRITE_LOG_ENTRY_SECONDS)
+}
+
 register_convex_histogram!(
     SUBSCRIPTION_LOG_INVALIDATE_SECONDS,
     "Time to invalidate segsstiptions when edvancing rh_ log",

crates/database/src/subscription.rs

@@ -27,15 +27,18 @@ use common::{
     errors::report_error,
     knobs::{
         SUBSCRIPTIONS_WORKER_QUEUE_SIZE,
+        SUBSCRIPTION_ADVANCE_LOG_TRACING_THRESHOLD,
         SUBSCRIPTION_INVALIDATION_DELAY_MULTIPLIER,
         SUBSCRIPTION_INVALIDATION_DELAY_THRESHOLD,
+        SUBSCRIPTION_PROCESS_LOG_ENTRY_TRACING_THRESHOLD,
     },
     runtime::{
         block_in_place,
         Runtime,
         SpawnHandle,
     },
     types::{
+        GenericIndexName,
         PersistenceVersion,
         SubscriberId,
         TabletIndexName,
@@ -328,9 +331,12 @@ impl SubscriptionManager {
                 let mut log_len = 0;
                 let mut num_writes = 0;
                 self.log.for_each(from_ts, next_ts, |_, writes| {
+                    let process_log_timer = metrics::subscription_process_write_log_entry_timer();
                     log_len += 1;
                     num_writes += writes.len();
-                    for (_, document_change) in writes {
+                    let mut tablet_ids = BTreeSet::new();
+                    for (resolved_id, document_change) in writes {
+                        tablet_ids.insert(resolved_id.tablet_id);
                         // We're applying a mutation to the document so if it already exists
                         // we need to remove it before writing the new version.
                         if let Some(ref old_document) = document_change.old_document {
@@ -352,9 +358,46 @@ impl SubscriptionManager {
                             );
                         }
                     }
+
+                    if process_log_timer.elapsed()
+                        > Duration::from_secs(*SUBSCRIPTION_PROCESS_LOG_ENTRY_TRACING_THRESHOLD)
+                    {
+                        tracing::info!(
+                            "[{next_ts}: advance_log] simple commit took {:?}, affected tables: \
+                             {tablet_ids:?}",
+                            process_log_timer.elapsed()
+                        );
+                    }
                 })?;
                 metrics::log_subscriptions_log_length(log_len);
                 metrics::log_subscriptions_log_writes(num_writes);
+                if _timer.elapsed()
+                    > Duration::from_secs(*SUBSCRIPTION_ADVANCE_LOG_TRACING_THRESHOLD)
+                {
+                    let subscribers_by_index: BTreeMap<&GenericIndexName<_>, usize> = self
+                        .subscriptions
+                        .indexed
+                        .iter()
+                        .map(|(key, (_fields, range_map))| (key, range_map.len()))
+                        .collect();
+                    let total_subscribers: usize = subscribers_by_index.values().sum();
+                    let search_len = self.subscriptions.search.filter_len();
+                    let fuzzy_len = self.subscriptions.search.fuzzy_len();
+                    tracing::info!(
+                        "[{next_ts} advance_log] Duration {}ms, indexes: {}, search filters: {}, \
+                         fuzzy search: {}",
+                        _timer.elapsed().as_millis(),
+                        self.subscriptions.indexed.len(),
+                        search_len,
+                        fuzzy_len
+                    );
+                    tracing::info!(
+                        "`[{next_ts} advance_log] Subscription map size: {total_subscribers}"
+                    );
+                    tracing::info!(
+                        "[{next_ts} advance_log] Subscribers by index {subscribers_by_index:?}"
+                    );
+                }
             }
 
             {

crates/indexing/src/interval.rs

@@ -27,6 +27,10 @@ impl<ID: Clone + Ord> IntervalMap<ID> {
         self.sets.is_empty()
     }
 
+    pub fn len(&self) -> usize {
+        self.sets.len()
+    }
+
     /// Insert the IntervalSet for the given ID.
     pub fn insert(&mut self, id: ID, set: IntervalSet) -> Option<IntervalSet> {
         self.sets.insert(id, set)

crates/search/src/query.rs

@@ -969,6 +969,14 @@ impl TextSearchSubscriptions {
         }
     }
 
+    pub fn filter_len(&self) -> usize {
+        self.filter_conditions.values().map(|m| m.len()).sum()
+    }
+
+    pub fn fuzzy_len(&self) -> usize {
+        self.fuzzy_searches.len()
+    }
+
     pub fn insert(&mut self, id: SubscriberId, index: &TabletIndexName, reads: &QueryReads) {
         self.filter_conditions
             .entry(index.clone())