From 05e0019221847e41423850c0eec8c85db3dbae77 Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Tue, 28 Jul 2026 14:38:02 +0200 Subject: [PATCH 1/7] fix: incorrect connection stats for reused muxes --- .../connection_metrics_hot_reload_test.go | 372 ++++++++++++++++++ router/core/graph_server.go | 37 +- router/core/graph_server_test.go | 82 ++++ router/core/router.go | 102 ++++- router/pkg/metric/connection_pool_stats.go | 18 +- .../metric/oltp_connection_metric_store.go | 2 +- .../metric/prom_connection_metric_store.go | 2 +- 7 files changed, 572 insertions(+), 43 deletions(-) create mode 100644 router-tests/telemetry/connection_metrics_hot_reload_test.go diff --git a/router-tests/telemetry/connection_metrics_hot_reload_test.go b/router-tests/telemetry/connection_metrics_hot_reload_test.go new file mode 100644 index 0000000000..946e0b0fa4 --- /dev/null +++ b/router-tests/telemetry/connection_metrics_hot_reload_test.go @@ -0,0 +1,372 @@ +package telemetry + +import ( + "context" + "fmt" + "net/http" + "slices" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router-tests/testenv" + "github.com/wundergraph/cosmo/router-tests/testutils" + "github.com/wundergraph/cosmo/router/core" + nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" + "github.com/wundergraph/cosmo/router/pkg/config" + "github.com/wundergraph/cosmo/router/pkg/controlplane/configpoller" + "github.com/wundergraph/cosmo/router/pkg/otel" + "github.com/wundergraph/cosmo/router/pkg/routerconfig" + "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "go.uber.org/zap/zapcore" + "google.golang.org/protobuf/proto" +) + +// A reused mux keeps the transports, and therefore the connection stats, of the +// server that built it. When those stats were graph-server-scoped, the outgoing +// server's shutdown unregistered their callback and the reused mux dropped out +// of router.http.client.active_connections. +// +// Only a feature flag changes here, so the base mux is reused and every +// post-reload request goes through it. +func TestConnectionMetrics_ReusedMuxAfterHotReload(t *testing.T) { + t.Parallel() + + const featureFlag = "connection-metrics-ff" + + metricReader := metric.NewManualReader() + poller := newReloadConfigPoller(map[string]string{featureFlag: "v1"}) + + testenv.Run(t, connectionMetricsReloadEnv(metricReader, poller), func(t *testing.T, xEnv *testenv.Environment) { + // Step 1: one request opens exactly one pooled connection. + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryEmployeesSubgraph}) + requireActiveConnections(t, metricReader, connectionsBySubgraph{"employees": 1}) + + baseline := len(xEnv.Observer().All()) + + // Step 2: reload with a change that only touches the feature flag. The + // base graph is untouched, so the base mux must be reused as-is. + require.NoError(t, poller.Emit( + poller.configWithFeatureFlags(map[string]string{featureFlag: "v2"}), + &routerconfig.Changes{ + AddedConfigs: map[string]struct{}{}, + RemovedConfigs: map[string]struct{}{}, + ChangedConfigs: map[string]struct{}{featureFlag: {}}, + }, + )) + + // Guard the premise of the test: if the base mux were rebuilt, its new + // transports would feed the new server's stats and the assertion below + // would pass without exercising the reuse path at all. + require.Contains(t, muxKeysFor(xEnv, baseline, muxReusedMessage), "", + "the base mux must be reused by the new graph server") + + // Step 3: drive traffic through the reused base mux only. The rebuilt + // feature-flag mux is never asked to serve a request, so it never dials + // a subgraph and contributes nothing to the metric. + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryEmployeesSubgraph}) + + // Step 4: still exactly one. Before the fix the callback covering the + // reused mux was unregistered and this reported nothing at all. + requireActiveConnections(t, metricReader, connectionsBySubgraph{"employees": 1}) + }) +} + +// Covers muxes that appear and disappear. One flag is removed and two added, so +// the mux count changes (3 -> 4); a symmetric swap could hide bugs that only +// show up when the population grows or shrinks. +// +// Every role drives traffic to a subgraph no other mux touches, so each data +// point is attributable to one mux: base -> products, reused flag -> test1, +// removed flag -> family, added A -> mood, added B -> hobbies. The mood and +// hobbies queries resolve an Employee entity, so employees is shared and never +// asserted on. +// +// The removed flag's subgraph must fall back to zero: the stats now live for the +// whole router lifetime, so a missing decrement would accumulate forever. +func TestConnectionMetrics_AddedAndRemovedMuxesAfterHotReload(t *testing.T) { + t.Parallel() + + const ( + ffReused = "kept-flag" + ffRemoved = "dropped-flag" + ffAddedA = "new-flag-a" + ffAddedB = "new-flag-b" + ) + + metricReader := metric.NewManualReader() + poller := newReloadConfigPoller(map[string]string{ + ffReused: "v1", + ffRemoved: "v1", + }) + + testenv.Run(t, connectionMetricsReloadEnv(metricReader, poller), func(t *testing.T, xEnv *testenv.Environment) { + // Step 1: open a connection from every mux of the initial server, each + // to its own subgraph. + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryProductsSubgraph}) + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryTest1Subgraph, Header: featureFlagHeader(ffReused)}) + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryFamilySubgraph, Header: featureFlagHeader(ffRemoved)}) + + requireActiveConnections(t, metricReader, connectionsBySubgraph{ + "products": 1, "test1": 1, "family": 1, + }) + + baseline := len(xEnv.Observer().All()) + + // Step 2: drop one feature flag and add two. The base graph and the kept + // flag are unchanged, so their muxes are reused; the dropped flag's mux + // is torn down with the previous server. + require.NoError(t, poller.Emit( + poller.configWithFeatureFlags(map[string]string{ + ffReused: "v1", + ffAddedA: "v1", + ffAddedB: "v1", + }), + &routerconfig.Changes{ + AddedConfigs: map[string]struct{}{ffAddedA: {}, ffAddedB: {}}, + RemovedConfigs: map[string]struct{}{ffRemoved: {}}, + ChangedConfigs: map[string]struct{}{}, + }, + )) + + // Guard the premise: without the expected reuse/teardown split the + // assertions below would not mean what they claim. + require.ElementsMatch(t, []string{"", ffReused}, muxKeysFor(xEnv, baseline, muxReusedMessage), + "the base mux and the unchanged feature-flag mux must be reused") + require.Equal(t, []string{ffRemoved}, muxKeysFor(xEnv, baseline, muxShutdownMessage), + "only the removed feature-flag mux must be torn down") + + // The old server closed every idle connection it owned, including those of + // the muxes it handed over. Released means decremented, not gone: the stats + // are insert-only, so each subgraph keeps reporting a zero. + requireActiveConnections(t, metricReader, connectionsBySubgraph{ + "products": 0, "test1": 0, "family": 0, + }) + + // Step 4: drive traffic through every mux of the new server — the two + // reused ones and the two freshly built ones. + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryProductsSubgraph}) + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryTest1Subgraph, Header: featureFlagHeader(ffReused)}) + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryMoodSubgraph, Header: featureFlagHeader(ffAddedA)}) + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryHobbiesSubgraph, Header: featureFlagHeader(ffAddedB)}) + + // One connection per subgraph: the two reused muxes re-dialled through the + // transports they kept, the two new ones through the new server's. employees + // stays at one because both new muxes share that transport. family, whose + // mux is gone, must not come back through a stale counter. + requireActiveConnections(t, metricReader, connectionsBySubgraph{ + "products": 1, "test1": 1, "mood": 1, "hobbies": 1, "employees": 1, "family": 0, + }) + }) +} + +// The first four resolve within a single subgraph, so a connection identifies +// the mux that served the request. The last two also reach employees. +const ( + queryEmployeesSubgraph = `query { employees { id } }` + queryProductsSubgraph = `query { factTypes }` + queryTest1Subgraph = `query { headerValue(name: "X-Test") }` + queryFamilySubgraph = `query { findEmployees { id } }` + queryMoodSubgraph = `query { employee(id: 1) { currentMood } }` + queryHobbiesSubgraph = `query { employee(id: 1) { hobbies { __typename } } }` +) + +const ( + muxReusedMessage = "graph mux is being reused by new graph server, skipping shutdown" + muxShutdownMessage = "shutting down graph mux" +) + +// connectionMetricsReloadEnv builds the shared test environment. +func connectionMetricsReloadEnv(reader *metric.ManualReader, poller *reloadConfigPoller) *testenv.Config { + return &testenv.Config{ + MetricReader: reader, + MetricOptions: testenv.MetricOptions{ + EnableOTLPConnectionMetrics: true, + }, + LogObservation: testenv.LogObservationConfig{ + Enabled: true, + LogLevel: zapcore.DebugLevel, + }, + RouterOptions: []core.Option{ + // A per-subgraph transport is what puts wg.subgraph.name on the data + // points; the shared base transport dials with an empty name. + core.WithSubgraphTransportOptions(core.NewSubgraphTransportOptions(config.TrafficShapingRules{ + Subgraphs: map[string]config.GlobalSubgraphRequestRule{ + "employees": {}, + "family": {}, + "hobbies": {}, + "mood": {}, + "products": {}, + "test1": {}, + }, + })), + }, + RouterConfig: &testenv.RouterConfig{ + // File-watch reloads report Changes=nil and rebuild every mux, so we + // need a poller to emit a partial change. + ConfigPollerFactory: func(cfg *nodev1.RouterConfig) configpoller.ConfigPoller { + poller.setBaseConfig(cfg) + return poller + }, + }, + } +} + +func featureFlagHeader(flag string) http.Header { + return http.Header{"X-Feature-Flag": []string{flag}} +} + +// requireActiveConnections polls until the gauge reports exactly want. Exact +// counts matter both ways: an undercount is the bug under test, an overcount +// would mean a connection counted twice across transport generations. +func requireActiveConnections(t *testing.T, reader *metric.ManualReader, want connectionsBySubgraph) { + t.Helper() + require.EventuallyWithT(t, func(c *assert.CollectT) { + assert.Equal(c, want, activeConnections(t, reader)) + }, 10*time.Second, 50*time.Millisecond) +} + +// activeConnections sums router.http.client.active_connections per subgraph. +// Unreported subgraphs read as zero, which is also how an unregistered callback +// looks: the SDK drops an instrument that produced no observations. +func activeConnections(t *testing.T, reader *metric.ManualReader) connectionsBySubgraph { + t.Helper() + + rm := metricdata.ResourceMetrics{} + require.NoError(t, reader.Collect(context.Background(), &rm)) + + conns := connectionsBySubgraph{} + + scope := testutils.GetMetricScopeByName(rm.ScopeMetrics, "cosmo.router.connections") + if scope == nil { + return conns + } + + m := testutils.GetMetricByName(scope, "router.http.client.active_connections") + if m == nil { + return conns + } + + gauge, ok := m.Data.(metricdata.Gauge[int64]) + require.True(t, ok, "router.http.client.active_connections must be an int64 gauge") + + for _, dp := range gauge.DataPoints { + subgraph, _ := dp.Attributes.Value(otel.WgSubgraphName) + conns[subgraph.AsString()] += dp.Value + } + return conns +} + +// connectionsBySubgraph maps subgraph name to active connections. The empty key +// holds connections dialed through the shared base transport. +type connectionsBySubgraph map[string]int64 + +// String renders the whole snapshot so failures show more than the one subgraph. +func (c connectionsBySubgraph) String() string { + names := make([]string, 0, len(c)) + for name := range c { + names = append(names, name) + } + sort.Strings(names) + + parts := make([]string, 0, len(names)) + for _, name := range names { + // Label separately from the map key, which is "" for the base transport. + label := name + if label == "" { + label = "" + } + parts = append(parts, fmt.Sprintf("%s=%d", label, c[name])) + } + return "{" + strings.Join(parts, " ") + "}" +} + +// muxKeysFor returns the mux keys logged with message after the from-th entry. +// The base graph mux uses the empty string as its key. +func muxKeysFor(xEnv *testenv.Environment, from int, message string) []string { + keys := make([]string, 0) + for _, e := range xEnv.Observer().All()[from:] { + if e.Message != message { + continue + } + for _, f := range e.Context { + if f.Key == "mux" && f.Type == zapcore.StringType { + keys = append(keys, f.String) + } + } + } + slices.Sort(keys) + return keys +} + +// reloadConfigPoller serves the config testenv built for the running subgraph +// servers and lets the test emit reloads with hand-picked Changes. +type reloadConfigPoller struct { + // initialFlags maps feature flag name to execution config version for the + // config served on startup. + initialFlags map[string]string + + mu sync.Mutex + baseConfig *nodev1.RouterConfig + handler func(response *routerconfig.Response) error +} + +func newReloadConfigPoller(initialFlags map[string]string) *reloadConfigPoller { + return &reloadConfigPoller{initialFlags: initialFlags} +} + +// setBaseConfig stores the config testenv assembled for the subgraph servers. +func (p *reloadConfigPoller) setBaseConfig(cfg *nodev1.RouterConfig) { + p.mu.Lock() + defer p.mu.Unlock() + p.baseConfig = cfg +} + +func (p *reloadConfigPoller) GetRouterConfig(_ context.Context) (*routerconfig.Response, error) { + return &routerconfig.Response{Config: p.configWithFeatureFlags(p.initialFlags)}, nil +} + +func (p *reloadConfigPoller) Subscribe(_ context.Context, handler func(response *routerconfig.Response) error) { + p.mu.Lock() + defer p.mu.Unlock() + p.handler = handler +} + +// Emit invokes the Subscribe handler synchronously, so once it returns the +// previous graph server has been fully shut down. +func (p *reloadConfigPoller) Emit(cfg *nodev1.RouterConfig, changes *routerconfig.Changes) error { + p.mu.Lock() + h := p.handler + p.mu.Unlock() + + if h == nil { + return fmt.Errorf("reloadConfigPoller: Subscribe was never called by the router") + } + return h(&routerconfig.Response{Config: cfg, Changes: changes}) +} + +// configWithFeatureFlags rebuilds the base config with exactly these flags. The +// base graph part is identical across calls, which is what makes it reusable. +func (p *reloadConfigPoller) configWithFeatureFlags(flags map[string]string) *nodev1.RouterConfig { + p.mu.Lock() + defer p.mu.Unlock() + + next := proto.Clone(p.baseConfig).(*nodev1.RouterConfig) + next.FeatureFlagConfigs = &nodev1.FeatureFlagRouterExecutionConfigs{ + ConfigByFeatureFlagName: make(map[string]*nodev1.FeatureFlagRouterExecutionConfig, len(flags)), + } + for flag, ffVersion := range flags { + // Each flag mirrors the base graph so any query works through any mux. + next.FeatureFlagConfigs.ConfigByFeatureFlagName[flag] = &nodev1.FeatureFlagRouterExecutionConfig{ + Version: ffVersion, + EngineConfig: proto.Clone(p.baseConfig.GetEngineConfig()).(*nodev1.EngineConfiguration), + Subgraphs: next.GetSubgraphs(), + } + } + return next +} diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 74209fb49a..2fde2379f2 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -104,7 +104,6 @@ type ( prometheusEngineMetrics *rmetric.EngineMetrics connectionMetrics *rmetric.ConnectionMetrics instanceData InstanceData - traceDialer *TraceDialer connector *grpcconnector.Connector circuitBreakerManager *circuit.Manager headerPropagation *HeaderPropagation @@ -160,14 +159,9 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig return nil, fmt.Errorf(`the compatibility version "%s" is not compatible with this router version`, response.Config.CompatibilityVersion) } - // Active-connection tracking via TraceDialer is only needed when ConnectionStats is on. - // The httptrace-based network metrics attach in the RoundTripper and don't require the dialer. - networkStatsEnabled := r.metricConfig.OpenTelemetry.NetworkStats || r.metricConfig.Prometheus.NetworkStats - connectionStatsEnabled := networkStatsEnabled || r.metricConfig.OpenTelemetry.ConnectionStats || r.metricConfig.Prometheus.ConnectionStats - var traceDialer *TraceDialer - if connectionStatsEnabled { - traceDialer = NewTraceDialer() - } + // Nil unless ConnectionStats is on. Owned by the router, not by this server: + // muxes reused across a reload keep the transports they were built with. + traceDialer := r.connectionTraceDialer() // Build subgraph client TLS configs (mTLS for outbound subgraph connections) defaultClientTLS, perSubgraphTLS, err := buildSubgraphHTTPTLSConfigs( @@ -220,7 +214,6 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig baseTransport: baseTransport, subgraphTransports: subgraphTransports, playgroundHandler: r.playgroundHandler, - traceDialer: traceDialer, baseRouterConfigVersion: response.Config.GetVersion(), graphMuxList: make(map[string]*graphMux, 1), instanceData: InstanceData{ @@ -276,20 +269,12 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig } } - if connectionStatsEnabled { - connStore, err := rmetric.NewConnectionMetricStore( - s.logger, - nil, - s.otlpMeterProvider, - s.promMeterProvider, - s.metricConfig, - s.traceDialer.connectionPoolStats, - ) - if err != nil { - return nil, err - } - s.connectionMetrics = connStore + // Created here so the transports above have seeded the max connection counts. + connStore, err := r.connectionMetricStore(traceDialer) + if err != nil { + return nil, err } + s.connectionMetrics = connStore if err := s.setupEngineStatistics(mappedMetricAttributes); err != nil { return nil, fmt.Errorf("failed to setup engine statistics: %w", err) @@ -2287,12 +2272,6 @@ func (s *graphServer) Shutdown(ctx context.Context) error { } } - if s.connectionMetrics != nil { - if aErr := s.connectionMetrics.Shutdown(ctx); aErr != nil { - finalErr = errors.Join(finalErr, aErr) - } - } - if s.otlpEngineMetrics != nil { if err := s.otlpEngineMetrics.Shutdown(); err != nil { finalErr = errors.Join(finalErr, err) diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index 7dbeda5b94..0e6bb3abe9 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -12,6 +12,7 @@ import ( "time" "weak" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.uber.org/zap" "github.com/go-chi/chi/v5" @@ -19,6 +20,7 @@ import ( "github.com/stretchr/testify/require" nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" "github.com/wundergraph/cosmo/router/pkg/config" + rmetric "github.com/wundergraph/cosmo/router/pkg/metric" "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource" "github.com/wundergraph/cosmo/router/pkg/routerconfig" ) @@ -1007,3 +1009,83 @@ func toKeys[K cmp.Ordered, V any](m map[K]V) []K { slices.Sort(keys) return keys } + +// The connection metric store is router-scoped and unregistered exactly once, by +// Router.Shutdown, which can race with an in-flight config reload. +func TestConnectionMetricStoreLifetime(t *testing.T) { + t.Parallel() + + newTestRouter := func() *Router { + r := &Router{} + r.logger = zap.NewNop() + r.metricConfig = &rmetric.Config{ + OpenTelemetry: rmetric.OpenTelemetry{ConnectionStats: true}, + } + r.otlpMeterProvider = sdkmetric.NewMeterProvider() + r.promMeterProvider = sdkmetric.NewMeterProvider() + return r + } + + t.Run("successive graph servers share one dialer and one store", func(t *testing.T) { + t.Parallel() + + r := newTestRouter() + + firstDialer := r.connectionTraceDialer() + require.NotNil(t, firstDialer) + firstStore, err := r.connectionMetricStore(firstDialer) + require.NoError(t, err) + require.NotNil(t, firstStore) + + // A mux reused across a reload counts into the first server's stats. + secondDialer := r.connectionTraceDialer() + secondStore, err := r.connectionMetricStore(secondDialer) + require.NoError(t, err) + + require.Same(t, firstDialer, secondDialer, "the trace dialer must survive a graph server swap") + require.Same(t, firstStore, secondStore, "the connection metric store must survive a graph server swap") + }) + + t.Run("a build losing the race with shutdown gets no store", func(t *testing.T) { + t.Parallel() + + r := newTestRouter() + + dialer := r.connectionTraceDialer() + store, err := r.connectionMetricStore(dialer) + require.NoError(t, err) + require.NotNil(t, store) + + require.NoError(t, r.shutdownConnectionMetrics(context.Background())) + + // Without the flag this would register a callback nobody unregisters. + late, err := r.connectionMetricStore(dialer) + require.NoError(t, err) + require.Nil(t, late, "a graph server built after shutdown must not create a new store") + }) + + t.Run("shutdown before any graph server was built still blocks creation", func(t *testing.T) { + t.Parallel() + + r := newTestRouter() + + require.NoError(t, r.shutdownConnectionMetrics(context.Background())) + + store, err := r.connectionMetricStore(r.connectionTraceDialer()) + require.NoError(t, err) + require.Nil(t, store) + }) + + t.Run("connection stats disabled yields no dialer and no store", func(t *testing.T) { + t.Parallel() + + r := newTestRouter() + r.metricConfig = &rmetric.Config{} + + require.Nil(t, r.connectionTraceDialer()) + + store, err := r.connectionMetricStore(nil) + require.NoError(t, err) + require.Nil(t, store) + }) +} diff --git a/router/core/router.go b/router/core/router.go index 63f86afc51..5504f1e277 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -94,15 +94,19 @@ type ( // Router is the main application instance. Router struct { Config - httpServer *server - modules []Module - EngineStats statistics.EngineStatistics - playgroundHandler func(http.Handler) http.Handler - proxy ProxyFunc - disableUsageTracking bool - usage UsageTracker - headerPropagation *HeaderPropagation - reloadPersistentState *ReloadPersistentState + httpServer *server + modules []Module + EngineStats statistics.EngineStatistics + playgroundHandler func(http.Handler) http.Handler + proxy ProxyFunc + disableUsageTracking bool + usage UsageTracker + headerPropagation *HeaderPropagation + reloadPersistentState *ReloadPersistentState + connectionStatsLock sync.Mutex + traceDialer *TraceDialer + connectionMetrics *rmetric.ConnectionMetrics + connectionStatsShutdown bool } UsageTracker interface { @@ -1878,6 +1882,10 @@ func (r *Router) Shutdown(ctx context.Context) error { } } + if subErr := r.shutdownConnectionMetrics(ctx); subErr != nil { + err.Append(fmt.Errorf("failed to shutdown connection metrics: %w", subErr)) + } + var wg sync.WaitGroup if r.prometheusServer != nil { @@ -2632,6 +2640,82 @@ func WithStreamsHandlerConfiguration(cfg config.StreamsHandlerConfiguration) Opt type ProxyFunc func(req *http.Request) (*url.URL, error) +// connectionStatsEnabled reports whether any exporter asks for subgraph +// connection statistics. +func (r *Router) connectionStatsEnabled() bool { + return r.metricConfig.OpenTelemetry.ConnectionStats || + r.metricConfig.Prometheus.ConnectionStats || + r.metricConfig.OpenTelemetry.NetworkStats || + r.metricConfig.Prometheus.NetworkStats +} + +// connectionTraceDialer returns the router-scoped TraceDialer, or nil when +// connection statistics are disabled. Router-scoped because a reused graph mux +// keeps the transports, and therefore the stats, of the server that built it. +func (r *Router) connectionTraceDialer() *TraceDialer { + if !r.connectionStatsEnabled() { + return nil + } + + r.connectionStatsLock.Lock() + defer r.connectionStatsLock.Unlock() + + if r.traceDialer == nil { + r.traceDialer = NewTraceDialer() + } + return r.traceDialer +} + +// connectionMetricStore returns the router-scoped connection metric store, or +// nil when traceDialer is nil. Created on the first graph server rather than in +// setupTelemetry: it seeds the max-connections gauge from the transports, which +// do not exist yet. Shut down in Router.Shutdown. +func (r *Router) connectionMetricStore(traceDialer *TraceDialer) (*rmetric.ConnectionMetrics, error) { + if traceDialer == nil { + return nil, nil + } + + r.connectionStatsLock.Lock() + defer r.connectionStatsLock.Unlock() + + if r.connectionStatsShutdown { + // Lost the race with Router.Shutdown; this server will never serve. + return nil, nil + } + + if r.connectionMetrics == nil { + store, err := rmetric.NewConnectionMetricStore( + r.logger, + nil, + r.otlpMeterProvider, + r.promMeterProvider, + r.metricConfig, + traceDialer.connectionPoolStats, + ) + if err != nil { + return nil, err + } + r.connectionMetrics = store + } + return r.connectionMetrics, nil +} + +// shutdownConnectionMetrics unregisters the store and blocks later creation. +// The flag is needed because a config reload can still be inside newGraphServer +// here: it checks r.shutdown before calling newServer. The lock is held across +// the teardown so such a reload either creates the store first or sees the flag. +func (r *Router) shutdownConnectionMetrics(ctx context.Context) error { + r.connectionStatsLock.Lock() + defer r.connectionStatsLock.Unlock() + + r.connectionStatsShutdown = true + + if r.connectionMetrics == nil { + return nil + } + return r.connectionMetrics.Shutdown(ctx) +} + func newHTTPTransport(opts *TransportRequestOptions, proxy ProxyFunc, traceDialer *TraceDialer, subgraph string, clientTLS *tls.Config) *http.Transport { dialer := &net.Dialer{ Timeout: opts.DialTimeout, diff --git a/router/pkg/metric/connection_pool_stats.go b/router/pkg/metric/connection_pool_stats.go index 998fc5cb51..4f1c53c69c 100644 --- a/router/pkg/metric/connection_pool_stats.go +++ b/router/pkg/metric/connection_pool_stats.go @@ -1,15 +1,18 @@ package metric import ( + "maps" "sync" "sync/atomic" ) +// ConnectionPoolStats tracks active connections per subgraph host. One instance +// is shared by every subgraph transport of a router, so all state is guarded. type ConnectionPoolStats struct { mu sync.RWMutex connsPerHost map[SubgraphHostKey]*atomic.Int64 - MaxConnsPerSubgraph map[string]int64 + maxConnsPerSubgraph map[string]int64 } type SubgraphHostKey struct { @@ -20,12 +23,21 @@ type SubgraphHostKey struct { func NewConnectionPoolStats() *ConnectionPoolStats { return &ConnectionPoolStats{ connsPerHost: make(map[SubgraphHostKey]*atomic.Int64), - MaxConnsPerSubgraph: make(map[string]int64), + maxConnsPerSubgraph: make(map[string]int64), } } func (t *ConnectionPoolStats) AddSubgraphHostCount(subgraph string, maxHostCount int64) { - t.MaxConnsPerSubgraph[subgraph] = maxHostCount + t.mu.Lock() + defer t.mu.Unlock() + t.maxConnsPerSubgraph[subgraph] = maxHostCount +} + +// GetMaxConnsPerSubgraph returns a snapshot of the configured per-subgraph max. +func (t *ConnectionPoolStats) GetMaxConnsPerSubgraph() map[string]int64 { + t.mu.RLock() + defer t.mu.RUnlock() + return maps.Clone(t.maxConnsPerSubgraph) } func (t *ConnectionPoolStats) GetCounter(key SubgraphHostKey) *atomic.Int64 { diff --git a/router/pkg/metric/oltp_connection_metric_store.go b/router/pkg/metric/oltp_connection_metric_store.go index dde622f76d..d56e9f961b 100644 --- a/router/pkg/metric/oltp_connection_metric_store.go +++ b/router/pkg/metric/oltp_connection_metric_store.go @@ -53,7 +53,7 @@ func newOtlpConnectionMetrics(logger *zap.Logger, meterProvider *metric.MeterPro } func (h *otlpConnectionMetrics) startInitMetrics(connStats *ConnectionPoolStats, attributes []attribute.KeyValue) error { - for subgraph, maxConns := range connStats.MaxConnsPerSubgraph { + for subgraph, maxConns := range connStats.GetMaxConnsPerSubgraph() { attrs := make([]attribute.KeyValue, 0, 1) if subgraph != "" { attrs = append(attrs, otel.WgSubgraphName.String(subgraph)) diff --git a/router/pkg/metric/prom_connection_metric_store.go b/router/pkg/metric/prom_connection_metric_store.go index 8ce1dfcc8b..80cdd580f9 100644 --- a/router/pkg/metric/prom_connection_metric_store.go +++ b/router/pkg/metric/prom_connection_metric_store.go @@ -53,7 +53,7 @@ func newPromConnectionMetrics(logger *zap.Logger, meterProvider *metric.MeterPro } func (h *promConnectionMetrics) startInitMetrics(connStats *ConnectionPoolStats, attributes []attribute.KeyValue) error { - for subgraph, maxConns := range connStats.MaxConnsPerSubgraph { + for subgraph, maxConns := range connStats.GetMaxConnsPerSubgraph() { attrs := make([]attribute.KeyValue, 0, 1) if subgraph != "" { attrs = append(attrs, otel.WgSubgraphName.String(subgraph)) From f00f6108099560ac4c64fe860d4bacbad0103173 Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Tue, 28 Jul 2026 15:37:46 +0200 Subject: [PATCH 2/7] chore: improve --- .../connection_metrics_hot_reload_test.go | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/router-tests/telemetry/connection_metrics_hot_reload_test.go b/router-tests/telemetry/connection_metrics_hot_reload_test.go index 946e0b0fa4..4b7795da72 100644 --- a/router-tests/telemetry/connection_metrics_hot_reload_test.go +++ b/router-tests/telemetry/connection_metrics_hot_reload_test.go @@ -227,20 +227,23 @@ func featureFlagHeader(flag string) http.Header { func requireActiveConnections(t *testing.T, reader *metric.ManualReader, want connectionsBySubgraph) { t.Helper() require.EventuallyWithT(t, func(c *assert.CollectT) { - assert.Equal(c, want, activeConnections(t, reader)) + assert.Equal(c, want, activeConnections(c, reader)) }, 10*time.Second, 50*time.Millisecond) } // activeConnections sums router.http.client.active_connections per subgraph. // Unreported subgraphs read as zero, which is also how an unregistered callback // looks: the SDK drops an instrument that produced no observations. -func activeConnections(t *testing.T, reader *metric.ManualReader) connectionsBySubgraph { - t.Helper() +// +// Takes an assert.TestingT because it runs on EventuallyWithT's goroutine, +// where a require-style FailNow is not allowed. +func activeConnections(t assert.TestingT, reader *metric.ManualReader) connectionsBySubgraph { + conns := connectionsBySubgraph{} rm := metricdata.ResourceMetrics{} - require.NoError(t, reader.Collect(context.Background(), &rm)) - - conns := connectionsBySubgraph{} + if !assert.NoError(t, reader.Collect(context.Background(), &rm)) { + return conns + } scope := testutils.GetMetricScopeByName(rm.ScopeMetrics, "cosmo.router.connections") if scope == nil { @@ -253,7 +256,9 @@ func activeConnections(t *testing.T, reader *metric.ManualReader) connectionsByS } gauge, ok := m.Data.(metricdata.Gauge[int64]) - require.True(t, ok, "router.http.client.active_connections must be an int64 gauge") + if !assert.True(t, ok, "router.http.client.active_connections must be an int64 gauge") { + return conns + } for _, dp := range gauge.DataPoints { subgraph, _ := dp.Attributes.Value(otel.WgSubgraphName) From 84e33b864a256e46e889e4bd799dd718fa9bd8b4 Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Tue, 28 Jul 2026 15:55:43 +0200 Subject: [PATCH 3/7] chore: improve --- router/core/graph_server_test.go | 4 ++- router/core/router.go | 42 ++++++++++++++------------------ 2 files changed, 21 insertions(+), 25 deletions(-) diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index 0e6bb3abe9..49f48be68f 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -1056,9 +1056,10 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { require.NoError(t, err) require.NotNil(t, store) + // Mirrors Router.Shutdown: the flag is set before the teardown runs. + r.shutdown.Store(true) require.NoError(t, r.shutdownConnectionMetrics(context.Background())) - // Without the flag this would register a callback nobody unregisters. late, err := r.connectionMetricStore(dialer) require.NoError(t, err) require.Nil(t, late, "a graph server built after shutdown must not create a new store") @@ -1069,6 +1070,7 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { r := newTestRouter() + r.shutdown.Store(true) require.NoError(t, r.shutdownConnectionMetrics(context.Background())) store, err := r.connectionMetricStore(r.connectionTraceDialer()) diff --git a/router/core/router.go b/router/core/router.go index 5504f1e277..69fc36769f 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -94,19 +94,18 @@ type ( // Router is the main application instance. Router struct { Config - httpServer *server - modules []Module - EngineStats statistics.EngineStatistics - playgroundHandler func(http.Handler) http.Handler - proxy ProxyFunc - disableUsageTracking bool - usage UsageTracker - headerPropagation *HeaderPropagation - reloadPersistentState *ReloadPersistentState - connectionStatsLock sync.Mutex - traceDialer *TraceDialer - connectionMetrics *rmetric.ConnectionMetrics - connectionStatsShutdown bool + httpServer *server + modules []Module + EngineStats statistics.EngineStatistics + playgroundHandler func(http.Handler) http.Handler + proxy ProxyFunc + disableUsageTracking bool + usage UsageTracker + headerPropagation *HeaderPropagation + reloadPersistentState *ReloadPersistentState + connectionStatsLock sync.Mutex + traceDialer *TraceDialer + connectionMetrics *rmetric.ConnectionMetrics } UsageTracker interface { @@ -2652,14 +2651,12 @@ func (r *Router) connectionStatsEnabled() bool { // connectionTraceDialer returns the router-scoped TraceDialer, or nil when // connection statistics are disabled. Router-scoped because a reused graph mux // keeps the transports, and therefore the stats, of the server that built it. +// Unlocked: only newGraphServer touches the dialer, and it never runs concurrently. func (r *Router) connectionTraceDialer() *TraceDialer { if !r.connectionStatsEnabled() { return nil } - r.connectionStatsLock.Lock() - defer r.connectionStatsLock.Unlock() - if r.traceDialer == nil { r.traceDialer = NewTraceDialer() } @@ -2678,8 +2675,8 @@ func (r *Router) connectionMetricStore(traceDialer *TraceDialer) (*rmetric.Conne r.connectionStatsLock.Lock() defer r.connectionStatsLock.Unlock() - if r.connectionStatsShutdown { - // Lost the race with Router.Shutdown; this server will never serve. + // Check if the router is shutting down to prevent creating asynchronous metrics. + if r.shutdown.Load() { return nil, nil } @@ -2700,16 +2697,13 @@ func (r *Router) connectionMetricStore(traceDialer *TraceDialer) (*rmetric.Conne return r.connectionMetrics, nil } -// shutdownConnectionMetrics unregisters the store and blocks later creation. -// The flag is needed because a config reload can still be inside newGraphServer -// here: it checks r.shutdown before calling newServer. The lock is held across -// the teardown so such a reload either creates the store first or sees the flag. +// shutdownConnectionMetrics unregisters the store. Must run after r.shutdown is +// set, so that a config reload still inside newGraphServer cannot create a new +// store afterwards. func (r *Router) shutdownConnectionMetrics(ctx context.Context) error { r.connectionStatsLock.Lock() defer r.connectionStatsLock.Unlock() - r.connectionStatsShutdown = true - if r.connectionMetrics == nil { return nil } From e26771549aee176e4cdc96d7246bca213a5f887d Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Wed, 29 Jul 2026 15:54:46 +0200 Subject: [PATCH 4/7] chore: record max connections on each graph server creation --- router/core/graph_server.go | 2 +- router/core/graph_server_test.go | 12 ++++++------ router/core/router.go | 6 +++++- router/pkg/metric/connection_metric_store.go | 14 ++++++++++++++ router/pkg/metric/oltp_connection_metric_store.go | 8 -------- router/pkg/metric/prom_connection_metric_store.go | 8 -------- 6 files changed, 26 insertions(+), 24 deletions(-) diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 2fde2379f2..3cc45d5f9e 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -270,7 +270,7 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig } // Created here so the transports above have seeded the max connection counts. - connStore, err := r.connectionMetricStore(traceDialer) + connStore, err := r.connectionMetricStore(routerCtx, traceDialer) if err != nil { return nil, err } diff --git a/router/core/graph_server_test.go b/router/core/graph_server_test.go index 49f48be68f..7894d7fe5a 100644 --- a/router/core/graph_server_test.go +++ b/router/core/graph_server_test.go @@ -1033,13 +1033,13 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { firstDialer := r.connectionTraceDialer() require.NotNil(t, firstDialer) - firstStore, err := r.connectionMetricStore(firstDialer) + firstStore, err := r.connectionMetricStore(t.Context(), firstDialer) require.NoError(t, err) require.NotNil(t, firstStore) // A mux reused across a reload counts into the first server's stats. secondDialer := r.connectionTraceDialer() - secondStore, err := r.connectionMetricStore(secondDialer) + secondStore, err := r.connectionMetricStore(t.Context(), secondDialer) require.NoError(t, err) require.Same(t, firstDialer, secondDialer, "the trace dialer must survive a graph server swap") @@ -1052,7 +1052,7 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { r := newTestRouter() dialer := r.connectionTraceDialer() - store, err := r.connectionMetricStore(dialer) + store, err := r.connectionMetricStore(t.Context(), dialer) require.NoError(t, err) require.NotNil(t, store) @@ -1060,7 +1060,7 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { r.shutdown.Store(true) require.NoError(t, r.shutdownConnectionMetrics(context.Background())) - late, err := r.connectionMetricStore(dialer) + late, err := r.connectionMetricStore(t.Context(), dialer) require.NoError(t, err) require.Nil(t, late, "a graph server built after shutdown must not create a new store") }) @@ -1073,7 +1073,7 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { r.shutdown.Store(true) require.NoError(t, r.shutdownConnectionMetrics(context.Background())) - store, err := r.connectionMetricStore(r.connectionTraceDialer()) + store, err := r.connectionMetricStore(t.Context(), r.connectionTraceDialer()) require.NoError(t, err) require.Nil(t, store) }) @@ -1086,7 +1086,7 @@ func TestConnectionMetricStoreLifetime(t *testing.T) { require.Nil(t, r.connectionTraceDialer()) - store, err := r.connectionMetricStore(nil) + store, err := r.connectionMetricStore(t.Context(), nil) require.NoError(t, err) require.Nil(t, store) }) diff --git a/router/core/router.go b/router/core/router.go index 69fc36769f..328c23de88 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -2667,7 +2667,7 @@ func (r *Router) connectionTraceDialer() *TraceDialer { // nil when traceDialer is nil. Created on the first graph server rather than in // setupTelemetry: it seeds the max-connections gauge from the transports, which // do not exist yet. Shut down in Router.Shutdown. -func (r *Router) connectionMetricStore(traceDialer *TraceDialer) (*rmetric.ConnectionMetrics, error) { +func (r *Router) connectionMetricStore(ctx context.Context, traceDialer *TraceDialer) (*rmetric.ConnectionMetrics, error) { if traceDialer == nil { return nil, nil } @@ -2694,6 +2694,10 @@ func (r *Router) connectionMetricStore(traceDialer *TraceDialer) (*rmetric.Conne } r.connectionMetrics = store } + + // Ensure to record max connections on each graph server creation. + r.connectionMetrics.RecordMaxConnections(ctx, traceDialer.connectionPoolStats) + return r.connectionMetrics, nil } diff --git a/router/pkg/metric/connection_metric_store.go b/router/pkg/metric/connection_metric_store.go index 68af12e92f..3800ab04db 100644 --- a/router/pkg/metric/connection_metric_store.go +++ b/router/pkg/metric/connection_metric_store.go @@ -78,6 +78,20 @@ func NewConnectionMetricStore( return connMetrics, nil } +// RecordMaxConnections records the configured per-subgraph connection ceiling. +// Must be called on every graph server build. +func (c *ConnectionMetrics) RecordMaxConnections(ctx context.Context, connectionPoolStats *ConnectionPoolStats) { + for subgraph, maxConns := range connectionPoolStats.GetMaxConnsPerSubgraph() { + attrs := make([]attribute.KeyValue, 0, 1) + if subgraph != "" { + attrs = append(attrs, otel.WgSubgraphName.String(subgraph)) + } + opts := otelmetric.WithAttributes(attrs...) + c.otlpConnectionMetrics.MeasureMaxConnections(ctx, maxConns, opts) + c.promConnectionMetrics.MeasureMaxConnections(ctx, maxConns, opts) + } +} + func (c *ConnectionMetrics) MeasureMaxConnections(ctx context.Context, reused bool, attrs ...attribute.KeyValue) { // Add the reused attribute to the base attributes reusedAttr := otel.WgClientReusedConnection.Bool(reused) diff --git a/router/pkg/metric/oltp_connection_metric_store.go b/router/pkg/metric/oltp_connection_metric_store.go index d56e9f961b..0846ec7692 100644 --- a/router/pkg/metric/oltp_connection_metric_store.go +++ b/router/pkg/metric/oltp_connection_metric_store.go @@ -53,14 +53,6 @@ func newOtlpConnectionMetrics(logger *zap.Logger, meterProvider *metric.MeterPro } func (h *otlpConnectionMetrics) startInitMetrics(connStats *ConnectionPoolStats, attributes []attribute.KeyValue) error { - for subgraph, maxConns := range connStats.GetMaxConnsPerSubgraph() { - attrs := make([]attribute.KeyValue, 0, 1) - if subgraph != "" { - attrs = append(attrs, otel.WgSubgraphName.String(subgraph)) - } - h.MeasureMaxConnections(context.Background(), maxConns, otelmetric.WithAttributes(attrs...)) - } - rc, err := h.meter.RegisterCallback(func(_ context.Context, o otelmetric.Observer) error { stats := connStats.GetStats() for key, activeConnections := range stats { diff --git a/router/pkg/metric/prom_connection_metric_store.go b/router/pkg/metric/prom_connection_metric_store.go index 80cdd580f9..c16c7b9b59 100644 --- a/router/pkg/metric/prom_connection_metric_store.go +++ b/router/pkg/metric/prom_connection_metric_store.go @@ -53,14 +53,6 @@ func newPromConnectionMetrics(logger *zap.Logger, meterProvider *metric.MeterPro } func (h *promConnectionMetrics) startInitMetrics(connStats *ConnectionPoolStats, attributes []attribute.KeyValue) error { - for subgraph, maxConns := range connStats.GetMaxConnsPerSubgraph() { - attrs := make([]attribute.KeyValue, 0, 1) - if subgraph != "" { - attrs = append(attrs, otel.WgSubgraphName.String(subgraph)) - } - h.MeasureMaxConnections(context.Background(), maxConns, otelmetric.WithAttributes(attrs...)) - } - rc, err := h.meter.RegisterCallback(func(_ context.Context, o otelmetric.Observer) error { stats := connStats.GetStats() for key, activeConnections := range stats { From 2862fc01bbf3f113165953fe9c5d05f56c52d785 Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Wed, 29 Jul 2026 15:58:01 +0200 Subject: [PATCH 5/7] chore: improve naming --- router/core/router.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/router/core/router.go b/router/core/router.go index 328c23de88..dfd262cdde 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -103,9 +103,9 @@ type ( usage UsageTracker headerPropagation *HeaderPropagation reloadPersistentState *ReloadPersistentState - connectionStatsLock sync.Mutex - traceDialer *TraceDialer + connectionMetricsLock sync.Mutex connectionMetrics *rmetric.ConnectionMetrics + traceDialer *TraceDialer } UsageTracker interface { @@ -2672,8 +2672,8 @@ func (r *Router) connectionMetricStore(ctx context.Context, traceDialer *TraceDi return nil, nil } - r.connectionStatsLock.Lock() - defer r.connectionStatsLock.Unlock() + r.connectionMetricsLock.Lock() + defer r.connectionMetricsLock.Unlock() // Check if the router is shutting down to prevent creating asynchronous metrics. if r.shutdown.Load() { @@ -2705,8 +2705,8 @@ func (r *Router) connectionMetricStore(ctx context.Context, traceDialer *TraceDi // set, so that a config reload still inside newGraphServer cannot create a new // store afterwards. func (r *Router) shutdownConnectionMetrics(ctx context.Context) error { - r.connectionStatsLock.Lock() - defer r.connectionStatsLock.Unlock() + r.connectionMetricsLock.Lock() + defer r.connectionMetricsLock.Unlock() if r.connectionMetrics == nil { return nil From f2b41c2632ffb7bbd1ca64d3fb602847ca86a311 Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Thu, 30 Jul 2026 11:26:46 +0200 Subject: [PATCH 6/7] chore: remove dead code and correctly set attributes --- router/pkg/metric/connection_metric_store.go | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/router/pkg/metric/connection_metric_store.go b/router/pkg/metric/connection_metric_store.go index 3800ab04db..64dd794ee3 100644 --- a/router/pkg/metric/connection_metric_store.go +++ b/router/pkg/metric/connection_metric_store.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "github.com/wundergraph/cosmo/router/pkg/otel" @@ -86,25 +87,12 @@ func (c *ConnectionMetrics) RecordMaxConnections(ctx context.Context, connection if subgraph != "" { attrs = append(attrs, otel.WgSubgraphName.String(subgraph)) } - opts := otelmetric.WithAttributes(attrs...) + opts := c.recordOpts(attrs) c.otlpConnectionMetrics.MeasureMaxConnections(ctx, maxConns, opts) c.promConnectionMetrics.MeasureMaxConnections(ctx, maxConns, opts) } } -func (c *ConnectionMetrics) MeasureMaxConnections(ctx context.Context, reused bool, attrs ...attribute.KeyValue) { - // Add the reused attribute to the base attributes - reusedAttr := otel.WgClientReusedConnection.Bool(reused) - allAttrs := append([]attribute.KeyValue{}, c.baseAttributes...) - allAttrs = append(allAttrs, reusedAttr) - allAttrs = append(allAttrs, attrs...) - - opts := otelmetric.WithAttributes(allAttrs...) - - c.otlpConnectionMetrics.MeasureMaxConnections(ctx, 1, opts) - c.promConnectionMetrics.MeasureMaxConnections(ctx, 1, opts) -} - func (c *ConnectionMetrics) MeasureConnectionAcquireDuration(ctx context.Context, duration float64, attrs ...attribute.KeyValue) { opts := c.recordOpts(attrs) c.otlpConnectionMetrics.MeasureConnectionAcquireDuration(ctx, duration, opts) @@ -142,8 +130,7 @@ func (c *ConnectionMetrics) MeasureTimeToFirstByte(ctx context.Context, duration } func (c *ConnectionMetrics) recordOpts(attrs []attribute.KeyValue) otelmetric.RecordOption { - copied := append([]attribute.KeyValue{}, c.baseAttributes...) - return otelmetric.WithAttributes(append(copied, attrs...)...) + return otelmetric.WithAttributes(append(slices.Clone(c.baseAttributes), attrs...)...) } // Flush flushes the metrics to the backend synchronously. From e2ddaa6d95cb60406676c8ad5b4510a5bc4286fc Mon Sep 17 00:00:00 2001 From: Ludwig Bedacht Date: Thu, 30 Jul 2026 13:02:10 +0200 Subject: [PATCH 7/7] chore: add test for new feature flag in execution config --- .../connection_metrics_hot_reload_test.go | 102 +++++++++++++++++- 1 file changed, 101 insertions(+), 1 deletion(-) diff --git a/router-tests/telemetry/connection_metrics_hot_reload_test.go b/router-tests/telemetry/connection_metrics_hot_reload_test.go index 4b7795da72..a490b1e4ff 100644 --- a/router-tests/telemetry/connection_metrics_hot_reload_test.go +++ b/router-tests/telemetry/connection_metrics_hot_reload_test.go @@ -165,6 +165,71 @@ func TestConnectionMetrics_AddedAndRemovedMuxesAfterHotReload(t *testing.T) { }) } +// A feature flag can route to a subgraph that no earlier execution config +// mentioned. Here myff introduces products_fg, which the base graph never +// references, so its first connection is opened by a mux that did not exist +// before the reload. +// +// Its data points still carry wg.subgraph.name because the transport set comes +// from static router configuration, not from the execution config: products_fg +// has a traffic-shaping entry, so its transport exists from boot even while no +// config mentions the subgraph. A subgraph without such an entry would instead +// share the unnamed base transport. +func TestConnectionMetrics_SubgraphAddedByFeatureFlagAfterHotReload(t *testing.T) { + t.Parallel() + + const ( + featureFlag = "myff" + // Routed to by the myff flag only; the base graph uses products instead. + newSubgraph = "products_fg" + ) + + metricReader := metric.NewManualReader() + // No feature flags initially, so nothing references products_fg yet. + poller := newReloadConfigPoller(nil) + + testenv.Run(t, connectionMetricsReloadEnv(metricReader, poller), func(t *testing.T, xEnv *testenv.Environment) { + // Guard the premise: the router must not have heard of products_fg yet, or + // the reload below is not introducing anything. + served, err := poller.GetRouterConfig(context.Background()) + require.NoError(t, err) + require.NotContains(t, subgraphNames(served.Config), newSubgraph, + "%s must be absent from the config the router booted with", newSubgraph) + + // Step 1: the base graph resolves factTypes through products. + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{Query: queryProductsSubgraph}) + requireActiveConnections(t, metricReader, connectionsBySubgraph{"products": 1}) + + baseline := len(xEnv.Observer().All()) + + next := poller.configWithNativeFeatureFlags(featureFlag) + // Guard the fixture: without the real flag config there is no new subgraph. + require.Contains(t, subgraphNames(next), newSubgraph, + "the %s flag config must introduce the %s subgraph", featureFlag, newSubgraph) + + // Step 2: add the flag. The base graph is untouched, so its mux is reused. + require.NoError(t, poller.Emit(next, &routerconfig.Changes{ + AddedConfigs: map[string]struct{}{featureFlag: {}}, + RemovedConfigs: map[string]struct{}{}, + ChangedConfigs: map[string]struct{}{}, + })) + require.Contains(t, muxKeysFor(xEnv, baseline, muxReusedMessage), "", + "the base mux must be reused by the new graph server") + + // Step 3: the same query through the new mux resolves through products_fg. + xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ + Query: queryProductsSubgraph, + Header: featureFlagHeader(featureFlag), + }) + + // products dropped to zero because the outgoing server closed its idle + // connections and nothing has asked the reused base mux for more. + requireActiveConnections(t, metricReader, connectionsBySubgraph{ + "products": 0, newSubgraph: 1, + }) + }) +} + // The first four resolve within a single subgraph, so a connection identifies // the mux that served the request. The last two also reach employees. const ( @@ -202,7 +267,9 @@ func connectionMetricsReloadEnv(reader *metric.ManualReader, poller *reloadConfi "hobbies": {}, "mood": {}, "products": {}, - "test1": {}, + // Only referenced by the myff feature flag, never by the base graph. + "products_fg": {}, + "test1": {}, }, })), }, @@ -291,6 +358,21 @@ func (c connectionsBySubgraph) String() string { return "{" + strings.Join(parts, " ") + "}" } +// subgraphNames lists every subgraph the config names, base graph and feature +// flags alike, i.e. everything the router could route to. +func subgraphNames(cfg *nodev1.RouterConfig) []string { + names := make([]string, 0) + for _, sg := range cfg.GetSubgraphs() { + names = append(names, sg.GetName()) + } + for _, ff := range cfg.GetFeatureFlagConfigs().GetConfigByFeatureFlagName() { + for _, sg := range ff.GetSubgraphs() { + names = append(names, sg.GetName()) + } + } + return names +} + // muxKeysFor returns the mux keys logged with message after the from-th entry. // The base graph mux uses the empty string as its key. func muxKeysFor(xEnv *testenv.Environment, from int, message string) []string { @@ -355,6 +437,24 @@ func (p *reloadConfigPoller) Emit(cfg *nodev1.RouterConfig, changes *routerconfi return h(&routerconfig.Response{Config: cfg, Changes: changes}) } +// configWithNativeFeatureFlags rebuilds the base config keeping only the named +// feature flags as testenv assembled them, so the subgraphs they route to are the +// real ones rather than copies of the base graph. +func (p *reloadConfigPoller) configWithNativeFeatureFlags(names ...string) *nodev1.RouterConfig { + p.mu.Lock() + defer p.mu.Unlock() + + next := proto.Clone(p.baseConfig).(*nodev1.RouterConfig) + kept := make(map[string]*nodev1.FeatureFlagRouterExecutionConfig, len(names)) + for _, name := range names { + if cfg, ok := next.GetFeatureFlagConfigs().GetConfigByFeatureFlagName()[name]; ok { + kept[name] = cfg + } + } + next.FeatureFlagConfigs = &nodev1.FeatureFlagRouterExecutionConfigs{ConfigByFeatureFlagName: kept} + return next +} + // configWithFeatureFlags rebuilds the base config with exactly these flags. The // base graph part is identical across calls, which is what makes it reusable. func (p *reloadConfigPoller) configWithFeatureFlags(flags map[string]string) *nodev1.RouterConfig {