From 84f798f921dfe7ede8512e122d6808f49fe85fb1 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 18:11:44 +0200 Subject: [PATCH 01/13] fix(router): release schema refs on config reload to prevent memory leak Stop storing schemaDocument in cached planWithMetaData entries so plan caches no longer pin the old router schema AST (~200MB) after CDN reloads. Also call OnRouterConfigReload before building a new graph server so slow-plan cache entries are extracted while the old graphMux is still referenced, matching the supervisor restart path. Co-authored-by: Cursor --- router/core/operation_planner.go | 7 +++---- router/core/router.go | 4 ++++ 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/router/core/operation_planner.go b/router/core/operation_planner.go index f9da57396f..037b23021a 100644 --- a/router/core/operation_planner.go +++ b/router/core/operation_planner.go @@ -18,9 +18,9 @@ import ( ) type planWithMetaData struct { - preparedPlan plan.Plan - operationDocument, schemaDocument *ast.Document - typeFieldUsageInfo []*graphqlschemausage.TypeFieldUsageInfo + preparedPlan plan.Plan + operationDocument *ast.Document + typeFieldUsageInfo []*graphqlschemausage.TypeFieldUsageInfo argumentUsageInfo []*graphqlmetricsv1.ArgumentUsageInfo content string operationName string @@ -96,7 +96,6 @@ func (p *OperationPlanner) planOperation(content string, name string, includeQue return &planWithMetaData{ preparedPlan: preparedPlan, operationDocument: &doc, - schemaDocument: p.executor.RouterSchema, }, nil } diff --git a/router/core/router.go b/router/core/router.go index 5dd087acd1..7e6dff4076 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -627,6 +627,10 @@ func (r *Router) serverTLSConfig() (*tls.Config, error) { // newGraphServer creates a new server. func (r *Router) newServer(ctx context.Context, response *routerconfig.Response) error { + // Extract slow-plan cache entries before building the new graph server, which + // overwrites ReloadPersistentState cache references and before the old graphMux shuts down. + r.reloadPersistentState.OnRouterConfigReload() + server, err := newGraphServer(ctx, r, response, r.proxy) if err != nil { r.logger.Error("Failed to create graph server. Keeping the old server", zap.Error(err)) From bfb52449bad5a74fc68f3b4d3897594f9bb3b9f7 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 19:09:27 +0200 Subject: [PATCH 02/13] fix(router): drain WS subs and skip plan cache OnEvict on mux shutdown Disable ristretto OnEvict migration into slowplancache when a graphMux is shutting down, since Close() clears every entry and the fallback cache is about to be closed anyway. Close websocket subscriptions synchronously before plan caches so preparedPlan and executor refs are released first. Co-authored-by: Cursor --- router/core/graph_server.go | 20 +++++++++++++++++++- router/core/websocket.go | 11 ++++++++++- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/router/core/graph_server.go b/router/core/graph_server.go index af206997be..8317191694 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -681,6 +681,9 @@ type graphMux struct { mux *chi.Mux reused atomic.Bool + wsHandler *WebsocketHandler + planCacheOnEvictEnabled atomic.Bool + planCache *ristretto.Cache[uint64, *planWithMetaData] planFallbackCache *slowplancache.Cache[*planWithMetaData] persistedOperationCache *ristretto.Cache[uint64, NormalizationCacheEntry] @@ -721,10 +724,14 @@ func (s *graphMux) buildOperationCaches(srv *graphServer) (computeSha256 bool, e BufferItems: 64, } if srv.cacheWarmup != nil && srv.cacheWarmup.Enabled && srv.cacheWarmup.InMemoryFallback { + s.planCacheOnEvictEnabled.Store(true) planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { // This could be called before planFallbackCache is set, but it's not a problem // because there is a nil guard inside, as well as items should not really be evicted // on startup + if !s.planCacheOnEvictEnabled.Load() { + return + } s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) } } @@ -958,10 +965,20 @@ func (s *graphMux) configureCacheMetrics(srv *graphServer, baseOtelAttributes [] } func (s *graphMux) Shutdown(ctx context.Context) error { + // Close websocket subscriptions synchronously before tearing down plan caches so + // active preparedPlan and executor references are released first. + if s.wsHandler != nil { + s.wsHandler.ShutdownConnections() + } + // cancel the graph muxes context to close its resources like websocket connections, resolvers, etc. s.cancel() + // ristretto Close() clears all entries and invokes OnEvict for each one. Disable + // migration into the slow-plan fallback cache during intentional mux shutdown. + s.planCacheOnEvictEnabled.Store(false) s.planCache.Close() + s.planFallbackCache.Wait() s.planFallbackCache.Close() s.persistedOperationCache.Close() s.normalizationCache.Close() @@ -1830,7 +1847,7 @@ func (s *graphServer) buildGraphMux( }) if s.webSocketConfiguration != nil && s.webSocketConfiguration.Enabled { - wsMiddleware := NewWebsocketMiddleware(graphMuxCtx, WebsocketMiddlewareOptions{ + wsMiddleware, wsHandler := NewWebsocketMiddleware(graphMuxCtx, WebsocketMiddlewareOptions{ OperationProcessor: operationProcessor, OperationBlocker: operationBlocker, Planner: operationPlanner, @@ -1850,6 +1867,7 @@ func (s *graphServer) buildGraphMux( DisableVariablesRemapping: s.engineExecutionConfiguration.DisableVariablesRemapping, ApolloCompatibilityFlags: s.apolloCompatibilityFlags, }) + gm.wsHandler = wsHandler // When the playground path is equal to the graphql path, we need to handle // ws upgrades and html requests on the same route. diff --git a/router/core/websocket.go b/router/core/websocket.go index 95f83864b6..e4cf5b85a0 100644 --- a/router/core/websocket.go +++ b/router/core/websocket.go @@ -70,7 +70,7 @@ type WebsocketMiddlewareOptions struct { ApolloCompatibilityFlags config.ApolloCompatibilityFlags } -func NewWebsocketMiddleware(ctx context.Context, opts WebsocketMiddlewareOptions) func(http.Handler) http.Handler { +func NewWebsocketMiddleware(ctx context.Context, opts WebsocketMiddlewareOptions) (func(http.Handler) http.Handler, *WebsocketHandler) { handler := &WebsocketHandler{ ctx: ctx, operationProcessor: opts.OperationProcessor, @@ -148,7 +148,16 @@ func NewWebsocketMiddleware(ctx context.Context, opts WebsocketMiddlewareOptions } handler.handleUpgradeRequest(w, r) }) + }, handler +} + +// ShutdownConnections closes all active websocket connections and unsubscribes +// any live GraphQL subscriptions before graph mux caches are torn down. +func (h *WebsocketHandler) ShutdownConnections() { + if h == nil { + return } + h.closeAllConnections() } // wsConnectionWrapper is a wrapper around websocket.Conn that allows From b6ce7cddf550a78f4cdd04c4c7b605d9d8169932 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 19:25:20 +0200 Subject: [PATCH 03/13] fix(router): register heap pprof handlers for in-use profiling Expose /debug/pprof/heap and related routes on the pprof server so forced-GC heap snapshots (heap?gc=1) work for memory leak diagnosis. Co-authored-by: Cursor --- router/pkg/profile/profile.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/router/pkg/profile/profile.go b/router/pkg/profile/profile.go index 8ee0239848..513505b6a8 100644 --- a/router/pkg/profile/profile.go +++ b/router/pkg/profile/profile.go @@ -43,6 +43,9 @@ func NewServer(addr string, log *zap.Logger) Server { mux.HandleFunc("/debug/pprof/profile", pprof.Profile) mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) mux.HandleFunc("/debug/pprof/trace", pprof.Trace) + mux.Handle("/debug/pprof/heap", pprof.Handler("heap")) + mux.Handle("/debug/pprof/allocs", pprof.Handler("allocs")) + mux.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine")) svr := &http.Server{ Addr: addr, From a5c8fe0dc1599617606c9502034f24b7f5bfa9e0 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 19:52:00 +0200 Subject: [PATCH 04/13] fix(router): re-read PPROF_ADDR from env after flag.Parse Flag defaults are captured at package init before embedders set PPROF_ADDR in main(), so platform-api ensurePprofAddr had no effect. Re-read env after flag.Parse() matches the existing CONFIG_PATH pattern. Co-authored-by: Cursor --- router/cmd/main.go | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/router/cmd/main.go b/router/cmd/main.go index 6ec9c9e1f1..e1909e4bf4 100644 --- a/router/cmd/main.go +++ b/router/cmd/main.go @@ -46,6 +46,19 @@ func Main() { // Parse flags before calling profile.Start(), since it may add flags flag.Parse() + // Re-read profiling env after flag.Parse() — flag defaults are captured at package + // init, before embedders (e.g. platform-api-cosmo-router) can Setenv in main(). + if *pprofListenAddr == "" { + if addr := os.Getenv("PPROF_ADDR"); addr != "" { + *pprofListenAddr = addr + } + } + if *pyroscopeAddr == "" { + if addr := os.Getenv("PYROSCOPE_ADDR"); addr != "" { + *pyroscopeAddr = addr + } + } + if *help { flag.PrintDefaults() os.Exit(0) From c9cbb1728fcc197262a5a9903231c6f63aac2e10 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 20:20:48 +0200 Subject: [PATCH 05/13] fix(router): release executor schema refs and reduce upstream WS overhead on reload Executor.Close() nils federation schema AST held after graph mux shutdown. Share one upstream subscription client across subgraph factories and disable upstream ping loops when client WebSocket is disabled. Co-authored-by: Cursor --- router/core/executor.go | 13 +++++++++ router/core/executor_test.go | 37 ++++++++++++++++++++++++ router/core/factoryresolver.go | 53 ++++++++++++++++++++++++---------- router/core/graph_server.go | 44 ++++++++++++++++++---------- 4 files changed, 117 insertions(+), 30 deletions(-) create mode 100644 router/core/executor_test.go diff --git a/router/core/executor.go b/router/core/executor.go index ae72771f96..f3f59c304c 100644 --- a/router/core/executor.go +++ b/router/core/executor.go @@ -51,6 +51,19 @@ type Executor struct { TrackUsageInfo bool } +// Close releases schema and planner references held by the executor so a replaced +// graph mux can be garbage-collected after config reload. +func (e *Executor) Close() { + if e == nil { + return + } + e.ClientSchema = nil + e.RouterSchema = nil + e.PlanConfig = plan.Configuration{} + e.RenameTypeNames = nil + e.Resolver = nil +} + type ExecutorBuildOptions struct { EngineConfig *nodev1.EngineConfiguration Subgraphs []*nodev1.Subgraph diff --git a/router/core/executor_test.go b/router/core/executor_test.go new file mode 100644 index 0000000000..a157b09a49 --- /dev/null +++ b/router/core/executor_test.go @@ -0,0 +1,37 @@ +package core + +import ( + "testing" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/graphql-go-tools/v2/pkg/ast" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/plan" +) + +func TestExecutorCloseReleasesSchemaReferences(t *testing.T) { + t.Parallel() + + executor := &Executor{ + ClientSchema: &ast.Document{}, + RouterSchema: &ast.Document{}, + PlanConfig: plan.Configuration{DataSources: []plan.DataSource{nil}}, + RenameTypeNames: nil, + } + + executor.Close() + + require.Nil(t, executor.ClientSchema) + require.Nil(t, executor.RouterSchema) + require.Empty(t, executor.PlanConfig.DataSources) + require.Nil(t, executor.RenameTypeNames) + require.Nil(t, executor.Resolver) +} + +func TestExecutorCloseNilSafe(t *testing.T) { + t.Parallel() + + var executor *Executor + require.NotPanics(t, func() { + executor.Close() + }) +} diff --git a/router/core/factoryresolver.go b/router/core/factoryresolver.go index d0a94d9a51..8d16b83b6b 100644 --- a/router/core/factoryresolver.go +++ b/router/core/factoryresolver.go @@ -7,6 +7,7 @@ import ( "net/http" "net/url" "slices" + "sync" "time" "github.com/buger/jsonparser" @@ -78,6 +79,9 @@ type DefaultFactoryResolver struct { transportFactory ApiTransportFactory defaultSubgraphRequestTimeout time.Duration subscriptionClientOptions []graphql_datasource.SubscriptionClientOption + + subscriptionClient graphql_datasource.GraphQLSubscriptionClient + subscriptionClientOnce sync.Once } func NewDefaultFactoryResolver( @@ -183,10 +187,7 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla if d.transportFactory == nil || d.baseTransport == nil { // dummy implementation for plan generator that doesn't make requests - subscriptionClient := graphql_datasource.NewGraphQLSubscriptionClient(d.engineCtx, - d.subscriptionClientOptions..., - ) - return graphql_datasource.NewFactory(d.engineCtx, http.DefaultClient, subscriptionClient) + return graphql_datasource.NewFactory(d.engineCtx, http.DefaultClient, d.sharedSubscriptionClient()) } defaultHTTPClient := &http.Client{ @@ -194,22 +195,44 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla Transport: d.transportFactory.RoundTripper(d.baseTransport), } - streamingClient := &http.Client{ - Transport: d.transportFactory.RoundTripper(d.baseTransport), - } - - subscriptionClient := graphql_datasource.NewGraphQLSubscriptionClient( - d.engineCtx, - append([]graphql_datasource.SubscriptionClientOption{graphql_datasource.WithUpgradeClient(defaultHTTPClient), graphql_datasource.WithStreamingClient(streamingClient)}, d.subscriptionClientOptions...)..., - ) - if subgraphClient, ok := d.subgraphHTTPClients[subgraphName]; ok { // it's intentional that we're not using the subgraphClient for subscriptions // custom subgraph clients are intended to be used for custom timeouts, which is not relevant for subscriptions - return graphql_datasource.NewFactory(d.engineCtx, subgraphClient, subscriptionClient) + return graphql_datasource.NewFactory(d.engineCtx, subgraphClient, d.sharedSubscriptionClient()) } - return graphql_datasource.NewFactory(d.engineCtx, defaultHTTPClient, subscriptionClient) + return graphql_datasource.NewFactory(d.engineCtx, defaultHTTPClient, d.sharedSubscriptionClient()) +} + +func (d *DefaultFactoryResolver) sharedSubscriptionClient() graphql_datasource.GraphQLSubscriptionClient { + d.subscriptionClientOnce.Do(func() { + if d.transportFactory == nil || d.baseTransport == nil { + d.subscriptionClient = graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, + d.subscriptionClientOptions..., + ) + return + } + + defaultHTTPClient := &http.Client{ + Timeout: d.defaultSubgraphRequestTimeout, + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + streamingClient := &http.Client{ + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + d.subscriptionClient = graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, + append([]graphql_datasource.SubscriptionClientOption{ + graphql_datasource.WithUpgradeClient(defaultHTTPClient), + graphql_datasource.WithStreamingClient(streamingClient), + }, d.subscriptionClientOptions...)..., + ) + }) + + return d.subscriptionClient } func (d *DefaultFactoryResolver) ResolveStaticFactory() (factory plan.PlannerFactory[staticdatasource.Configuration], err error) { diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 8317191694..5ea8d088b9 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -682,6 +682,7 @@ type graphMux struct { reused atomic.Bool wsHandler *WebsocketHandler + executor *Executor planCacheOnEvictEnabled atomic.Bool planCache *ristretto.Cache[uint64, *planWithMetaData] @@ -988,6 +989,11 @@ func (s *graphMux) Shutdown(ctx context.Context) error { s.validationCache.Close() s.operationHashCache.Close() + if s.executor != nil { + s.executor.Close() + s.executor = nil + } + var err error if s.accessLogsFileLogger != nil { @@ -1463,22 +1469,29 @@ func (s *graphServer) buildGraphMux( return nil, fmt.Errorf("failed to process retry options: %w", err) } + subscriptionClientOptions := &SubscriptionClientOptions{ + PingInterval: s.engineExecutionConfiguration.WebSocketClientPingInterval, + PingTimeout: s.engineExecutionConfiguration.WebSocketClientPingTimeout, + WriteTimeout: s.engineExecutionConfiguration.WebSocketClientWriteTimeout, + AckTimeout: s.engineExecutionConfiguration.WebSocketClientAckTimeout, + ReadLimit: int64(s.engineExecutionConfiguration.WebSocketClientReadLimit), + DefaultErrorExtensionCode: s.subgraphErrorPropagation.DefaultExtensionCode, + } + // Client-facing WebSocket subscriptions are disabled; skip upstream ping loops + // that would otherwise start one goroutine per subgraph datasource factory. + if s.webSocketConfiguration != nil && !s.webSocketConfiguration.Enabled { + subscriptionClientOptions.PingInterval = 0 + } + ecb := &ExecutorConfigurationBuilder{ - introspection: s.introspection, - baseURL: s.baseURL, - baseTripper: s.baseTransport, - subgraphTrippers: subgraphTippers, - pluginHost: s.connector, - logger: s.logger, - trackUsageInfo: s.graphqlMetricsConfig.Enabled || s.metricConfig.Prometheus.PromSchemaFieldUsage.Enabled, - subscriptionClientOptions: &SubscriptionClientOptions{ - PingInterval: s.engineExecutionConfiguration.WebSocketClientPingInterval, - PingTimeout: s.engineExecutionConfiguration.WebSocketClientPingTimeout, - WriteTimeout: s.engineExecutionConfiguration.WebSocketClientWriteTimeout, - AckTimeout: s.engineExecutionConfiguration.WebSocketClientAckTimeout, - ReadLimit: int64(s.engineExecutionConfiguration.WebSocketClientReadLimit), - DefaultErrorExtensionCode: s.subgraphErrorPropagation.DefaultExtensionCode, - }, + introspection: s.introspection, + baseURL: s.baseURL, + baseTripper: s.baseTransport, + subgraphTrippers: subgraphTippers, + pluginHost: s.connector, + logger: s.logger, + trackUsageInfo: s.graphqlMetricsConfig.Enabled || s.metricConfig.Prometheus.PromSchemaFieldUsage.Enabled, + subscriptionClientOptions: subscriptionClientOptions, transportOptions: &TransportOptions{ SubgraphTransportOptions: s.subgraphTransportOptions, PreHandlers: s.preOriginHandlers, @@ -1514,6 +1527,7 @@ func (s *graphServer) buildGraphMux( if err != nil { return nil, fmt.Errorf("failed to build plan configuration: %w", err) } + gm.executor = executor s.pubSubProviders = providers if pubSubStartupErr := s.startupPubSubProviders(s.graphServerCtx); pubSubStartupErr != nil { From d80b9add6e3e4a1b442014e0edd397021a23f419 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 20:23:33 +0200 Subject: [PATCH 06/13] fix(router): use noop upstream subscription client when subscriptions unused Skip WSTransport/SSE initialization when the router schema has no subscription root fields or when client WebSocket and pubsub events are both disabled. Co-authored-by: Cursor --- router/core/executor.go | 41 ++++--- router/core/factoryresolver.go | 9 ++ router/core/graph_server.go | 1 + .../core/noop_graphql_subscription_client.go | 105 ++++++++++++++++++ .../noop_graphql_subscription_client_test.go | 89 +++++++++++++++ router/core/transport.go | 2 + 6 files changed, 233 insertions(+), 14 deletions(-) create mode 100644 router/core/noop_graphql_subscription_client.go create mode 100644 router/core/noop_graphql_subscription_client_test.go diff --git a/router/core/executor.go b/router/core/executor.go index f3f59c304c..e600187c56 100644 --- a/router/core/executor.go +++ b/router/core/executor.go @@ -75,10 +75,11 @@ type ExecutorBuildOptions struct { TraceClientRequired bool PluginsEnabled bool InstanceData InstanceData + WebSocketConfiguration *config.WebSocketConfiguration } func (b *ExecutorConfigurationBuilder) Build(ctx context.Context, opts *ExecutorBuildOptions) (*Executor, []pubsub_datasource.Provider, error) { - planConfig, providers, err := b.buildPlannerConfiguration(ctx, opts.EngineConfig, opts.Subgraphs, opts.RouterEngineConfig, opts.PluginsEnabled) + planConfig, providers, err := b.buildPlannerConfiguration(ctx, opts) if err != nil { return nil, nil, fmt.Errorf("failed to build planner configuration: %w", err) } @@ -228,29 +229,41 @@ func (b *ExecutorConfigurationBuilder) Build(ctx context.Context, opts *Executor }, providers, nil } -func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Context, engineConfig *nodev1.EngineConfiguration, subgraphs []*nodev1.Subgraph, routerEngineCfg *RouterEngineConfiguration, pluginsEnabled bool) (*plan.Configuration, []pubsub_datasource.Provider, error) { +func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Context, opts *ExecutorBuildOptions) (*plan.Configuration, []pubsub_datasource.Provider, error) { // this loader is used to take the engine config and create a plan config // the plan config is what the engine uses to turn a GraphQL Request into an execution plan // the plan config is stateful as it carries connection pools and other things + subscriptionClientOptions := b.subscriptionClientOptions + if subscriptionClientOptions == nil { + subscriptionClientOptions = &SubscriptionClientOptions{} + } + resolvedSubscriptionClientOptions := *subscriptionClientOptions + resolvedSubscriptionClientOptions.UseNoopClient = shouldUseNoopUpstreamSubscriptionClient( + opts.EngineConfig.GetGraphqlSchema(), + opts.EngineConfig, + opts.RouterEngineConfig.Events, + opts.WebSocketConfiguration, + ) + loader := NewLoader(ctx, b.trackUsageInfo, NewDefaultFactoryResolver( ctx, b.transportOptions, - b.subscriptionClientOptions, + &resolvedSubscriptionClientOptions, b.baseTripper, b.subgraphTrippers, b.pluginHost, b.logger, - routerEngineCfg.Execution.EnableNetPoll, + opts.RouterEngineConfig.Execution.EnableNetPoll, b.instanceData, ), b.logger, b.subscriptionHooks) // this generates the plan config using the data source factories from the config package - planConfig, providers, err := loader.Load(engineConfig, subgraphs, routerEngineCfg, pluginsEnabled) + planConfig, providers, err := loader.Load(opts.EngineConfig, opts.Subgraphs, opts.RouterEngineConfig, opts.PluginsEnabled) if err != nil { return nil, nil, fmt.Errorf("failed to load configuration: %w", err) } - debug := &routerEngineCfg.Execution.Debug + debug := &opts.RouterEngineConfig.Execution.Debug planConfig.Debug = plan.DebugConfiguration{ PrintOperationTransformations: debug.PrintOperationTransformations, PrintOperationEnableASTRefs: debug.PrintOperationEnableASTRefs, @@ -261,19 +274,19 @@ func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Con PlanningVisitor: debug.PlanningVisitor, DatasourceVisitor: debug.DatasourceVisitor, } - planConfig.MinifySubgraphOperations = routerEngineCfg.Execution.MinifySubgraphOperations + planConfig.MinifySubgraphOperations = opts.RouterEngineConfig.Execution.MinifySubgraphOperations - planConfig.EnableOperationNamePropagation = routerEngineCfg.Execution.EnableSubgraphFetchOperationName + planConfig.EnableOperationNamePropagation = opts.RouterEngineConfig.Execution.EnableSubgraphFetchOperationName - planConfig.BuildFetchReasons = routerEngineCfg.Execution.EnableRequireFetchReasons || routerEngineCfg.Execution.ValidateRequiredExternalFields - planConfig.ValidateRequiredExternalFields = routerEngineCfg.Execution.ValidateRequiredExternalFields - planConfig.RelaxSubgraphOperationFieldSelectionMergingNullability = routerEngineCfg.Execution.RelaxSubgraphOperationFieldSelectionMergingNullability + planConfig.BuildFetchReasons = opts.RouterEngineConfig.Execution.EnableRequireFetchReasons || opts.RouterEngineConfig.Execution.ValidateRequiredExternalFields + planConfig.ValidateRequiredExternalFields = opts.RouterEngineConfig.Execution.ValidateRequiredExternalFields + planConfig.RelaxSubgraphOperationFieldSelectionMergingNullability = opts.RouterEngineConfig.Execution.RelaxSubgraphOperationFieldSelectionMergingNullability // Enable cost computation when cost control is enabled - if routerEngineCfg.CostControl != nil && routerEngineCfg.CostControl.Enabled { + if opts.RouterEngineConfig.CostControl != nil && opts.RouterEngineConfig.CostControl.Enabled { planConfig.ComputeCosts = true - planConfig.StaticCostDefaultListSize = routerEngineCfg.CostControl.EstimatedListSize - planConfig.IgnoreImplementingTypeWeights = routerEngineCfg.CostControl.IgnoreImplementingTypeWeights + planConfig.StaticCostDefaultListSize = opts.RouterEngineConfig.CostControl.EstimatedListSize + planConfig.IgnoreImplementingTypeWeights = opts.RouterEngineConfig.CostControl.IgnoreImplementingTypeWeights } return planConfig, providers, nil diff --git a/router/core/factoryresolver.go b/router/core/factoryresolver.go index 8d16b83b6b..63675acb91 100644 --- a/router/core/factoryresolver.go +++ b/router/core/factoryresolver.go @@ -79,6 +79,7 @@ type DefaultFactoryResolver struct { transportFactory ApiTransportFactory defaultSubgraphRequestTimeout time.Duration subscriptionClientOptions []graphql_datasource.SubscriptionClientOption + useNoopSubscriptionClient bool subscriptionClient graphql_datasource.GraphQLSubscriptionClient subscriptionClientOnce sync.Once @@ -135,7 +136,9 @@ func NewDefaultFactoryResolver( graphql_datasource.WithLogger(factoryLogger), } + useNoopSubscriptionClient := false if subscriptionClientOptions != nil { + useNoopSubscriptionClient = subscriptionClientOptions.UseNoopClient if subscriptionClientOptions.PingInterval > 0 { options = append(options, graphql_datasource.WithPingInterval(subscriptionClientOptions.PingInterval)) } @@ -168,6 +171,7 @@ func NewDefaultFactoryResolver( transportFactory: transportFactory, defaultSubgraphRequestTimeout: transportOptions.SubgraphTransportOptions.RequestTimeout, subscriptionClientOptions: options, + useNoopSubscriptionClient: useNoopSubscriptionClient, } } @@ -206,6 +210,11 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla func (d *DefaultFactoryResolver) sharedSubscriptionClient() graphql_datasource.GraphQLSubscriptionClient { d.subscriptionClientOnce.Do(func() { + if d.useNoopSubscriptionClient { + d.subscriptionClient = noopGraphQLSubscriptionClientInstance + return + } + if d.transportFactory == nil || d.baseTransport == nil { d.subscriptionClient = graphql_datasource.NewGraphQLSubscriptionClient( d.engineCtx, diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 5ea8d088b9..f3522084a3 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -1522,6 +1522,7 @@ func (s *graphServer) buildGraphMux( HeartbeatInterval: s.subscriptionHeartbeatInterval, PluginsEnabled: s.plugins.Enabled, InstanceData: s.instanceData, + WebSocketConfiguration: s.webSocketConfiguration, }, ) if err != nil { diff --git a/router/core/noop_graphql_subscription_client.go b/router/core/noop_graphql_subscription_client.go new file mode 100644 index 0000000000..79af8d33f5 --- /dev/null +++ b/router/core/noop_graphql_subscription_client.go @@ -0,0 +1,105 @@ +package core + +import ( + "errors" + + nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" + "github.com/wundergraph/cosmo/router/pkg/config" + "github.com/wundergraph/graphql-go-tools/v2/pkg/ast" + "github.com/wundergraph/graphql-go-tools/v2/pkg/astparser" + "github.com/wundergraph/graphql-go-tools/v2/pkg/asttransform" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/graphql_datasource" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve" +) + +var errUpstreamGraphQLSubscriptionsDisabled = errors.New("upstream GraphQL subscriptions are disabled") + +// noopGraphQLSubscriptionClient satisfies graphql-go-tools NewFactory's non-nil +// subscription client requirement without initializing upstream WS/SSE transports. +type noopGraphQLSubscriptionClient struct{} + +func (c *noopGraphQLSubscriptionClient) Subscribe(_ *resolve.Context, _ graphql_datasource.GraphQLSubscriptionOptions, _ resolve.SubscriptionUpdater) error { + return errUpstreamGraphQLSubscriptionsDisabled +} + +var noopGraphQLSubscriptionClientInstance graphql_datasource.GraphQLSubscriptionClient = &noopGraphQLSubscriptionClient{} + +func shouldUseNoopUpstreamSubscriptionClient( + graphqlSchema string, + engineConfig *nodev1.EngineConfiguration, + eventsConfig config.EventsConfiguration, + webSocketConfiguration *config.WebSocketConfiguration, +) bool { + if !schemaHasSubscriptionRootFields(graphqlSchema) { + return true + } + if !clientWebSocketSubscriptionsEnabled(webSocketConfiguration) && !eventSubscriptionsEnabled(engineConfig, eventsConfig) { + return true + } + return false +} + +func schemaHasSubscriptionRootFields(graphqlSchema string) bool { + if graphqlSchema == "" { + return false + } + + doc, report := astparser.ParseGraphqlDocumentString(graphqlSchema) + if report.HasErrors() { + return false + } + if err := asttransform.MergeDefinitionWithBaseSchema(&doc); err != nil { + return false + } + + return subscriptionRootFieldCount(&doc) > 0 +} + +func subscriptionRootFieldCount(doc *ast.Document) int { + if doc.Index.SubscriptionTypeName == nil { + return 0 + } + + node, ok := doc.Index.FirstNodeByNameBytes(doc.Index.SubscriptionTypeName) + if !ok || node.Kind != ast.NodeKindObjectTypeDefinition { + return 0 + } + + return len(doc.ObjectTypeDefinitions[node.Ref].FieldsDefinition.Refs) +} + +func clientWebSocketSubscriptionsEnabled(webSocketConfiguration *config.WebSocketConfiguration) bool { + if webSocketConfiguration == nil { + return true + } + return webSocketConfiguration.Enabled +} + +func eventSubscriptionsEnabled(engineConfig *nodev1.EngineConfiguration, eventsConfig config.EventsConfiguration) bool { + if len(eventsConfig.Providers.Nats) > 0 || + len(eventsConfig.Providers.Kafka) > 0 || + len(eventsConfig.Providers.Redis) > 0 { + return true + } + + if engineConfig == nil { + return false + } + + for _, ds := range engineConfig.GetDatasourceConfigurations() { + if ds.GetKind() == nodev1.DataSourceKind_PUBSUB { + return true + } + customEvents := ds.GetCustomEvents() + if customEvents == nil { + continue + } + if len(customEvents.GetNats()) > 0 || + len(customEvents.GetKafka()) > 0 || + len(customEvents.GetRedis()) > 0 { + return true + } + } + + return false +} diff --git a/router/core/noop_graphql_subscription_client_test.go b/router/core/noop_graphql_subscription_client_test.go new file mode 100644 index 0000000000..a4a286c62c --- /dev/null +++ b/router/core/noop_graphql_subscription_client_test.go @@ -0,0 +1,89 @@ +package core + +import ( + "testing" + + nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" + "github.com/wundergraph/cosmo/router/pkg/config" + "github.com/stretchr/testify/require" + "github.com/wundergraph/graphql-go-tools/v2/pkg/engine/datasource/graphql_datasource" +) + +func TestShouldUseNoopUpstreamSubscriptionClient_NoSubscriptionRootFields(t *testing.T) { + schema := `type Query { hello: String }` + + require.True(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + nil, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: true}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_EmptySubscriptionType(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { }` + + require.True(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + nil, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: true}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_ClientWSDisabledWithoutEvents(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { onUpdate: String }` + + require.True(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + &nodev1.EngineConfiguration{}, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: false}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_ClientWSDisabledWithPubSubDatasource(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { onUpdate: String }` + + engineConfig := &nodev1.EngineConfiguration{ + DatasourceConfigurations: []*nodev1.DataSourceConfiguration{ + {Kind: nodev1.DataSourceKind_PUBSUB}, + }, + } + + require.False(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + engineConfig, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: false}, + )) +} + +func TestShouldUseNoopUpstreamSubscriptionClient_UpstreamSubscriptionsNeeded(t *testing.T) { + schema := `type Query { hello: String } +type Subscription { onUpdate: String }` + + require.False(t, shouldUseNoopUpstreamSubscriptionClient( + schema, + &nodev1.EngineConfiguration{}, + config.EventsConfiguration{}, + &config.WebSocketConfiguration{Enabled: true}, + )) +} + +func TestNoopGraphQLSubscriptionClient_SubscribeReturnsError(t *testing.T) { + err := noopGraphQLSubscriptionClientInstance.Subscribe(nil, graphql_datasource.GraphQLSubscriptionOptions{}, nil) + require.ErrorIs(t, err, errUpstreamGraphQLSubscriptionsDisabled) +} + +func TestSharedSubscriptionClient_UsesNoopWhenConfigured(t *testing.T) { + resolver := &DefaultFactoryResolver{ + useNoopSubscriptionClient: true, + } + + client := resolver.sharedSubscriptionClient() + require.Same(t, noopGraphQLSubscriptionClientInstance, client) +} diff --git a/router/core/transport.go b/router/core/transport.go index 609229c341..32afe6745e 100644 --- a/router/core/transport.go +++ b/router/core/transport.go @@ -223,6 +223,8 @@ type SubscriptionClientOptions struct { AckTimeout time.Duration ReadLimit int64 DefaultErrorExtensionCode string + // UseNoopClient skips upstream WS/SSE transport initialization when subscriptions are not needed. + UseNoopClient bool } func NewTransport(opts *TransportOptions) *TransportFactory { From d2d20a910328ceb5114435e8155b69c9fd0695b7 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 21:31:26 +0200 Subject: [PATCH 07/13] fix(router): nil graphMux caches after shutdown to allow GC on reload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close and drop Ristretto cache pointers, wsHandler, and mux after graphMux shutdown, and remove shut-down muxes from graphMuxList. Local benchmark: ~74 MB/reload → ~12 MB/reload retained inuse (same-content manifest reloads). Co-authored-by: Cursor --- router/core/graph_server.go | 42 ++++++++++++++++++++++++++++--------- 1 file changed, 32 insertions(+), 10 deletions(-) diff --git a/router/core/graph_server.go b/router/core/graph_server.go index f3522084a3..9dd1b8350a 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -965,6 +965,30 @@ func (s *graphMux) configureCacheMetrics(srv *graphServer, baseOtelAttributes [] return nil } +func closeRistrettoCacheUint64[V any](cache **ristretto.Cache[uint64, V]) { + if *cache != nil { + (*cache).Close() + *cache = nil + } +} + +// releaseOperationCaches drops references to closed Ristretto caches so the old +// graphMux can be collected after shutdown (Close clears entries but retains structs). +func (s *graphMux) releaseOperationCaches() { + closeRistrettoCacheUint64(&s.planCache) + if s.planFallbackCache != nil { + s.planFallbackCache.Close() + s.planFallbackCache = nil + } + closeRistrettoCacheUint64(&s.persistedOperationCache) + closeRistrettoCacheUint64(&s.normalizationCache) + closeRistrettoCacheUint64(&s.variablesNormalizationCache) + closeRistrettoCacheUint64(&s.remapVariablesCache) + closeRistrettoCacheUint64(&s.complexityCalculationCache) + closeRistrettoCacheUint64(&s.validationCache) + closeRistrettoCacheUint64(&s.operationHashCache) +} + func (s *graphMux) Shutdown(ctx context.Context) error { // Close websocket subscriptions synchronously before tearing down plan caches so // active preparedPlan and executor references are released first. @@ -978,22 +1002,19 @@ func (s *graphMux) Shutdown(ctx context.Context) error { // ristretto Close() clears all entries and invokes OnEvict for each one. Disable // migration into the slow-plan fallback cache during intentional mux shutdown. s.planCacheOnEvictEnabled.Store(false) - s.planCache.Close() - s.planFallbackCache.Wait() - s.planFallbackCache.Close() - s.persistedOperationCache.Close() - s.normalizationCache.Close() - s.variablesNormalizationCache.Close() - s.remapVariablesCache.Close() - s.complexityCalculationCache.Close() - s.validationCache.Close() - s.operationHashCache.Close() + if s.planFallbackCache != nil { + s.planFallbackCache.Wait() + } if s.executor != nil { s.executor.Close() s.executor = nil } + s.releaseOperationCaches() + s.wsHandler = nil + s.mux = nil + var err error if s.accessLogsFileLogger != nil { @@ -2205,6 +2226,7 @@ func (s *graphServer) Shutdown(ctx context.Context) error { if err := mux.Shutdown(ctx); err != nil { finalErr = errors.Join(finalErr, err) } + delete(s.graphMuxList, name) } // Close idle connections on base and subgraph transports From 3b6aadb277ce03e316dc475d479c8f8e9cc325de Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 21:34:25 +0200 Subject: [PATCH 08/13] fix(router): skip unchanged manifest reloads and release stale execution config Hash mapper.json before re-assembling; skip reload when content is unchanged (mtime-only touches). After a successful reload, swap staticExecutionConfig and proto.Reset the previous config so decoded protojson strings can be collected. --- router/core/router.go | 27 +++++++++++++++++++++++++ router/core/router_config.go | 3 +++ router/pkg/routerconfig/routerconfig.go | 11 ++++++++++ 3 files changed, 41 insertions(+) diff --git a/router/core/router.go b/router/core/router.go index 7e6dff4076..95eaa45b1a 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -27,6 +27,7 @@ import ( sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.uber.org/zap" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/graphqlmetrics/v1/graphqlmetricsv1connect" nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1" @@ -1102,6 +1103,11 @@ func (r *Router) bootstrap(ctx context.Context) error { } r.staticExecutionConfig = executionConfig + + if hash, hashErr := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path); hashErr == nil { + r.lastManifestMapperHash = hash + r.manifestMapperHashSeen = true + } } if err := r.buildClients(ctx); err != nil { @@ -1721,6 +1727,18 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) return } + mapperHash, err := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path) + if err != nil { + ll.Error("Failed to hash manifest mapper", zap.Error(err)) + return + } + + if r.manifestMapperHashSeen && mapperHash == r.lastManifestMapperHash { + ll.Debug("Manifest mapper unchanged, skipping reload", + zap.String("path", r.manifestConfig.Path)) + return + } + cfg, err := routerconfig.AssembleStaticExecutionConfigFromManifest( r.manifestConfig.Path, routerconfig.AssembleConfigRules{ SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, @@ -1738,6 +1756,15 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) ll.Error("Failed to update server with new config", zap.Error(err)) return } + + r.lastManifestMapperHash = mapperHash + r.manifestMapperHashSeen = true + + if old := r.staticExecutionConfig; old != nil && old != cfg { + proto.Reset(old) + } + r.staticExecutionConfig = cfg + r.trackExecutionConfigUsage(cfg, true) }, }) diff --git a/router/core/router_config.go b/router/core/router_config.go index 6b380ede12..062ae1409f 100644 --- a/router/core/router_config.go +++ b/router/core/router_config.go @@ -153,6 +153,9 @@ type Config struct { grpcPluginDialOptions []grpc.DialOption tracingAttributes []config.CustomAttribute subscriptionHooks subscriptionHooks + // lastManifestMapperHash skips manifest reload when mapper.json content is unchanged. + lastManifestMapperHash [32]byte + manifestMapperHashSeen bool } // Usage returns an anonymized version of the config for usage tracking diff --git a/router/pkg/routerconfig/routerconfig.go b/router/pkg/routerconfig/routerconfig.go index 71ba30e846..bda805ac0f 100644 --- a/router/pkg/routerconfig/routerconfig.go +++ b/router/pkg/routerconfig/routerconfig.go @@ -18,6 +18,7 @@ package routerconfig import ( + "crypto/sha256" "encoding/json" "fmt" "io/fs" @@ -106,6 +107,16 @@ func readMapperFile(path string) (map[string]string, error) { return mapper, nil } +// ManifestMapperSHA256 returns the SHA-256 digest of mapper.json bytes. +// Used to skip manifest reload when only the file mtime changed. +func ManifestMapperSHA256(manifestConfigPath string) ([32]byte, error) { + data, err := os.ReadFile(filepath.Join(manifestConfigPath, "mapper.json")) + if err != nil { + return [32]byte{}, fmt.Errorf("failed to read mapper file: %w", err) + } + return sha256.Sum256(data), nil +} + // assembleConfig assembles the router execution config from the base config and the feature flag configs. // The base config is the latest.json file in the manifest directory. // The feature flag configs are the feature-flags/.json files in the manifest directory. From 2c32b45df78a4f12bca7b8795478f9e328e79e41 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 21:44:00 +0200 Subject: [PATCH 09/13] fix(router): reuse graph muxes on manifest reload and release shutdown refs Pass Changes/Hashes from mapper.json graph hashes on the manifest watcher path so unchanged base or feature-flag muxes survive config reloads. Nil graphServer and graphMux metric fields after shutdown to drop retained references sooner. --- router/core/graph_server.go | 20 ++++++ router/core/router.go | 43 +++++++++++-- router/core/router_config.go | 2 + router/pkg/routerconfig/manifest_hashes.go | 61 +++++++++++++++++++ .../pkg/routerconfig/manifest_hashes_test.go | 44 +++++++++++++ 5 files changed, 164 insertions(+), 6 deletions(-) create mode 100644 router/pkg/routerconfig/manifest_hashes.go create mode 100644 router/pkg/routerconfig/manifest_hashes_test.go diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 9dd1b8350a..537f325510 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -1051,8 +1051,15 @@ func (s *graphMux) Shutdown(ctx context.Context) error { if aErr := s.prometheusMetricsExporter.Shutdown(ctx); aErr != nil { err = errors.Join(err, aErr) } + s.prometheusMetricsExporter = nil } + s.otelCacheMetrics = nil + s.prometheusCacheMetrics = nil + s.metricStore = nil + s.streamMetricStore = nil + s.accessLogsFileLogger = nil + if err != nil { return fmt.Errorf("shutdown graph mux: %w", err) } @@ -2245,8 +2252,21 @@ func (s *graphServer) Shutdown(ctx context.Context) error { if err := s.connector.StopAllProviders(); err != nil { finalErr = errors.Join(finalErr, err) } + s.connector = nil } + s.mux = nil + s.graphMuxList = nil + s.baseTransport = nil + s.subgraphTransports = nil + s.pubSubProviders = nil + s.circuitBreakerManager = nil + s.traceDialer = nil + s.runtimeMetrics = nil + s.otlpEngineMetrics = nil + s.prometheusEngineMetrics = nil + s.connectionMetrics = nil + return finalErr } diff --git a/router/core/router.go b/router/core/router.go index 95eaa45b1a..bea4051023 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -1108,6 +1108,14 @@ func (r *Router) bootstrap(ctx context.Context) error { r.lastManifestMapperHash = hash r.manifestMapperHashSeen = true } + + rules := routerconfig.AssembleConfigRules{ + SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, + IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, + } + if graphs, graphErr := routerconfig.ReadManifestMapperGraphs(r.manifestConfig.Path, rules); graphErr == nil { + r.lastManifestGraphHashes = graphs + } } if err := r.buildClients(ctx); err != nil { @@ -1607,6 +1615,13 @@ func (r *Router) Start(ctx context.Context) error { return nil } +func (r *Router) manifestAssembleRules() routerconfig.AssembleConfigRules { + return routerconfig.AssembleConfigRules{ + SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, + IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, + } +} + func (r *Router) startWithStaticExecutionConfig(ctx context.Context) error { r.trackExecutionConfigUsage(r.staticExecutionConfig, true) @@ -1614,7 +1629,12 @@ func (r *Router) startWithStaticExecutionConfig(ctx context.Context) error { return err } - if err := r.newServer(ctx, &routerconfig.Response{Config: r.staticExecutionConfig}); err != nil { + _, hashes := routerconfig.ComputeGraphChangesAndHashes(nil, r.lastManifestGraphHashes) + if err := r.newServer(ctx, &routerconfig.Response{ + Config: r.staticExecutionConfig, + Changes: nil, + Hashes: hashes, + }); err != nil { return err } @@ -1739,11 +1759,17 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) return } + rules := r.manifestAssembleRules() + mapperGraphs, err := routerconfig.ReadManifestMapperGraphs(r.manifestConfig.Path, rules) + if err != nil { + ll.Error("Failed to read manifest mapper graphs", zap.Error(err)) + return + } + + changes, hashes := routerconfig.ComputeGraphChangesAndHashes(r.lastManifestGraphHashes, mapperGraphs) + cfg, err := routerconfig.AssembleStaticExecutionConfigFromManifest( - r.manifestConfig.Path, routerconfig.AssembleConfigRules{ - SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, - IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, - }) + r.manifestConfig.Path, rules) if err != nil { ll.Error("Failed to assemble static execution config from manifest", zap.Error(err)) @@ -1752,13 +1778,18 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) ll.Info("Manifest config changed. Updating server with new config", zap.String("path", r.manifestConfig.Path)) - if err := r.newServer(ctx, &routerconfig.Response{Config: cfg}); err != nil { + if err := r.newServer(ctx, &routerconfig.Response{ + Config: cfg, + Changes: changes, + Hashes: hashes, + }); err != nil { ll.Error("Failed to update server with new config", zap.Error(err)) return } r.lastManifestMapperHash = mapperHash r.manifestMapperHashSeen = true + r.lastManifestGraphHashes = mapperGraphs if old := r.staticExecutionConfig; old != nil && old != cfg { proto.Reset(old) diff --git a/router/core/router_config.go b/router/core/router_config.go index 062ae1409f..116d4addbd 100644 --- a/router/core/router_config.go +++ b/router/core/router_config.go @@ -156,6 +156,8 @@ type Config struct { // lastManifestMapperHash skips manifest reload when mapper.json content is unchanged. lastManifestMapperHash [32]byte manifestMapperHashSeen bool + // lastManifestGraphHashes tracks per-graph CDN hashes from mapper.json for mux reuse. + lastManifestGraphHashes map[string]string } // Usage returns an anonymized version of the config for usage tracking diff --git a/router/pkg/routerconfig/manifest_hashes.go b/router/pkg/routerconfig/manifest_hashes.go new file mode 100644 index 0000000000..cb5d6619d9 --- /dev/null +++ b/router/pkg/routerconfig/manifest_hashes.go @@ -0,0 +1,61 @@ +package routerconfig + +import ( + "path/filepath" +) + +// ReadManifestMapperGraphs loads mapper.json graph hashes (base key "") and applies +// the same ignored-feature-flag filtering as config assembly. +func ReadManifestMapperGraphs(manifestConfigPath string, rules AssembleConfigRules) (map[string]string, error) { + mapper, err := readMapperFile(filepath.Join(manifestConfigPath, "mapper.json")) + if err != nil { + return nil, err + } + + for _, ff := range rules.IgnoredFeatureFlags { + delete(mapper, ff) + } + + return mapper, nil +} + +// ComputeGraphChangesAndHashes compares known mapper graph hashes with the current +// set. A nil known map indicates the initial load (Changes nil, rebuild everything). +func ComputeGraphChangesAndHashes(known, current map[string]string) (*Changes, map[string]HashInfo) { + hashes := make(map[string]HashInfo, len(current)) + if known == nil { + for name, hash := range current { + hashes[name] = HashInfo{NewHash: hash} + } + return nil, hashes + } + + changes := &Changes{ + AddedConfigs: make(map[string]struct{}), + RemovedConfigs: make(map[string]struct{}), + ChangedConfigs: make(map[string]struct{}), + } + + for name, hash := range current { + oldHash, exists := known[name] + if !exists { + changes.AddedConfigs[name] = struct{}{} + hashes[name] = HashInfo{NewHash: hash} + continue + } + if oldHash != hash { + changes.ChangedConfigs[name] = struct{}{} + hashes[name] = HashInfo{NewHash: hash, OldHash: oldHash} + continue + } + hashes[name] = HashInfo{OldHash: oldHash, NewHash: hash} + } + + for name := range known { + if _, exists := current[name]; !exists { + changes.RemovedConfigs[name] = struct{}{} + } + } + + return changes, hashes +} diff --git a/router/pkg/routerconfig/manifest_hashes_test.go b/router/pkg/routerconfig/manifest_hashes_test.go new file mode 100644 index 0000000000..ef65cf8e47 --- /dev/null +++ b/router/pkg/routerconfig/manifest_hashes_test.go @@ -0,0 +1,44 @@ +package routerconfig + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestComputeGraphChangesAndHashes_initial(t *testing.T) { + current := map[string]string{"": "base-v1", "ff1": "ff-v1"} + + changes, hashes := ComputeGraphChangesAndHashes(nil, current) + + assert.Nil(t, changes) + require.Len(t, hashes, 2) + assert.Equal(t, HashInfo{NewHash: "base-v1"}, hashes[""]) + assert.Equal(t, HashInfo{NewHash: "ff-v1"}, hashes["ff1"]) +} + +func TestComputeGraphChangesAndHashes_baseChangedFFUnchanged(t *testing.T) { + known := map[string]string{"": "base-v1", "ff1": "ff-v1"} + current := map[string]string{"": "base-v2", "ff1": "ff-v1"} + + changes, hashes := ComputeGraphChangesAndHashes(known, current) + + require.NotNil(t, changes) + assert.Contains(t, changes.ChangedConfigs, "") + assert.NotContains(t, changes.ChangedConfigs, "ff1") + assert.Equal(t, HashInfo{OldHash: "base-v1", NewHash: "base-v2"}, hashes[""]) + assert.Equal(t, HashInfo{OldHash: "ff-v1", NewHash: "ff-v1"}, hashes["ff1"]) +} + +func TestComputeGraphChangesAndHashes_ffAddedAndRemoved(t *testing.T) { + known := map[string]string{"": "base-v1", "ff-old": "old"} + current := map[string]string{"": "base-v1", "ff-new": "new"} + + changes, hashes := ComputeGraphChangesAndHashes(known, current) + + require.NotNil(t, changes) + assert.Contains(t, changes.AddedConfigs, "ff-new") + assert.Contains(t, changes.RemovedConfigs, "ff-old") + assert.Equal(t, HashInfo{OldHash: "base-v1", NewHash: "base-v1"}, hashes[""]) +} From 73fc25d6845b4196f499b15e5525c692a98b2b80 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Wed, 24 Jun 2026 22:09:14 +0200 Subject: [PATCH 10/13] Revert "fix(router): reuse graph muxes on manifest reload and release shutdown refs" This reverts commit 2c32b45df78a4f12bca7b8795478f9e328e79e41. --- router/core/graph_server.go | 20 ------ router/core/router.go | 43 ++----------- router/core/router_config.go | 2 - router/pkg/routerconfig/manifest_hashes.go | 61 ------------------- .../pkg/routerconfig/manifest_hashes_test.go | 44 ------------- 5 files changed, 6 insertions(+), 164 deletions(-) delete mode 100644 router/pkg/routerconfig/manifest_hashes.go delete mode 100644 router/pkg/routerconfig/manifest_hashes_test.go diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 537f325510..9dd1b8350a 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -1051,15 +1051,8 @@ func (s *graphMux) Shutdown(ctx context.Context) error { if aErr := s.prometheusMetricsExporter.Shutdown(ctx); aErr != nil { err = errors.Join(err, aErr) } - s.prometheusMetricsExporter = nil } - s.otelCacheMetrics = nil - s.prometheusCacheMetrics = nil - s.metricStore = nil - s.streamMetricStore = nil - s.accessLogsFileLogger = nil - if err != nil { return fmt.Errorf("shutdown graph mux: %w", err) } @@ -2252,21 +2245,8 @@ func (s *graphServer) Shutdown(ctx context.Context) error { if err := s.connector.StopAllProviders(); err != nil { finalErr = errors.Join(finalErr, err) } - s.connector = nil } - s.mux = nil - s.graphMuxList = nil - s.baseTransport = nil - s.subgraphTransports = nil - s.pubSubProviders = nil - s.circuitBreakerManager = nil - s.traceDialer = nil - s.runtimeMetrics = nil - s.otlpEngineMetrics = nil - s.prometheusEngineMetrics = nil - s.connectionMetrics = nil - return finalErr } diff --git a/router/core/router.go b/router/core/router.go index bea4051023..95eaa45b1a 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -1108,14 +1108,6 @@ func (r *Router) bootstrap(ctx context.Context) error { r.lastManifestMapperHash = hash r.manifestMapperHashSeen = true } - - rules := routerconfig.AssembleConfigRules{ - SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, - IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, - } - if graphs, graphErr := routerconfig.ReadManifestMapperGraphs(r.manifestConfig.Path, rules); graphErr == nil { - r.lastManifestGraphHashes = graphs - } } if err := r.buildClients(ctx); err != nil { @@ -1615,13 +1607,6 @@ func (r *Router) Start(ctx context.Context) error { return nil } -func (r *Router) manifestAssembleRules() routerconfig.AssembleConfigRules { - return routerconfig.AssembleConfigRules{ - SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, - IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, - } -} - func (r *Router) startWithStaticExecutionConfig(ctx context.Context) error { r.trackExecutionConfigUsage(r.staticExecutionConfig, true) @@ -1629,12 +1614,7 @@ func (r *Router) startWithStaticExecutionConfig(ctx context.Context) error { return err } - _, hashes := routerconfig.ComputeGraphChangesAndHashes(nil, r.lastManifestGraphHashes) - if err := r.newServer(ctx, &routerconfig.Response{ - Config: r.staticExecutionConfig, - Changes: nil, - Hashes: hashes, - }); err != nil { + if err := r.newServer(ctx, &routerconfig.Response{Config: r.staticExecutionConfig}); err != nil { return err } @@ -1759,17 +1739,11 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) return } - rules := r.manifestAssembleRules() - mapperGraphs, err := routerconfig.ReadManifestMapperGraphs(r.manifestConfig.Path, rules) - if err != nil { - ll.Error("Failed to read manifest mapper graphs", zap.Error(err)) - return - } - - changes, hashes := routerconfig.ComputeGraphChangesAndHashes(r.lastManifestGraphHashes, mapperGraphs) - cfg, err := routerconfig.AssembleStaticExecutionConfigFromManifest( - r.manifestConfig.Path, rules) + r.manifestConfig.Path, routerconfig.AssembleConfigRules{ + SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, + IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, + }) if err != nil { ll.Error("Failed to assemble static execution config from manifest", zap.Error(err)) @@ -1778,18 +1752,13 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) ll.Info("Manifest config changed. Updating server with new config", zap.String("path", r.manifestConfig.Path)) - if err := r.newServer(ctx, &routerconfig.Response{ - Config: cfg, - Changes: changes, - Hashes: hashes, - }); err != nil { + if err := r.newServer(ctx, &routerconfig.Response{Config: cfg}); err != nil { ll.Error("Failed to update server with new config", zap.Error(err)) return } r.lastManifestMapperHash = mapperHash r.manifestMapperHashSeen = true - r.lastManifestGraphHashes = mapperGraphs if old := r.staticExecutionConfig; old != nil && old != cfg { proto.Reset(old) diff --git a/router/core/router_config.go b/router/core/router_config.go index 116d4addbd..062ae1409f 100644 --- a/router/core/router_config.go +++ b/router/core/router_config.go @@ -156,8 +156,6 @@ type Config struct { // lastManifestMapperHash skips manifest reload when mapper.json content is unchanged. lastManifestMapperHash [32]byte manifestMapperHashSeen bool - // lastManifestGraphHashes tracks per-graph CDN hashes from mapper.json for mux reuse. - lastManifestGraphHashes map[string]string } // Usage returns an anonymized version of the config for usage tracking diff --git a/router/pkg/routerconfig/manifest_hashes.go b/router/pkg/routerconfig/manifest_hashes.go deleted file mode 100644 index cb5d6619d9..0000000000 --- a/router/pkg/routerconfig/manifest_hashes.go +++ /dev/null @@ -1,61 +0,0 @@ -package routerconfig - -import ( - "path/filepath" -) - -// ReadManifestMapperGraphs loads mapper.json graph hashes (base key "") and applies -// the same ignored-feature-flag filtering as config assembly. -func ReadManifestMapperGraphs(manifestConfigPath string, rules AssembleConfigRules) (map[string]string, error) { - mapper, err := readMapperFile(filepath.Join(manifestConfigPath, "mapper.json")) - if err != nil { - return nil, err - } - - for _, ff := range rules.IgnoredFeatureFlags { - delete(mapper, ff) - } - - return mapper, nil -} - -// ComputeGraphChangesAndHashes compares known mapper graph hashes with the current -// set. A nil known map indicates the initial load (Changes nil, rebuild everything). -func ComputeGraphChangesAndHashes(known, current map[string]string) (*Changes, map[string]HashInfo) { - hashes := make(map[string]HashInfo, len(current)) - if known == nil { - for name, hash := range current { - hashes[name] = HashInfo{NewHash: hash} - } - return nil, hashes - } - - changes := &Changes{ - AddedConfigs: make(map[string]struct{}), - RemovedConfigs: make(map[string]struct{}), - ChangedConfigs: make(map[string]struct{}), - } - - for name, hash := range current { - oldHash, exists := known[name] - if !exists { - changes.AddedConfigs[name] = struct{}{} - hashes[name] = HashInfo{NewHash: hash} - continue - } - if oldHash != hash { - changes.ChangedConfigs[name] = struct{}{} - hashes[name] = HashInfo{NewHash: hash, OldHash: oldHash} - continue - } - hashes[name] = HashInfo{OldHash: oldHash, NewHash: hash} - } - - for name := range known { - if _, exists := current[name]; !exists { - changes.RemovedConfigs[name] = struct{}{} - } - } - - return changes, hashes -} diff --git a/router/pkg/routerconfig/manifest_hashes_test.go b/router/pkg/routerconfig/manifest_hashes_test.go deleted file mode 100644 index ef65cf8e47..0000000000 --- a/router/pkg/routerconfig/manifest_hashes_test.go +++ /dev/null @@ -1,44 +0,0 @@ -package routerconfig - -import ( - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func TestComputeGraphChangesAndHashes_initial(t *testing.T) { - current := map[string]string{"": "base-v1", "ff1": "ff-v1"} - - changes, hashes := ComputeGraphChangesAndHashes(nil, current) - - assert.Nil(t, changes) - require.Len(t, hashes, 2) - assert.Equal(t, HashInfo{NewHash: "base-v1"}, hashes[""]) - assert.Equal(t, HashInfo{NewHash: "ff-v1"}, hashes["ff1"]) -} - -func TestComputeGraphChangesAndHashes_baseChangedFFUnchanged(t *testing.T) { - known := map[string]string{"": "base-v1", "ff1": "ff-v1"} - current := map[string]string{"": "base-v2", "ff1": "ff-v1"} - - changes, hashes := ComputeGraphChangesAndHashes(known, current) - - require.NotNil(t, changes) - assert.Contains(t, changes.ChangedConfigs, "") - assert.NotContains(t, changes.ChangedConfigs, "ff1") - assert.Equal(t, HashInfo{OldHash: "base-v1", NewHash: "base-v2"}, hashes[""]) - assert.Equal(t, HashInfo{OldHash: "ff-v1", NewHash: "ff-v1"}, hashes["ff1"]) -} - -func TestComputeGraphChangesAndHashes_ffAddedAndRemoved(t *testing.T) { - known := map[string]string{"": "base-v1", "ff-old": "old"} - current := map[string]string{"": "base-v1", "ff-new": "new"} - - changes, hashes := ComputeGraphChangesAndHashes(known, current) - - require.NotNil(t, changes) - assert.Contains(t, changes.AddedConfigs, "ff-new") - assert.Contains(t, changes.RemovedConfigs, "ff-old") - assert.Equal(t, HashInfo{OldHash: "base-v1", NewHash: "base-v1"}, hashes[""]) -} From c22045e1797d39eab18b3d42eb4641475c6aa91a Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Thu, 25 Jun 2026 17:32:35 +0200 Subject: [PATCH 11/13] chore(router): drop profiling/pyroscope changes from PR Remove late env re-read for PPROF/PYROSCOPE in main.go and extra pprof handlers in profile.go so this PR stays focused on config reload fixes. --- router/cmd/main.go | 13 ------------- router/pkg/profile/profile.go | 3 --- 2 files changed, 16 deletions(-) diff --git a/router/cmd/main.go b/router/cmd/main.go index e1909e4bf4..6ec9c9e1f1 100644 --- a/router/cmd/main.go +++ b/router/cmd/main.go @@ -46,19 +46,6 @@ func Main() { // Parse flags before calling profile.Start(), since it may add flags flag.Parse() - // Re-read profiling env after flag.Parse() — flag defaults are captured at package - // init, before embedders (e.g. platform-api-cosmo-router) can Setenv in main(). - if *pprofListenAddr == "" { - if addr := os.Getenv("PPROF_ADDR"); addr != "" { - *pprofListenAddr = addr - } - } - if *pyroscopeAddr == "" { - if addr := os.Getenv("PYROSCOPE_ADDR"); addr != "" { - *pyroscopeAddr = addr - } - } - if *help { flag.PrintDefaults() os.Exit(0) diff --git a/router/pkg/profile/profile.go b/router/pkg/profile/profile.go index 513505b6a8..8ee0239848 100644 --- a/router/pkg/profile/profile.go +++ b/router/pkg/profile/profile.go @@ -43,9 +43,6 @@ func NewServer(addr string, log *zap.Logger) Server { mux.HandleFunc("/debug/pprof/profile", pprof.Profile) mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) mux.HandleFunc("/debug/pprof/trace", pprof.Trace) - mux.Handle("/debug/pprof/heap", pprof.Handler("heap")) - mux.Handle("/debug/pprof/allocs", pprof.Handler("allocs")) - mux.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine")) svr := &http.Server{ Addr: addr, From 8373bec7b636ce651159d681a289f25806f75489 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Thu, 25 Jun 2026 17:37:17 +0200 Subject: [PATCH 12/13] fix(router): gate memory-leak fixes behind mondaytweaks constants Centralize all monday.com config-reload leak fixes in mondaytweaks.go so they are easy to audit and disable individually. Restore profiling helpers from stash behind separate tweak flags. --- router/cmd/main.go | 23 ++++++- router/core/executor.go | 15 +++-- router/core/factoryresolver.go | 44 +++++++++++- router/core/graph_server.go | 90 +++++++++++++++++-------- router/core/router.go | 44 +++++++----- router/docs/Profiling.md | 7 ++ router/pkg/mondaytweaks/mondaytweaks.go | 58 ++++++++++++++++ router/pkg/profile/profile.go | 7 ++ router/pkg/profile/pyroscope.go | 84 +++++++++++++++++++++++ router/pkg/profile/pyroscope_test.go | 75 +++++++++++++++++++++ 10 files changed, 394 insertions(+), 53 deletions(-) create mode 100644 router/pkg/profile/pyroscope.go create mode 100644 router/pkg/profile/pyroscope_test.go diff --git a/router/cmd/main.go b/router/cmd/main.go index 6ec9c9e1f1..094db3dbd6 100644 --- a/router/cmd/main.go +++ b/router/cmd/main.go @@ -19,6 +19,7 @@ import ( "github.com/wundergraph/cosmo/router/internal/versioninfo" "github.com/wundergraph/cosmo/router/pkg/config" "github.com/wundergraph/cosmo/router/pkg/logging" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" "github.com/wundergraph/cosmo/router/pkg/profile" "github.com/wundergraph/cosmo/router/pkg/watcher" @@ -46,6 +47,21 @@ func Main() { // Parse flags before calling profile.Start(), since it may add flags flag.Parse() + // Re-read profiling env after flag.Parse() — flag defaults are captured at package + // init, before embedders (e.g. platform-api-cosmo-router) can Setenv in main(). + if mondaytweaks.RereadProfilingEnvAfterFlagParse { + if *pprofListenAddr == "" { + if addr := os.Getenv("PPROF_ADDR"); addr != "" { + *pprofListenAddr = addr + } + } + if *pyroscopeAddr == "" { + if addr := os.Getenv("PYROSCOPE_ADDR"); addr != "" { + *pyroscopeAddr = addr + } + } + } + if *help { flag.PrintDefaults() os.Exit(0) @@ -122,11 +138,14 @@ func Main() { logger := baseLogger.With(zap.String("component", "pyroscope")) logger.Info("starting pyroscope server") + applicationName := profile.PyroscopeApplicationName(result.Config.Telemetry.ServiceName) + tags := profile.PyroscopeTags() + pyro, err := pyroscope.Start(pyroscope.Config{ - ApplicationName: "wundergraph.cosmo.router", + ApplicationName: applicationName, ServerAddress: *pyroscopeAddr, Logger: logger.Sugar(), - Tags: map[string]string{"hostname": os.Getenv("HOSTNAME")}, + Tags: tags, ProfileTypes: []pyroscope.ProfileType{ pyroscope.ProfileCPU, diff --git a/router/core/executor.go b/router/core/executor.go index e600187c56..69e4b02cf7 100644 --- a/router/core/executor.go +++ b/router/core/executor.go @@ -8,6 +8,7 @@ import ( "go.uber.org/zap" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" 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/grpcconnector" @@ -239,12 +240,14 @@ func (b *ExecutorConfigurationBuilder) buildPlannerConfiguration(ctx context.Con subscriptionClientOptions = &SubscriptionClientOptions{} } resolvedSubscriptionClientOptions := *subscriptionClientOptions - resolvedSubscriptionClientOptions.UseNoopClient = shouldUseNoopUpstreamSubscriptionClient( - opts.EngineConfig.GetGraphqlSchema(), - opts.EngineConfig, - opts.RouterEngineConfig.Events, - opts.WebSocketConfiguration, - ) + if mondaytweaks.UseNoopUpstreamSubscriptionClientWhenUnused { + resolvedSubscriptionClientOptions.UseNoopClient = shouldUseNoopUpstreamSubscriptionClient( + opts.EngineConfig.GetGraphqlSchema(), + opts.EngineConfig, + opts.RouterEngineConfig.Events, + opts.WebSocketConfiguration, + ) + } loader := NewLoader(ctx, b.trackUsageInfo, NewDefaultFactoryResolver( ctx, diff --git a/router/core/factoryresolver.go b/router/core/factoryresolver.go index 63675acb91..46db85e1b8 100644 --- a/router/core/factoryresolver.go +++ b/router/core/factoryresolver.go @@ -18,6 +18,7 @@ import ( 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/grpcconnector" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" rmetric "github.com/wundergraph/cosmo/router/pkg/metric" "github.com/wundergraph/cosmo/router/pkg/pubsub" pubsub_datasource "github.com/wundergraph/cosmo/router/pkg/pubsub/datasource" @@ -191,7 +192,7 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla if d.transportFactory == nil || d.baseTransport == nil { // dummy implementation for plan generator that doesn't make requests - return graphql_datasource.NewFactory(d.engineCtx, http.DefaultClient, d.sharedSubscriptionClient()) + return graphql_datasource.NewFactory(d.engineCtx, http.DefaultClient, d.subscriptionClientForFactory()) } defaultHTTPClient := &http.Client{ @@ -202,10 +203,47 @@ func (d *DefaultFactoryResolver) ResolveGraphqlFactory(subgraphName string) (pla if subgraphClient, ok := d.subgraphHTTPClients[subgraphName]; ok { // it's intentional that we're not using the subgraphClient for subscriptions // custom subgraph clients are intended to be used for custom timeouts, which is not relevant for subscriptions - return graphql_datasource.NewFactory(d.engineCtx, subgraphClient, d.sharedSubscriptionClient()) + return graphql_datasource.NewFactory(d.engineCtx, subgraphClient, d.subscriptionClientForFactory()) } - return graphql_datasource.NewFactory(d.engineCtx, defaultHTTPClient, d.sharedSubscriptionClient()) + return graphql_datasource.NewFactory(d.engineCtx, defaultHTTPClient, d.subscriptionClientForFactory()) +} + +func (d *DefaultFactoryResolver) subscriptionClientForFactory() graphql_datasource.GraphQLSubscriptionClient { + if mondaytweaks.ShareUpstreamSubscriptionClient { + return d.sharedSubscriptionClient() + } + return d.newSubscriptionClient() +} + +func (d *DefaultFactoryResolver) newSubscriptionClient() graphql_datasource.GraphQLSubscriptionClient { + if d.useNoopSubscriptionClient { + return noopGraphQLSubscriptionClientInstance + } + + if d.transportFactory == nil || d.baseTransport == nil { + return graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, + d.subscriptionClientOptions..., + ) + } + + defaultHTTPClient := &http.Client{ + Timeout: d.defaultSubgraphRequestTimeout, + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + streamingClient := &http.Client{ + Transport: d.transportFactory.RoundTripper(d.baseTransport), + } + + return graphql_datasource.NewGraphQLSubscriptionClient( + d.engineCtx, + append([]graphql_datasource.SubscriptionClientOption{ + graphql_datasource.WithUpgradeClient(defaultHTTPClient), + graphql_datasource.WithStreamingClient(streamingClient), + }, d.subscriptionClientOptions...)..., + ) } func (d *DefaultFactoryResolver) sharedSubscriptionClient() graphql_datasource.GraphQLSubscriptionClient { diff --git a/router/core/graph_server.go b/router/core/graph_server.go index 9dd1b8350a..1e47962df3 100644 --- a/router/core/graph_server.go +++ b/router/core/graph_server.go @@ -23,6 +23,7 @@ import ( "github.com/golang-jwt/jwt/v5" "github.com/klauspost/compress/gzhttp" "github.com/klauspost/compress/gzip" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" "github.com/wundergraph/cosmo/router/pkg/routerconfig" "go.opentelemetry.io/otel/attribute" otelmetric "go.opentelemetry.io/otel/metric" @@ -725,15 +726,21 @@ func (s *graphMux) buildOperationCaches(srv *graphServer) (computeSha256 bool, e BufferItems: 64, } if srv.cacheWarmup != nil && srv.cacheWarmup.Enabled && srv.cacheWarmup.InMemoryFallback { - s.planCacheOnEvictEnabled.Store(true) - planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { - // This could be called before planFallbackCache is set, but it's not a problem - // because there is a nil guard inside, as well as items should not really be evicted - // on startup - if !s.planCacheOnEvictEnabled.Load() { - return + if mondaytweaks.SkipPlanCacheOnEvictDuringMuxShutdown { + s.planCacheOnEvictEnabled.Store(true) + planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { + // This could be called before planFallbackCache is set, but it's not a problem + // because there is a nil guard inside, as well as items should not really be evicted + // on startup + if !s.planCacheOnEvictEnabled.Load() { + return + } + s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) + } + } else { + planCacheConfig.OnEvict = func(item *ristretto.Item[*planWithMetaData]) { + s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) } - s.planFallbackCache.Set(item.Key, item.Value, item.Value.planningDuration) } } s.planCache, err = ristretto.NewCache[uint64, *planWithMetaData](planCacheConfig) @@ -989,31 +996,53 @@ func (s *graphMux) releaseOperationCaches() { closeRistrettoCacheUint64(&s.operationHashCache) } +func (s *graphMux) closeOperationCachesLegacy() { + s.planCache.Close() + s.planFallbackCache.Close() + s.persistedOperationCache.Close() + s.normalizationCache.Close() + s.variablesNormalizationCache.Close() + s.remapVariablesCache.Close() + s.complexityCalculationCache.Close() + s.validationCache.Close() + s.operationHashCache.Close() +} + func (s *graphMux) Shutdown(ctx context.Context) error { - // Close websocket subscriptions synchronously before tearing down plan caches so - // active preparedPlan and executor references are released first. - if s.wsHandler != nil { - s.wsHandler.ShutdownConnections() + if mondaytweaks.DrainWebsocketSubscriptionsBeforeCacheClose { + // Close websocket subscriptions synchronously before tearing down plan caches so + // active preparedPlan and executor references are released first. + if s.wsHandler != nil { + s.wsHandler.ShutdownConnections() + } } // cancel the graph muxes context to close its resources like websocket connections, resolvers, etc. s.cancel() - // ristretto Close() clears all entries and invokes OnEvict for each one. Disable - // migration into the slow-plan fallback cache during intentional mux shutdown. - s.planCacheOnEvictEnabled.Store(false) - if s.planFallbackCache != nil { - s.planFallbackCache.Wait() + if mondaytweaks.SkipPlanCacheOnEvictDuringMuxShutdown { + // ristretto Close() clears all entries and invokes OnEvict for each one. Disable + // migration into the slow-plan fallback cache during intentional mux shutdown. + s.planCacheOnEvictEnabled.Store(false) + if s.planFallbackCache != nil { + s.planFallbackCache.Wait() + } } - if s.executor != nil { - s.executor.Close() - s.executor = nil + if mondaytweaks.CloseExecutorOnGraphMuxShutdown { + if s.executor != nil { + s.executor.Close() + s.executor = nil + } } - s.releaseOperationCaches() - s.wsHandler = nil - s.mux = nil + if mondaytweaks.NilGraphMuxCachesOnShutdown { + s.releaseOperationCaches() + s.wsHandler = nil + s.mux = nil + } else { + s.closeOperationCachesLegacy() + } var err error @@ -1500,7 +1529,8 @@ func (s *graphServer) buildGraphMux( } // Client-facing WebSocket subscriptions are disabled; skip upstream ping loops // that would otherwise start one goroutine per subgraph datasource factory. - if s.webSocketConfiguration != nil && !s.webSocketConfiguration.Enabled { + if mondaytweaks.DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled && + s.webSocketConfiguration != nil && !s.webSocketConfiguration.Enabled { subscriptionClientOptions.PingInterval = 0 } @@ -1549,7 +1579,9 @@ func (s *graphServer) buildGraphMux( if err != nil { return nil, fmt.Errorf("failed to build plan configuration: %w", err) } - gm.executor = executor + if mondaytweaks.CloseExecutorOnGraphMuxShutdown { + gm.executor = executor + } s.pubSubProviders = providers if pubSubStartupErr := s.startupPubSubProviders(s.graphServerCtx); pubSubStartupErr != nil { @@ -1903,7 +1935,9 @@ func (s *graphServer) buildGraphMux( DisableVariablesRemapping: s.engineExecutionConfiguration.DisableVariablesRemapping, ApolloCompatibilityFlags: s.apolloCompatibilityFlags, }) - gm.wsHandler = wsHandler + if mondaytweaks.DrainWebsocketSubscriptionsBeforeCacheClose { + gm.wsHandler = wsHandler + } // When the playground path is equal to the graphql path, we need to handle // ws upgrades and html requests on the same route. @@ -2226,7 +2260,9 @@ func (s *graphServer) Shutdown(ctx context.Context) error { if err := mux.Shutdown(ctx); err != nil { finalErr = errors.Join(finalErr, err) } - delete(s.graphMuxList, name) + if mondaytweaks.NilGraphMuxCachesOnShutdown { + delete(s.graphMuxList, name) + } } // Close idle connections on base and subgraph transports diff --git a/router/core/router.go b/router/core/router.go index 95eaa45b1a..e1fe1cfbeb 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -19,6 +19,7 @@ import ( "connectrpc.com/connect" "github.com/mitchellh/mapstructure" "github.com/nats-io/nuid" + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" "github.com/wundergraph/cosmo/router/pkg/routerconfig" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" @@ -630,7 +631,9 @@ func (r *Router) serverTLSConfig() (*tls.Config, error) { func (r *Router) newServer(ctx context.Context, response *routerconfig.Response) error { // Extract slow-plan cache entries before building the new graph server, which // overwrites ReloadPersistentState cache references and before the old graphMux shuts down. - r.reloadPersistentState.OnRouterConfigReload() + if mondaytweaks.CallOnRouterConfigReloadOnHotReload { + r.reloadPersistentState.OnRouterConfigReload() + } server, err := newGraphServer(ctx, r, response, r.proxy) if err != nil { @@ -1104,7 +1107,7 @@ func (r *Router) bootstrap(ctx context.Context) error { r.staticExecutionConfig = executionConfig - if hash, hashErr := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path); hashErr == nil { + if hash, hashErr := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path); hashErr == nil && mondaytweaks.SkipManifestReloadWhenMapperUnchanged { r.lastManifestMapperHash = hash r.manifestMapperHashSeen = true } @@ -1727,16 +1730,18 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) return } - mapperHash, err := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path) - if err != nil { - ll.Error("Failed to hash manifest mapper", zap.Error(err)) - return - } + if mondaytweaks.SkipManifestReloadWhenMapperUnchanged { + mapperHash, err := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path) + if err != nil { + ll.Error("Failed to hash manifest mapper", zap.Error(err)) + return + } - if r.manifestMapperHashSeen && mapperHash == r.lastManifestMapperHash { - ll.Debug("Manifest mapper unchanged, skipping reload", - zap.String("path", r.manifestConfig.Path)) - return + if r.manifestMapperHashSeen && mapperHash == r.lastManifestMapperHash { + ll.Debug("Manifest mapper unchanged, skipping reload", + zap.String("path", r.manifestConfig.Path)) + return + } } cfg, err := routerconfig.AssembleStaticExecutionConfigFromManifest( @@ -1757,11 +1762,20 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) return } - r.lastManifestMapperHash = mapperHash - r.manifestMapperHashSeen = true + if mondaytweaks.SkipManifestReloadWhenMapperUnchanged { + mapperHash, err := routerconfig.ManifestMapperSHA256(r.manifestConfig.Path) + if err != nil { + ll.Error("Failed to hash manifest mapper", zap.Error(err)) + return + } + r.lastManifestMapperHash = mapperHash + r.manifestMapperHashSeen = true + } - if old := r.staticExecutionConfig; old != nil && old != cfg { - proto.Reset(old) + if mondaytweaks.ResetExecutionConfigProtoOnReload { + if old := r.staticExecutionConfig; old != nil && old != cfg { + proto.Reset(old) + } } r.staticExecutionConfig = cfg r.trackExecutionConfigUsage(cfg, true) diff --git a/router/docs/Profiling.md b/router/docs/Profiling.md index 7a1baacdd3..5db0a728c3 100644 --- a/router/docs/Profiling.md +++ b/router/docs/Profiling.md @@ -118,5 +118,12 @@ To use Pyroscope for continuous profiling of the router: 2. Run the router with either `PYROSCOPE_ADDR=http://localhost:4040` or `-pyroscope-addr http://localhost:4040` to enable sending continuous profiling data to Pyroscope. You can view this data in Grafana. + + Optional Pyroscope environment variables: + + - `PYROSCOPE_APPLICATION_NAME` — overrides the application name sent to Pyroscope. When unset, + falls back to `telemetry.service_name` from config, then `wundergraph.cosmo.router`. + - `PYROSCOPE_TAGS` — comma-separated `key=value` pairs appended to built-in tags (e.g. + `region=us-east-1,team=platform`). Custom tags override built-in tags on key collision. 3. Visit the drilldown profiles section in Grafana at `http://localhost:9300` 4. Select the router from the service dropdown diff --git a/router/pkg/mondaytweaks/mondaytweaks.go b/router/pkg/mondaytweaks/mondaytweaks.go index 226bbbf8c4..1689663e04 100644 --- a/router/pkg/mondaytweaks/mondaytweaks.go +++ b/router/pkg/mondaytweaks/mondaytweaks.go @@ -10,4 +10,62 @@ const ( // struct itself is GC'd — which may be delayed by goroutines still referencing // the owning graphMux — causing ~200-300 MB of retained memory per config reload. ClearSlowPlanCacheOnClose = true + + // OmitSchemaDocumentFromCachedPlans removes the unused schemaDocument field from + // planWithMetaData (compile-time structural change in operation_planner.go). + OmitSchemaDocumentFromCachedPlans = true + + // CallOnRouterConfigReloadOnHotReload invokes ReloadPersistentState.OnRouterConfigReload + // at the start of Router.newServer(), matching the supervisor restart path. + CallOnRouterConfigReloadOnHotReload = true + + // SkipPlanCacheOnEvictDuringMuxShutdown disables ristretto OnEvict migration into + // slowplancache while a graphMux is shutting down intentionally. + SkipPlanCacheOnEvictDuringMuxShutdown = true + + // DrainWebsocketSubscriptionsBeforeCacheClose closes client websocket subscriptions + // synchronously before plan caches are torn down on graphMux shutdown. + DrainWebsocketSubscriptionsBeforeCacheClose = true + + // CloseExecutorOnGraphMuxShutdown nils federation schema refs held by Executor after + // graphMux drain, allowing the old graph generation to be garbage-collected. + CloseExecutorOnGraphMuxShutdown = true + + // NilGraphMuxCachesOnShutdown closes and nils Ristretto caches on shut-down graphMux, + // drops wsHandler/mux references, and removes the mux from graphMuxList. + NilGraphMuxCachesOnShutdown = true + + // ResetExecutionConfigProtoOnReload proto.Resets the previous staticExecutionConfig + // after a successful manifest reload so decoded protojson strings can be collected. + ResetExecutionConfigProtoOnReload = true + + // SkipManifestReloadWhenMapperUnchanged skips manifest watcher reload when mapper.json + // bytes are unchanged. Disabled: latest.json / feature-flag files can change without + // mapper.json changing, which would serve stale config. + SkipManifestReloadWhenMapperUnchanged = false + + // ShareUpstreamSubscriptionClient uses one upstream GraphQLSubscriptionClient per + // DefaultFactoryResolver instead of one per subgraph factory (behavior-altering). + ShareUpstreamSubscriptionClient = true + + // UseNoopUpstreamSubscriptionClientWhenUnused skips upstream WS/SSE transport init + // when subscriptions are not used (behavior-altering). + UseNoopUpstreamSubscriptionClientWhenUnused = true + + // DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled sets PingInterval=0 on + // upstream subscription clients when client-facing websocket is disabled. + DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled = true + + // RegisterHeapPprofRoutes exposes /debug/pprof/heap and related routes on the + // cosmo pprof server (staging diagnostics). + RegisterHeapPprofRoutes = true + + // RereadProfilingEnvAfterFlagParse re-reads PPROF_ADDR/PYROSCOPE_ADDR after flag.Parse + // so platform-api-cosmo-router embed main() can Setenv before routercmd.Main(). + RereadProfilingEnvAfterFlagParse = true + + // ResolvePyroscopeNameAndTagsFromEnv reads PYROSCOPE_APPLICATION_NAME and + // PYROSCOPE_TAGS when starting the Pyroscope client. When disabled, uses the + // upstream hardcoded application name and hostname-only tags. + ResolvePyroscopeNameAndTagsFromEnv = true ) diff --git a/router/pkg/profile/profile.go b/router/pkg/profile/profile.go index 8ee0239848..bb1ddb4f9e 100644 --- a/router/pkg/profile/profile.go +++ b/router/pkg/profile/profile.go @@ -10,6 +10,8 @@ import ( runtimePprof "runtime/pprof" "go.uber.org/zap" + + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" ) type Profiler interface { @@ -43,6 +45,11 @@ func NewServer(addr string, log *zap.Logger) Server { mux.HandleFunc("/debug/pprof/profile", pprof.Profile) mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) mux.HandleFunc("/debug/pprof/trace", pprof.Trace) + if mondaytweaks.RegisterHeapPprofRoutes { + mux.Handle("/debug/pprof/heap", pprof.Handler("heap")) + mux.Handle("/debug/pprof/allocs", pprof.Handler("allocs")) + mux.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine")) + } svr := &http.Server{ Addr: addr, diff --git a/router/pkg/profile/pyroscope.go b/router/pkg/profile/pyroscope.go new file mode 100644 index 0000000000..60f20d716c --- /dev/null +++ b/router/pkg/profile/pyroscope.go @@ -0,0 +1,84 @@ +package profile + +import ( + "os" + "strings" + + "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" +) + +const ( + DefaultPyroscopeApplicationName = "wundergraph.cosmo.router" + + EnvPyroscopeApplicationName = "PYROSCOPE_APPLICATION_NAME" + EnvPyroscopeTags = "PYROSCOPE_TAGS" +) + +// PyroscopeApplicationName resolves the Pyroscope application name. +// When mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv is enabled, precedence is: +// PYROSCOPE_APPLICATION_NAME > telemetryServiceName > default. +// Otherwise returns the upstream hardcoded default. +func PyroscopeApplicationName(telemetryServiceName string) string { + if !mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv { + return DefaultPyroscopeApplicationName + } + + if name := os.Getenv(EnvPyroscopeApplicationName); name != "" { + return name + } + if telemetryServiceName != "" { + return telemetryServiceName + } + return DefaultPyroscopeApplicationName +} + +// PyroscopeTags builds Pyroscope tags for the running process. +// When mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv is enabled, PYROSCOPE_TAGS +// (comma-separated key=value pairs) is merged with HOSTNAME. Custom tags override +// built-in tags on key collision. Otherwise returns hostname-only tags. +func PyroscopeTags() map[string]string { + if !mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv { + return pyroscopeHostnameTag() + } + + tags := pyroscopeHostnameTag() + for key, value := range ParseKeyValueEnv(os.Getenv(EnvPyroscopeTags)) { + tags[key] = value + } + return tags +} + +func pyroscopeHostnameTag() map[string]string { + tags := map[string]string{} + if hostname := os.Getenv("HOSTNAME"); hostname != "" { + tags["hostname"] = hostname + } + return tags +} + +// ParseKeyValueEnv parses comma-separated key=value pairs. +func ParseKeyValueEnv(raw string) map[string]string { + tags := map[string]string{} + raw = strings.TrimSpace(raw) + if raw == "" { + return tags + } + + for part := range strings.SplitSeq(raw, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + key, value, ok := strings.Cut(part, "=") + if !ok { + continue + } + key = strings.TrimSpace(key) + value = strings.TrimSpace(value) + if key == "" || value == "" { + continue + } + tags[key] = value + } + return tags +} diff --git a/router/pkg/profile/pyroscope_test.go b/router/pkg/profile/pyroscope_test.go new file mode 100644 index 0000000000..ba7280ac5a --- /dev/null +++ b/router/pkg/profile/pyroscope_test.go @@ -0,0 +1,75 @@ +package profile + +import ( + "testing" +) + +func TestPyroscopeApplicationName(t *testing.T) { + tests := []struct { + name string + pyroscopeAppName string + telemetryServiceName string + want string + }{ + { + name: "PYROSCOPE_APPLICATION_NAME wins", + pyroscopeAppName: "custom-service", + want: "custom-service", + }, + { + name: "telemetry service name is second", + telemetryServiceName: "platform-api-cosmo-router", + want: "platform-api-cosmo-router", + }, + { + name: "default when unset", + want: DefaultPyroscopeApplicationName, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(EnvPyroscopeApplicationName, tc.pyroscopeAppName) + + if got := PyroscopeApplicationName(tc.telemetryServiceName); got != tc.want { + t.Fatalf("PyroscopeApplicationName() = %q, want %q", got, tc.want) + } + }) + } +} + +func TestPyroscopeTags(t *testing.T) { + t.Setenv("HOSTNAME", "pod-123") + t.Setenv(EnvPyroscopeTags, "region=us-east-1, team=platform") + + got := PyroscopeTags() + if got["hostname"] != "pod-123" { + t.Fatalf("hostname tag = %q, want %q", got["hostname"], "pod-123") + } + if got["region"] != "us-east-1" { + t.Fatalf("region tag = %q, want %q", got["region"], "us-east-1") + } + if got["team"] != "platform" { + t.Fatalf("team tag = %q, want %q", got["team"], "platform") + } +} + +func TestPyroscopeTags_CustomOverridesBuiltIn(t *testing.T) { + t.Setenv("HOSTNAME", "pod-123") + t.Setenv(EnvPyroscopeTags, "hostname=override-host") + + got := PyroscopeTags() + if got["hostname"] != "override-host" { + t.Fatalf("hostname tag = %q, want override from PYROSCOPE_TAGS", got["hostname"]) + } +} + +func TestParseKeyValueEnv(t *testing.T) { + got := ParseKeyValueEnv(" region=us-east-1 ,team=platform,invalid,=bad,key=") + if len(got) != 2 { + t.Fatalf("ParseKeyValueEnv() len = %d, want 2 (%v)", len(got), got) + } + if got["region"] != "us-east-1" || got["team"] != "platform" { + t.Fatalf("ParseKeyValueEnv() = %v", got) + } +} From 8941675406ed5c0bed59acd70a326f0863724a74 Mon Sep 17 00:00:00 2001 From: Adam Rutkowski Date: Thu, 25 Jun 2026 17:40:59 +0200 Subject: [PATCH 13/13] chore(router): drop profiling and pyroscope mondaytweaks Remove PPROF/PYROSCOPE env re-read, heap pprof routes, and Pyroscope name/tag helpers so the PR stays focused on config reload memory fixes. --- router/cmd/main.go | 23 +------ router/docs/Profiling.md | 7 --- router/pkg/mondaytweaks/mondaytweaks.go | 13 ---- router/pkg/profile/profile.go | 7 --- router/pkg/profile/pyroscope.go | 84 ------------------------- router/pkg/profile/pyroscope_test.go | 75 ---------------------- 6 files changed, 2 insertions(+), 207 deletions(-) delete mode 100644 router/pkg/profile/pyroscope.go delete mode 100644 router/pkg/profile/pyroscope_test.go diff --git a/router/cmd/main.go b/router/cmd/main.go index 094db3dbd6..6ec9c9e1f1 100644 --- a/router/cmd/main.go +++ b/router/cmd/main.go @@ -19,7 +19,6 @@ import ( "github.com/wundergraph/cosmo/router/internal/versioninfo" "github.com/wundergraph/cosmo/router/pkg/config" "github.com/wundergraph/cosmo/router/pkg/logging" - "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" "github.com/wundergraph/cosmo/router/pkg/profile" "github.com/wundergraph/cosmo/router/pkg/watcher" @@ -47,21 +46,6 @@ func Main() { // Parse flags before calling profile.Start(), since it may add flags flag.Parse() - // Re-read profiling env after flag.Parse() — flag defaults are captured at package - // init, before embedders (e.g. platform-api-cosmo-router) can Setenv in main(). - if mondaytweaks.RereadProfilingEnvAfterFlagParse { - if *pprofListenAddr == "" { - if addr := os.Getenv("PPROF_ADDR"); addr != "" { - *pprofListenAddr = addr - } - } - if *pyroscopeAddr == "" { - if addr := os.Getenv("PYROSCOPE_ADDR"); addr != "" { - *pyroscopeAddr = addr - } - } - } - if *help { flag.PrintDefaults() os.Exit(0) @@ -138,14 +122,11 @@ func Main() { logger := baseLogger.With(zap.String("component", "pyroscope")) logger.Info("starting pyroscope server") - applicationName := profile.PyroscopeApplicationName(result.Config.Telemetry.ServiceName) - tags := profile.PyroscopeTags() - pyro, err := pyroscope.Start(pyroscope.Config{ - ApplicationName: applicationName, + ApplicationName: "wundergraph.cosmo.router", ServerAddress: *pyroscopeAddr, Logger: logger.Sugar(), - Tags: tags, + Tags: map[string]string{"hostname": os.Getenv("HOSTNAME")}, ProfileTypes: []pyroscope.ProfileType{ pyroscope.ProfileCPU, diff --git a/router/docs/Profiling.md b/router/docs/Profiling.md index 5db0a728c3..7a1baacdd3 100644 --- a/router/docs/Profiling.md +++ b/router/docs/Profiling.md @@ -118,12 +118,5 @@ To use Pyroscope for continuous profiling of the router: 2. Run the router with either `PYROSCOPE_ADDR=http://localhost:4040` or `-pyroscope-addr http://localhost:4040` to enable sending continuous profiling data to Pyroscope. You can view this data in Grafana. - - Optional Pyroscope environment variables: - - - `PYROSCOPE_APPLICATION_NAME` — overrides the application name sent to Pyroscope. When unset, - falls back to `telemetry.service_name` from config, then `wundergraph.cosmo.router`. - - `PYROSCOPE_TAGS` — comma-separated `key=value` pairs appended to built-in tags (e.g. - `region=us-east-1,team=platform`). Custom tags override built-in tags on key collision. 3. Visit the drilldown profiles section in Grafana at `http://localhost:9300` 4. Select the router from the service dropdown diff --git a/router/pkg/mondaytweaks/mondaytweaks.go b/router/pkg/mondaytweaks/mondaytweaks.go index 1689663e04..ab53923cf1 100644 --- a/router/pkg/mondaytweaks/mondaytweaks.go +++ b/router/pkg/mondaytweaks/mondaytweaks.go @@ -55,17 +55,4 @@ const ( // DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled sets PingInterval=0 on // upstream subscription clients when client-facing websocket is disabled. DisableUpstreamSubscriptionPingWhenClientWebSocketDisabled = true - - // RegisterHeapPprofRoutes exposes /debug/pprof/heap and related routes on the - // cosmo pprof server (staging diagnostics). - RegisterHeapPprofRoutes = true - - // RereadProfilingEnvAfterFlagParse re-reads PPROF_ADDR/PYROSCOPE_ADDR after flag.Parse - // so platform-api-cosmo-router embed main() can Setenv before routercmd.Main(). - RereadProfilingEnvAfterFlagParse = true - - // ResolvePyroscopeNameAndTagsFromEnv reads PYROSCOPE_APPLICATION_NAME and - // PYROSCOPE_TAGS when starting the Pyroscope client. When disabled, uses the - // upstream hardcoded application name and hostname-only tags. - ResolvePyroscopeNameAndTagsFromEnv = true ) diff --git a/router/pkg/profile/profile.go b/router/pkg/profile/profile.go index bb1ddb4f9e..8ee0239848 100644 --- a/router/pkg/profile/profile.go +++ b/router/pkg/profile/profile.go @@ -10,8 +10,6 @@ import ( runtimePprof "runtime/pprof" "go.uber.org/zap" - - "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" ) type Profiler interface { @@ -45,11 +43,6 @@ func NewServer(addr string, log *zap.Logger) Server { mux.HandleFunc("/debug/pprof/profile", pprof.Profile) mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol) mux.HandleFunc("/debug/pprof/trace", pprof.Trace) - if mondaytweaks.RegisterHeapPprofRoutes { - mux.Handle("/debug/pprof/heap", pprof.Handler("heap")) - mux.Handle("/debug/pprof/allocs", pprof.Handler("allocs")) - mux.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine")) - } svr := &http.Server{ Addr: addr, diff --git a/router/pkg/profile/pyroscope.go b/router/pkg/profile/pyroscope.go deleted file mode 100644 index 60f20d716c..0000000000 --- a/router/pkg/profile/pyroscope.go +++ /dev/null @@ -1,84 +0,0 @@ -package profile - -import ( - "os" - "strings" - - "github.com/wundergraph/cosmo/router/pkg/mondaytweaks" -) - -const ( - DefaultPyroscopeApplicationName = "wundergraph.cosmo.router" - - EnvPyroscopeApplicationName = "PYROSCOPE_APPLICATION_NAME" - EnvPyroscopeTags = "PYROSCOPE_TAGS" -) - -// PyroscopeApplicationName resolves the Pyroscope application name. -// When mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv is enabled, precedence is: -// PYROSCOPE_APPLICATION_NAME > telemetryServiceName > default. -// Otherwise returns the upstream hardcoded default. -func PyroscopeApplicationName(telemetryServiceName string) string { - if !mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv { - return DefaultPyroscopeApplicationName - } - - if name := os.Getenv(EnvPyroscopeApplicationName); name != "" { - return name - } - if telemetryServiceName != "" { - return telemetryServiceName - } - return DefaultPyroscopeApplicationName -} - -// PyroscopeTags builds Pyroscope tags for the running process. -// When mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv is enabled, PYROSCOPE_TAGS -// (comma-separated key=value pairs) is merged with HOSTNAME. Custom tags override -// built-in tags on key collision. Otherwise returns hostname-only tags. -func PyroscopeTags() map[string]string { - if !mondaytweaks.ResolvePyroscopeNameAndTagsFromEnv { - return pyroscopeHostnameTag() - } - - tags := pyroscopeHostnameTag() - for key, value := range ParseKeyValueEnv(os.Getenv(EnvPyroscopeTags)) { - tags[key] = value - } - return tags -} - -func pyroscopeHostnameTag() map[string]string { - tags := map[string]string{} - if hostname := os.Getenv("HOSTNAME"); hostname != "" { - tags["hostname"] = hostname - } - return tags -} - -// ParseKeyValueEnv parses comma-separated key=value pairs. -func ParseKeyValueEnv(raw string) map[string]string { - tags := map[string]string{} - raw = strings.TrimSpace(raw) - if raw == "" { - return tags - } - - for part := range strings.SplitSeq(raw, ",") { - part = strings.TrimSpace(part) - if part == "" { - continue - } - key, value, ok := strings.Cut(part, "=") - if !ok { - continue - } - key = strings.TrimSpace(key) - value = strings.TrimSpace(value) - if key == "" || value == "" { - continue - } - tags[key] = value - } - return tags -} diff --git a/router/pkg/profile/pyroscope_test.go b/router/pkg/profile/pyroscope_test.go deleted file mode 100644 index ba7280ac5a..0000000000 --- a/router/pkg/profile/pyroscope_test.go +++ /dev/null @@ -1,75 +0,0 @@ -package profile - -import ( - "testing" -) - -func TestPyroscopeApplicationName(t *testing.T) { - tests := []struct { - name string - pyroscopeAppName string - telemetryServiceName string - want string - }{ - { - name: "PYROSCOPE_APPLICATION_NAME wins", - pyroscopeAppName: "custom-service", - want: "custom-service", - }, - { - name: "telemetry service name is second", - telemetryServiceName: "platform-api-cosmo-router", - want: "platform-api-cosmo-router", - }, - { - name: "default when unset", - want: DefaultPyroscopeApplicationName, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - t.Setenv(EnvPyroscopeApplicationName, tc.pyroscopeAppName) - - if got := PyroscopeApplicationName(tc.telemetryServiceName); got != tc.want { - t.Fatalf("PyroscopeApplicationName() = %q, want %q", got, tc.want) - } - }) - } -} - -func TestPyroscopeTags(t *testing.T) { - t.Setenv("HOSTNAME", "pod-123") - t.Setenv(EnvPyroscopeTags, "region=us-east-1, team=platform") - - got := PyroscopeTags() - if got["hostname"] != "pod-123" { - t.Fatalf("hostname tag = %q, want %q", got["hostname"], "pod-123") - } - if got["region"] != "us-east-1" { - t.Fatalf("region tag = %q, want %q", got["region"], "us-east-1") - } - if got["team"] != "platform" { - t.Fatalf("team tag = %q, want %q", got["team"], "platform") - } -} - -func TestPyroscopeTags_CustomOverridesBuiltIn(t *testing.T) { - t.Setenv("HOSTNAME", "pod-123") - t.Setenv(EnvPyroscopeTags, "hostname=override-host") - - got := PyroscopeTags() - if got["hostname"] != "override-host" { - t.Fatalf("hostname tag = %q, want override from PYROSCOPE_TAGS", got["hostname"]) - } -} - -func TestParseKeyValueEnv(t *testing.T) { - got := ParseKeyValueEnv(" region=us-east-1 ,team=platform,invalid,=bad,key=") - if len(got) != 2 { - t.Fatalf("ParseKeyValueEnv() len = %d, want 2 (%v)", len(got), got) - } - if got["region"] != "us-east-1" || got["team"] != "platform" { - t.Fatalf("ParseKeyValueEnv() = %v", got) - } -}