Merge pull request #34340 from aljoscha/push-ozkukkxmlxwo

adapter: add histogram metrics for more coordinator methods

Author: aljoscha

src/adapter/src/coord/catalog_implications.rs

@@ -28,7 +28,7 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use fail::fail_point;
 use itertools::Itertools;
@@ -82,6 +82,8 @@ impl Coordinator {
         ctx: Option<&mut ExecuteContext>,
         catalog_updates: Vec<ParsedStateUpdate>,
     ) -> Result<(), AdapterError> {
+        let start = Instant::now();
+
         let mut catalog_implications: BTreeMap<CatalogItemId, CatalogImplication> = BTreeMap::new();
         let mut cluster_commands: BTreeMap<ClusterId, CatalogImplication> = BTreeMap::new();
         let mut cluster_replica_commands: BTreeMap<(ClusterId, ReplicaId), CatalogImplication> =
@@ -144,6 +146,10 @@ impl Coordinator {
         )
         .await?;
 
+        self.metrics
+            .apply_catalog_implications_seconds
+            .observe(start.elapsed().as_secs_f64());
+
         Ok(())
     }
 

src/adapter/src/coord/ddl.rs

@@ -13,7 +13,7 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::pin::Pin;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use fail::fail_point;
 use maplit::{btreemap, btreeset};
@@ -67,8 +67,15 @@ impl Coordinator {
         session: Option<&Session>,
         ops: Vec<catalog::Op>,
     ) -> Result<(), AdapterError> {
-        self.catalog_transact_with_context(session.map(|session| session.conn_id()), None, ops)
-            .await
+        let start = Instant::now();
+        let result = self
+            .catalog_transact_with_context(session.map(|session| session.conn_id()), None, ops)
+            .await;
+        self.metrics
+            .catalog_transact_seconds
+            .with_label_values(&["catalog_transact"])
+            .observe(start.elapsed().as_secs_f64());
+        result
     }
 
     /// Same as [`Self::catalog_transact_with_context`] but takes a [`Session`]
@@ -93,6 +100,8 @@ impl Coordinator {
             ) -> Pin<Box<dyn Future<Output = ()> + 'a>>
             + 'static,
     {
+        let start = Instant::now();
+
         let (table_updates, catalog_updates) = self
             .catalog_transact_inner(ctx.as_ref().map(|ctx| ctx.session().conn_id()), ops)
             .await?;
@@ -129,6 +138,11 @@ impl Coordinator {
         )
         .await;
 
+        self.metrics
+            .catalog_transact_seconds
+            .with_label_values(&["catalog_transact_with_side_effects"])
+            .observe(start.elapsed().as_secs_f64());
+
         Ok(())
     }
 
@@ -146,6 +160,8 @@ impl Coordinator {
         ctx: Option<&mut ExecuteContext>,
         ops: Vec<catalog::Op>,
     ) -> Result<(), AdapterError> {
+        let start = Instant::now();
+
         let conn_id = conn_id.or_else(|| ctx.as_ref().map(|ctx| ctx.session().conn_id()));
 
         let (table_updates, catalog_updates) = self.catalog_transact_inner(conn_id, ops).await?;
@@ -176,6 +192,11 @@ impl Coordinator {
             "coordinator inconsistency detected"
         );
 
+        self.metrics
+            .catalog_transact_seconds
+            .with_label_values(&["catalog_transact_with_context"])
+            .observe(start.elapsed().as_secs_f64());
+
         Ok(())
     }
 
