Just allow attaching a metric meter late

Author: Sushisource

core/src/lib.rs

@@ -55,7 +55,10 @@ use std::sync::Arc;
 use temporal_client::{ConfiguredClient, TemporalServiceClientWithMetrics};
 use temporal_sdk_core_api::{
     errors::{CompleteActivityError, PollActivityError, PollWfError},
-    telemetry::{metrics::TemporalMeter, TelemetryOptions},
+    telemetry::{
+        metrics::{CoreMeter, TemporalMeter},
+        TelemetryOptions,
+    },
     Worker as WorkerTrait,
 };
 use temporal_sdk_core_protos::coresdk::ActivityHeartbeat;
@@ -278,6 +281,12 @@ impl CoreRuntime {
     pub fn telemetry(&self) -> &TelemetryInstance {
         &self.telemetry
     }
+
+    /// Some metric meters cannot be initialized until after a tokio runtime has started and after
+    /// other telemetry has initted (ex: prometheus). They can be attached here.
+    pub fn attach_late_init_metrics(&mut self, meter: Arc<dyn CoreMeter + 'static>) {
+        self.telemetry.attach_late_init_metrics(meter)
+    }
 }
 
 impl Drop for CoreRuntime {

core/src/telemetry/mod.rs

@@ -77,6 +77,12 @@ impl TelemetryInstance {
         self.trace_subscriber.clone()
     }
 
+    /// Some metric meters cannot be initialized until after a tokio runtime has started and after
+    /// other telemetry has initted (ex: prometheus). They can be attached here.
+    pub fn attach_late_init_metrics(&mut self, meter: Arc<dyn CoreMeter + 'static>) {
+        self.metrics = Some(meter);
+    }
+
     /// Returns our wrapper for metric meters, can be used to, ex: initialize clients
     pub fn get_metric_meter(&self) -> Option<TemporalMeter> {
         self.metrics.clone().map(|m| {

tests/integ_tests/metrics_tests.rs

@@ -423,3 +423,30 @@ async fn query_of_closed_workflow_doesnt_tick_terminal_metric(
         .expect("Must find matching metric");
     assert!(matching_line.ends_with('1'));
 }
+
+#[test]
+fn runtime_new() {
+    let mut rt = CoreRuntime::new(
+        get_integ_telem_options(),
+        tokio::runtime::Builder::new_multi_thread(),
+    )
+    .unwrap();
+    let handle = rt.tokio_handle();
+    let _rt = handle.enter();
+    let (telemopts, addr, _aborter) = prom_metrics();
+    rt.attach_late_init_metrics(telemopts.metrics.unwrap());
+    let opts = get_integ_server_options();
+    handle.block_on(async {
+        let mut raw_client = opts
+            .connect_no_namespace(rt.metric_meter(), None)
+            .await
+            .unwrap();
+        assert!(raw_client.get_client().capabilities().is_some());
+        let _ = raw_client
+            .list_namespaces(ListNamespacesRequest::default())
+            .await
+            .unwrap();
+        let body = get_text(format!("http://{addr}/metrics")).await;
+        assert!(body.contains("temporal_request"));
+    });
+}