feat(worker): Make OTel export period configurable (#977)

Author: mjameswh

packages/core-bridge/index.d.ts

@@ -3,6 +3,18 @@ import type { TLSConfig } from '@temporalio/common/lib/internal-non-workflow';
 
 export { TLSConfig };
 
+type Shadow<Base, New> = Base extends object
+  ? New extends object
+    ? {
+        [K in keyof Base | keyof New]: K extends keyof Base
+          ? K extends keyof New
+            ? Shadow<Base[K], New[K]>
+            : Base[K]
+          : New[K];
+      }
+    : New
+  : New;
+
 export interface RetryOptions {
   /** Initial wait time before the first retry. */
   initialInterval: number;
@@ -98,9 +110,22 @@ export interface OtelCollectorExporter {
      * Optional set of HTTP request headers to send to Collector (e.g. for authentication)
      */
     headers?: Record<string, string>;
+    /**
+     * Specify how frequently in metrics should be exported.
+     *
+     * @format number of milliseconds or {@link https://www.npmjs.com/package/ms | ms-formatted string}
+     * @defaults 1 second
+     */
+    metricsExportInterval?: string | number;
   };
 }
 
+/** @experimental */
+export type CompiledOtelTraceExporter = Shadow<OtelCollectorExporter, { otel: { metricsExportInterval?: never } }>;
+
+/** @experimental */
+export type CompiledOtelMetricsExporter = Shadow<OtelCollectorExporter, { otel: { metricsExportInterval: number } }>;
+
 /**
  * Prometheus metrics exporter options
  *
@@ -224,8 +249,10 @@ export type CompiledTelemetryOptions = {
   );
   tracing?: {
     filter: string;
-  } & TraceExporter;
-  metrics?: MetricsExporter;
+  } & CompiledOtelTraceExporter;
+  metrics?: {
+    temporality?: 'cumulative' | 'delta';
+  } & (PrometheusMetricsExporter | CompiledOtelMetricsExporter);
 };
 
 export interface WorkerOptions {

packages/core-bridge/sdk-core

@@ -231,8 +231,14 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
                     } else {
                         Default::default()
                     };
+                let metric_periodicity = Some(Duration::from_millis(js_value_getter!(
+                    cx,
+                    otel,
+                    "metricPeriodicity",
+                    JsNumber
+                ) as u64));
                 telemetry_opts
-                    .metrics(MetricsExporter::Otel(OtelCollectorOptions { url, headers }));
+                    .metrics(MetricsExporter::Otel(OtelCollectorOptions { url, headers, metric_periodicity }));
             } else {
                 cx.throw_type_error(
                     "Invalid telemetryOptions.metrics, missing `prometheus` or `otel` option",
@@ -256,7 +262,7 @@ impl ObjectHandleConversionsExt for Handle<'_, JsObject> {
                     };
                 telemetry_opts.tracing(TraceExportConfig {
                     filter,
-                    exporter: TraceExporter::Otel(OtelCollectorOptions { url, headers }),
+                    exporter: TraceExporter::Otel(OtelCollectorOptions { url, headers, metric_periodicity: None }),
                 });
             } else {
                 cx.throw_type_error(

packages/core-bridge/src/conversions.rs

@@ -7,6 +7,8 @@ import {
   TelemetryOptions,
   CompiledTelemetryOptions,
   ForwardLogger,
+  MetricsExporter,
+  OtelCollectorExporter,
 } from '@temporalio/core-bridge';
 import { filterNullAndUndefined, normalizeTlsConfig } from '@temporalio/common/lib/internal-non-workflow';
 import { IllegalStateError } from '@temporalio/common';
@@ -24,13 +26,18 @@ import { History } from '@temporalio/common/lib/proto-utils';
 import * as v8 from 'v8';
 import * as fs from 'fs';
 import * as os from 'os';
+import { msToNumber } from '@temporalio/common/lib/time';
 
 export { History };
 
 function isForwardingLogger(opts: TelemetryOptions['logging']): opts is ForwardLogger {
   return Object.hasOwnProperty.call(opts, 'forward');
 }
 
+function isOtelCollectorExporter(opts: MetricsExporter): opts is OtelCollectorExporter {
+  return Object.hasOwnProperty.call(opts, 'otel');
+}
+
 /**
  * Options used to create a Core runtime
  */
@@ -207,7 +214,7 @@ export class Runtime {
   // eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types
   protected static compileOptions(options: RuntimeOptions): CompiledRuntimeOptions {
     // eslint-disable-next-line deprecation/deprecation
-    const { logging, tracing, tracingFilter, ...otherTelemetryOpts } = options.telemetryOptions ?? {};
+    const { logging, tracing, metrics, tracingFilter, ...otherTelemetryOpts } = options.telemetryOptions ?? {};
 
     const defaultFilter = tracingFilter ?? makeTelemetryFilterString({ core: 'INFO', other: 'INFO' });
     const loggingFilter = logging?.filter;
@@ -232,9 +239,27 @@ export class Runtime {
         tracing: tracing?.otel && {
           filter: defaultFilter,
           otel: {
-            ...filterNullAndUndefined(tracing.otel),
+            url: tracing.otel.url,
+            headers: tracing.otel.headers,
+            metricsExportInterval: undefined,
           },
         },
+        metrics: metrics && {
+          temporality: metrics.temporality,
+          ...(isOtelCollectorExporter(metrics)
+            ? {
+                otel: {
+                  url: metrics.otel.url,
+                  headers: metrics.otel.headers,
+                  metricsExportInterval: msToNumber(metrics.otel.metricsExportInterval ?? '1s'),
+                },
+              }
+            : {
+                prometheus: {
+                  bindAddress: metrics.prometheus.bindAddress,
+                },
+              }),
+        },
         ...filterNullAndUndefined(otherTelemetryOpts ?? {}),
       },
       logger: options.logger ?? new DefaultLogger('INFO'),