refactor: drop the forked HTTP middleware for the shared one (#904)

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/klauspost/cpuid/v2 v2.4.0
 	github.com/miekg/dns v1.1.73
-	github.com/mulgadc/bluebottle v1.18.1-0.20260826044308-1a1e1242c217
+	github.com/mulgadc/bluebottle v1.18.1-0.20260826232854-f91084812667
 	github.com/mulgadc/northstar v1.18.0
 	github.com/mulgadc/predastore v1.18.1-0.20260826045109-26ce1b091112
 	github.com/mulgadc/viperblock v1.18.0

go.sum

@@ -1331,6 +1331,8 @@ github.com/mulgadc/bluebottle v1.18.0 h1:aNrogJBz59gJAxCwpERGBC5bGXkSoUCCCBbNvna
 github.com/mulgadc/bluebottle v1.18.0/go.mod h1:KkUyzqfKxyI8PvKR27EVqJnVm4BSbajl1Q/u+PQxAYw=
 github.com/mulgadc/bluebottle v1.18.1-0.20260826044308-1a1e1242c217 h1:zRKEFadLkUQxNCK2L3MYSQxRmu4VEEHWTxMwh04Qmeo=
 github.com/mulgadc/bluebottle v1.18.1-0.20260826044308-1a1e1242c217/go.mod h1:KkUyzqfKxyI8PvKR27EVqJnVm4BSbajl1Q/u+PQxAYw=
+github.com/mulgadc/bluebottle v1.18.1-0.20260826232854-f91084812667 h1:3q1aWTrGgQBSicNQ68uM/wYFkmaNld76wCr/VsnLC4s=
+github.com/mulgadc/bluebottle v1.18.1-0.20260826232854-f91084812667/go.mod h1:KkUyzqfKxyI8PvKR27EVqJnVm4BSbajl1Q/u+PQxAYw=
 github.com/mulgadc/northstar v1.18.0 h1:ksRjIKE6qtrw/ZoYaJh8JRBh+xdnuDw8FsmFLFCiXbg=
 github.com/mulgadc/northstar v1.18.0/go.mod h1:z6iol77R1vsIQyIBc1JQboUPhSJ6eG4bipzfbZZbLEM=
 github.com/mulgadc/predastore v1.18.0 h1:RW1Qw1SM2iYxqi6HaHLv5Qb1h1vhZSO3x6pOhPAb9mw=

spinifex/otelsetup/http.go

@@ -1,104 +1,38 @@
-// Package otelsetup carries spinifex's HTTP request instrumentation: a tracing
-// middleware and the request counter/duration histogram it feeds. The OTel
-// bootstrap itself lives in bluebottle/pkg/otelsetup, which service entrypoints
-// call directly.
+// Package otelsetup carries spinifex's request instrumentation: the request
+// counter/duration histogram, and a thin binding of the shared HTTP middleware
+// to those instruments. The middleware body and the OTel bootstrap itself live
+// in bluebottle/pkg/otelsetup, which service entrypoints call directly.
 package otelsetup
 
 import (
 	"context"
-	"fmt"
 	"net/http"
-	"time"
 
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/codes"
-	"go.opentelemetry.io/otel/propagation"
-	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
-	"go.opentelemetry.io/otel/trace"
+	bbotel "github.com/mulgadc/bluebottle/pkg/otelsetup"
 )
 
-const httpTracerName = "github.com/mulgadc/spinifex/spinifex/otelsetup"
+// OutcomeForStatus classifies a response for the outcome dimension. Re-exported
+// so the NATS path, which classifies an awserrors HTTP status, provably shares
+// one implementation with the HTTP path.
+func OutcomeForStatus(status int) string { return bbotel.OutcomeForStatus(status) }
 
-// statusRecorder captures the response status for span attributes.
-type statusRecorder struct {
-	http.ResponseWriter
+// SetRequestAction sets the logical action recorded on request metrics for the
+// in-flight request. No-op when the request did not pass through
+// HTTPMiddleware.
+func SetRequestAction(ctx context.Context, action string) { bbotel.SetRequestAction(ctx, action) }
 
-	status int
-}
-
-func (w *statusRecorder) WriteHeader(code int) {
-	w.status = code
-	w.ResponseWriter.WriteHeader(code)
-}
-
-// Unwrap returns the wrapped ResponseWriter, letting http.ResponseController
-// (and any other Unwrap-aware caller) walk past statusRecorder to reach the
-// real flusher underneath.
-func (w *statusRecorder) Unwrap() http.ResponseWriter { return w.ResponseWriter }
-
-// Flush implements http.Flusher by delegating to the wrapped ResponseWriter.
-// Without this, statusRecorder blocks every downstream Flush call (including
-// chi's WrapResponseWriter, which wraps statusRecorder) from ever reaching
-// the real writer: the call succeeds as a no-op but no bytes reach the
-// socket, silently breaking streaming handlers.
-func (w *statusRecorder) Flush() {
-	_ = http.NewResponseController(w.ResponseWriter).Flush()
-}
-
-// OutcomeForStatus classifies a response for the outcome dimension. 4xx is
-// client_error rather than error so a service's error rate stays server-fault
-// only, while client rejections (an IMDS 401, an unresolvable ENI's 404) stop
-// being counted as successes. Exported because the NATS path classifies against
-// the same statuses and the two must agree.
-func OutcomeForStatus(status int) string {
-	switch {
-	case status >= http.StatusInternalServerError:
-		return "error"
-	case status >= http.StatusBadRequest:
-		return "client_error"
-	default:
-		return "success"
-	}
-}
-
-// HTTPMiddleware opens a server span per request, honoring an inbound W3C
-// traceparent header, and records request count/duration metrics. Handlers
-// rename the span (and SetRequestAction) once they resolve a logical
-// operation (e.g. the SigV4 action). No-op unless Init configured export.
+// HTTPMiddleware binds the shared server middleware to spinifex's conventions:
+// health probes record metrics without rooting a trace, and requests land on
+// the same instruments the NATS paths write to rather than bluebottle's.
 func HTTPMiddleware(serverName string) func(http.Handler) http.Handler {
-	return func(next http.Handler) http.Handler {
-		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			// Health probes fire every few seconds per service and would root a
-			// trace each — record metrics only.
-			if r.URL.Path == "/health" {
-				start := time.Now()
-				rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
-				next.ServeHTTP(rec, r)
-				RecordRequest(r.Context(), r.Method+" /health", OutcomeForStatus(rec.status), time.Since(start))
-				return
-			}
-			ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header))
-			action := &requestAction{name: r.Method}
-			ctx = context.WithValue(ctx, requestActionKey{}, action)
-			ctx, span := otel.Tracer(httpTracerName).Start(ctx, r.Method+" "+r.URL.Path,
-				trace.WithSpanKind(trace.SpanKindServer),
-				trace.WithAttributes(
-					semconv.HTTPRequestMethodKey.String(r.Method),
-					semconv.URLPath(r.URL.Path),
-					attribute.String("server.name", serverName),
-				))
-			defer span.End()
-
-			start := time.Now()
-			rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
-			next.ServeHTTP(rec, r.WithContext(ctx))
-
-			span.SetAttributes(semconv.HTTPResponseStatusCode(rec.status))
-			if rec.status >= http.StatusInternalServerError {
-				span.SetStatus(codes.Error, fmt.Sprintf("HTTP %d", rec.status))
-			}
-			RecordRequest(ctx, action.name, OutcomeForStatus(rec.status), time.Since(start))
-		})
-	}
+	return bbotel.HTTPMiddleware(serverName,
+		bbotel.WithUntracedPaths("/health"),
+		bbotel.WithRecorder(recordSharedRequest))
+}
+
+// recordSharedRequest funnels the shared middleware's metric onto spinifex's
+// counter. Status code and byte counts are dropped: they would add series the
+// spinifex dashboards do not read.
+func recordSharedRequest(ctx context.Context, m bbotel.RequestMetric) {
+	RecordRequest(ctx, m.Action, m.Outcome, m.Elapsed)
 }

