First pass

Author: Sushisource

client/Cargo.toml

@@ -33,7 +33,9 @@ uuid = { version = "1.1", features = ["v4"] }
 
 [dependencies.temporal-sdk-core-protos]
 path = "../sdk-core-protos"
-version = "0.1"
+
+[dependencies.temporal-sdk-core-api]
+path = "../core-api"
 
 [dev-dependencies]
 assert_matches = "1"

client/src/lib.rs

@@ -13,7 +13,6 @@ mod retry;
 mod workflow_handle;
 
 pub use crate::retry::{CallType, RetryClient, RETRYABLE_ERROR_CODES};
-pub use metrics::ClientMetricProvider;
 pub use raw::{HealthService, OperatorService, TestService, WorkflowService};
 pub use temporal_sdk_core_protos::temporal::api::{
     enums::v1::ArchivalState,
@@ -44,6 +43,7 @@ use std::{
     sync::Arc,
     time::{Duration, Instant},
 };
+use temporal_sdk_core_api::telemetry::metrics::CoreMeter;
 use temporal_sdk_core_protos::{
     coresdk::{workflow_commands::QueryResult, IntoPayloadsExt},
     grpc::health::v1::health_client::HealthClient,
@@ -294,7 +294,7 @@ impl ClientOptions {
     pub async fn connect(
         &self,
         namespace: impl Into<String>,
-        metrics_meter: Option<&dyn ClientMetricProvider>,
+        metrics_meter: Option<&dyn CoreMeter>,
         headers: Option<Arc<RwLock<HashMap<String, String>>>>,
     ) -> Result<RetryClient<Client>, ClientInitError> {
         let client = self
@@ -312,7 +312,7 @@ impl ClientOptions {
     /// See [RetryClient] for more
     pub async fn connect_no_namespace(
         &self,
-        metrics_meter: Option<&dyn ClientMetricProvider>,
+        metrics_meter: Option<&dyn CoreMeter>,
         headers: Option<Arc<RwLock<HashMap<String, String>>>>,
     ) -> Result<RetryClient<ConfiguredClient<TemporalServiceClientWithMetrics>>, ClientInitError>
     {
@@ -327,7 +327,7 @@ impl ClientOptions {
         let service = ServiceBuilder::new()
             .layer_fn(|channel| GrpcMetricSvc {
                 inner: channel,
-                metrics: metrics_meter.map(|mm| MetricsContext::new(vec![], mm)),
+                metrics: metrics_meter.map(|mm| MetricsContext::new(mm)),
             })
             .service(channel);
         let headers = headers.unwrap_or_default();

client/src/metrics.rs

@@ -1,49 +1,38 @@
 use crate::{AttachMetricLabels, LONG_POLL_METHOD_NAMES};
 use futures::{future::BoxFuture, FutureExt};
-use opentelemetry::{
-    metrics::{Counter, Histogram},
-    KeyValue,
-};
 use std::{
     sync::Arc,
     task::{Context, Poll},
     time::{Duration, Instant},
 };
+use temporal_sdk_core_api::telemetry::metrics::{
+    CoreMeter, Counter, Histogram, MetricAttributes, MetricKeyValue, MetricsAttributesOptions,
+};
 use tonic::{body::BoxBody, transport::Channel};
 use tower::Service;
 
 /// Used to track context associated with metrics, and record/update them
 // Possible improvement: make generic over some type tag so that methods are only exposed if the
 // appropriate k/vs have already been set.
-#[derive(Clone, Debug)]
+#[derive(Clone, derive_more::DebugCustom)]
+#[debug(fmt = "MetricsContext {{ attribs: {kvs:?}, poll_is_long: {poll_is_long} }}")]
 pub struct MetricsContext {
-    ctx: opentelemetry::Context,
-    kvs: Arc<Vec<KeyValue>>,
+    kvs: MetricAttributes,
     poll_is_long: bool,
 
-    svc_request: Counter<u64>,
-    svc_request_failed: Counter<u64>,
-    long_svc_request: Counter<u64>,
-    long_svc_request_failed: Counter<u64>,
-
-    svc_request_latency: Histogram<u64>,
-    long_svc_request_latency: Histogram<u64>,
-}
+    svc_request: Arc<dyn Counter>,
+    svc_request_failed: Arc<dyn Counter>,
+    long_svc_request: Arc<dyn Counter>,
+    long_svc_request_failed: Arc<dyn Counter>,
 
-/// Things that can provide metrics for the client implement this. Trait exists to avoid having
-/// to make a whole new lower-level crate just for a tiny shared wrapper around OTel meters.
-pub trait ClientMetricProvider: Send + Sync {
-    /// Construct a counter metric
-    fn counter(&self, name: &'static str) -> Counter<u64>;
-    /// Construct a histogram metric
-    fn histogram(&self, name: &'static str) -> Histogram<u64>;
+    svc_request_latency: Arc<dyn Histogram>,
+    long_svc_request_latency: Arc<dyn Histogram>,
 }
 
 impl MetricsContext {
-    pub(crate) fn new(kvs: Vec<KeyValue>, metric_provider: &dyn ClientMetricProvider) -> Self {
+    pub(crate) fn new(metric_provider: &dyn CoreMeter) -> Self {
         Self {
-            ctx: opentelemetry::Context::current(),
-            kvs: Arc::new(kvs),
+            kvs: metric_provider.new_attributes(MetricsAttributesOptions::default()),
             poll_is_long: false,
             svc_request: metric_provider.counter("request"),
             svc_request_failed: metric_provider.counter("request_failure"),
@@ -55,15 +44,15 @@ impl MetricsContext {
     }
 
     /// Extend an existing metrics context with new attributes, returning a new one
-    pub(crate) fn with_new_attrs(&self, new_kvs: impl IntoIterator<Item = KeyValue>) -> Self {
+    pub(crate) fn with_new_attrs(&self, new_kvs: impl IntoIterator<Item = MetricKeyValue>) -> Self {
         let mut r = self.clone();
         r.add_new_attrs(new_kvs);
         r
     }
 
     /// Add new attributes to the context, mutating it
-    pub(crate) fn add_new_attrs(&mut self, new_kvs: impl IntoIterator<Item = KeyValue>) {
-        Arc::make_mut(&mut self.kvs).extend(new_kvs);
+    pub(crate) fn add_new_attrs(&mut self, new_kvs: impl IntoIterator<Item = MetricKeyValue>) {
+        self.kvs.add_new_attrs(new_kvs);
     }
 
     pub(crate) fn set_is_long_poll(&mut self) {
@@ -73,29 +62,29 @@ impl MetricsContext {
     /// A request to the temporal service was made
     pub(crate) fn svc_request(&self) {
         if self.poll_is_long {
-            self.long_svc_request.add(&self.ctx, 1, &self.kvs);
+            self.long_svc_request.add(1, &self.kvs);
         } else {
-            self.svc_request.add(&self.ctx, 1, &self.kvs);
+            self.svc_request.add(1, &self.kvs);
         }
     }
 
     /// A request to the temporal service failed
     pub(crate) fn svc_request_failed(&self) {
         if self.poll_is_long {
-            self.long_svc_request_failed.add(&self.ctx, 1, &self.kvs);
+            self.long_svc_request_failed.add(1, &self.kvs);
         } else {
-            self.svc_request_failed.add(&self.ctx, 1, &self.kvs);
+            self.svc_request_failed.add(1, &self.kvs);
         }
     }
 
     /// Record service request latency
     pub(crate) fn record_svc_req_latency(&self, dur: Duration) {
         if self.poll_is_long {
             self.long_svc_request_latency
-                .record(&self.ctx, dur.as_millis() as u64, &self.kvs);
+                .record(dur.as_millis() as u64, &self.kvs);
         } else {
             self.svc_request_latency
-                .record(&self.ctx, dur.as_millis() as u64, &self.kvs);
+                .record(dur.as_millis() as u64, &self.kvs);
         }
     }
 }
@@ -104,16 +93,16 @@ const KEY_NAMESPACE: &str = "namespace";
 const KEY_SVC_METHOD: &str = "operation";
 const KEY_TASK_QUEUE: &str = "task_queue";
 
-pub(crate) fn namespace_kv(ns: String) -> KeyValue {
-    KeyValue::new(KEY_NAMESPACE, ns)
+pub(crate) fn namespace_kv(ns: String) -> MetricKeyValue {
+    MetricKeyValue::new(KEY_NAMESPACE, ns)
 }
 
-pub(crate) fn task_queue_kv(tq: String) -> KeyValue {
-    KeyValue::new(KEY_TASK_QUEUE, tq)
+pub(crate) fn task_queue_kv(tq: String) -> MetricKeyValue {
+    MetricKeyValue::new(KEY_TASK_QUEUE, tq)
 }
 
-pub(crate) fn svc_operation(op: String) -> KeyValue {
-    KeyValue::new(KEY_SVC_METHOD, op)
+pub(crate) fn svc_operation(op: String) -> MetricKeyValue {
+    MetricKeyValue::new(KEY_SVC_METHOD, op)
 }
 
 /// Implements metrics functionality for gRPC (really, any http) calls

client/src/raw.rs

@@ -9,6 +9,7 @@ use crate::{
     LONG_POLL_TIMEOUT,
 };
 use futures::{future::BoxFuture, FutureExt, TryFutureExt};
+use temporal_sdk_core_api::telemetry::metrics::MetricKeyValue;
 use temporal_sdk_core_protos::{
     grpc::health::v1::{health_client::HealthClient, *},
     temporal::api::{
@@ -199,10 +200,10 @@ fn req_cloner<T: Clone>(cloneme: &Request<T>) -> Request<T> {
 
 #[derive(Debug)]
 pub(super) struct AttachMetricLabels {
-    pub(super) labels: Vec<opentelemetry::KeyValue>,
+    pub(super) labels: Vec<MetricKeyValue>,
 }
 impl AttachMetricLabels {
-    pub fn new(kvs: impl Into<Vec<opentelemetry::KeyValue>>) -> Self {
+    pub fn new(kvs: impl Into<Vec<MetricKeyValue>>) -> Self {
         Self { labels: kvs.into() }
     }
     pub fn namespace(ns: impl Into<String>) -> Self {

core-api/Cargo.toml

@@ -12,9 +12,14 @@ categories = ["development-tools"]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[features]
+otel_impls = ["opentelemetry"]
+
 [dependencies]
 async-trait = "0.1"
 derive_builder = "0.12"
+derive_more = "0.99"
+opentelemetry = { version = "0.18", optional = true }
 prost-types = "0.11"
 serde = { version = "1.0", default_features = false, features = ["derive"] }
 serde_json = "1.0"
@@ -26,8 +31,4 @@ url = "2.3"
 
 [dependencies.temporal-sdk-core-protos]
 path = "../sdk-core-protos"
-version = "0.1"
-
-[dependencies.temporal-client]
-path = "../client"
-version = "0.1"
+version = "0.1"

core-api/src/telemetry.rs

@@ -1,6 +1,10 @@
+pub mod metrics;
+
+use crate::telemetry::metrics::CoreMeter;
 use std::{
     collections::HashMap,
     net::SocketAddr,
+    sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 use tracing_core::Level;
@@ -82,6 +86,9 @@ pub enum MetricsExporter {
     Otel(OtelCollectorOptions),
     /// Expose metrics directly via an embedded http server bound to the provided address.
     Prometheus(SocketAddr),
+    /// Export metrics calls to lang where it can inject them into whatever metrics system the
+    /// user likes
+    Lang(Arc<dyn CoreMeter>),
 }
 
 /// Control where logs go