feat(metrics): add UpDownCounter metric instrument type

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<dyn CoreMeter> {
     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<GaugeF64> 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<dyn MetricAttributable<Box<dyn UpDownCounterBase>> + Send + Sync>,
+    Arc<dyn UpDownCounterBase>,
+>;
+
+/// 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<dyn MetricAttributable<Box<dyn UpDownCounterBase>> + 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<dyn UpDownCounterBase>
+                })
+        });
+        bound.adds(value);
+    }
+}
+impl MetricAttributable<UpDownCounter> for UpDownCounter {
+    fn with_attributes(
+        &self,
+        attributes: &MetricAttributes,
+    ) -> Result<UpDownCounter, Box<dyn std::error::Error>> {
+        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,32 @@ mod otel {
             }
         }
     }
+
+    impl MetricAttributable<Box<dyn UpDownCounterBase>>
+        for opentelemetry::metrics::UpDownCounter<i64>
+    {
+        fn with_attributes(
+            &self,
+            attributes: &MetricAttributes,
+        ) -> Result<Box<dyn UpDownCounterBase>, Box<dyn std::error::Error>> {
+            Ok(Box::new(InstrumentWithAttributes {
+                inner: self.clone(),
+                attributes: attributes.clone(),
+            }))
+        }
+    }
+
+    impl UpDownCounterBase
+        for InstrumentWithAttributes<opentelemetry::metrics::UpDownCounter<i64>>
+    {
+        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 +1448,9 @@ impl<CM: CoreMeter> CoreMeter for PrefixedMetricsMeter<CM> {
         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)
+    }
 }

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 (e.g. inflight counts).
+    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.

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 {

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<Box<dyn GaugeBase>> for PromMetric<IntGaugeVec> {
     }
 }
 
+impl UpDownCounterBase for CorePromIntGauge {
+    fn adds(&self, value: i64) {
+        self.0.add(value);
+    }
+}
+impl MetricAttributable<Box<dyn UpDownCounterBase>> for PromMetric<IntGaugeVec> {
+    fn with_attributes(
+        &self,
+        attributes: &MetricAttributes,
+    ) -> Result<Box<dyn UpDownCounterBase>, Box<dyn std::error::Error>> {
+        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::<IntGaugeVec>::new(
+            metric_name,
+            params.description.to_string(),
+            self.registry.clone(),
+        )))
+    }
 }
 
 impl CorePrometheusMeter {

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);

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<Box<dyn metrics::UpDownCounterBase>> for CustomMetric {
+    fn with_attributes(
+        &self,
+        atts: &metrics::MetricAttributes,
+    ) -> Result<Box<dyn metrics::UpDownCounterBase>, Box<dyn Error>> {
+        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()