Oximeter: add queue depth metric. (#10736)

In #10683, we added a metric counting the number of samples evicted from
the oximeter collector's database_batcher queue. Operators can use that
metric to detect that samples are being lost, but not that samples are
potentially about to be lost. In other words, we have to wait for
something to go wrong to take action.

This patch adds a queue depth metric to the collector as well: a
histogram recording the current length of the queue each time we push a
new batch of samples onto it. Operators can use this metric to detect
when the queue depth is approaching its maximum length.

Part of #10552.

Note: builds on #10683. Can just review the second commit, or review
both commits here and ignore #10683.

And I promise I'm almost done thinking about #10552. This is one of the
last papercuts before I'll feel like this is sorted.

Author: jmcarp

Cargo.lock

@@ -50,6 +50,7 @@ nexus-client.workspace = true
 expectorate.workspace = true
 httpmock.workspace = true
 omicron-test-utils.workspace = true
+oximeter-test-utils.workspace = true
 openapi-lint.workspace = true
 openapiv3.workspace = true
 proptest.workspace = true

oximeter/collector/Cargo.toml

@@ -21,6 +21,8 @@ use nexus_client::Client as NexusClient;
 use nexus_client::types::IdSortMode;
 use omicron_common::backoff;
 use omicron_common::backoff::BackoffError;
+use oximeter::Sample;
+use oximeter::types::ProducerResultsItem;
 use oximeter_db::Client;
 use oximeter_db::DbWrite;
 use oximeter_types::producer::ProducerDetails;
@@ -45,6 +47,7 @@ use std::sync::Mutex;
 use std::sync::MutexGuard;
 use std::time::Duration;
 use tokio::sync::mpsc;
+use tokio::time::interval;
 use uuid::Uuid;
 
 /// The internal agent the oximeter server uses to collect metrics from producers.
@@ -93,7 +96,7 @@ impl OximeterAgent {
             "collector_ip" => address.ip().to_string(),
         ));
         let insertion_log = log.new(o!("component" => "results-sink"));
-        let instertion_log_cluster =
+        let insertion_log_cluster =
             log.new(o!("component" => "results-sink-cluster"));
 
         // Determine the version of the database.
@@ -139,17 +142,23 @@ impl OximeterAgent {
             collector_port: address.port(),
         };
 
+        let collector_stats = self_stats::CollectorStats::new(Utc::now());
+
         // Spawn the task for aggregating and inserting all metrics to a
         // single node ClickHouse installation.