spinifex/otelsetup/http_test.go

@@ -1,13 +1,15 @@
 package otelsetup
 
 import (
+	"context"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
-	bbotel "github.com/mulgadc/bluebottle/pkg/otelsetup"
 	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 	"go.opentelemetry.io/otel/trace"
@@ -24,152 +26,91 @@ func withRecorder(t *testing.T) *tracetest.SpanRecorder {
 	return sr
 }
 
-// flusherSpy records whether Flush was called, satisfying http.Flusher so a
-// statusRecorder wrapping it can prove the flush chain reaches it.
-type flusherSpy struct {
-	http.ResponseWriter
+// The global meter binds its instruments to the first real provider installed
+// in the process and ignores later ones, so the reader is installed once and
+// each test filters by its own action name.
+var metricReader = func() *sdkmetric.ManualReader {
+	r := sdkmetric.NewManualReader()
+	otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(r)))
+	return r
+}()
 
-	flushed bool
-}
+// requestPointsFor returns the outcome of every mulga.requests point recorded
+// under action's rpc.method attribute. Points are cumulative for the life of
+// the test binary, so actions must be unique per test.
+func requestPointsFor(t *testing.T, action string) []string {
+	t.Helper()
 
-func (f *flusherSpy) Flush() { f.flushed = true }
-
-// TestStatusRecorderFlushReachesUnderlyingFlusher guards the §0 fix: before
-// Unwrap()/Flush() were added, http.NewResponseController(rec).Flush() would
-// fail with ErrNotSupported (or, via a Flusher-only outer wrapper such as
-// chi's WrapResponseWriter, silently no-op) because statusRecorder itself was
-// not a Flusher and exposed no way to reach the one it wraps.
-func TestStatusRecorderFlushReachesUnderlyingFlusher(t *testing.T) {
-	spy := &flusherSpy{ResponseWriter: httptest.NewRecorder()}
-	rec := &statusRecorder{ResponseWriter: spy, status: http.StatusOK}
-
-	if err := http.NewResponseController(rec).Flush(); err != nil {
-		t.Fatalf("Flush: %v", err)
-	}
-	if !spy.flushed {
-		t.Error("flush did not reach the underlying Flusher spy")
-	}
-}
-
-func TestHTTPMiddlewareSpanPerRequest(t *testing.T) {
-	sr := withRecorder(t)
-
-	h := HTTPMiddleware("test-server")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if !trace.SpanContextFromContext(r.Context()).IsValid() {
-			t.Error("handler context has no span")
-		}
-		w.WriteHeader(http.StatusTeapot)
-	}))
-	h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/foo", nil))
-
-	spans := sr.Ended()
-	if len(spans) != 1 {
-		t.Fatalf("ended spans = %d, want 1", len(spans))
-	}
-	span := spans[0]
-	if span.Name() != "POST /foo" {
-		t.Errorf("span name = %q, want %q", span.Name(), "POST /foo")
-	}
-	got := map[string]string{}
-	for _, kv := range span.Attributes() {
-		got[string(kv.Key)] = kv.Value.String()
-	}
-	if got["http.response.status_code"] != "418" {
-		t.Errorf("status attr = %q, want 418", got["http.response.status_code"])
-	}
-	if got["server.name"] != "test-server" {
-		t.Errorf("server.name = %q", got["server.name"])
-	}
-	if span.Status().Code == codes.Error {
-		t.Error("4xx must not mark span as error")
-	}
-}
-
-func TestHTTPMiddleware5xxSetsErrorStatus(t *testing.T) {
-	sr := withRecorder(t)
-
-	h := HTTPMiddleware("test-server")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		w.WriteHeader(http.StatusBadGateway)
-	}))
-	h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/x", nil))
-
-	spans := sr.Ended()
-	if len(spans) != 1 {
-		t.Fatalf("ended spans = %d, want 1", len(spans))
-	}
-	if spans[0].Status().Code != codes.Error {
-		t.Errorf("span status = %v, want Error on 5xx", spans[0].Status().Code)
-	}
-}
-
-func TestHTTPMiddlewareExtractsTraceparent(t *testing.T) {
-	sr := withRecorder(t)
-	// Extraction needs the W3C propagator installed (the shared Init does this
-	// even without an endpoint).
-	t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "")
-	if _, err := bbotel.Init(t.Context(), "test-svc"); err != nil {
-		t.Fatalf("Init: %v", err)
+	var rm metricdata.ResourceMetrics
+	if err := metricReader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("Collect: %v", err)
 	}
 
