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()) +}