Update Core & pass through new metrics options (#1383)

Co-authored-by: James Watkins-Harvey <james.watkinsharvey@temporal.io>

Author: Sushisource

packages/core-bridge/Cargo.lock

@@ -20,10 +20,10 @@ incremental = false
 futures = { version = "0.3", features = ["executor"] }
 log = "0.4"
 neon = { version = "0.10", default-features = false, features = ["napi-6", "event-queue-api"] }
-opentelemetry = "0.18"
+opentelemetry = "0.22"
 parking_lot = "0.12"
-prost = "0.11"
-prost-types = "0.11"
+prost = "0.12"
+prost-types = "0.12"
 serde_json = "1.0"
 tokio = "1.13"
 once_cell = "1.19"

packages/core-bridge/Cargo.toml

@@ -158,6 +158,17 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
         if let Some(tls_cfg) = tls_cfg {
             client_options.tls_cfg(tls_cfg);
         }
+        let headers = match js_optional_getter!(cx, self, "metadata", JsObject) {
+            None => None,
+            Some(h) => Some(h.as_hash_map_of_string_to_string(cx).map_err(|reason| {
+                cx.throw_type_error::<_, HashMap<String, String>>(format!(
+                    "Invalid metadata: {}",
+                    reason
+                ))
+                .unwrap_err()
+            })?),
+        };
+        client_options.headers(headers);
 
         Ok(client_options
             .client_name("temporal-typescript".to_string())
@@ -211,6 +222,22 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
                     }
                 };
 
+                if let Some(counters_total_suffix) =
+                    js_optional_value_getter!(cx, prom, "countersTotalSuffix", JsBoolean)
+                {
+                    options.counters_total_suffix(counters_total_suffix);
+                }
+                if let Some(unit_suffix) =
+                    js_optional_value_getter!(cx, prom, "unitSuffix", JsBoolean)
+                {
+                    options.unit_suffix(unit_suffix);
+                }
+                if let Some(use_seconds_for_durations) =
+                    js_optional_value_getter!(cx, prom, "useSecondsForDurations", JsBoolean)
+                {
+                    options.use_seconds_for_durations(use_seconds_for_durations);
+                }
+
                 let options = options.build().map_err(|e| {
                     cx.throw_type_error::<_, TelemetryOptions>(format!(
                         "Failed to build prometheus exporter options: {:?}",
@@ -240,6 +267,12 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
                     }
                 };
 
+                if let Some(use_seconds_for_durations) =
+                    js_optional_value_getter!(cx, otel, "useSecondsForDurations", JsBoolean)
+                {
+                    options.use_seconds_for_durations(use_seconds_for_durations);
+                }
+
                 if let Some(ref headers) = js_optional_getter!(cx, otel, "headers", JsObject) {
                     options.headers(headers.as_hash_map_of_string_to_string(cx)?);
                 };

packages/core-bridge/sdk-core

@@ -1,28 +1,23 @@
 use crate::{conversions::*, errors::*, helpers::*, worker::*};
-use neon::context::Context;
-use neon::prelude::*;
-use parking_lot::RwLock;
-use std::cell::Cell;
+use neon::{context::Context, prelude::*};
 use std::{
-    cell::RefCell,
+    cell::{Cell, RefCell},
     collections::HashMap,
     ops::Deref,
     sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 use temporal_client::{ClientInitError, ConfiguredClient, TemporalServiceClientWithMetrics};
-use temporal_sdk_core::api::telemetry::CoreTelemetry;
-use temporal_sdk_core::CoreRuntime;
 use temporal_sdk_core::{
+    api::telemetry::CoreTelemetry,
     ephemeral_server::EphemeralServer as CoreEphemeralServer,
     init_replay_worker, init_worker,
     replay::{HistoryForReplay, ReplayWorkerInput},
-    ClientOptions, RetryClient, WorkerConfig,
+    ClientOptions, CoreRuntime, RetryClient, WorkerConfig,
 };
-use tokio::sync::oneshot;
 use tokio::sync::{
     mpsc::{channel, unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},
-    Mutex,
+    oneshot, Mutex,
 };
 use tokio_stream::wrappers::ReceiverStream;
 
@@ -64,7 +59,6 @@ pub enum RuntimeRequest {
     CreateClient {
         runtime: Arc<RuntimeHandle>,
         options: ClientOptions,
-        headers: Option<HashMap<String, String>>,
         /// Used to send the result back into JS
         callback: Root<JsFunction>,
     },
@@ -166,13 +160,12 @@ pub fn start_bridge_loop(
                 RuntimeRequest::CreateClient {
                     runtime,
                     options,
-                    headers,
                     callback,
                 } => {
                     let mm = core_runtime.telemetry().get_metric_meter();
                     core_runtime.tokio_handle().spawn(async move {
                         match options
-                            .connect_no_namespace(mm, headers.map(|h| Arc::new(RwLock::new(h))))
+                            .connect_no_namespace(mm)
                             .await
                         {
                             Err(err) => {
@@ -337,8 +330,7 @@ pub fn start_bridge_loop(
                                     format!("Failed to start test server: {}", err),
                                 )
                             },
-                        )
-                        .await
+                        ).await
                     });
                 }
                 RuntimeRequest::PushReplayHistory {
@@ -350,8 +342,8 @@ pub fn start_bridge_loop(
                         let sendfut = async move {
                             tx.send(pushme).await.map_err(|e| {
                                 format!(
-                    "Receive side of history replay channel is gone. This is an sdk bug. {:?}",
-                    e
+                                    "Receive side of history replay channel is gone. This is an sdk bug. {:?}",
+                                    e
                                 )
                             })
                         };
@@ -361,8 +353,7 @@ pub fn start_bridge_loop(
                                 UNEXPECTED_ERROR,
                                 format!("Error pushing replay history {}", err),
                             )
-                        })
-                        .await
+                        }).await
                     });
                 }
             }