-        tokio::spawn(async move {
-            crate::results_sink::database_batcher(
-                insertion_log,
-                client,
-                db_config.batch_size,
-                Duration::from_secs(db_config.batch_interval),
-                collection_task_wrapper.single_rx,
-            )
-            .await
+        tokio::spawn({
+            let sink_stats = collector_stats.single_stats.clone();
+            async move {
+                crate::results_sink::database_batcher(
+                    insertion_log,
+                    client,
+                    db_config.batch_size,
+                    Duration::from_secs(db_config.batch_interval),
+                    collection_task_wrapper.single_rx,
+                    sink_stats,
+                )
+                .await
+            }
         });
 
         // Our internal testing rack will be running a ClickHouse cluster
@@ -184,28 +193,35 @@ impl OximeterAgent {
 
         // Spawn the task for aggregating and inserting all metrics to a
         // replicated cluster ClickHouse installation
-        tokio::spawn(async move {
+        tokio::spawn({
             results_sink::database_batcher(
-                instertion_log_cluster,
+                insertion_log_cluster,
                 cluster_client,
                 db_config.batch_size,
                 Duration::from_secs(db_config.batch_interval),
                 collection_task_wrapper.cluster_rx,
+                collector_stats.cluster_stats.clone(),
             )
-            .await
         });
 
         let self_ = Self {
             id,
-            log,
+            log: log.clone(),
             collection_target,
-            result_sender: collection_task_wrapper.wrapper_tx,
+            result_sender: collection_task_wrapper.wrapper_tx.clone(),
             collection_tasks: Arc::new(Mutex::new(BTreeMap::new())),
             refresh_interval,
             refresh_task: Arc::new(Mutex::new(None)),
             last_refresh_time: Arc::new(Mutex::new(None)),
         };
 
+        tokio::spawn(emit_self_stats(
+            collector_stats,
+            collection_target,
+            collection_task_wrapper.wrapper_tx,
+            log,
+        ));
+
         Ok(self_)
     }
 
@@ -269,6 +285,11 @@ impl OximeterAgent {
                 client.init_replicated_db().await?;
             }
 
+            let sink_stats = Arc::new(self_stats::CollectorSinkStats::new(
+                "standalone".into(),
+                Utc::now(),
+            ));
+
             // Spawn the task for aggregating and inserting all metrics
             tokio::spawn(async move {
                 results_sink::database_batcher(
@@ -277,6 +298,7 @@ impl OximeterAgent {
                     db_config.batch_size,
                     Duration::from_secs(db_config.batch_interval),
                     collection_task_wrapper.single_rx,
+                    sink_stats,
                 )
                 .await
             });
@@ -524,6 +546,55 @@ impl CollectionTaskSenderWrapper {
     }
 }
 
+// Emit stats about the collector agent.
+//
+// Note: If oximeter isn't able to write metrics to ClickHouse,
+// self-stats about oximeter's inability to write metrics will
+// also never reach ClickHouse. However, we may still be able to
+// emit these metrics if the write path to ClickHouse is partially
+// degraded. And because the relevant counters live in oximeter's
+// memory, even if ClickHouse becomes fully unavailable and then
+// recovers, the metrics will also eventually reach the database.
+//
+// TODO(#10552): Emit a metric about failed database writes as well.
+async fn emit_self_stats(
+    collector_stats: self_stats::CollectorStats,
+    collection_target: self_stats::OximeterCollector,
+    wrapper_tx: CollectionTaskSenderWrapper,
+    log: Logger,
+) {
+    let mut sink_stats_ticker = interval(self_stats::COLLECTION_INTERVAL);
+    loop {
+        sink_stats_ticker.tick().await;
+
+        let mut samples: Vec<Sample> = Vec::new();
+
+        for sink_stats in
+            [&collector_stats.single_stats, &collector_stats.cluster_stats]
+        {
+            match sink_stats.samples(&collection_target) {
+                Ok(s) => samples.extend(s),
+                Err(e) => error!(
+                    log,
+                    "failed to build database sink self-stats";
+                    "sink" => sink_stats.label.clone(),
+                    InlineErrorChain::new(&e),
+                ),
+            };
+        }
+
+        let _ = wrapper_tx
+            .send(
+                CollectionTaskOutput {
+                    results: vec![ProducerResultsItem::Ok(samples)],
+                    was_forced_collection: false,
+                },
+                &log,
+            )
+            .await;
+    }
+}
+
 #[derive(Debug)]
 pub struct CollectionTaskWrapper {
     wrapper_tx: CollectionTaskSenderWrapper,

oximeter/collector/src/agent.rs

@@ -9,7 +9,9 @@
 
 use crate::collection_task::CollectionTaskOutput;
 use crate::probes;
+use crate::self_stats;
 use oximeter::Sample;
+use oximeter::histogram::Record;
 use oximeter::queue::BoundedQueue;
 use oximeter::types::ProducerResultsItem;
 use oximeter_db::Client;
@@ -41,6 +43,7 @@ pub async fn database_batcher(
     batch_size: usize,
     batch_interval: Duration,
     mut rx: mpsc::Receiver<CollectionTaskOutput>,
+    sink_stats: Arc<self_stats::CollectorSinkStats>,
 ) {
     // Construct a handoff point between the batch task here, and the database
     // insertion task.
@@ -109,7 +112,7 @@ pub async fn database_batcher(
                         probes::results__sink__item__processed!(|| n_samples);
 
                         // Append the current batch to the handoff buffer.
-                        let n_dropped =
+                        let (n_dropped, queue_depth) =
                             batch_tx.send_and_notify(batch, was_forced_collection);
                         if n_dropped > 0 {
                             probes::dropped__old__samples!(|| n_dropped);
@@ -118,7 +121,13 @@ pub async fn database_batcher(
                                 "sample buffer full, dropped oldest samples";
                                 "n_dropped" => n_dropped,
                             );
+                            *sink_stats.samples_dropped.lock().unwrap() += n_dropped.try_into().unwrap();
                         }
+                        // Note: Histogram::u64::sample should never error.
+                        // The `sample` method only returns a `HistogramError`
+                        // if its argument is non-finite, and `u64` values are
+                        // always finite.
+                        sink_stats.queue_depth.lock().unwrap().sample(queue_depth.try_into().unwrap()).unwrap();
                     }
                     None => {
                         warn!(log, "result queue closed, exiting");
@@ -195,7 +204,7 @@ impl<T> BatchSender<T> {
         &self,
         samples: Vec<T>,
         was_forced_collection: bool,
-    ) -> usize {
+    ) -> (usize, usize) {
         let mut batch = self.handoff.batch.lock().unwrap();
 
         let n_dropped = batch.extend(samples);
@@ -204,7 +213,7 @@ impl<T> BatchSender<T> {
         if was_forced_collection || batch.len() >= self.batch_size {
             self.notify_inserter();
         }
-        n_dropped
+        (n_dropped, batch.len())
     }
 }
 
@@ -311,6 +320,8 @@ pub async fn logger(log: Logger, mut rx: mpsc::Receiver<CollectionTaskOutput>) {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use chrono::Utc;
+    use omicron_test_utils::dev::test_setup_log;
     use proptest::prelude::*;
     use tokio::time::timeout;
 
@@ -322,7 +333,7 @@ mod tests {
         let (batch_tx, batch_rx) = batch_handoff::<()>(BATCH_SIZE);
         let n_samples = 3;
         let samples = vec![(); n_samples];
-        let n_dropped = batch_tx.send_and_notify(samples, true);
+        let (n_dropped, _) = batch_tx.send_and_notify(samples, true);
         assert_eq!(n_dropped, 0);
         assert_eq!(batch_tx.handoff.batch.lock().unwrap().len(), n_samples);
 
@@ -348,7 +359,7 @@ mod tests {
             // Insert the first batch of samples.
             let (batch_tx, batch_rx) = batch_handoff::<usize>(BATCH_SIZE);
             let samples = (0..current_size).collect();
-            let n_dropped = batch_tx.send_and_notify(samples, false);
+            let (n_dropped, _) = batch_tx.send_and_notify(samples, false);
             let expected_n_dropped = current_size.saturating_sub(MAX_BUFFER_SIZE);
             let expected_n_samples = current_size.min(MAX_BUFFER_SIZE);
             assert_eq!(n_dropped, expected_n_dropped);
@@ -384,7 +395,7 @@ mod tests {
             // Push the new set of samples, which start at the end of the first
             // batch, even if we've dropped some.
             let new_samples = (current_size..(current_size + new_size)).collect();
-            let n_dropped = batch_tx.send_and_notify(new_samples, false);
+            let (n_dropped, _) = batch_tx.send_and_notify(new_samples, false);
 
             // We expect to drop all the samples beyond the maximum buffer size.
             // We need to include what we have in the buffer already (some of
@@ -422,4 +433,51 @@ mod tests {
             }
         }
     }
+
+    #[tokio::test]
+    async fn batcher_increments_dropped_counter() {
+        let logctx = test_setup_log("batcher_increments_dropped_counter");
+        let log = &logctx.log;
+
+        // Construct a database batcher with a dummy ClickHouse client.
+        let client = Client::new("[::1]:0".parse().unwrap(), log);
+        let (tx, rx) = mpsc::channel(1);
+        let sink_stats = Arc::new(self_stats::CollectorSinkStats::new(
+            "test".into(),
+            Utc::now(),
+        ));
+        let batcher = tokio::spawn(database_batcher(
+            log.clone(),
+            client,
+            BATCH_SIZE,
+            Duration::from_secs(3600),
+            rx,
+            sink_stats.clone(),
+        ));
+
+        // Insert `want_dropped` too many samples into the batcher.
+        let want_dropped = 50;
+        let samples = (0..MAX_BUFFER_SIZE + want_dropped)
+            .map(|_| oximeter_test_utils::make_sample())
+            .collect();
+        tx.send(CollectionTaskOutput {
+            results: vec![ProducerResultsItem::Ok(samples)],
+            was_forced_collection: false,
+        })
+        .await
+        .unwrap();
+
+        // Drop the sender so that the batcher loop exits.
+        drop(tx);
+        batcher.await.unwrap();
+
+        let samples_dropped = sink_stats.samples_dropped.lock().unwrap();
+        assert_eq!(samples_dropped.value(), want_dropped as u64);
+
+        let queue_depth = sink_stats.queue_depth.lock().unwrap();
+        assert_eq!(queue_depth.n_samples(), 1);
+        assert_eq!(queue_depth.max(), MAX_BUFFER_SIZE as u64);
+
+        logctx.cleanup_successful();
+    }
 }