From 13edad337015df2b169fb47dee0b5c9f77f55768 Mon Sep 17 00:00:00 2001 From: Javier Rodriguez Date: Thu, 17 Sep 2026 12:45:49 +0200 Subject: [PATCH] fix(controlplane): restore OTel server spans for gRPC and HTTP The gRPC OTel stats handler was silently dropped whenever server.grpc.max_recv_msg_size was configured: Kratos' grpc.Options assigns its option slice instead of appending to it, so the second call overwrote the first and no gRPC server span was ever produced. Both raw options now travel in a single grpc.Options call. The HTTP transport had no instrumentation at all, so transcoded requests produced no server span and no traceparent extraction, leaving traces rooted at whichever application middleware ran first. Kratos' tracing.Server() middleware now heads the HTTP-only chain. Assisted-by: Claude Code Signed-off-by: Javier Rodriguez Chainloop-Trace-Sessions: 250d029e-f05e-422c-a4b2-fb568aa06da2 --- app/controlplane/cmd/wire_gen.go | 1 + app/controlplane/internal/server/grpc.go | 28 ++- app/controlplane/internal/server/http.go | 34 +++- .../internal/server/tracing_test.go | 184 ++++++++++++++++++ 4 files changed, 237 insertions(+), 10 deletions(-) create mode 100644 app/controlplane/internal/server/tracing_test.go diff --git a/app/controlplane/cmd/wire_gen.go b/app/controlplane/cmd/wire_gen.go index a06ddc2e2..c4ecb0af2 100644 --- a/app/controlplane/cmd/wire_gen.go +++ b/app/controlplane/cmd/wire_gen.go @@ -369,6 +369,7 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr BootstrapConfig: bootstrap, Credentials: readerWriter, Validator: validator, + TracerProvider: tracerProvider, } grpcServer, err := server.NewGRPCServer(opts) if err != nil { diff --git a/app/controlplane/internal/server/grpc.go b/app/controlplane/internal/server/grpc.go index cc5a527c4..d0017202a 100644 --- a/app/controlplane/internal/server/grpc.go +++ b/app/controlplane/internal/server/grpc.go @@ -49,6 +49,7 @@ import ( protovalidateMiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate" grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" + "go.opentelemetry.io/otel/trace" grpcLib "google.golang.org/grpc" ) @@ -98,6 +99,7 @@ type Opts struct { BootstrapConfig *conf.Bootstrap Credentials credentials.ReaderWriter Validator protovalidate.Validator + TracerProvider trace.TracerProvider } var ( @@ -109,6 +111,26 @@ var ( allButOrganizationOperationsSkipRegexp = regexp.MustCompile("/controlplane.v1.OrganizationService/Create|/controlplane.v1.UserService/ListMemberships|/controlplane.v1.ContextService/Current|/controlplane.v1.AuthService/DeleteAccount") ) +// rawGRPCServerOptions returns the plain gRPC server options that Kratos forwards +// to the underlying grpc-go server. +// +// They MUST all be passed in a single grpc.Options call: that Kratos option +// assigns the slice rather than appending to it, so a second call silently drops +// whatever the first one installed. +func rawGRPCServerOptions(c *conf.Server_GRPC, tp trace.TracerProvider) []grpcLib.ServerOption { + // The stats handler is what opens the transport-level server span for gRPC, + // and for grpc-web too, since grpcweb re-dispatches through grpc.Server.ServeHTTP. + options := []grpcLib.ServerOption{ + grpcLib.StatsHandler(otelgrpc.NewServerHandler(otelgrpc.WithTracerProvider(tp))), + } + + if v := c.GetMaxRecvMsgSize(); v > 0 { + options = append(options, grpcLib.MaxRecvMsgSize(int(v))) + } + + return options +} + // NewGRPCServer new a gRPC server. func NewGRPCServer(opts *Opts) (*grpc.Server, error) { // Opt-in histogram metrics for the interceptor @@ -121,7 +143,7 @@ func NewGRPCServer(opts *Opts) (*grpc.Server, error) { grpc.UnaryInterceptor( protovalidateMiddleware.UnaryServerInterceptor(opts.Validator), ), - grpc.Options(grpcLib.StatsHandler(otelgrpc.NewServerHandler())), + grpc.Options(rawGRPCServerOptions(opts.ServerConfig.GetGrpc(), opts.TracerProvider)...), } if v := opts.ServerConfig.Grpc.Network; v != "" { @@ -148,10 +170,6 @@ func NewGRPCServer(opts *Opts) (*grpc.Server, error) { } } - if v := opts.ServerConfig.Grpc.GetMaxRecvMsgSize(); v > 0 { - serverOpts = append(serverOpts, grpc.Options(grpcLib.MaxRecvMsgSize(int(v)))) - } - srv := grpc.NewServer(serverOpts...) v1.RegisterWorkflowServiceServer(srv, opts.WorkflowSvc) v1.RegisterStatusServiceServer(srv, service.NewStatusService(opts.AuthSvc.AuthURLs.Login, Version, opts.CASClientUseCase, opts.BootstrapConfig)) diff --git a/app/controlplane/internal/server/http.go b/app/controlplane/internal/server/http.go index a1be3af4e..8a4190b55 100644 --- a/app/controlplane/internal/server/http.go +++ b/app/controlplane/internal/server/http.go @@ -27,7 +27,9 @@ import ( v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1" "github.com/chainloop-dev/chainloop/app/controlplane/internal/service" "github.com/go-kratos/kratos/v2/middleware" + "github.com/go-kratos/kratos/v2/middleware/tracing" "github.com/improbable-eng/grpc-web/go/grpcweb" + "go.opentelemetry.io/otel/trace" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" @@ -40,12 +42,8 @@ import ( // NewHTTPServer new an HTTP server. func NewHTTPServer(opts *Opts, grpcSrv *grpc.Server) (*http.Server, error) { - middlewares := craftMiddleware(opts) - // important, the validation middleware should be the last one - middlewares = append(middlewares, protoValidateHTTPMiddleware(opts.Validator)) - var serverOpts = []http.ServerOption{ - http.Middleware(middlewares...), + http.Middleware(httpServerMiddlewares(opts)...), } // Prevent unmatched routes from falling through to http.DefaultServeMux, // which would otherwise expose net/http/pprof's /debug/pprof/* endpoints @@ -107,6 +105,32 @@ func NewHTTPServer(opts *Opts, grpcSrv *grpc.Server) (*http.Server, error) { return wrappedServer, nil } +// httpServerMiddlewares returns the middleware chain for the HTTP transport. +func httpServerMiddlewares(opts *Opts) []middleware.Middleware { + appMiddlewares := craftMiddleware(opts) + + middlewares := make([]middleware.Middleware, 0, len(appMiddlewares)+2) + // The tracing middleware must come first so the transport-level server span is + // the root of the trace instead of whichever application middleware runs first. + middlewares = append(middlewares, serverTracingMiddleware(opts.TracerProvider)) + middlewares = append(middlewares, appMiddlewares...) + // important, the validation middleware should be the last one + middlewares = append(middlewares, protoValidateHTTPMiddleware(opts.Validator)) + + return middlewares +} + +// serverTracingMiddleware opens the transport-level server span for HTTP requests +// and extracts any incoming trace context so the request joins the caller's trace. +// +// It is deliberately scoped to the HTTP transport: gRPC (and grpc-web, which +// re-dispatches through grpc.Server.ServeHTTP) gets its server span from the +// otelgrpc stats handler installed in NewGRPCServer, so adding this to the shared +// middleware chain would produce a duplicate span per gRPC request. +func serverTracingMiddleware(tp trace.TracerProvider) middleware.Middleware { + return tracing.Server(tracing.WithTracerProvider(tp)) +} + // Custom kraos middleware based on the protovalidate middleware // https://pkg.go.dev/github.com/grpc-ecosystem/go-grpc-middleware/v2@v2.1.0/interceptors/protovalidate#UnaryServerInterceptor // but tailored specifically for the http server diff --git a/app/controlplane/internal/server/tracing_test.go b/app/controlplane/internal/server/tracing_test.go new file mode 100644 index 000000000..7184bd242 --- /dev/null +++ b/app/controlplane/internal/server/tracing_test.go @@ -0,0 +1,184 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "context" + "net" + "net/http/httptest" + "testing" + + v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1" + conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1" + "github.com/go-kratos/kratos/v2/transport/grpc" + "github.com/go-kratos/kratos/v2/transport/http" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" + grpcLib "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +// newRecordingTracerProvider returns a TracerProvider that keeps every finished +// span in memory so tests can assert on them. +func newRecordingTracerProvider() (trace.TracerProvider, *tracetest.SpanRecorder) { + recorder := tracetest.NewSpanRecorder() + return sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder)), recorder +} + +// statusServiceStub is a minimal StatusService implementation, enough to drive a +// request through the transport under test. +type statusServiceStub struct { + v1.UnimplementedStatusServiceServer +} + +func (s *statusServiceStub) Statusz(_ context.Context, _ *v1.StatuszRequest) (*v1.StatuszResponse, error) { + return &v1.StatuszResponse{}, nil +} + +func (s *statusServiceStub) Infoz(_ context.Context, _ *v1.InfozRequest) (*v1.InfozResponse, error) { + return &v1.InfozResponse{}, nil +} + +// TestRawGRPCServerOptions covers the regression where setting +// server.grpc.max_recv_msg_size silently dropped the OTel stats handler: Kratos' +// grpc.Options assigns instead of appending, so both raw options have to travel +// in a single call. +func TestRawGRPCServerOptions(t *testing.T) { + const statuszOperation = "controlplane.v1.StatusService/Statusz" + + testCases := []struct { + name string + maxRecvMsgSize int32 + // wantResourceExhausted expects the request to be rejected for exceeding + // the configured max_recv_msg_size. + wantResourceExhausted bool + }{ + { + name: "no max_recv_msg_size configured", + maxRecvMsgSize: 0, + }, + { + name: "max_recv_msg_size configured", + maxRecvMsgSize: 1024 * 1024, + }, + { + // StatuszRequest{Readiness: true} serializes to 2 bytes, so a 1 byte + // limit proves the option is still applied alongside the stats handler. + name: "max_recv_msg_size enforced", + maxRecvMsgSize: 1, + wantResourceExhausted: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + tp, recorder := newRecordingTracerProvider() + + lis, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + srv := grpc.NewServer( + grpc.Listener(lis), + grpc.Options(rawGRPCServerOptions(&conf.Server_GRPC{MaxRecvMsgSize: tc.maxRecvMsgSize}, tp)...), + ) + v1.RegisterStatusServiceServer(srv, &statusServiceStub{}) + + go func() { _ = srv.Start(context.Background()) }() + t.Cleanup(func() { _ = srv.Stop(context.Background()) }) + + conn, err := grpcLib.NewClient(lis.Addr().String(), grpcLib.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + _, err = v1.NewStatusServiceClient(conn).Statusz(context.Background(), &v1.StatuszRequest{Readiness: true}) + if tc.wantResourceExhausted { + require.Error(t, err) + assert.Equal(t, codes.ResourceExhausted, status.Code(err)) + } else { + require.NoError(t, err) + } + + // Regardless of the outcome the transport must have opened a server span. + spans := recorder.Ended() + require.Len(t, spans, 1) + assert.Equal(t, statuszOperation, spans[0].Name()) + assert.Equal(t, trace.SpanKindServer, spans[0].SpanKind()) + }) + } +} + +// TestHTTPServerTracingMiddleware asserts the HTTP transport opens a server span +// rooted at the incoming trace context, so transcoded requests no longer root at +// whichever application middleware happens to run first. +func TestHTTPServerTracingMiddleware(t *testing.T) { + tp, recorder := newRecordingTracerProvider() + + srv := http.NewServer(http.Middleware(serverTracingMiddleware(tp))) + v1.RegisterStatusServiceHTTPServer(srv, &statusServiceStub{}) + + ts := httptest.NewServer(srv) + t.Cleanup(ts.Close) + + req := httptest.NewRequest("GET", "/statusz", nil) + req.RequestURI = "" + req.URL, _ = req.URL.Parse(ts.URL + "/statusz") + // A remote parent, to prove the propagator extracts incoming trace context. + req.Header.Set("traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01") + + resp, err := ts.Client().Do(req) + require.NoError(t, err) + t.Cleanup(func() { _ = resp.Body.Close() }) + require.Equal(t, 200, resp.StatusCode) + + spans := recorder.Ended() + require.Len(t, spans, 1) + assert.Equal(t, v1.OperationStatusServiceStatusz, spans[0].Name()) + assert.Equal(t, trace.SpanKindServer, spans[0].SpanKind()) + assert.Equal(t, "4bf92f3577b34da6a3ce929d0e0e4736", spans[0].SpanContext().TraceID().String()) + assert.Equal(t, "00f067aa0ba902b7", spans[0].Parent().SpanID().String()) +} + +// TestHTTPServerMiddlewaresOpenWithTracing pins the tracing middleware to the +// head of the HTTP chain: anything running before it would become the trace root. +func TestHTTPServerMiddlewaresOpenWithTracing(t *testing.T) { + tp, recorder := newRecordingTracerProvider() + + opts := &Opts{ + TracerProvider: tp, + AuthConfig: &conf.Auth{}, + ServerConfig: &conf.Server{}, + } + + middlewares := httpServerMiddlewares(opts) + require.NotEmpty(t, middlewares) + + srv := http.NewServer(http.Middleware(middlewares[0])) + v1.RegisterStatusServiceHTTPServer(srv, &statusServiceStub{}) + + rec := httptest.NewRecorder() + srv.ServeHTTP(rec, httptest.NewRequest("GET", "/statusz", nil)) + require.Equal(t, 200, rec.Code) + + spans := recorder.Ended() + require.Len(t, spans, 1) + assert.Equal(t, v1.OperationStatusServiceStatusz, spans[0].Name()) + assert.Equal(t, trace.SpanKindServer, spans[0].SpanKind()) +}