diff --git a/crates/common/src/telemetry/metrics.rs b/crates/common/src/telemetry/metrics.rs index 2865bb5bd..89351146d 100644 --- a/crates/common/src/telemetry/metrics.rs +++ b/crates/common/src/telemetry/metrics.rs @@ -169,6 +169,9 @@ pub trait CoreMeter: Send + Sync + Debug { /// Create a new gauge instrument recording `f64` values. fn gauge_f64(&self, params: MetricParameters) -> GaugeF64; + + /// Create a new up-down counter instrument recording `i64` values. + fn up_down_counter(&self, params: MetricParameters) -> UpDownCounter; } /// Provides a generic way to record metrics in memory. @@ -351,6 +354,10 @@ impl CoreMeter for Arc { fn gauge_f64(&self, params: MetricParameters) -> GaugeF64 { self.as_ref().gauge_f64(params) } + + fn up_down_counter(&self, params: MetricParameters) -> UpDownCounter { + self.as_ref().up_down_counter(params) + } } /// Attributes which are provided every time a call to record a specific metric is made. @@ -979,6 +986,81 @@ impl MetricAttributable for GaugeF64 { } } +/// Base trait for up-down counter implementations. +pub trait UpDownCounterBase: Send + Sync { + /// Add a signed value to the up-down counter. + fn adds(&self, value: i64); +} + +/// The lazy-bound inner type for [`UpDownCounter`]. +pub type UpDownCounterImpl = LazyBoundMetric< + Arc> + Send + Sync>, + Arc, +>; + +/// An up-down counter metric instrument that supports both positive and negative deltas. +#[derive(Clone)] +pub struct UpDownCounter { + primary: UpDownCounterImpl, +} +impl UpDownCounter { + /// Create a new up-down counter from an attributable metric source. + pub fn new( + inner: Arc> + Send + Sync>, + ) -> Self { + Self { + primary: LazyBoundMetric { + metric: inner, + attributes: MetricAttributes::Empty, + bound_cache: OnceLock::new(), + }, + } + } + + /// Add a signed `i64` value with the given attributes. + pub fn add(&self, value: i64, attributes: &MetricAttributes) { + match self.primary.metric.with_attributes(attributes) { + Ok(base) => base.adds(value), + Err(e) => { + dbg_panic!("Failed to initialize primary metric, will drop values: {e:?}"); + } + } + } + + /// Replace the attributes on the primary metric. + pub fn update_attributes(&mut self, new_attributes: MetricAttributes) { + self.primary.update_attributes(new_attributes.clone()); + } +} +impl UpDownCounterBase for UpDownCounter { + fn adds(&self, value: i64) { + let bound = self.primary.bound_cache.get_or_init(|| { + self.primary + .metric + .with_attributes(&self.primary.attributes) + .map(Into::into) + .unwrap_or_else(|e| { + dbg_panic!("Failed to initialize primary metric, will drop values: {e:?}"); + Arc::new(NoOpInstrument) as Arc + }) + }); + bound.adds(value); + } +} +impl MetricAttributable for UpDownCounter { + fn with_attributes( + &self, + attributes: &MetricAttributes, + ) -> Result> { + let primary = LazyBoundMetric { + metric: self.primary.metric.clone(), + attributes: attributes.clone(), + bound_cache: OnceLock::new(), + }; + Ok(UpDownCounter { primary }) + } +} + /// A [`CoreMeter`] implementation that discards all metric recordings. #[derive(Debug)] pub struct NoOpCoreMeter; @@ -1025,6 +1107,10 @@ impl CoreMeter for NoOpCoreMeter { fn gauge_f64(&self, _: MetricParameters) -> GaugeF64 { GaugeF64::new(Arc::new(NoOpInstrument)) } + + fn up_down_counter(&self, _: MetricParameters) -> UpDownCounter { + UpDownCounter::new(Arc::new(NoOpInstrument)) + } } macro_rules! impl_metric_attributable { @@ -1043,6 +1129,12 @@ macro_rules! impl_metric_attributable { /// A no-op metric instrument that discards all recordings. pub struct NoOpInstrument; macro_rules! impl_no_op { + ($base_trait:ident, signed) => { + impl_metric_attributable!($base_trait, NoOpInstrument, NoOpInstrument); + impl $base_trait for NoOpInstrument { + fn adds(&self, _: i64) {} + } + }; ($base_trait:ident, $value_type:ty) => { impl_metric_attributable!($base_trait, NoOpInstrument, NoOpInstrument); impl $base_trait for NoOpInstrument { @@ -1062,6 +1154,7 @@ impl_no_op!(HistogramF64Base, f64); impl_no_op!(HistogramDurationBase, Duration); impl_no_op!(GaugeBase, u64); impl_no_op!(GaugeF64Base, f64); +impl_no_op!(UpDownCounterBase, signed); #[cfg(test)] mod tests { @@ -1238,6 +1331,28 @@ mod otel { } } } + + impl MetricAttributable> for opentelemetry::metrics::UpDownCounter { + fn with_attributes( + &self, + attributes: &MetricAttributes, + ) -> Result, Box> { + Ok(Box::new(InstrumentWithAttributes { + inner: self.clone(), + attributes: attributes.clone(), + })) + } + } + + impl UpDownCounterBase for InstrumentWithAttributes> { + fn adds(&self, value: i64) { + if let MetricAttributes::OTel { kvs } = &self.attributes { + self.inner.add(value, kvs); + } else { + dbg_panic!("Must use OTel attributes with an OTel metric implementation"); + } + } + } } /// Maintains a mapping of metric labels->values with a defined ordering, used for Prometheus labels @@ -1329,4 +1444,9 @@ impl CoreMeter for PrefixedMetricsMeter { params.name = (self.prefix.clone() + &*params.name).into(); self.meter.gauge_f64(params) } + + fn up_down_counter(&self, mut params: MetricParameters) -> UpDownCounter { + params.name = (self.prefix.clone() + &*params.name).into(); + self.meter.up_down_counter(params) + } } diff --git a/crates/common/src/telemetry/metrics/core.rs b/crates/common/src/telemetry/metrics/core.rs index b2054ce5b..1719a8e5b 100644 --- a/crates/common/src/telemetry/metrics/core.rs +++ b/crates/common/src/telemetry/metrics/core.rs @@ -63,6 +63,8 @@ pub enum MetricKind { HistogramF64, /// A histogram that records [`Duration`] values. HistogramDuration, + /// An `i64` counter that can go up or down. + UpDownCounter, } /// The value to record when updating a metric instrument. #[derive(Debug, Clone, Copy)] @@ -77,6 +79,8 @@ pub enum MetricUpdateVal { ValueF64(f64), /// A duration observation (for duration histograms). Duration(Duration), + /// A signed `i64` delta (for up-down counters). + SignedDelta(i64), } /// Collects buffered metric events for later replay into a real metrics backend. diff --git a/crates/common/src/telemetry/otel.rs b/crates/common/src/telemetry/otel.rs index 4f31ee8d9..0668ee37f 100644 --- a/crates/common/src/telemetry/otel.rs +++ b/crates/common/src/telemetry/otel.rs @@ -5,7 +5,7 @@ use super::{ ACTIVITY_EXEC_LATENCY_HISTOGRAM_NAME, ACTIVITY_SCHED_TO_START_LATENCY_HISTOGRAM_NAME, CoreMeter, Counter, DEFAULT_MS_BUCKETS, DEFAULT_S_BUCKETS, Gauge, GaugeF64, Histogram, HistogramBase, HistogramDuration, HistogramDurationBase, HistogramF64, HistogramF64Base, - MetricAttributable, MetricAttributes, MetricParameters, NewAttributes, + MetricAttributable, MetricAttributes, MetricParameters, NewAttributes, UpDownCounter, WORKFLOW_E2E_LATENCY_HISTOGRAM_NAME, WORKFLOW_TASK_EXECUTION_LATENCY_HISTOGRAM_NAME, WORKFLOW_TASK_REPLAY_LATENCY_HISTOGRAM_NAME, WORKFLOW_TASK_SCHED_TO_START_LATENCY_HISTOGRAM_NAME, default_buckets_for, @@ -235,6 +235,16 @@ impl CoreMeter for CoreOtelMeter { .build(), )) } + + fn up_down_counter(&self, params: MetricParameters) -> UpDownCounter { + UpDownCounter::new(Arc::new( + self.meter + .i64_up_down_counter(params.name) + .with_unit(params.unit) + .with_description(params.description) + .build(), + )) + } } impl CoreOtelMeter { diff --git a/crates/common/src/telemetry/prometheus_meter.rs b/crates/common/src/telemetry/prometheus_meter.rs index 4e487ec6c..82f8a7725 100644 --- a/crates/common/src/telemetry/prometheus_meter.rs +++ b/crates/common/src/telemetry/prometheus_meter.rs @@ -4,7 +4,7 @@ use crate::{ CoreMeter, Counter, CounterBase, Gauge, GaugeBase, GaugeF64, GaugeF64Base, Histogram, HistogramBase, HistogramDuration, HistogramDurationBase, HistogramF64, HistogramF64Base, MetricAttributable, MetricAttributes, MetricParameters, NewAttributes, OrderedPromLabelSet, - default_buckets_for, + UpDownCounter, UpDownCounterBase, default_buckets_for, }, }; use anyhow::anyhow; @@ -385,6 +385,29 @@ impl MetricAttributable> for PromMetric { } } +impl UpDownCounterBase for CorePromIntGauge { + fn adds(&self, value: i64) { + self.0.add(value); + } +} +impl MetricAttributable> for PromMetric { + fn with_attributes( + &self, + attributes: &MetricAttributes, + ) -> Result, Box> { + let labels = self.extract_prometheus_labels(attributes)?; + let vector = self.get_or_create_vector(labels, |name, desc, label_names| { + let opts = Opts::new(name, desc); + IntGaugeVec::new(opts, label_names).map_err(Into::into) + })?; + if let Ok(g) = vector.get_metric_with(&labels.as_prom_labels()) { + Ok(Box::new(CorePromIntGauge(g))) + } else { + Err(self.label_mismatch_err(attributes).into()) + } + } +} + struct CorePromGauge(prometheus::Gauge); impl GaugeF64Base for CorePromGauge { fn records(&self, value: f64) { @@ -580,6 +603,15 @@ impl CoreMeter for CorePrometheusMeter { self.registry.clone(), ))) } + + fn up_down_counter(&self, params: MetricParameters) -> UpDownCounter { + let metric_name = params.name.to_string(); + UpDownCounter::new(Arc::new(PromMetric::::new( + metric_name, + params.description.to_string(), + self.registry.clone(), + ))) + } } impl CorePrometheusMeter { diff --git a/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h b/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h index ed03dd47a..11178167f 100644 --- a/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h +++ b/crates/sdk-core-c-bridge/include/temporal-sdk-core-c-bridge.h @@ -27,6 +27,7 @@ typedef enum TemporalCoreMetricKind { HistogramDuration, GaugeInteger, GaugeFloat, + UpDownCounterInteger, } TemporalCoreMetricKind; typedef enum TemporalCoreForwardedLogLevel { @@ -937,6 +938,10 @@ void temporal_core_metric_record_float(const struct TemporalCoreMetric *metric, double value, const struct TemporalCoreMetricAttributes *attrs); +void temporal_core_metric_record_integer_signed(const struct TemporalCoreMetric *metric, + int64_t value, + const struct TemporalCoreMetricAttributes *attrs); + void temporal_core_metric_record_duration(const struct TemporalCoreMetric *metric, uint64_t value_ms, const struct TemporalCoreMetricAttributes *attrs); diff --git a/crates/sdk-core-c-bridge/src/metric.rs b/crates/sdk-core-c-bridge/src/metric.rs index b32cdf2a0..3ba6ebea9 100644 --- a/crates/sdk-core-c-bridge/src/metric.rs +++ b/crates/sdk-core-c-bridge/src/metric.rs @@ -136,6 +136,7 @@ pub enum MetricKind { HistogramDuration, GaugeInteger, GaugeFloat, + UpDownCounterInteger, } pub enum Metric { @@ -145,6 +146,7 @@ pub enum Metric { HistogramDuration(metrics::HistogramDuration), GaugeInteger(metrics::Gauge), GaugeFloat(metrics::GaugeF64), + UpDownCounterInteger(metrics::UpDownCounter), } #[unsafe(no_mangle)] @@ -167,6 +169,9 @@ pub extern "C" fn temporal_core_metric_new( } MetricKind::GaugeInteger => Metric::GaugeInteger(meter.core.gauge(options.into())), MetricKind::GaugeFloat => Metric::GaugeFloat(meter.core.gauge_f64(options.into())), + MetricKind::UpDownCounterInteger => { + Metric::UpDownCounterInteger(meter.core.up_down_counter(options.into())) + } })) } @@ -208,6 +213,20 @@ pub extern "C" fn temporal_core_metric_record_float( } } +#[unsafe(no_mangle)] +pub extern "C" fn temporal_core_metric_record_integer_signed( + metric: *const Metric, + value: i64, + attrs: *const MetricAttributes, +) { + let metric = unsafe { &*metric }; + let attrs = unsafe { &*attrs }; + match metric { + Metric::UpDownCounterInteger(counter) => counter.add(value, &attrs.core), + _ => panic!("Not a signed integer type"), + } +} + #[unsafe(no_mangle)] pub extern "C" fn temporal_core_metric_record_duration( metric: *const Metric, @@ -359,6 +378,12 @@ impl metrics::CoreMeter for CustomMetricMeterRef { fn gauge_f64(&self, params: metrics::MetricParameters) -> metrics::GaugeF64 { metrics::GaugeF64::new(Arc::new(self.new_metric(params, MetricKind::GaugeFloat))) } + + fn up_down_counter(&self, params: metrics::MetricParameters) -> metrics::UpDownCounter { + metrics::UpDownCounter::new(Arc::new( + self.new_metric(params, MetricKind::UpDownCounterInteger), + )) + } } impl CustomMetricMeterRef { @@ -652,6 +677,29 @@ impl metrics::GaugeF64Base for CustomMetric { } } +impl metrics::MetricAttributable> for CustomMetric { + fn with_attributes( + &self, + atts: &metrics::MetricAttributes, + ) -> Result, Box> { + Ok(Box::new(CustomMetric { + meter_impl: self.meter_impl.clone(), + metric: self.metric.clone(), + bound_attributes: Some(atts.clone()), + })) + } +} + +impl metrics::UpDownCounterBase for CustomMetric { + fn adds(&self, value: i64) { + unsafe { + let meter = &*(self.meter_impl.0); + let attr_ptr = self.attr_ptr(); + (meter.metric_record_integer)(*self.metric, value as u64, attr_ptr); + } + } +} + fn raw_custom_metric_attributes(attributes: &metrics::MetricAttributes) -> *const libc::c_void { if let metrics::MetricAttributes::Dynamic(v) = attributes { v.clone() diff --git a/crates/sdk-core/src/telemetry/metrics.rs b/crates/sdk-core/src/telemetry/metrics.rs index 0f80702f9..98edbfed1 100644 --- a/crates/sdk-core/src/telemetry/metrics.rs +++ b/crates/sdk-core/src/telemetry/metrics.rs @@ -900,6 +900,12 @@ where fn gauge_f64(&self, params: MetricParameters) -> GaugeF64 { GaugeF64::new(Arc::new(self.new_instrument(params, MetricKind::GaugeF64))) } + + fn up_down_counter(&self, params: MetricParameters) -> UpDownCounter { + UpDownCounter::new(Arc::new( + self.new_instrument(params, MetricKind::UpDownCounter), + )) + } } impl MetricCallBufferer for MetricsCallBuffer where @@ -1030,6 +1036,22 @@ where })) } } + +impl MetricAttributable> for BufferInstrument +where + I: BufferInstrumentRef + Send + Sync + Clone + 'static, +{ + fn with_attributes( + &self, + attributes: &MetricAttributes, + ) -> Result, Box> { + Ok(Box::new(InstrumentWithAttributes { + inner: self.clone(), + attributes: attributes.clone(), + })) + } +} + impl CounterBase for InstrumentWithAttributes> where I: BufferInstrumentRef + Send + Sync + Clone + 'static, @@ -1085,6 +1107,16 @@ where } } +impl UpDownCounterBase for InstrumentWithAttributes> +where + I: BufferInstrumentRef + Send + Sync + Clone + 'static, +{ + fn adds(&self, value: i64) { + self.inner + .send(MetricUpdateVal::SignedDelta(value), &self.attributes) + } +} + #[cfg(test)] mod tests { use super::*; @@ -1203,7 +1235,7 @@ mod tests { #[test] fn metric_buffer() { - let call_buffer = MetricsCallBuffer::new(10); + let call_buffer = MetricsCallBuffer::new(12); let ctr = call_buffer.counter(MetricParameters { name: "ctr".into(), description: "a counter".into(), @@ -1224,6 +1256,11 @@ mod tests { description: "a duration histogram".into(), unit: "seconds".into(), }); + let up_down_ctr = call_buffer.up_down_counter(MetricParameters { + name: "up_down_ctr".into(), + description: "an up down counter".into(), + unit: "widgets".into(), + }); let attrs_1 = call_buffer.new_attributes(NewAttributes { attributes: vec![MetricKeyValue::new("hi", "yo")], }); @@ -1234,6 +1271,7 @@ mod tests { histo.record(2, &attrs_1); gauge.record(3, &attrs_2); histo_dur.record(Duration::from_secs_f64(1.2), &attrs_1); + up_down_ctr.add(-3, &attrs_2); let mut calls = call_buffer.retrieve(); calls.reverse(); @@ -1281,6 +1319,17 @@ mod tests { => populate_into ); hist_4.set(Arc::new(DummyInstrumentRef(4))).unwrap(); + let up_down_5 = assert_matches!( + calls.pop(), + Some(MetricEvent::Create { + params, + populate_into, + kind: MetricKind::UpDownCounter + }) + if params.name == "up_down_ctr" + => populate_into + ); + up_down_5.set(Arc::new(DummyInstrumentRef(5))).unwrap(); let a1 = assert_matches!( calls.pop(), Some(MetricEvent::CreateAttributes { @@ -1340,5 +1389,14 @@ mod tests { if DummyCustomAttrs::as_id(&attributes) == 1 && instrument.get().0 == 4 && d == Duration::from_secs_f64(1.2) ); + assert_matches!( + calls.pop(), + Some(MetricEvent::Update{ + instrument, + attributes, + update: MetricUpdateVal::SignedDelta(-3) + }) + if DummyCustomAttrs::as_id(&attributes) == 2 && instrument.get().0 == 5 + ); } } diff --git a/crates/sdk-core/tests/integ_tests/metrics_tests.rs b/crates/sdk-core/tests/integ_tests/metrics_tests.rs index 0f3115b79..53cd146bf 100644 --- a/crates/sdk-core/tests/integ_tests/metrics_tests.rs +++ b/crates/sdk-core/tests/integ_tests/metrics_tests.rs @@ -56,7 +56,8 @@ use temporalio_common::{ TaskQueueLabelStrategy, TelemetryOptions, build_otlp_metric_exporter, metrics::{ CoreMeter, CounterBase, Gauge, GaugeBase, HistogramBase, MetricKeyValue, - MetricParameters, NewAttributes, WORKFLOW_TASK_EXECUTION_LATENCY_HISTOGRAM_NAME, + MetricParameters, NewAttributes, UpDownCounterBase, + WORKFLOW_TASK_EXECUTION_LATENCY_HISTOGRAM_NAME, }, start_prometheus_metric_exporter, }, @@ -1300,10 +1301,17 @@ async fn test_prometheus_endpoint_integration() { description: "Number of active test connections".into(), unit: "".into(), }); + let up_down_counter = meter.up_down_counter(MetricParameters { + name: "test_inflight_operations".into(), + description: "Number of inflight test operations".into(), + unit: "".into(), + }); counter.adds(5); histogram.records(100); gauge.records(10); + up_down_counter.adds(5); + up_down_counter.adds(-2); let url = format!("http://{addr}/metrics"); let response = tokio::time::timeout(Duration::from_secs(10), reqwest::get(&url)) @@ -1330,6 +1338,8 @@ async fn test_prometheus_endpoint_integration() { assert!(body.contains("test_active_connections 10"),); assert!(body.contains("test_request_duration_ms_count 1"),); assert!(body.contains("test_request_duration_ms_sum 100"),); + assert!(body.contains("test_inflight_operations"),); + assert!(body.contains("test_inflight_operations 3"),); } #[tokio::test]