-	h := HTTPMiddleware("test-server")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
-	req := httptest.NewRequest(http.MethodGet, "/y", nil)
-	req.Header.Set("Traceparent", "00-0102030405060708090a0b0c0d0e0f10-0a0b0c0d0e0f0102-01")
-	h.ServeHTTP(httptest.NewRecorder(), req)
-
-	spans := sr.Ended()
-	if len(spans) != 1 {
-		t.Fatalf("ended spans = %d, want 1", len(spans))
-	}
-	if got := spans[0].SpanContext().TraceID().String(); got != "0102030405060708090a0b0c0d0e0f10" {
-		t.Errorf("trace id = %s, want inbound traceparent trace id", got)
-	}
-	if got := spans[0].Parent().SpanID().String(); got != "0a0b0c0d0e0f0102" {
-		t.Errorf("parent span id = %s, want inbound span id", got)
-	}
-}
-
-// TestOutcomeForStatusSeparatesClientErrors guards the classification the
-// dashboards split on: 4xx must not be counted as success (which hid every
-// IMDS 401 and 404) and must not be folded into error, which is server fault.
-func TestOutcomeForStatusSeparatesClientErrors(t *testing.T) {
-	tests := []struct {
-		status int
-		want   string
-	}{
-		{status: http.StatusOK, want: "success"},
-		{status: http.StatusNoContent, want: "success"},
-		{status: http.StatusNotModified, want: "success"},
-		{status: http.StatusBadRequest, want: "client_error"},
-		{status: http.StatusUnauthorized, want: "client_error"},
-		{status: http.StatusForbidden, want: "client_error"},
-		{status: http.StatusNotFound, want: "client_error"},
-		{status: http.StatusInternalServerError, want: "error"},
-		{status: http.StatusBadGateway, want: "error"},
-	}
-
-	for _, tc := range tests {
-		if got := OutcomeForStatus(tc.status); got != tc.want {
-			t.Errorf("OutcomeForStatus(%d) = %q, want %q", tc.status, got, tc.want)
+	var outcomes []string
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			sum, ok := m.Data.(metricdata.Sum[int64])
+			if m.Name != "mulga.requests" || !ok {
+				continue
+			}
+			for _, dp := range sum.DataPoints {
+				if v, found := dp.Attributes.Value(attribute.Key(actionAttrKey)); !found || v.AsString() != action {
+					continue
+				}
+				v, _ := dp.Attributes.Value(attribute.Key("outcome"))
+				outcomes = append(outcomes, v.AsString())
+			}
 		}
 	}
+	return outcomes
 }
 
