feat: local prometheus sdk metrics

Author: andreas-karlsson

confidence-resolver/src/telemetry.rs

@@ -1,3 +1,4 @@
+use core::fmt;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
@@ -117,6 +118,93 @@ pub struct HistogramSnapshot {
     pub buckets: Vec<u64>,
 }
 
+impl TelemetrySnapshot {
+    /// Format the snapshot as Prometheus exposition text.
+    ///
+    /// All values are cumulative counters, matching what Prometheus expects.
+    /// The histogram uses cumulative `le` buckets derived from the exponential
+    /// bucket boundaries.
+    ///
+    /// `instance` is included as an `instance="..."` label on every metric,
+    /// allowing outputs from multiple WASM instances to be concatenated into
+    /// a single scrape endpoint.
+    pub fn to_prometheus(&self, instance: &str) -> String {
+        let mut out = String::new();
+        // fmt::Write for String is infallible; ignore the Ok.
+        let _ = self.write_prometheus(&mut out, instance);
+        out
+    }
+
+    fn write_prometheus(&self, w: &mut dyn fmt::Write, instance: &str) -> fmt::Result {
+        self.write_histogram(w, instance)?;
+        self.write_resolve_rates(w, instance)
+    }
+
+    fn write_histogram(&self, w: &mut dyn fmt::Write, instance: &str) -> fmt::Result {
+        if self.latency.count == 0 {
+            return Ok(());
+        }
+
+        writeln!(
+            w,
+            "# TYPE confidence_resolve_latency_microseconds histogram"
+        )?;
+
+        let mut cumulative: u64 = 0;
+        for (i, &count) in self.latency.buckets.iter().enumerate() {
+            cumulative = cumulative.wrapping_add(count);
+            if cumulative == 0 {
+                continue;
+            }
+            let upper = ((i as f64 + 1.0) * LN_RATIO).exp();
+            writeln!(
+                w,
+                "confidence_resolve_latency_microseconds_bucket{{instance=\"{instance}\",le=\"{upper:.6e}\"}} {cumulative}"
+            )?;
+        }
+        writeln!(
+            w,
+            "confidence_resolve_latency_microseconds_bucket{{instance=\"{instance}\",le=\"+Inf\"}} {}",
+            self.latency.count
+        )?;
+        writeln!(
+            w,
+            "confidence_resolve_latency_microseconds_sum{{instance=\"{instance}\"}} {}",
+            self.latency.sum
+        )?;
+        writeln!(
+            w,
+            "confidence_resolve_latency_microseconds_count{{instance=\"{instance}\"}} {}",
+            self.latency.count
+        )?;
+
+        Ok(())
+    }
+
+    fn write_resolve_rates(&self, w: &mut dyn fmt::Write, instance: &str) -> fmt::Result {
+        let has_any = self.resolve_rates.iter().any(|&c| c > 0);
+        if !has_any {
+            return Ok(());
+        }
+
+        writeln!(w, "# TYPE confidence_resolves_total counter")?;
+        for (i, &count) in self.resolve_rates.iter().enumerate() {
+            if count == 0 {
+                continue;
+            }
+            let label = ResolveReason::try_from(i as i32)
+                .map(|r| r.as_str_name())
+                .unwrap_or("UNKNOWN");
+            writeln!(
+                w,
+                "confidence_resolves_total{{instance=\"{instance}\",reason=\"{label}\"}} {count}"
+            )?;
+        }
+
+        Ok(())
+    }
+}
+
 /// Concurrent telemetry collector.
 ///
 /// All methods are safe to call from multiple threads without locking.
@@ -525,4 +613,84 @@ mod tests {
         assert!(delta.resolve_latency.is_none()); // no counter activity
         assert_eq!(delta.state_age.unwrap().last_state_update, 1000);
     }