@@ -197,6 +218,8 @@ impl Coordinator {
             + Sync
             + 'static,
     {
+        let start = Instant::now();
+
         let Some(Transaction {
             ops:
                 TransactionOps::DDL {
@@ -208,13 +231,22 @@ impl Coordinator {
             ..
         }) = ctx.session().transaction().inner()
         else {
-            return self
+            let result = self
                 .catalog_transact_with_side_effects(Some(ctx), ops, side_effect)
                 .await;
+            self.metrics
+                .catalog_transact_seconds
+                .with_label_values(&["catalog_transact_with_ddl_transaction"])
+                .observe(start.elapsed().as_secs_f64());
+            return result;
         };
 
         // Make sure our Catalog hasn't changed since openning the transaction.
         if self.catalog().transient_revision() != *txn_revision {
+            self.metrics
+                .catalog_transact_seconds
+                .with_label_values(&["catalog_transact_with_ddl_transaction"])
+                .observe(start.elapsed().as_secs_f64());
             return Err(AdapterError::DDLTransactionRace);
         }
 
@@ -227,7 +259,7 @@ impl Coordinator {
         // Run our Catalog transaction, but abort before committing.
         let result = self.catalog_transact(Some(ctx.session()), all_ops).await;
 
-        match result {
+        let result = match result {
             // We purposefully fail with this error to prevent committing the transaction.
             Err(AdapterError::TransactionDryRun { new_ops, new_state }) => {
                 // Sets these ops to our transaction, bailing if the Catalog has changed since we
@@ -244,7 +276,14 @@ impl Coordinator {
             }
             Ok(_) => unreachable!("unexpected success!"),
             Err(e) => Err(e),
-        }
+        };
+
+        self.metrics
+            .catalog_transact_seconds
+            .with_label_values(&["catalog_transact_with_ddl_transaction"])
+            .observe(start.elapsed().as_secs_f64());
+
+        result
     }
 
     /// Perform a catalog transaction. [`Coordinator::ship_dataflow`] must be

src/adapter/src/metrics.rs

@@ -50,6 +50,8 @@ pub struct Metrics {
     pub pgwire_ensure_transaction_seconds: HistogramVec,
     pub catalog_snapshot_seconds: HistogramVec,
     pub pgwire_recv_scheduling_delay_ms: HistogramVec,
+    pub catalog_transact_seconds: HistogramVec,
+    pub apply_catalog_implications_seconds: Histogram,
 }
 
 impl Metrics {
@@ -221,6 +223,17 @@ impl Metrics {
                 help: "The time between a pgwire connection's receiver task being woken up by incoming data and getting polled.",
                 var_labels: ["message_type"],
                 buckets: histogram_milliseconds_buckets(0.128, 512000.),
+            )),
+            catalog_transact_seconds: registry.register(metric!(
+                name: "mz_catalog_transact_seconds",
+                help: "The time it takes to run various catalog transact methods.",
+                var_labels: ["method"],
+                buckets: histogram_seconds_buckets(0.001, 32.0),
+            )),
+            apply_catalog_implications_seconds: registry.register(metric!(
+                name: "mz_apply_catalog_implications_seconds",
+                help: "The time it takes to apply catalog implications.",
+                buckets: histogram_seconds_buckets(0.001, 32.0),
             ))
         }
     }

src/catalog/src/durable.rs

@@ -12,7 +12,7 @@
 use std::fmt::Debug;
 use std::num::NonZeroI64;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
 use itertools::Itertools;
@@ -211,6 +211,9 @@ pub trait ReadOnlyDurableCatalogState: Debug + Send + Sync {
     /// NB: We may remove this in later iterations of Pv2.
     fn epoch(&self) -> Epoch;
 
+    /// Returns the metrics for this catalog state.
+    fn metrics(&self) -> &Metrics;
+
     /// Politely releases all external resources that can only be released in an async context.
     async fn expire(self: Box<Self>);
 
@@ -323,12 +326,16 @@ pub trait DurableCatalogState: ReadOnlyDurableCatalogState {
         amount: u64,
         commit_ts: Timestamp,
     ) -> Result<Vec<u64>, CatalogError> {
+        let start = Instant::now();
         if amount == 0 {
             return Ok(Vec::new());
         }
         let mut txn = self.transaction().await?;
         let ids = txn.get_and_increment_id_by(id_type.to_string(), amount)?;
         txn.commit_internal(commit_ts).await?;
+        self.metrics()
+            .allocate_id_seconds
+            .observe(start.elapsed().as_secs_f64());
         Ok(ids)
     }
 

src/catalog/src/durable/metrics.rs

@@ -11,7 +11,8 @@
 
 use mz_ore::metric;
 use mz_ore::metrics::{IntCounter, MetricsRegistry};
-use prometheus::{Counter, IntGaugeVec};
+use mz_ore::stats::histogram_seconds_buckets;
+use prometheus::{Counter, Histogram, IntGaugeVec};
 
 #[derive(Debug, Clone)]
 pub struct Metrics {
@@ -23,6 +24,7 @@ pub struct Metrics {
     pub syncs: IntCounter,
     pub sync_latency_seconds: Counter,
     pub collection_entries: IntGaugeVec,
+    pub allocate_id_seconds: Histogram,
 }
 
 impl Metrics {
@@ -62,6 +64,11 @@ impl Metrics {
                 help: "Total number of entries, after consolidation, per catalog collection.",
                 var_labels: ["collection"],
             )),
+            allocate_id_seconds: registry.register(metric!(
+                name: "mz_catalog_allocate_id_seconds",
+                help: "The time it takes to allocate IDs in the durable catalog.",
+                buckets: histogram_seconds_buckets(0.001, 32.0),
+            )),
         }
     }
 }

src/catalog/src/durable/persist.rs

@@ -1599,6 +1599,10 @@ impl ReadOnlyDurableCatalogState for PersistCatalogState {
             .epoch
     }
 
+    fn metrics(&self) -> &Metrics {
+        &self.metrics
+    }
+
     #[mz_ore::instrument(level = "debug")]
     async fn expire(self: Box<Self>) {
         self.expire().await