From cc25bad357906c7c14abbd18764a97f81a7b9d95 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Thu, 17 Sep 2026 12:50:00 +0200 Subject: [PATCH] fix(trace): keep gRPC OTel stats handler when max_recv_msg_size is set kratos' grpc.Options assigns s.grpcOpts instead of appending, so the second grpc.Options(grpcLib.MaxRecvMsgSize(...)) call in NewGRPCServer silently wiped the otelgrpc stats handler whenever max_recv_msg_size was configured, dropping all gRPC server spans. Pass all raw grpc-go server options in a single grpc.Options call. Also add kratos tracing.Server() as the outermost middleware of the HTTP server so transcoded REST requests produce a transport-level server span instead of rooting traces at the first middleware span. Fixes chainloop-dev/chainloop#3448 Signed-off-by: Miguel Martinez Trivino --- app/controlplane/internal/server/grpc.go | 8 +- app/controlplane/internal/server/http.go | 49 ++++-- app/controlplane/internal/server/otel_test.go | 162 ++++++++++++++++++ go.mod | 2 +- 4 files changed, 202 insertions(+), 19 deletions(-) create mode 100644 app/controlplane/internal/server/otel_test.go diff --git a/app/controlplane/internal/server/grpc.go b/app/controlplane/internal/server/grpc.go index cc5a527c4..ede33825f 100644 --- a/app/controlplane/internal/server/grpc.go +++ b/app/controlplane/internal/server/grpc.go @@ -121,7 +121,6 @@ func NewGRPCServer(opts *Opts) (*grpc.Server, error) { grpc.UnaryInterceptor( protovalidateMiddleware.UnaryServerInterceptor(opts.Validator), ), - grpc.Options(grpcLib.StatsHandler(otelgrpc.NewServerHandler())), } if v := opts.ServerConfig.Grpc.Network; v != "" { @@ -148,9 +147,14 @@ func NewGRPCServer(opts *Opts) (*grpc.Server, error) { } } + // Raw grpc-go server options must be passed in a single grpc.Options call: + // kratos' grpc.Options assigns s.grpcOpts instead of appending, so a second + // call (e.g. for MaxRecvMsgSize) would silently wipe the OTel stats handler. + grpcOpts := []grpcLib.ServerOption{grpcLib.StatsHandler(otelgrpc.NewServerHandler())} if v := opts.ServerConfig.Grpc.GetMaxRecvMsgSize(); v > 0 { - serverOpts = append(serverOpts, grpc.Options(grpcLib.MaxRecvMsgSize(int(v)))) + grpcOpts = append(grpcOpts, grpcLib.MaxRecvMsgSize(int(v))) } + serverOpts = append(serverOpts, grpc.Options(grpcOpts...)) srv := grpc.NewServer(serverOpts...) v1.RegisterWorkflowServiceServer(srv, opts.WorkflowSvc) diff --git a/app/controlplane/internal/server/http.go b/app/controlplane/internal/server/http.go index a1be3af4e..80665f9e5 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/contrib/instrumentation/net/http/otelhttp" "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(httpMiddlewares(opts)...), } // Prevent unmatched routes from falling through to http.DefaultServeMux, // which would otherwise expose net/http/pprof's /debug/pprof/* endpoints @@ -66,18 +64,22 @@ func NewHTTPServer(opts *Opts, grpcSrv *grpc.Server) (*http.Server, error) { httpSrv := http.NewServer(serverOpts...) // Relax secure cookie settings in development mode opts.AuthSvc.SetDevMode(Version == "dev") - // NOTE: these non-grpc transcoded methods DO NOT RUN the middlewares - httpSrv.Handle(service.AuthLoginPath, middlewares_http.Logging(opts.Logger, opts.AuthSvc.RegisterLoginHandler())) - httpSrv.Handle(service.AuthCallbackPath, middlewares_http.Logging(opts.Logger, opts.AuthSvc.RegisterCallbackHandler())) + // NOTE: these non-grpc transcoded methods DO NOT RUN the middlewares, so + // they are wrapped with otelhttp individually to still produce server spans. + httpSrv.Handle(service.AuthLoginPath, + otelhttp.NewHandler(middlewares_http.Logging(opts.Logger, opts.AuthSvc.RegisterLoginHandler()), "HTTP "+service.AuthLoginPath)) + httpSrv.Handle(service.AuthCallbackPath, + otelhttp.NewHandler(middlewares_http.Logging(opts.Logger, opts.AuthSvc.RegisterCallbackHandler()), "HTTP "+service.AuthCallbackPath)) httpSrv.Handle(service.PrometheusMetricsPath, - middlewares_http.Logging(opts.Logger, - middlewares_http.AuthFromAuthorizationHeader( - loadJWTKeyFunc(opts.AuthConfig.GetGeneratedJwsHmacSecret()), - apiTokenCustomClaims(), - apitoken.SigningMethod, - opts.PrometheusSvc, - ), - )) + otelhttp.NewHandler( + middlewares_http.Logging(opts.Logger, + middlewares_http.AuthFromAuthorizationHeader( + loadJWTKeyFunc(opts.AuthConfig.GetGeneratedJwsHmacSecret()), + apiTokenCustomClaims(), + apitoken.SigningMethod, + opts.PrometheusSvc, + ), + ), "HTTP "+service.PrometheusMetricsPath)) statusSvc := service.NewStatusService(opts.AuthSvc.AuthURLs.Login, Version, opts.CASClientUseCase, opts.BootstrapConfig) v1.RegisterStatusServiceHTTPServer(httpSrv, statusSvc) v1.RegisterReferrerServiceHTTPServer(httpSrv, service.NewReferrerService(opts.ReferrerUseCase)) @@ -107,6 +109,21 @@ func NewHTTPServer(opts *Opts, grpcSrv *grpc.Server) (*http.Server, error) { return wrappedServer, nil } +// httpMiddlewares returns the middleware chain for the HTTP server. +// Transport-level tracing comes first: the gRPC server is instrumented via the +// otelgrpc stats handler, but transcoded HTTP requests get no server span +// otherwise, and tracing must be the outermost middleware so every other span +// (middlewares, biz, data) nests under it. Note that the handlers registered +// via httpSrv.Handle (auth login/callback, metrics) bypass this chain too, so +// they are wrapped with otelhttp individually instead. /openapi.yaml and +// /download/{digest} are kratos Context handlers that bypass the chain and +// remain untraced (low-value static/redirect endpoints). +func httpMiddlewares(opts *Opts) []middleware.Middleware { + middlewares := append([]middleware.Middleware{tracing.Server()}, craftMiddleware(opts)...) + // important, the validation middleware should be the last one + return append(middlewares, protoValidateHTTPMiddleware(opts.Validator)) +} + // 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/otel_test.go b/app/controlplane/internal/server/otel_test.go new file mode 100644 index 000000000..cf0e66fd6 --- /dev/null +++ b/app/controlplane/internal/server/otel_test.go @@ -0,0 +1,162 @@ +// +// 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" + "io" + "net/http" + "testing" + "time" + + "buf.build/go/protovalidate" + conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1" + "github.com/chainloop-dev/chainloop/app/controlplane/internal/service" + "github.com/go-kratos/kratos/v2/log" + kratoshttp "github.com/go-kratos/kratos/v2/transport/http" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + grpcLib "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/health/grpc_health_v1" +) + +// withInMemoryTracing installs an in-memory OTel tracer provider for the +// duration of the test and returns the exporter. +func withInMemoryTracing(t *testing.T) *tracetest.InMemoryExporter { + t.Helper() + exporter := tracetest.NewInMemoryExporter() + tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exporter)) + prev := otel.GetTracerProvider() + otel.SetTracerProvider(tp) + t.Cleanup(func() { otel.SetTracerProvider(prev) }) + return exporter +} + +func spanNames(exporter *tracetest.InMemoryExporter) []string { + spans := exporter.GetSpans() + names := make([]string, 0, len(spans)) + for _, s := range spans { + names = append(names, s.Name) + } + return names +} + +// The gRPC server must produce server spans even when max_recv_msg_size is +// configured. kratos' grpc.Options assigns s.grpcOpts instead of appending, +// so a second grpc.Options call in NewGRPCServer would silently drop the +// OTel stats handler. This exercises the real NewGRPCServer constructor. +func TestGRPCServerSpanWithMaxRecvMsgSize(t *testing.T) { + exporter := withInMemoryTracing(t) + + validator, err := protovalidate.New() + require.NoError(t, err) + + srv, err := NewGRPCServer(&Opts{ + ServerConfig: &conf.Server{ + Grpc: &conf.Server_GRPC{ + Addr: "127.0.0.1:0", + MaxRecvMsgSize: 8 * 1024 * 1024, + }, + }, + AuthConfig: &conf.Auth{}, + Logger: log.NewStdLogger(io.Discard), + Validator: validator, + AuthSvc: &service.AuthService{AuthURLs: &service.AuthURLs{Login: "http://localhost"}}, + }) + require.NoError(t, err) + + // Endpoint() binds the listener so the address is known before Start. + endpoint, err := srv.Endpoint() + require.NoError(t, err) + go func() { _ = srv.Start(context.Background()) }() + defer func() { _ = srv.Stop(context.Background()) }() + + conn, err := grpcLib.NewClient(endpoint.Host, grpcLib.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + defer conn.Close() + + // The request itself is rejected by the auth middleware, but the + // transport-level stats handler records the server span regardless. + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + _, _ = grpc_health_v1.NewHealthClient(conn).Check(ctx, &grpc_health_v1.HealthCheckRequest{}) + cancel() + if names := spanNames(exporter); len(names) > 0 { + assert.Contains(t, names, "grpc.health.v1.Health/Check") + return + } + time.Sleep(50 * time.Millisecond) + } + require.Fail(t, "expected gRPC server span from otelgrpc stats handler") +} + +// The HTTP server must produce a transport-level server span so transcoded +// REST requests don't root their traces at the first middleware span. This +// exercises the production httpMiddlewares chain used by NewHTTPServer. +func TestHTTPServerSpanFromTracingMiddleware(t *testing.T) { + exporter := withInMemoryTracing(t) + + validator, err := protovalidate.New() + require.NoError(t, err) + + srv := kratoshttp.NewServer( + kratoshttp.Address("127.0.0.1:0"), + kratoshttp.Middleware(httpMiddlewares(&Opts{ + AuthConfig: &conf.Auth{}, + Logger: log.NewStdLogger(io.Discard), + Validator: validator, + })...), + ) + r := srv.Route("/") + // Mirror generated handlers: kratos runs the middleware chain only when + // the handler invokes ctx.Middleware (transport/http/context.go). + r.GET("/ping", func(ctx kratoshttp.Context) error { + h := ctx.Middleware(func(context.Context, any) (any, error) { + return "ok", nil + }) + out, err := h(ctx, &grpc_health_v1.HealthCheckRequest{}) + if err != nil { + return err + } + return ctx.String(200, out.(string)) + }) + + endpoint, err := srv.Endpoint() + require.NoError(t, err) + go func() { _ = srv.Start(context.Background()) }() + defer func() { _ = srv.Stop(context.Background()) }() + + // The request is rejected by the auth middleware (401), but the tracing + // middleware is outermost and records the server span regardless. + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + resp, err := http.Get("http://" + endpoint.Host + "/ping") + if err == nil { + _ = resp.Body.Close() + } + if names := spanNames(exporter); len(names) > 0 { + assert.Contains(t, names, "/ping") + return + } + time.Sleep(50 * time.Millisecond) + } + require.Fail(t, "expected HTTP server span from tracing middleware") +} diff --git a/go.mod b/go.mod index 902c4c13d..04d2dc18f 100644 --- a/go.mod +++ b/go.mod @@ -114,6 +114,7 @@ require ( github.com/moby/moby/api v1.55.0 github.com/sergi/go-diff v1.4.0 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.70.0 + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.70.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.45.0 go.opentelemetry.io/otel/sdk v1.45.0 ) @@ -330,7 +331,6 @@ require ( gitlab.com/gitlab-org/security-products/analyzers/ruleset/v3 v3.3.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.70.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.45.0 // indirect go.opentelemetry.io/otel/metric v1.45.0 // indirect go.opentelemetry.io/otel/sdk/metric v1.45.0 // indirect