@@ -453,24 +444,10 @@ pub fn client_new(mut cx: FunctionContext) -> JsResult<JsUndefined> {
     let callback = cx.argument::<JsFunction>(2)?;
 
     let client_options = opts.as_client_options(&mut cx)?;
-    let headers = match js_optional_getter!(&mut cx, &opts, "metadata", JsObject) {
-        None => None,
-        Some(h) => Some(
-            h.as_hash_map_of_string_to_string(&mut cx)
-                .map_err(|reason| {
-                    cx.throw_type_error::<_, HashMap<String, String>>(format!(
-                        "Invalid metadata: {}",
-                        reason
-                    ))
-                    .unwrap_err()
-                })?,
-        ),
-    };
 
     let request = RuntimeRequest::CreateClient {
         runtime: (**runtime).clone(),
         options: client_options,
-        headers,
         callback: callback.root(&mut cx),
     };
     if let Err(err) = runtime.sender.send(request) {

packages/core-bridge/src/conversions.rs

@@ -122,6 +122,10 @@ export interface OtelCollectorExporter {
      * @defaults 1 second
      */
     metricsExportInterval?: Duration;
+    /**
+     * If set to true, the exporter will use seconds for durations instead of milliseconds.
+     */
+    useSecondsForDurations?: boolean;
   };
 }
 
@@ -142,6 +146,19 @@ export interface PrometheusMetricsExporter {
      * Metrics will be available for scraping under the standard `/metrics` route.
      */
     bindAddress: string;
+    /**
+     * If set to true, all counter names will include a "_total" suffix.
+     */
+    countersTotalSuffix?: boolean;
+    /**
+     * If set to true, all histograms will include the unit in their name as a suffix.
+     * EX: "_milliseconds"
+     */
+    unitSuffix?: boolean;
+    /**
+     * If set to true, the exporter will use seconds for durations instead of milliseconds.
+     */
+    useSecondsForDurations?: boolean;
   };
 }
 

packages/core-bridge/src/runtime.rs

@@ -33,6 +33,7 @@ test('NativeConnection passes headers provided in options', async (t) => {
       callback: grpc.sendUnaryData<temporal.api.workflowservice.v1.IGetSystemInfoResponse>
     ) {
       const [value] = call.metadata.get('initial');
+      console.log(call.metadata);
       if (value === 'true') {
         gotInitialHeader = true;
       }

packages/core-bridge/ts/index.ts

@@ -64,3 +64,50 @@ test.serial('Exporting Prometheus metrics from Core works', async (t) => {
     await localEnv.teardown();
   }
 });
+
+test.serial('Exporting Prometheus metrics from Core works with lots of options', async (t) => {
+  const port = await getRandomPort();
+  Runtime.install({
+    telemetryOptions: {
+      metrics: {
+        prometheus: {
+          bindAddress: `127.0.0.1:${port}`,
+          countersTotalSuffix: true,
+          unitSuffix: true,
+          useSecondsForDurations: true,
+        },
+      },
+    },
+  });
+  const localEnv = await TestWorkflowEnvironment.createLocal();
+  try {
+    const worker = await Worker.create({
+      connection: localEnv.nativeConnection,
+      workflowsPath: require.resolve('./workflows'),
+      taskQueue: 'test-prometheus',
+    });
+    const client = new WorkflowClient({
+      connection: localEnv.connection,
+    });
+    await worker.runUntil(async () => {
+      await client.execute(workflows.successString, {
+        taskQueue: 'test-prometheus',
+        workflowId: uuid4(),
+      });
+      const resp = await fetch(`http://127.0.0.1:${port}/metrics`);
+      const text = await resp.text();
+      // Verify use seconds & unit suffix
+      t.assert(
+        text.includes(
+          'temporal_workflow_task_replay_latency_seconds_bucket{namespace="default",' +
+            'service_name="temporal-core-sdk",task_queue="test-prometheus",' +
+            'workflow_type="successString",le="0.001"}'
+        )
+      );
+      // Verify 'total' suffix
+      t.assert(text.includes('temporal_worker_start_total'));
+    });
+  } finally {
+    await localEnv.teardown();
+  }
+});

packages/test/src/test-native-connection-headers.ts

@@ -241,11 +241,15 @@ export class Runtime {
                   url: metrics.otel.url,
                   headers: metrics.otel.headers ?? {},
                   metricsExportInterval: msToNumber(metrics.otel.metricsExportInterval ?? '1s'),
+                  useSecondsForDurations: metrics.otel.useSecondsForDurations,
                 },
               }
             : {
                 prometheus: {
                   bindAddress: metrics.prometheus.bindAddress,
+                  unitSuffix: metrics.prometheus.unitSuffix,
+                  countersTotalSuffix: metrics.prometheus.countersTotalSuffix,
+                  useSecondsForDurations: metrics.prometheus.useSecondsForDurations,
                 },
               }),
         },