+
+    #[test]
+    fn prometheus_empty_snapshot() {
+        let snap = TelemetrySnapshot::default();
+        assert_eq!(snap.to_prometheus("w0"), "");
+    }
+
+    #[test]
+    fn prometheus_histogram_and_counters() {
+        let tel = Telemetry::new();
+        tel.record_latency_us(100);
+        tel.record_latency_us(500);
+        tel.mark_resolve(ResolveReason::Match);
+        tel.mark_resolve(ResolveReason::Match);
+        tel.mark_resolve(ResolveReason::NoSegmentMatch);
+
+        let prom = tel.snapshot().to_prometheus("w0");
+
+        // Histogram header
+        assert!(prom.contains("# TYPE confidence_resolve_latency_microseconds histogram"));
+        // Must have +Inf bucket with instance label
+        assert!(prom.contains(
+            r#"confidence_resolve_latency_microseconds_bucket{instance="w0",le="+Inf"} 2"#
+        ));
+        // Sum and count with instance label
+        assert!(prom.contains(r#"confidence_resolve_latency_microseconds_sum{instance="w0"} 600"#));
+        assert!(prom.contains(r#"confidence_resolve_latency_microseconds_count{instance="w0"} 2"#));
+
+        // Counters with instance label
+        assert!(prom.contains("# TYPE confidence_resolves_total counter"));
+        assert!(prom.contains(
+            r#"confidence_resolves_total{instance="w0",reason="RESOLVE_REASON_MATCH"} 2"#
+        ));
+        assert!(prom.contains(
+            r#"confidence_resolves_total{instance="w0",reason="RESOLVE_REASON_NO_SEGMENT_MATCH"} 1"#
+        ));
+        // Zero-count reasons should be omitted
+        assert!(!prom.contains("FLAG_ARCHIVED"));
+    }
+
+    #[test]
+    fn prometheus_cumulative_buckets() {
+        let tel = Telemetry::new();
+        // Two observations in different buckets
+        tel.record_latency_us(10);
+        tel.record_latency_us(10_000);
+
+        let prom = tel.snapshot().to_prometheus("w0");
+
+        // Parse all le buckets and verify they are monotonically non-decreasing
+        let bucket_counts: Vec<u64> = prom
+            .lines()
+            .filter(|l| l.contains("le=\"") && !l.contains("+Inf"))
+            .map(|l| l.rsplit_once(' ').unwrap().1.parse::<u64>().unwrap())
+            .collect();
+
+        for pair in bucket_counts.windows(2) {
+            assert!(
+                pair[1] >= pair[0],
+                "Prometheus buckets must be cumulative, got {} then {}",
+                pair[0],
+                pair[1]
+            );
+        }
+    }
+
+    #[test]
+    fn prometheus_distinct_instances_concatenate() {
+        let tel = Telemetry::new();
+        tel.record_latency_us(100);
+        tel.mark_resolve(ResolveReason::Match);
+
+        let snap = tel.snapshot();
+        let mut combined = snap.to_prometheus("w0");
+        combined.push_str(&snap.to_prometheus("w1"));
+
+        // Both instances present
+        assert!(combined.contains(r#"instance="w0""#));
+        assert!(combined.contains(r#"instance="w1""#));
+    }
 }

openfeature-provider/go/confidence/internal/local_resolver/assets/confidence_resolver.wasm

@@ -20,6 +20,7 @@ type LocalResolver interface {
 	ResolveProcess(*wasm.ResolveProcessRequest) (*wasm.ResolveProcessResponse, error)
 	FlushAllLogs() error
 	FlushAssignLogs() error
+	PrometheusSnapshot() (string, error)
 	Close(context.Context) error
 }
 

openfeature-provider/go/confidence/internal/local_resolver/local_resolver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -91,6 +92,20 @@ func (s *PooledResolver) FlushAssignLogs() error {
 	})
 }
 
+// PrometheusSnapshot implements LocalResolver.
+func (s *PooledResolver) PrometheusSnapshot() (string, error) {
+	var b strings.Builder
+	err := s.maintenance(func(lr LocalResolver) error {
+		text, err := lr.PrometheusSnapshot()
+		if err != nil {
+			return err
+		}
+		b.WriteString(text)
+		return nil
+	})
+	return b.String(), err
+}
+
 func (s *PooledResolver) Close(ctx context.Context) error {
 	return s.maintenance(func(lr LocalResolver) error {
 		return lr.Close(ctx)

openfeature-provider/go/confidence/internal/local_resolver/pool.go

@@ -117,6 +117,13 @@ func (r *RecoveringResolver) FlushAssignLogs() (err error) {
 	return
 }
 
+func (r *RecoveringResolver) PrometheusSnapshot() (text string, err error) {
+	r.withRecover("PrometheusSnapshot", &err, func(lr LocalResolver) {
+		text, err = lr.PrometheusSnapshot()
+	})
+	return
+}
+
 func (r *RecoveringResolver) Close(ctx context.Context) error {
 	// For Close, if we panic, don't recreate during shutdown; just surface error.
 	defer func() {

openfeature-provider/go/confidence/internal/local_resolver/recover.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	_ "embed"
@@ -18,6 +20,9 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
+// instanceCounter is a package-level counter for auto-assigning unique instance IDs.
+var instanceCounter atomic.Int64
+
 // defaultWasmBytes contains the embedded WASM resolver module.
 // This file is automatically populated during the build process from wasm/confidence_resolver.wasm.
 // The WASM file is built from the Rust source in wasm/rust-guest/ and must be kept in sync.
@@ -32,9 +37,10 @@ type LogSink func(logs *resolverv1.WriteFlagLogsRequest)
 func NoOpLogSink(logs *resolverv1.WriteFlagLogsRequest) {}
 
 type WasmResolver struct {
-	instance api.Module
-	logSink  LogSink
-	mu       *sync.Mutex
+	instance   api.Module
+	logSink    LogSink
+	mu         *sync.Mutex
+	instanceID string
 }
 
 var _ LocalResolver = (*WasmResolver)(nil)
@@ -67,6 +73,18 @@ func (r *WasmResolver) FlushAssignLogs() error {
 	return err
 }
 
+func (r *WasmResolver) PrometheusSnapshot() (string, error) {
+	req := &wasm.PrometheusSnapshotRequest{
+		Instance: r.instanceID,
+	}
+	resp := &wasm.PrometheusSnapshotResponse{}
+	err := r.call("wasm_msg_guest_prometheus_snapshot", req, resp)
+	if err != nil {
+		return "", err
+	}
+	return resp.GetText(), nil
+}
+
 func (r *WasmResolver) Close(ctx context.Context) error {
 	// TODO we should call flush assigned until it doesn't flush any more
 	r.FlushAllLogs()
@@ -158,10 +176,12 @@ func (wrf *WasmResolverFactory) New() LocalResolver {
 	if err != nil {
 		panic(err)
 	}
+	id := instanceCounter.Add(1)
 	return &WasmResolver{
-		instance: instance,
-		logSink:  wrf.logSink,
-		mu:       &sync.Mutex{},
+		instance:   instance,
+		logSink:    wrf.logSink,
+		mu:         &sync.Mutex{},
+		instanceID: fmt.Sprintf("%d", id),
 	}
 }