-// TestHTTPMiddleware4xxIsNotSpanError pins the split between the two signals:
-// a client error is a distinct metric outcome but must not mark the span
-// failed, or every 404 would show as a broken request in the trace view.
-func TestHTTPMiddleware4xxIsNotSpanError(t *testing.T) {
-	sr := withRecorder(t)
+// TestHTTPMiddlewareRecordsOnSpinifexInstruments pins the WithRecorder binding:
+// requests must land on spinifex's rpc.method counter, the one the NATS paths
+// also write to, not on bluebottle's s3.action instruments.
+func TestHTTPMiddlewareRecordsOnSpinifexInstruments(t *testing.T) {
+	withRecorder(t)
 
-	h := HTTPMiddleware("test")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	h := HTTPMiddleware("test-server")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		SetRequestAction(r.Context(), "test.http.recorded")
 		w.WriteHeader(http.StatusUnauthorized)
 	}))
 	h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/latest/api/token", nil))
 
-	spans := sr.Ended()
-	if len(spans) != 1 {
-		t.Fatalf("got %d spans, want 1", len(spans))
-	}
-	if got := spans[0].Status().Code; got == codes.Error {
-		t.Errorf("span status = %v, want not Error for a 4xx", got)
+	if got := requestPointsFor(t, "test.http.recorded"); len(got) != 1 || got[0] != "client_error" {
+		t.Errorf("recorded outcomes = %v, want [client_error]", got)
+	}
+}
+
+// TestHTTPMiddlewareHealthProbeSkipsTrace guards the behaviour the fork existed
+// for: health probes fire every few seconds per service and must be measured
+// without rooting a trace each.
+func TestHTTPMiddlewareHealthProbeSkipsTrace(t *testing.T) {
+	sr := withRecorder(t)
+
+	h := HTTPMiddleware("test-server")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if trace.SpanContextFromContext(r.Context()).IsValid() {
+			t.Error("health probe rooted a span")
+		}
+	}))
+	h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/health", nil))
+
+	if spans := sr.Ended(); len(spans) != 0 {
+		t.Fatalf("ended spans = %d, want 0", len(spans))
+	}
+	if got := requestPointsFor(t, "GET /health"); len(got) != 1 || got[0] != "success" {
+		t.Errorf("recorded outcomes = %v, want [success]", got)
+	}
+}
+
+// TestHTTPMiddlewareStillTracesRealRequests keeps the probe skip path-scoped.
+func TestHTTPMiddlewareStillTracesRealRequests(t *testing.T) {
+	sr := withRecorder(t)
+
+	h := HTTPMiddleware("test-server")(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/foo", nil))
+
+	if spans := sr.Ended(); len(spans) != 1 {
+		t.Fatalf("ended spans = %d, want 1", len(spans))
 	}
 }

spinifex/otelsetup/metrics.go

@@ -10,6 +10,10 @@ import (
 	"go.opentelemetry.io/otel/metric"
 )
 
+// meterName is the instrumentation scope every spinifex request metric shares,
+// so the HTTP and NATS paths land on one set of instruments.
+const meterName = "github.com/mulgadc/spinifex/spinifex/otelsetup"
+
 // actionAttrKey names the logical operation on request metrics. Values must
 // stay low-cardinality: resolved action names only, never resource IDs.
 const actionAttrKey = "rpc.method"
@@ -32,7 +36,7 @@ var (
 // global meter delegates to the real provider once Init installs it.
 func requestInstruments() (metric.Int64Counter, metric.Float64Histogram) {
 	instrumentsOnce.Do(func() {
-		m := otel.Meter(httpTracerName)
+		m := otel.Meter(meterName)
 		var err error
 		requestCounter, err = m.Int64Counter("mulga.requests",
 			metric.WithDescription("Count of service requests handled."),
@@ -74,7 +78,7 @@ func RecordRequest(ctx context.Context, action, outcome string, elapsed time.Dur
 func RecordResourceLeak(ctx context.Context, kind string) {
 	leakOnce.Do(func() {
 		var err error
-		leakCounter, err = otel.Meter(httpTracerName).Int64Counter("mulga.resource.leaked",
+		leakCounter, err = otel.Meter(meterName).Int64Counter("mulga.resource.leaked",
 			metric.WithDescription("Count of resources teardown abandoned without reclaiming."),
 			metric.WithUnit("{resource}"))
 		if err != nil {
@@ -86,18 +90,3 @@ func RecordResourceLeak(ctx context.Context, kind string) {
 			attribute.NewSet(attribute.String(leakKindAttrKey, kind))))
 	}
 }
-
-// requestActionKey carries a mutable per-request holder so downstream
-// middleware can name the request after routing/auth resolves the action.
-type requestActionKey struct{}
-
-type requestAction struct{ name string }
-
-// SetRequestAction sets the logical action recorded on request metrics for
-// the in-flight request. No-op when the request did not pass through
-// HTTPMiddleware.
-func SetRequestAction(ctx context.Context, action string) {
-	if h, ok := ctx.Value(requestActionKey{}).(*requestAction); ok && action != "" {
-		h.name = action
-	}
-}