INC-1252: Clean up adapter metrics keyed by clusters

src/adapter/src/client.rs

@@ -7,7 +7,6 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use std::borrow::Cow;
 use std::collections::BTreeMap;
 use std::fmt::{Debug, Display, Formatter};
 use std::future::Future;
@@ -2187,25 +2186,12 @@ impl RecordFirstRowStream {
         let session = client.session.as_ref().expect("session invariant");
         let isolation_level = *session.vars().transaction_isolation();
         let name_hint = ApplicationNameHint::from_str(session.application_name());
-        let instance = match instance_id {
-            Some(i) => Cow::Owned(i.to_string()),
-            None => Cow::Borrowed("none"),
-        };
-        let strategy = match strategy {
-            Some(s) => s.name(),
-            None => "none",
-        };
 
         client
             .inner()
             .metrics()
-            .time_to_first_row_seconds
-            .with_label_values(&[
-                instance.as_ref(),
-                isolation_level.as_variant_str(),
-                strategy,
-                name_hint.as_str(),
-            ])
+            .by_cluster
+            .time_to_first_row_seconds(instance_id, isolation_level, strategy, name_hint)
     }
 
     /// If you want to match [`RecordFirstRowStream`]'s logic but don't need

src/adapter/src/coord/catalog_implications.rs

@@ -1030,6 +1030,10 @@ impl Coordinator {
                 self.drop_vpc_endpoints_in_background(vpc_endpoints_to_drop)
             }
 
+            let clusters_losing_replicas: BTreeSet<_> = cluster_replicas_to_drop
+                .iter()
+                .map(|(cluster_id, _)| *cluster_id)
+                .collect();
             if !cluster_replicas_to_drop.is_empty() {
                 fail::fail_point!("after_catalog_drop_replica");
 
@@ -1038,8 +1042,20 @@ impl Coordinator {
                 }
             }
             if !clusters_to_drop.is_empty() {
-                for cluster_id in clusters_to_drop {
-                    self.controller.drop_cluster(cluster_id);
+                for cluster_id in &clusters_to_drop {
+                    self.controller.drop_cluster(*cluster_id);
+                }
+            }
+            // A dropped cluster, or one left without replicas, cannot serve
+            // peeks, so its peek series are stale. They come back on the first
+            // peek once a cluster has a replica again.
+            for cluster_id in clusters_losing_replicas.into_iter().chain(clusters_to_drop) {
+                let has_replicas = self
+                    .catalog()
+                    .try_get_cluster(cluster_id)
+                    .is_some_and(|cluster| cluster.replicas().next().is_some());
+                if !has_replicas {
+                    self.metrics.by_cluster.remove_cluster(cluster_id);
                 }
             }
 

src/adapter/src/coord/timestamp_selection.rs

@@ -698,15 +698,12 @@ impl Coordinator {
             isolation_level,
         )?;
         self.metrics
-            .determine_timestamp
-            .with_label_values(&[
-                match det.respond_immediately() {
-                    true => "true",
-                    false => "false",
-                },
-                isolation_level.as_variant_str(),
-                &compute_instance.to_string(),
-            ])
+            .by_cluster
+            .determine_timestamp(
+                compute_instance,
+                det.respond_immediately(),
+                *isolation_level,
+            )
             .inc();
         if !det.respond_immediately()
             && isolation_level.is_bounded_staleness()
@@ -725,8 +722,8 @@ impl Coordinator {
                 )?;
                 if let Some(serializable) = serializable_det.timestamp_context.timestamp() {
                     self.metrics
-                        .timestamp_difference_for_bounded_staleness_ms
-                        .with_label_values(&[compute_instance.to_string().as_str()])
+                        .by_cluster
+                        .timestamp_difference_for_bounded_staleness_ms(compute_instance)
                         .observe(f64::cast_lossy(u64::from(
                             serializable.saturating_sub(*bs_ts),
                         )));

src/adapter/src/frontend_peek.rs

@@ -1532,14 +1532,12 @@ impl PeekClient {
 
         session
             .metrics()
-            .determine_timestamp(&[
-                match det.respond_immediately() {
-                    true => "true",
-                    false => "false",
-                },
-                isolation_level.as_variant_str(),
-                &compute_instance.to_string(),
-            ])
+            .by_cluster()
+            .determine_timestamp(
+                compute_instance,
+                det.respond_immediately(),
+                *isolation_level,
+            )
             .inc();
         if !det.respond_immediately()
             && isolation_level.is_bounded_staleness()
@@ -1562,9 +1560,8 @@ impl PeekClient {
                 if let Some(serializable) = serializable_det.timestamp_context.timestamp() {
                     session
                         .metrics()
-                        .timestamp_difference_for_bounded_staleness_ms(&[compute_instance
-                            .to_string()
-                            .as_ref()])
+                        .by_cluster()
+                        .timestamp_difference_for_bounded_staleness_ms(compute_instance)
                         .observe(f64::cast_lossy(u64::from(
                             serializable.saturating_sub(*bs_ts),
                         )));