From 02358a79d798b80f1b47a64a5d59787628ce045a Mon Sep 17 00:00:00 2001 From: mintaka Date: Thu, 27 Aug 2026 23:37:25 -0400 Subject: [PATCH 1/2] feat(otel): emit server-side spans + traceresponse header (RIG-2889) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire OTel emission into the compass-server binary and its RPC doors (T4b T2). The endpoint is env-only (`OTEL_EXPORTER_OTLP_ENDPOINT`); empty = tracing off, so the shipped socket-only path stays zero-overhead. - **cmd/compass-server:** bootstrap `SetupTracerProvider`/`SetupMeterProvider` off `ServeConfig.OtelEndpoint` (populated from the env knob in `buildServeConfig`), defer both shutdowns before `Serve`. - **serve.go / network_door.go:** mount `otelconnect` (outermost) + the trace-response interceptor on every door chain — socket + dev `CompassService`, `CommsService`, and the network shared bearer/admin-gate chain — so the RPC span envelopes the security interceptors and their ordering is unchanged relative to itself. Both interceptors are inert no-ops when no provider is installed. The interceptor + comms-handler construction lives in `buildDoors` (where `otelIC` is consumed), keeping `Serve` within its complexity budget. - **comms.go:** stamp `compass.message.id` onto the handler span in `PostMessage` and `RespondToAsk` (no-op when no span is active). - **CORS:** expose the `traceresponse` header on `devCORS` + `networkCORS` so a cross-origin browser can read the response trace id. Adds `connectrpc.com/otelconnect v0.9.0`; refreshes the nix `vendorHash` for the moved module set. Spec-impact: none. Refs RIG-2889. Co-authored-by: Matt Wilkinson --- flake.nix | 2 +- go/cmd/compass-server/main.go | 25 +++ go/go.mod | 3 +- go/go.sum | 2 + go/internal/comms/comms.go | 10 + go/server/cors_pgtest_test.go | 8 +- go/server/network_door.go | 14 +- go/server/otel_emission_pgtest_test.go | 299 +++++++++++++++++++++++++ go/server/serve.go | 37 ++- guest-image/default.nix | 2 +- 10 files changed, 389 insertions(+), 13 deletions(-) create mode 100644 go/server/otel_emission_pgtest_test.go diff --git a/flake.nix b/flake.nix index 158c8403c..bb62a4cb5 100644 --- a/flake.nix +++ b/flake.nix @@ -49,7 +49,7 @@ # touched (guest-image/default.nix:82-87). vendorHash pins the fetched set — # the whole module graph, so it matches guestd's proxyVendor hash. Recompute # with lib.fakeHash on a go.mod/go.sum move. - vendorHash = "sha256-di6nYpDDZblu6TPpendW88l5LMRZNzSE2t2tiWfms9c="; + vendorHash = "sha256-GHZsEfvnu1tY6Bd7Fxg7SEWEI+HS0NlQuBbm6pz/UK4="; in { packages = forAllSystems ( diff --git a/go/cmd/compass-server/main.go b/go/cmd/compass-server/main.go index 086a7ac31..b5c2fad67 100644 --- a/go/cmd/compass-server/main.go +++ b/go/cmd/compass-server/main.go @@ -19,6 +19,7 @@ import ( "syscall" "time" + "github.com/RigelBuild/compass/go/internal/otel" "github.com/RigelBuild/compass/go/server" ) @@ -91,6 +92,27 @@ func run() error { stopDrainLog := logOnDrainSignal() defer stopDrainLog() + // OTel emission (T4b): endpoint-gated off the ENV-only knob. When + // OTEL_EXPORTER_OTLP_ENDPOINT is empty, Setup* install no provider and return + // no-op shutdowns, so this is zero-overhead on the shipped socket-only path. + // The shutdowns flush under the drain ctx, deferred before Serve so they run + // on graceful shutdown. + otelCfg := otel.Config{ + ServiceName: "compass-server", + ServiceVersion: version, + Endpoint: cfg.OtelEndpoint, + } + traceShutdown, err := otel.SetupTracerProvider(ctx, otelCfg) + if err != nil { + return fmt.Errorf("otel: tracer provider: %w", err) + } + defer func() { _ = traceShutdown(ctx) }() // best-effort flush on drain; a collector error here is not actionable at exit + meterShutdown, err := otel.SetupMeterProvider(ctx, otelCfg) + if err != nil { + return fmt.Errorf("otel: meter provider: %w", err) + } + defer func() { _ = meterShutdown(ctx) }() // best-effort flush on drain; a collector error here is not actionable at exit + return server.Serve(ctx, cfg) } @@ -193,6 +215,9 @@ func buildServeConfig(args []string) (server.ServeConfig, bool, error) { AdminHandle: *f.adminHandle, CORSAllowedOrigin: *f.corsAllowedOrigin, PublicURL: firstNonEmpty(*f.publicURL, os.Getenv("COMPASS_PUBLIC_URL")), + // ENV-ONLY knob (Matt 2026-08-28): the OTLP exporter and the enable-gate + // read one source, so no --otel-endpoint flag. Empty = tracing off. + OtelEndpoint: os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"), }, false, nil } diff --git a/go/go.mod b/go/go.mod index e0abab89b..460e6873e 100644 --- a/go/go.mod +++ b/go/go.mod @@ -17,6 +17,7 @@ go 1.25.0 require ( connectrpc.com/connect v1.20.0 connectrpc.com/cors v0.1.0 + connectrpc.com/otelconnect v0.9.0 github.com/BurntSushi/toml v1.6.0 github.com/cachix/secretspec/secretspec-go v0.15.0 github.com/hashicorp/golang-lru/v2 v2.0.7 @@ -34,6 +35,7 @@ require ( go.opentelemetry.io/otel v1.46.0 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.46.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.46.0 + go.opentelemetry.io/otel/metric v1.46.0 go.opentelemetry.io/otel/sdk v1.46.0 go.opentelemetry.io/otel/sdk/metric v1.46.0 go.opentelemetry.io/otel/trace v1.46.0 @@ -89,7 +91,6 @@ require ( github.com/zeebo/xxh3 v1.1.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.46.0 // indirect - go.opentelemetry.io/otel/metric v1.46.0 // indirect go.opentelemetry.io/proto/otlp v1.11.0 // indirect golang.org/x/crypto v0.55.0 // indirect golang.org/x/sys v0.47.0 // indirect diff --git a/go/go.sum b/go/go.sum index 230a32276..ea81bd03b 100644 --- a/go/go.sum +++ b/go/go.sum @@ -2,6 +2,8 @@ connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ= connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4= connectrpc.com/cors v0.1.0 h1:f3gTXJyDZPrDIZCQ567jxfD9PAIpopHiRDnJRt3QuOQ= connectrpc.com/cors v0.1.0/go.mod h1:v8SJZCPfHtGH1zsm+Ttajpozd4cYIUryl4dFB6QEpfg= +connectrpc.com/otelconnect v0.9.0 h1:NggB3pzRC3pukQWaYbRHJulxuXvmCKCKkQ9hbrHAWoA= +connectrpc.com/otelconnect v0.9.0/go.mod h1:AEkVLjCPXra+ObGFCOClcJkNjS7zPaQSqvO0lCyjfZc= github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/adrg/xdg v0.5.3 h1:xRnxJXne7+oWDatRhR1JLnvuccuIeCoBu2rtuLqQB78= diff --git a/go/internal/comms/comms.go b/go/internal/comms/comms.go index c3e72ecd8..714ee4c14 100644 --- a/go/internal/comms/comms.go +++ b/go/internal/comms/comms.go @@ -21,6 +21,8 @@ import ( "fmt" "connectrpc.com/connect" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "github.com/RigelBuild/compass/go/events" compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" @@ -349,6 +351,10 @@ func (c *Comms) PostMessage( if err != nil { return nil, edgeError(err) } + // Stamp the appended message's id onto the handler's RPC span (the + // otelconnect origin span mounted on this service), so a trace filters to + // one message. A no-op when no span is active (no provider installed). + trace.SpanFromContext(ctx).SetAttributes(attribute.String("compass.message.id", string(msg.ID))) // Publish only on a genuine insert: an idempotent retry returns the stored // row unchanged (inserted=false), so re-fanning MessagePosted would emit a // spurious live state-change for a row that did not change. @@ -386,6 +392,10 @@ func (c *Comms) RespondToAsk( if err != nil { return nil, edgeError(err) } + // Stamp the answer message's id onto the handler's RPC span (the delivery + // origin for the answer post), matching PostMessage. A no-op when no span is + // active. + trace.SpanFromContext(ctx).SetAttributes(attribute.String("compass.message.id", string(answerMsg.ID))) // MessageUpdated carries the ask's new answered state to the UI; it is NOT a // delivery trigger. MessagePosted for the answer message IS the delivery // trigger — it fans out on the normal message rail, so an offline or diff --git a/go/server/cors_pgtest_test.go b/go/server/cors_pgtest_test.go index 2e941fb2e..548e587ce 100644 --- a/go/server/cors_pgtest_test.go +++ b/go/server/cors_pgtest_test.go @@ -28,6 +28,8 @@ import ( "net/http/httptest" "testing" + "connectrpc.com/otelconnect" + "github.com/RigelBuild/compass/go/events" compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/gen/compass/v1/compassv1connect" @@ -67,7 +69,11 @@ func buildDoorHandler(t *testing.T, corsOrigin string) http.Handler { CORSAllowedOrigin: corsOrigin, } secretsSvc := newSecretsService(st, nil, nil) - srv, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil) + otelIC, err := otelconnect.NewInterceptor() + if err != nil { + t.Fatalf("otelconnect.NewInterceptor: %v", err) + } + srv, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC) if err != nil { t.Fatalf("buildNetworkServer: %v", err) } diff --git a/go/server/network_door.go b/go/server/network_door.go index 430571769..39a03db77 100644 --- a/go/server/network_door.go +++ b/go/server/network_door.go @@ -20,11 +20,13 @@ import ( "connectrpc.com/connect" connectcors "connectrpc.com/cors" + "connectrpc.com/otelconnect" "github.com/rs/cors" "github.com/RigelBuild/compass/go/gen/compass/v1/compassv1connect" "github.com/RigelBuild/compass/go/internal/auth" "github.com/RigelBuild/compass/go/internal/gen/compass/v1/compassv1internalconnect" + "github.com/RigelBuild/compass/go/internal/otel" "github.com/RigelBuild/compass/go/internal/runnerhub" "github.com/RigelBuild/compass/go/internal/secrets" "github.com/RigelBuild/compass/go/internal/store" @@ -140,7 +142,7 @@ func networkCORS(origin string) *cors.Cors { AllowedOrigins: []string{origin}, AllowedMethods: connectcors.AllowedMethods(), AllowedHeaders: append(connectcors.AllowedHeaders(), "Authorization"), - ExposedHeaders: connectcors.ExposedHeaders(), + ExposedHeaders: append(connectcors.ExposedHeaders(), "traceresponse"), AllowCredentials: false, }) } @@ -241,6 +243,7 @@ func buildNetworkServer( adminID store.AccountID, netTLS *tls.Config, resolver secrets.Resolver, + otelIC *otelconnect.Interceptor, ) (*http.Server, error) { handle := cfg.resolvedAdminHandle() stateDir := cfg.StateDir @@ -260,7 +263,16 @@ func buildNetworkServer( slog.Info("network door bootstrap admin token written", "path", tokenPath, "handle", handle, "listen", cfg.Listen) + // otelconnect (outermost) produces the RPC span and NewTraceResponseInterceptor + // stamps the traceresponse header, prepended to the shared bearer + admin-gate + // chain so every network-door service (CompassService, CommsService, and + // SecretsService, which rides the same chain) carries them. Both are inert + // no-ops when no provider is installed (empty OtelEndpoint). Ordering: otel + // first keeps the security-critical Bearer→AdminGate order unchanged relative + // to itself. interceptors := connect.WithInterceptors( + otelIC, + otel.NewTraceResponseInterceptor(), auth.BearerInterceptor(st), auth.BearerStreamInterceptor(st), auth.NewAdminGate(adminID), diff --git a/go/server/otel_emission_pgtest_test.go b/go/server/otel_emission_pgtest_test.go new file mode 100644 index 000000000..19e8a6981 --- /dev/null +++ b/go/server/otel_emission_pgtest_test.go @@ -0,0 +1,299 @@ +//go:build pgtest && unix + +package server + +// T4b server-emission integration: the OTel wiring T2 adds to the shipped +// socket door and the network-door CORS policy, exercised end-to-end. +// +// The emission path is only observable through a REAL provider: otelconnect and +// NewTraceResponseInterceptor both read the GLOBAL tracer provider (otelconnect +// interceptor.go:56 otel.GetTracerProvider), so a test that would prove a span +// carries compass.message.id and the response carries a matching traceresponse +// header must install a global SDK provider with an in-memory exporter, then +// drive PostMessage over the production socket door (Serve). resetGlobals-style +// save/restore keeps the installed globals from leaking into sibling tests. +// +// Store-gated (Serve opens the store of record and PostMessage's D9 membership +// gate is store-enforced), so behind `//go:build pgtest && unix` via the shared +// pgtest harness (pgtest.RequireDSN → an isolated-schema DSN, or t.Skip when no +// runtime). Hermetic: t.TempDir() socket + state paths, no fixed ports. +// +// context.Background() is the test root (rule://go-thread-context _test.go +// exemption): threaded into store.Open, the pre-seed, and every RPC below. + +import ( + "context" + "net/http" + "net/http/httptest" + "path/filepath" + "regexp" + "strings" + "testing" + + "connectrpc.com/connect" + "connectrpc.com/otelconnect" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + + "github.com/RigelBuild/compass/go/events" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/gen/compass/v1/compassv1connect" + "github.com/RigelBuild/compass/go/internal/comms" + "github.com/RigelBuild/compass/go/internal/pgtest" + "github.com/RigelBuild/compass/go/internal/store" +) + +// messageIDAttr is the span attribute PostMessage/RespondToAsk stamp the +// appended message's id onto (comms.go), and traceResponseHdr is the W3C +// trace-response header the interceptor sets (otel.NewTraceResponseInterceptor). +const ( + messageIDAttr = "compass.message.id" + traceResponseHdr = "traceresponse" + exposeHeadersHdr = "Access-Control-Expose-Headers" + corsOriginForTest = "https://app.example.com" +) + +// installGlobalSpanExporter installs a global SDK TracerProvider feeding an +// in-memory exporter and a W3C propagator (what otelconnect and the trace- +// response interceptor read), restoring the prior globals on cleanup so an +// enabled-path test never leaks its provider into a sibling. It mirrors the +// enabled path SetupTracerProvider installs, but with a SyncSpanProcessor onto +// an InMemoryExporter so the test reads spans without a collector or a flush +// race. Returns the exporter the assertions read. +func installGlobalSpanExporter(t *testing.T) *tracetest.InMemoryExporter { + t.Helper() + prevTP := otel.GetTracerProvider() + prevProp := otel.GetTextMapPropagator() + exp := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) + otel.SetTracerProvider(tp) + otel.SetTextMapPropagator(propagation.TraceContext{}) + t.Cleanup(func() { + _ = tp.Shutdown(context.Background()) // best-effort flush; sync exporter already holds the spans, error not actionable in test + otel.SetTracerProvider(prevTP) + otel.SetTextMapPropagator(prevProp) + }) + return exp +} + +// seedAdminChannel opens the store, finds-or-creates the bootstrap admin (the +// account the socket door attributes every RPC to), and creates one OPEN channel +// the admin is a member of — so PostMessage's D9 membership gate admits a post +// over the socket. It returns the DSN (so Serve opens the SAME isolated schema) +// and the channel id. Serve's own BootstrapAdmin later finds this same admin. +func seedAdminChannel(t *testing.T, ctx context.Context) (dsn string, channelID string) { + t.Helper() + dsn = pgtest.RequireDSN(t) + st, err := store.Open(ctx, dsn) + if err != nil { + t.Fatalf("store Open: %v", err) + } + t.Cleanup(st.Close) + admin, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: bootstrapAdminHandle, DisplayName: bootstrapAdminDisplayName}) + if err != nil { + t.Fatalf("BootstrapAdmin: %v", err) + } + ch, err := st.CreateChannel(ctx, admin.ID, store.NewChannel{ + Name: "otel-emission", Kind: store.ChannelKindChannel, + MemberAccountIDs: []store.AccountID{admin.ID}, + }) + if err != nil { + t.Fatalf("CreateChannel: %v", err) + } + return dsn, string(ch.ID) +} + +// postOverSocket posts one text message to channelID over the shipped socket +// door and returns the connect response (whose header carries traceresponse if +// the interceptor set one) and the created message id. +func postOverSocket(t *testing.T, ctx context.Context, socketPath, channelID string) (*connect.Response[compassv1.PostMessageResponse], string) { + t.Helper() + client := newUDSCommsClient(t, socketPath) + resp, err := client.PostMessage(ctx, connect.NewRequest(&compassv1.PostMessageRequest{ + Container: &compassv1.PostMessageRequest_ChannelId{ChannelId: channelID}, + Topic: &compassv1.PostMessageRequest_TopicName{TopicName: "general"}, + Blocks: []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: "hello otel"}}}, + })) + if err != nil { + t.Fatalf("PostMessage over socket: %v", err) + } + return resp, resp.Msg.GetMessage().GetId() +} + +// TestServerEmissionSocketDoorTracesPostMessage is the enabled-path emission +// contract on the SHIPPED socket door: a PostMessage RPC produces a server span +// carrying the compass.message.id of the appended message, and the response +// carries a traceresponse header whose trace id equals that span's trace id. +// Drives Serve (the production socket door with the otelconnect + trace-response +// interceptors T2 mounts) against a real store, with a global in-memory span +// exporter installed so the span is observable. +func TestServerEmissionSocketDoorTracesPostMessage(t *testing.T) { + ctx := context.Background() // test root (rule://go-thread-context _test.go exemption) + exp := installGlobalSpanExporter(t) + dsn, channelID := seedAdminChannel(t, ctx) + + socketPath := serveOTelSocket(t, "otel-emission-test", dsn) + + resp, msgID := postOverSocket(t, ctx, socketPath, channelID) + if msgID == "" { + t.Fatal("PostMessage returned an empty message id") + } + + // (a) exactly one server span carries the appended message's id. + span := spanWithMessageID(t, exp.GetSpans(), msgID) + + // (b) the traceresponse header's trace id equals that span's trace id. + hdr := resp.Header().Get(traceResponseHdr) + if hdr == "" { + t.Fatal("response carried no traceresponse header on the enabled path") + } + wantTraceID := span.SpanContext.TraceID().String() + // grammar: 00-<32hex traceid>-<16hex spanid>-<2hex flags> + if !w3cTraceResponse.MatchString(hdr) { + t.Fatalf("traceresponse %q is not W3C 00-… grammar", hdr) + } + if gotTraceID := hdr[3:35]; gotTraceID != wantTraceID { + t.Fatalf("traceresponse trace id = %q, want the span's trace id %q (header %q)", gotTraceID, wantTraceID, hdr) + } +} + +// TestServerEmissionDisabledProducesNoSpansNoHeader is the disabled-path +// contract: with NO global provider installed (the shipped default when +// OTEL_EXPORTER_OTLP_ENDPOINT is unset), a PostMessage over the socket door +// produces ZERO exported spans and NO traceresponse header — otelconnect falls +// back to the global no-op provider, so no span is ever recorded and the +// interceptor finds no valid span context to stamp. +func TestServerEmissionDisabledProducesNoSpansNoHeader(t *testing.T) { + ctx := context.Background() // test root (rule://go-thread-context _test.go exemption) + // Save/restore globals but install NO SDK provider: an in-memory exporter is + // attached to a provider that is NEVER set global, so it can only receive a + // span if the door wrongly recorded one against some other provider. + prevTP := otel.GetTracerProvider() + prevProp := otel.GetTextMapPropagator() + exp := tracetest.NewInMemoryExporter() + _ = sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) // deliberately NOT set global + t.Cleanup(func() { + otel.SetTracerProvider(prevTP) + otel.SetTextMapPropagator(prevProp) + }) + + dsn, channelID := seedAdminChannel(t, ctx) + socketPath := serveOTelSocket(t, "otel-disabled-test", dsn) + + resp, _ := postOverSocket(t, ctx, socketPath, channelID) + + if spans := exp.GetSpans(); len(spans) != 0 { + t.Fatalf("disabled path exported %d spans, want 0", len(spans)) + } + if hdr := resp.Header().Get(traceResponseHdr); hdr != "" { + t.Fatalf("disabled path set traceresponse = %q, want none", hdr) + } +} + +// TestNetworkDoorExposesTraceResponseHeader pins the CORS half of T2: the +// network door's Access-Control-Expose-Headers must include traceresponse so a +// cross-origin browser can READ the header the trace-response interceptor sets. +// Drives the production buildNetworkServer (via buildDoorHandler's sibling shape) +// and inspects the actual-request CORS response, where rs/cors emits +// Access-Control-Expose-Headers (not on the preflight). +func TestNetworkDoorExposesTraceResponseHeader(t *testing.T) { + ctx := context.Background() // test root (rule://go-thread-context _test.go exemption) + st, admin, _ := newNetworkStore(t) + + bus := events.NewBus[busPayload]() + t.Cleanup(bus.Close) + svc := newService("otel-cors-test", bus, st, nil, nil, nil, nil) + commsBus := events.NewBus[*compassv1.SubscribeCommsResponse]() + t.Cleanup(commsBus.Close) + commsSvc := comms.NewComms(st, commsBus, admin) + secretsSvc := newSecretsService(st, nil, nil) + otelIC, err := otelconnect.NewInterceptor() + if err != nil { + t.Fatalf("otelconnect.NewInterceptor: %v", err) + } + srv, err := buildNetworkServer(ctx, ServeConfig{ + StateDir: t.TempDir(), + CORSAllowedOrigin: corsOriginForTest, + }, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC) + if err != nil { + t.Fatalf("buildNetworkServer: %v", err) + } + + // An actual (non-preflight) cross-origin GET carries the CORS + // Access-Control-Expose-Headers response header; rs/cors joins the exposed + // set into one comma-separated value. + req := httptest.NewRequest(http.MethodGet, compassv1connect.CompassServiceGetServerInfoProcedure, nil) + req.Header.Set("Origin", corsOriginForTest) + rec := httptest.NewRecorder() + srv.Handler.ServeHTTP(rec, req) + + exposed := rec.Result().Header.Get(exposeHeadersHdr) + if exposed == "" { + t.Fatalf("%s absent on a cross-origin actual request; the network door exposes no headers", exposeHeadersHdr) + } + if !headerListContains(exposed, traceResponseHdr) { + t.Fatalf("%s = %q, want it to include %q so a browser can read the trace-response header", exposeHeadersHdr, exposed, traceResponseHdr) + } +} + +// serveOTelSocket starts a production Serve on a fresh t.TempDir() socket with +// the given version + DSN, waits for the socket to bind, and returns its path. +// Serve is driven on a background goroutine bounded by a cancel deferred via +// t.Cleanup so the server drains at test end. It gates on a served RPC +// (waitListening + the caller's first PostMessage) rather than a sleep. +func serveOTelSocket(t *testing.T, version, dsn string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "compass.sock") + ctx, cancel := context.WithCancel(context.Background()) // test root (rule://go-thread-context _test.go exemption) + errCh := make(chan error, 1) + go func() { + errCh <- Serve(ctx, ServeConfig{SocketPath: path, Version: version, DatabaseDSN: dsn}) + }() + t.Cleanup(func() { + cancel() + select { + case <-errCh: // Serve returns nil on a clean ctx-cancel drain; a fault is covered by the serve_pgtest tests + case <-timeAfter(): + t.Fatal("Serve did not return after ctx cancel") + } + }) + waitListening(t, path) + return path +} + +// spanWithMessageID returns the single exported span carrying the given +// compass.message.id attribute, failing if zero or more than one match — the +// PostMessage origin span is unique per post, so a duplicate means the id +// leaked onto an unrelated span. +func spanWithMessageID(t *testing.T, spans tracetest.SpanStubs, msgID string) tracetest.SpanStub { + t.Helper() + var match []tracetest.SpanStub + for _, s := range spans { + for _, a := range s.Attributes { + if string(a.Key) == messageIDAttr && a.Value.AsString() == msgID { + match = append(match, s) + } + } + } + if len(match) != 1 { + t.Fatalf("found %d spans carrying %s=%q, want exactly 1 (spans exported: %d)", len(match), messageIDAttr, msgID, len(spans)) + } + return match[0] +} + +// headerListContains reports whether a comma-separated header value (as rs/cors +// joins Access-Control-Expose-Headers) contains target, case-insensitively. +func headerListContains(list, target string) bool { + for _, part := range strings.Split(list, ",") { + if strings.EqualFold(strings.TrimSpace(part), target) { + return true + } + } + return false +} + +// w3cTraceResponse matches the traceresponse header grammar the interceptor +// emits: 00-<32hex traceid>-<16hex spanid>-<2hex flags>. +var w3cTraceResponse = regexp.MustCompile(`^00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$`) diff --git a/go/server/serve.go b/go/server/serve.go index 90521a50a..4465293e2 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -29,6 +29,7 @@ import ( "connectrpc.com/connect" connectcors "connectrpc.com/cors" + "connectrpc.com/otelconnect" "github.com/rs/cors" "golang.org/x/sync/errgroup" @@ -40,6 +41,7 @@ import ( "github.com/RigelBuild/compass/go/internal/comms" "github.com/RigelBuild/compass/go/internal/forge" "github.com/RigelBuild/compass/go/internal/ingest" + "github.com/RigelBuild/compass/go/internal/otel" "github.com/RigelBuild/compass/go/internal/runnerhub" "github.com/RigelBuild/compass/go/internal/secrets" "github.com/RigelBuild/compass/go/internal/store" @@ -108,6 +110,11 @@ type ServeConfig struct { // leaves the driver off — today's behavior, zero new requirements on // existing deployments. See ForgeConfig and forgePollingEnabled. Forge ForgeConfig + // OtelEndpoint is the OTLP collector endpoint (OTEL_EXPORTER_OTLP_ENDPOINT); + // empty = tracing off. It gates OTel emission on both binaries: when unset, + // the bootstrap installs no provider and the RPC interceptors are inert + // no-ops (no active span, so no traceresponse header). + OtelEndpoint string } // ForgeConfig configures the board-ingestion poll driver (SEA-1810, DL-053). @@ -455,7 +462,6 @@ func Serve(ctx context.Context, cfg ServeConfig) error { commsBus := events.NewBus[*compassv1.SubscribeCommsResponse]() defer commsBus.Close() commsSvc := comms.NewComms(st, commsBus, admin.ID) - commsPath, commsHandler := compassv1connect.NewCommsServiceHandler(commsSvc) // Register the coordination-channel reconcile as the store's in-tx hook, so // the two parent-edge writers auto-provision/reconcile a manager's // coordination channel atomically with the tree edge (SEA-1722 T5). Wired here @@ -506,7 +512,7 @@ func Serve(ctx context.Context, cfg ServeConfig) error { // loopback, optional authenticated network). On a net-door build error the // listeners this Serve bound are still ours to close. doors, err := buildDoors(ctx, cfg, svc, commsSvc, secretsSvc, hub, st, admin.ID, resolver, - commsPath, commsHandler, devListener, netListener, netTLS) + devListener, netListener, netTLS) if err != nil { udsListener.Close() //nolint:errcheck,gosec // teardown on an already-failing startup path — nothing actionable remains (errcheck + its gosec G104 twin) listeners.close() @@ -608,12 +614,25 @@ func buildDoors( st *store.Store, adminID store.AccountID, resolver secrets.Resolver, - commsPath string, - commsHandler http.Handler, devListener net.Listener, netListener net.Listener, netTLS *tls.Config, ) (serveDoors, error) { + // otelconnect produces the server RPC span (and, once a MeterProvider is + // installed, RPC duration/count metrics); NewTraceResponseInterceptor stamps + // the span's trace id onto the "traceresponse" response header. Both are + // inert no-ops when OtelEndpoint is empty (no provider ⇒ no active span), so + // they are mounted unconditionally. otelconnect goes FIRST (outermost) in + // every chain so the span envelopes the security-critical interceptors and + // the AdminGate→Ambient ordering is unchanged relative to itself. + otelIC, err := otelconnect.NewInterceptor() + if err != nil { + return serveDoors{}, fmt.Errorf("otel: rpc interceptor: %w", err) + } + // CommsService rides the socket + dev doors and the network shared chain; it + // mounts the same otelconnect + trace-response pair as CompassService. + commsPath, commsHandler := compassv1connect.NewCommsServiceHandler(commsSvc, + connect.WithInterceptors(otelIC, otel.NewTraceResponseInterceptor())) // Shipped door: the Unix socket serves native gRPC (cleartext HTTP/2), // gRPC-Web, and Connect off the one connect-go handler. No CORS — the socket // is same-origin (the shell's webview / a native client). The 0600 socket is @@ -623,7 +642,8 @@ func buildDoors( // natively (http.Protocols); these connections are tracked by // http.Server.Shutdown and drain with the rest on shutdown. socketPath, socketHandler := compassv1connect.NewCompassServiceHandler(svc, - connect.WithInterceptors(auth.AmbientIdentity(adminID), auth.AmbientStreamInterceptor(adminID))) + connect.WithInterceptors(otelIC, otel.NewTraceResponseInterceptor(), + auth.AmbientIdentity(adminID), auth.AmbientStreamInterceptor(adminID))) // SecretsService rides the same ambient-identity pair so its handler reads a // caller via CallerFrom (the bootstrap admin, a user): the socket's 0600 mode // is the credential, and admin being a user satisfies the user-only writes. @@ -659,7 +679,8 @@ func buildDoors( var devServer *http.Server if devListener != nil { devPath, devHandler := compassv1connect.NewCompassServiceHandler(svc, - connect.WithInterceptors(auth.NewAdminGate(adminID), auth.AmbientIdentity(adminID), auth.AmbientStreamInterceptor(adminID))) + connect.WithInterceptors(otelIC, otel.NewTraceResponseInterceptor(), + auth.NewAdminGate(adminID), auth.AmbientIdentity(adminID), auth.AmbientStreamInterceptor(adminID))) // SecretsService on the dev door: same admin-gate + ambient chain as // CompassService. The gate classifies its 3 procedures authenticatedOpen, // so it passes them, then the ambient pair attaches the bootstrap admin @@ -681,7 +702,7 @@ func buildDoors( // On a build error the listeners this Serve bound are still ours to close. var netServer *http.Server if netListener != nil { - s, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, hub, st, adminID, netTLS, resolver) + s, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, hub, st, adminID, netTLS, resolver, otelIC) if err != nil { return serveDoors{}, err } @@ -760,7 +781,7 @@ func devCORS() *cors.Cors { AllowedOrigins: []string{"*"}, AllowedMethods: connectcors.AllowedMethods(), AllowedHeaders: connectcors.AllowedHeaders(), - ExposedHeaders: connectcors.ExposedHeaders(), + ExposedHeaders: append(connectcors.ExposedHeaders(), "traceresponse"), AllowCredentials: false, }) } diff --git a/guest-image/default.nix b/guest-image/default.nix index 9b061fef4..b8983c29a 100644 --- a/guest-image/default.nix +++ b/guest-image/default.nix @@ -96,7 +96,7 @@ let }; subPackages = [ "cmd/compass-guestd" ]; proxyVendor = true; - vendorHash = "sha256-di6nYpDDZblu6TPpendW88l5LMRZNzSE2t2tiWfms9c="; + vendorHash = "sha256-GHZsEfvnu1tY6Bd7Fxg7SEWEI+HS0NlQuBbm6pz/UK4="; env.CGO_ENABLED = 0; ldflags = [ "-s" From 84de91fa46db013135739eea73d597add693753d Mon Sep 17 00:00:00 2001 From: mintaka Date: Fri, 28 Aug 2026 01:25:59 -0400 Subject: [PATCH 2/2] fix(otel): flush server providers on drain + real disabled-path assert (RIG-2889) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round 1 on PR #713. ### M1 — shutdown flush dropped on graceful drain The deferred `traceShutdown`/`meterShutdown` ran with the signal-derived `ctx`, which is already cancelled by the time they fire (the SIGTERM that ends `Serve` is the same signal that cancels `ctx`). A cancelled ctx makes the SDK `Shutdown` abort its final `ForceFlush` immediately, so the last buffered batch of spans/metrics was silently dropped exactly on the drain path the defers exist to flush — contradicting the frozen design contract (bounded 2s shutdown flush, mirroring the agent). Each defer now derives a fresh `context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)` at fire time (not at bootstrap, where an absolute deadline would expire mid-run), matching the `context.WithoutCancel` precedent already in `internal/runner/run.go`. ### L1 — vacuous disabled-path span assertion `TestServerEmissionDisabledProducesNoSpansNoHeader` wired an in-memory exporter to a provider it never set global and never referenced, so its zero-spans check could never fire. Dropped the dead exporter; the traceresponse-header-absence assertion is the real disabled-path proof (no recording span ⇒ no header). Renamed to `TestServerEmissionDisabledSetsNoTraceResponseHeader` to match what it proves. Spec-impact: none. Refs RIG-2889 Co-authored-by: Matt Wilkinson --- go/cmd/compass-server/main.go | 20 ++++++++++++++---- go/server/otel_emission_pgtest_test.go | 29 +++++++++++++------------- 2 files changed, 31 insertions(+), 18 deletions(-) diff --git a/go/cmd/compass-server/main.go b/go/cmd/compass-server/main.go index b5c2fad67..2f5ee2684 100644 --- a/go/cmd/compass-server/main.go +++ b/go/cmd/compass-server/main.go @@ -95,8 +95,6 @@ func run() error { // OTel emission (T4b): endpoint-gated off the ENV-only knob. When // OTEL_EXPORTER_OTLP_ENDPOINT is empty, Setup* install no provider and return // no-op shutdowns, so this is zero-overhead on the shipped socket-only path. - // The shutdowns flush under the drain ctx, deferred before Serve so they run - // on graceful shutdown. otelCfg := otel.Config{ ServiceName: "compass-server", ServiceVersion: version, @@ -106,12 +104,26 @@ func run() error { if err != nil { return fmt.Errorf("otel: tracer provider: %w", err) } - defer func() { _ = traceShutdown(ctx) }() // best-effort flush on drain; a collector error here is not actionable at exit + // The drain ctx is already cancelled by the time these defers fire (the signal + // that ends Serve is the same one that cancels ctx), so a raw ctx.Shutdown + // would abort its final ForceFlush and drop the last batch. Sever the + // cancellation and bound the flush at 2s (design.md: mirror the agent's 2s + // shutdown bound), derived HERE at fire time so the deadline is not consumed + // by the process lifetime. + defer func() { + sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second) + defer cancel() + _ = traceShutdown(sctx) // best-effort flush on drain; a collector error here is not actionable at exit + }() meterShutdown, err := otel.SetupMeterProvider(ctx, otelCfg) if err != nil { return fmt.Errorf("otel: meter provider: %w", err) } - defer func() { _ = meterShutdown(ctx) }() // best-effort flush on drain; a collector error here is not actionable at exit + defer func() { + sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second) + defer cancel() + _ = meterShutdown(sctx) // best-effort flush on drain; a collector error here is not actionable at exit + }() return server.Serve(ctx, cfg) } diff --git a/go/server/otel_emission_pgtest_test.go b/go/server/otel_emission_pgtest_test.go index 19e8a6981..3af1f3ec7 100644 --- a/go/server/otel_emission_pgtest_test.go +++ b/go/server/otel_emission_pgtest_test.go @@ -159,21 +159,22 @@ func TestServerEmissionSocketDoorTracesPostMessage(t *testing.T) { } } -// TestServerEmissionDisabledProducesNoSpansNoHeader is the disabled-path +// TestServerEmissionDisabledSetsNoTraceResponseHeader is the disabled-path // contract: with NO global provider installed (the shipped default when -// OTEL_EXPORTER_OTLP_ENDPOINT is unset), a PostMessage over the socket door -// produces ZERO exported spans and NO traceresponse header — otelconnect falls -// back to the global no-op provider, so no span is ever recorded and the -// interceptor finds no valid span context to stamp. -func TestServerEmissionDisabledProducesNoSpansNoHeader(t *testing.T) { +// OTEL_EXPORTER_OTLP_ENDPOINT is unset), a PostMessage over the socket door sets +// NO traceresponse header — otelconnect falls back to the global no-op provider, +// so no span is ever recorded and the interceptor finds no valid span context to +// stamp. (Span absence is proven transitively: no header can be stamped without a +// recording span; a direct in-memory-exporter check would be vacuous, since no +// code path routes spans into a non-global provider.) +func TestServerEmissionDisabledSetsNoTraceResponseHeader(t *testing.T) { ctx := context.Background() // test root (rule://go-thread-context _test.go exemption) - // Save/restore globals but install NO SDK provider: an in-memory exporter is - // attached to a provider that is NEVER set global, so it can only receive a - // span if the door wrongly recorded one against some other provider. + // Save/restore globals but install NO SDK provider: the shipped disabled + // state. otelconnect and the trace-response interceptor both read the global + // tracer provider, so with only the default no-op global in place the door can + // record no span and the interceptor finds no valid span context to stamp. prevTP := otel.GetTracerProvider() prevProp := otel.GetTextMapPropagator() - exp := tracetest.NewInMemoryExporter() - _ = sdktrace.NewTracerProvider(sdktrace.WithSyncer(exp)) // deliberately NOT set global t.Cleanup(func() { otel.SetTracerProvider(prevTP) otel.SetTextMapPropagator(prevProp) @@ -184,9 +185,9 @@ func TestServerEmissionDisabledProducesNoSpansNoHeader(t *testing.T) { resp, _ := postOverSocket(t, ctx, socketPath, channelID) - if spans := exp.GetSpans(); len(spans) != 0 { - t.Fatalf("disabled path exported %d spans, want 0", len(spans)) - } + // The real disabled-path proof: no valid span context ⇒ no traceresponse + // header. (An in-memory exporter cannot prove absence here — no code path + // routes spans into a non-global provider, so its emptiness is vacuous.) if hdr := resp.Header().Get(traceResponseHdr); hdr != "" { t.Fatalf("disabled path set traceresponse = %q, want none", hdr) }