up: support labeled HTTP metrics

Author: dio

docs/observability.md

@@ -74,6 +74,37 @@ func onResponse(w *up.Writer, chunk *up.ResponseChunk) {
 }
 ```
 
+For labeled metrics, define tag keys at config time and pass values in the
+same order when recording:
+
+```go
+var tokenUsage up.MetricID
+
+func init() {
+    up.RegisterWithConfig("my-filter",
+        func(h up.ConfigHandle) error {
+            var err error
+            tokenUsage, err = h.DefineHistogram("gen_ai.client.token.usage",
+                "gen_ai.operation.name",
+                "gen_ai.provider.name",
+                "gen_ai.token.type",
+            )
+            return err
+        },
+        onRequest, onResponse,
+    )
+}
+
+func onResponse(w *up.Writer, chunk *up.ResponseChunk) {
+    if chunk.EndStream {
+        w.RecordHistogramLabels(tokenUsage, 42, "chat", "openai", "input")
+    }
+}
+```
+
+The label values must match the metric's tag-key order and count. Keep labels
+low-cardinality; do not use request IDs, user IDs, raw paths, or API keys.
+
 **Counter** — monotonically increasing, never decreases, reset on Envoy restart.
 Good for total requests, errors, retries.
 

up/README.md

@@ -72,10 +72,12 @@ and individual headers via `r.Header(name)` or all headers via `r.AllHeaders()`.
 - `w.SetUpstreamOverrideHost(host)` — override the upstream host for this request
 - `w.AddSpanTag(key, value)` — annotate the active tracing span
 - `w.IncrementCounter(id, delta)` — increment an Envoy counter defined at config time
+- `w.IncrementCounterLabels(id, delta, labels...)` — increment a labeled Envoy counter
 - `w.IncrementGauge(id, delta)` — increment an Envoy gauge defined at config time
 - `w.DecrementGauge(id, delta)` — decrement an Envoy gauge defined at config time
 - `w.SetGauge(id, value)` — set an Envoy gauge to an absolute value
 - `w.RecordHistogram(id, value)` — record a histogram observation
+- `w.RecordHistogramLabels(id, value, labels...)` — record a labeled histogram observation
 - `w.GetBufferedBody()` — read the buffered request body (requires `RegisterWithMutableBody`)
 - `w.SetBufferedBody(body)` — replace the buffered request body
 - `w.HTTPCallout(req, callback)` — make an async HTTP callout to an Envoy cluster

up/async.go

@@ -168,8 +168,9 @@ type filterStateMutation struct {
 // Multiple increments to the same MetricID are applied in order; they are
 // not coalesced.
 type counterMutation struct {
-	id    MetricID
-	delta uint64
+	id     MetricID
+	delta  uint64
+	labels []string
 }
 
 // gaugeMutation is a deferred gauge operation (set, increment, or decrement).
@@ -189,8 +190,9 @@ const (
 
 // histogramMutation is a deferred RecordHistogram operation.
 type histogramMutation struct {
-	id    MetricID
-	value uint64
+	id     MetricID
+	value  uint64
+	labels []string
 }
 
 // upstreamOverrideMutation is a deferred SetUpstreamOverrideHost operation.

up/filter.go

@@ -312,7 +312,7 @@ func (f *filter) flush(continueReq bool) {
 	}
 	f.filterState = f.filterState[:0]
 	for _, m := range f.counters {
-		f.handle.IncrementCounterValue(shared.MetricID(m.id), m.delta)
+		f.handle.IncrementCounterValue(shared.MetricID(m.id), m.delta, m.labels...)
 	}
 	f.counters = f.counters[:0]
 	for _, m := range f.gauges {
@@ -327,7 +327,7 @@ func (f *filter) flush(continueReq bool) {
 	}
 	f.gauges = f.gauges[:0]
 	for _, m := range f.histograms {
-		f.handle.RecordHistogramValue(shared.MetricID(m.id), m.value)
+		f.handle.RecordHistogramValue(shared.MetricID(m.id), m.value, m.labels...)
 	}
 	f.histograms = f.histograms[:0]
 	for _, m := range f.dynamicMetadata {

up/metrics_test.go

@@ -0,0 +1,62 @@
+package up
+
+import (
+	"testing"
+
+	"github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/shared"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dio/transit/up/testutil"
+)
+
+func TestWriter_MetricLabelsDirectWrite(t *testing.T) {
+	h := testutil.NewFilterHandle()
+	w := NewWriter(h)
+
+	w.IncrementCounterLabels(11, 2, "chat", "openai")
+	w.RecordHistogramLabels(12, 42, "chat", "openai", "input")
+
+	require.Equal(t, []testutil.MetricRecord{
+		{ID: shared.MetricID(11), Value: 2, Labels: []string{"chat", "openai"}},
+	}, h.Counters)
+	require.Equal(t, []testutil.MetricRecord{
+		{ID: shared.MetricID(12), Value: 42, Labels: []string{"chat", "openai", "input"}},
+	}, h.Histograms)
+}
+
+func TestWriter_MetricLabelsQueuedFlush(t *testing.T) {
+	h := testutil.NewFilterHandle()
+	f := &filter{handle: h}
+	w := &Writer{f: f}
+
+	counterLabels := []string{"chat", "openai"}
+	histogramLabels := []string{"chat", "openai", "input"}
+	w.IncrementCounterLabels(21, 3, counterLabels...)
+	w.RecordHistogramLabels(22, 84, histogramLabels...)
+
+	counterLabels[0] = "mutated"
+	histogramLabels[0] = "mutated"
+	f.flush(false)
+
+	require.Equal(t, []testutil.MetricRecord{
+		{ID: shared.MetricID(21), Value: 3, Labels: []string{"chat", "openai"}},
+	}, h.Counters)
+	require.Equal(t, []testutil.MetricRecord{
+		{ID: shared.MetricID(22), Value: 84, Labels: []string{"chat", "openai", "input"}},
+	}, h.Histograms)
+}
+
+func TestWriter_MetricNoLabelMethodsRemainUnlabeled(t *testing.T) {
+	h := testutil.NewFilterHandle()
+	w := NewWriter(h)
+
+	w.IncrementCounter(31, 4)
+	w.RecordHistogram(32, 168)
+
+	require.Equal(t, []testutil.MetricRecord{
+		{ID: shared.MetricID(31), Value: 4},
+	}, h.Counters)
+	require.Equal(t, []testutil.MetricRecord{
+		{ID: shared.MetricID(32), Value: 168},
+	}, h.Histograms)
+}

up/testutil/testutil.go

@@ -20,6 +20,13 @@ type LocalResponse struct {
 	Detail  string
 }
 
+// MetricRecord records a metric mutation emitted by a fake filter handle.
+type MetricRecord struct {
+	ID     shared.MetricID
+	Value  uint64
+	Labels []string
+}
+
 // FilterHandleOption configures a FakeFilterHandle.
 type FilterHandleOption func(*FakeFilterHandle)
 
@@ -99,6 +106,8 @@ type FakeFilterHandle struct {
 
 	// LocalResponses records every SendLocalResponse call.
 	LocalResponses   []LocalResponse
+	Counters         []MetricRecord
+	Histograms       []MetricRecord
 	ContinuedReq     int
 	ContinueRequestC chan struct{}
 	LocalResponseC   chan struct{}
@@ -399,9 +408,12 @@ func (h *FakeFilterHandle) ResetHttpStream(_ uint64)
 func (h *FakeFilterHandle) SetDownstreamWatermarkCallbacks(_ shared.DownstreamWatermarkCallbacks) {}
 func (h *FakeFilterHandle) ClearDownstreamWatermarkCallbacks()                                    {}
 
-// -- Metrics (no-ops) --
+// -- Metrics --
 
-func (h *FakeFilterHandle) RecordHistogramValue(_ shared.MetricID, _ uint64, _ ...string) shared.MetricsResult {
+func (h *FakeFilterHandle) RecordHistogramValue(id shared.MetricID, value uint64, labels ...string) shared.MetricsResult {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.Histograms = append(h.Histograms, MetricRecord{ID: id, Value: value, Labels: append([]string(nil), labels...)})
 	return shared.MetricsSuccess
 }
 func (h *FakeFilterHandle) SetGaugeValue(_ shared.MetricID, _ uint64, _ ...string) shared.MetricsResult {
@@ -413,7 +425,10 @@ func (h *FakeFilterHandle) IncrementGaugeValue(_ shared.MetricID, _ uint64, _ ..
 func (h *FakeFilterHandle) DecrementGaugeValue(_ shared.MetricID, _ uint64, _ ...string) shared.MetricsResult {
 	return shared.MetricsSuccess
 }
-func (h *FakeFilterHandle) IncrementCounterValue(_ shared.MetricID, _ uint64, _ ...string) shared.MetricsResult {
+func (h *FakeFilterHandle) IncrementCounterValue(id shared.MetricID, value uint64, labels ...string) shared.MetricsResult {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.Counters = append(h.Counters, MetricRecord{ID: id, Value: value, Labels: append([]string(nil), labels...)})
 	return shared.MetricsSuccess
 }
 

up/writer.go

@@ -428,11 +428,17 @@ func (w *Writer) SetResponseBody(data []byte) {
 
 // IncrementCounter queues (or immediately applies) a counter increment.
 func (w *Writer) IncrementCounter(id MetricID, delta uint64) {
+	w.IncrementCounterLabels(id, delta)
+}
+
+// IncrementCounterLabels queues (or immediately applies) a counter increment
+// with label values matching the tag keys used when the counter was defined.
+func (w *Writer) IncrementCounterLabels(id MetricID, delta uint64, labelValues ...string) {
 	if w.queued() {
-		w.f.counters = append(w.f.counters, counterMutation{id: id, delta: delta})
+		w.f.counters = append(w.f.counters, counterMutation{id: id, delta: delta, labels: cloneLabels(labelValues)})
 		return
 	}
-	w.f.handle.IncrementCounterValue(shared.MetricID(id), delta)
+	w.f.handle.IncrementCounterValue(shared.MetricID(id), delta, labelValues...)
 }
 
 // IncrementGauge queues (or immediately applies) a gauge increment.
@@ -464,11 +470,25 @@ func (w *Writer) SetGauge(id MetricID, value uint64) {
 
 // RecordHistogram queues (or immediately applies) a histogram observation.
 func (w *Writer) RecordHistogram(id MetricID, value uint64) {
+	w.RecordHistogramLabels(id, value)
+}
+
+// RecordHistogramLabels queues (or immediately applies) a histogram
+// observation with label values matching the tag keys used when the histogram
+// was defined.
+func (w *Writer) RecordHistogramLabels(id MetricID, value uint64, labelValues ...string) {
 	if w.queued() {
-		w.f.histograms = append(w.f.histograms, histogramMutation{id: id, value: value})
+		w.f.histograms = append(w.f.histograms, histogramMutation{id: id, value: value, labels: cloneLabels(labelValues)})
 		return
 	}
-	w.f.handle.RecordHistogramValue(shared.MetricID(id), value)
+	w.f.handle.RecordHistogramValue(shared.MetricID(id), value, labelValues...)
+}
+
+func cloneLabels(labels []string) []string {
+	if len(labels) == 0 {
+		return nil
+	}
+	return append([]string(nil), labels...)
 }
 
 // Go upgrades this request to asynchronous mode. fn runs in a new goroutine