catalog: Prevent duplicate catalog metrics reg (#23541)

This commit prevents catalog metrics from being registered multiple
times and panicking.

Resolves #23465

Author: jkosh44

src/catalog-debug/src/main.rs

@@ -242,10 +242,8 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
             };
             let persist_client = persist_clients.open(persist_location).await?;
             let organization_id = args.organization_id.expect("required for persist");
-            Box::new(
-                persist_backed_catalog_state(persist_client, organization_id, &metrics_registry)
-                    .await,
-            )
+            let metrics = Arc::new(mz_catalog::durable::Metrics::new(&metrics_registry));
+            Box::new(persist_backed_catalog_state(persist_client, organization_id, metrics).await)
         }
     };
 

src/catalog/src/durable.rs

@@ -12,13 +12,15 @@
 use async_trait::async_trait;
 use std::fmt::Debug;
 use std::num::NonZeroI64;
+use std::sync::Arc;
 use std::time::Duration;
 use uuid::Uuid;
 
 use mz_stash::DebugStashFactory;
 
 use crate::durable::debug::{DebugCatalogState, Trace};
 pub use crate::durable::error::{CatalogError, DurableCatalogError};
+pub use crate::durable::impls::persist::metrics::Metrics;
 use crate::durable::impls::persist::PersistHandle;
 use crate::durable::impls::shadow::OpenableShadowCatalogState;
 use crate::durable::impls::stash::{OpenableConnection, TestOpenableConnection};
@@ -276,9 +278,9 @@ pub fn test_stash_backed_catalog_state(
 pub async fn persist_backed_catalog_state(
     persist_client: PersistClient,
     organization_id: Uuid,
-    registry: &MetricsRegistry,
+    metrics: Arc<Metrics>,
 ) -> PersistHandle {
-    PersistHandle::new(persist_client, organization_id, registry).await
+    PersistHandle::new(persist_client, organization_id, metrics).await
 }
 
 /// Creates an openable durable catalog state implemented using persist that is meant to be used in
@@ -287,7 +289,8 @@ pub async fn test_persist_backed_catalog_state(
     persist_client: PersistClient,
     organization_id: Uuid,
 ) -> PersistHandle {
-    persist_backed_catalog_state(persist_client, organization_id, &MetricsRegistry::new()).await
+    let metrics = Arc::new(Metrics::new(&MetricsRegistry::new()));
+    persist_backed_catalog_state(persist_client, organization_id, metrics).await
 }
 
 /// Creates an openable durable catalog state implemented using both the stash and persist, that

src/catalog/src/durable/impls/persist.rs

@@ -7,7 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-mod metrics;
+pub(crate) mod metrics;
 pub(crate) mod state_update;
 
 use std::collections::BTreeMap;
@@ -23,7 +23,7 @@ use futures::StreamExt;
 use itertools::Itertools;
 use mz_audit_log::{VersionedEvent, VersionedStorageUsage};
 use mz_ore::collections::CollectionExt;
-use mz_ore::metrics::{MetricsFutureExt, MetricsRegistry};
+use mz_ore::metrics::MetricsFutureExt;
 use mz_ore::now::EpochMillis;
 use mz_ore::retry::{Retry, RetryResult};
 use mz_ore::{soft_assert, soft_assert_eq, soft_assert_ne};
@@ -103,7 +103,7 @@ impl PersistHandle {
     pub(crate) async fn new(
         persist_client: PersistClient,
         organization_id: Uuid,
-        registry: &MetricsRegistry,
+        metrics: Arc<Metrics>,
     ) -> PersistHandle {
         const SEED: usize = 1;
         let shard_id = Self::shard_id(organization_id, SEED);
@@ -117,7 +117,6 @@ impl PersistHandle {
             )
             .await
             .expect("invalid usage");
-        let metrics = Arc::new(Metrics::new(registry));
         PersistHandle {
             write_handle,
             read_handle,

src/environmentd/src/bin/environmentd/main.rs

@@ -946,7 +946,7 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
             },
             CatalogKind::Persist => CatalogConfig::Persist {
                 persist_clients,
-                metrics_registry: metrics_registry.clone(),
+                metrics: Arc::new(mz_catalog::durable::Metrics::new(&metrics_registry)),
             },
             CatalogKind::Shadow => CatalogConfig::Shadow {
                 url: args.adapter_stash_url.expect("required for shadow catalog"),

src/environmentd/src/lib.rs

@@ -261,8 +261,8 @@ pub enum CatalogConfig {
     Persist {
         /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient.
         persist_clients: Arc<PersistClientCache>,
-        /// Metrics registry.
-        metrics_registry: MetricsRegistry,
+        /// Persist catalog metrics.
+        metrics: Arc<mz_catalog::durable::Metrics>,
     },
     /// The catalog contents are stored in both persist and the stash and their contents are
     /// compared. This is mostly used for testing purposes.
@@ -728,7 +728,7 @@ async fn catalog_opener(
         }
         CatalogConfig::Persist {
             persist_clients,
-            metrics_registry,
+            metrics,
         } => {
             info!("Using persist backed catalog");
             let persist_client = persist_clients
@@ -739,7 +739,7 @@ async fn catalog_opener(
                 mz_catalog::durable::persist_backed_catalog_state(
                     persist_client,
                     environment_id.organization_id(),
-                    metrics_registry,
+                    Arc::clone(metrics),
                 )
                 .await,
             )