feat: add observability on subgraph connections and resolvers

router-tests/observability/prometheus_test.go

@@ -4193,6 +4193,7 @@ func TestPrometheus(t *testing.T) {
 			MetricOptions: testenv.MetricOptions{
 				PrometheusEngineStatsOptions: testenv.EngineStatOptions{
 					EnableSubscription: true,
+					EnableResolver:     true,
 				},
 			},
 		}, func(t *testing.T, xEnv *testenv.Environment) {
@@ -4244,18 +4245,22 @@ func TestPrometheus(t *testing.T) {
 
 			require.NoError(t, err)
 			mf := findEngineMetrics(promMetrics)
-			require.Len(t, mf, 4)
+			require.Len(t, mf, 6)
 
 			// Connection stats
 			connectionMetrics := findMetricFamilyByName(mf, "router_engine_connections")
 			subscriptionMetrics := findMetricFamilyByName(mf, "router_engine_subscriptions")
 			triggerMetrics := findMetricFamilyByName(mf, "router_engine_triggers")
 			messagesSentCounter := findMetricFamilyByName(mf, "router_engine_messages_sent_total")
+			resolversMaxConcurrentMetrics := findMetricFamilyByName(mf, "router_engine_resolvers_max_concurrent")
+			resolversInflightMetrics := findMetricFamilyByName(mf, "router_engine_resolvers_inflight")
 
 			require.NotNil(t, connectionMetrics)
 			require.NotNil(t, subscriptionMetrics)
 			require.NotNil(t, triggerMetrics)
 			require.NotNil(t, messagesSentCounter)
+			require.NotNil(t, resolversMaxConcurrentMetrics)
+			require.NotNil(t, resolversInflightMetrics)
 
 			// We only provide base attributes here. In the testing scenario we don't have any additional attributes
 			// that can increase the cardinality.
@@ -4275,6 +4280,14 @@ func TestPrometheus(t *testing.T) {
 			require.GreaterOrEqual(t, messagesSentCounter.Metric[0].GetCounter().GetValue(), float64(1))
 			require.ElementsMatch(t, baseAttributes, messagesSentCounter.Metric[0].Label)
 
+			require.Len(t, resolversMaxConcurrentMetrics.Metric, 1)
+			require.Greater(t, resolversMaxConcurrentMetrics.Metric[0].GetGauge().GetValue(), float64(0))
+			require.ElementsMatch(t, baseAttributes, resolversMaxConcurrentMetrics.Metric[0].Label)
+
+			require.Len(t, resolversInflightMetrics.Metric, 1)
+			require.GreaterOrEqual(t, resolversInflightMetrics.Metric[0].GetGauge().GetValue(), float64(0))
+			require.ElementsMatch(t, baseAttributes, resolversInflightMetrics.Metric[0].Label)
+
 			// close the connection
 			require.NoError(t, conn.Close())
 
@@ -4283,7 +4296,7 @@ func TestPrometheus(t *testing.T) {
 			promMetrics, err = promRegistry.Gather()
 			require.NoError(t, err)
 			mf = findEngineMetrics(promMetrics)
-			require.Len(t, mf, 4)
+			require.Len(t, mf, 6)
 
 			connectionMetrics = findMetricFamilyByName(mf, "router_engine_connections")
 			subscriptionMetrics = findMetricFamilyByName(mf, "router_engine_subscriptions")
@@ -5304,6 +5317,48 @@ func TestFlakyPrometheusRouterConnectionMetrics(t *testing.T) {
 		})
 	})
 
+	t.Run("validate http client trace metrics are present when enhanced_connection_stats is enabled", func(t *testing.T) {
+		t.Parallel()
+
+		promRegistry := prometheus.NewRegistry()
+		metricReader := metric.NewManualReader()
+
+		testenv.Run(t, &testenv.Config{
+			MetricReader:       metricReader,
+			PrometheusRegistry: promRegistry,
+			MetricOptions: testenv.MetricOptions{
+				EnablePrometheusEnhancedConnectionStats: true,
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `query { employees { id isAvailable } }`,
+			})
+
+			mf, err := promRegistry.Gather()
+			require.NoError(t, err)
+
+			// TTFB fires on every HTTP request. TCP connect fires on first
+			// connection to the test subgraph. DNS/TLS phases are skipped here
+			// because subgraphs are HTTP and listen on 127.0.0.1 (no name
+			// resolution, no TLS handshake) — Prometheus only emits histogram
+			// families that have recorded at least one observation, so we don't
+			// assert presence of dns_lookup or tls_handshake here. The
+			// instruments are still created; manual verification with a real
+			// remote HTTPS subgraph would surface them.
+			ttfb := findMetricFamilyByName(mf, "router_http_client_time_to_first_byte")
+			require.NotNil(t, ttfb)
+			require.NotEmpty(t, ttfb.GetMetric())
+
+			tcpConnect := findMetricFamilyByName(mf, "router_http_client_tcp_connect_duration")
+			require.NotNil(t, tcpConnect)
+			require.NotEmpty(t, tcpConnect.GetMetric())
+
+			// And the existing acquire_duration metric must still be there since
+			// enabling enhanced_connection_stats implies enabling the connection metric store.
+			require.NotNil(t, findMetricFamilyByName(mf, "router_http_client_connection_acquire_duration"))
+		})
+	})
+
 }
 
 func TestExcludeAttributesWithCustomExporterPrometheus(t *testing.T) {
@@ -5532,6 +5587,7 @@ func TestExcludeAttributesWithCustomExporterPrometheus(t *testing.T) {
 					MetricOptions: testenv.MetricOptions{
 						PrometheusEngineStatsOptions: testenv.EngineStatOptions{
 							EnableSubscription: true,
+							EnableResolver:     true,
 						},
 					},
 					DisableSimulateCloudExporter: usingCustomExporter != UseCloudExporter,
@@ -5604,18 +5660,22 @@ func TestExcludeAttributesWithCustomExporterPrometheus(t *testing.T) {
 
 					require.NoError(t, err)
 					mf := findEngineMetrics(promMetrics)
-					require.Len(t, mf, 4)
+					require.Len(t, mf, 6)
 
 					// Connection stats
 					connectionMetrics := findMetricFamilyByName(mf, "router_engine_connections")
 					subscriptionMetrics := findMetricFamilyByName(mf, "router_engine_subscriptions")
 					triggerMetrics := findMetricFamilyByName(mf, "router_engine_triggers")
 					messagesSentCounter := findMetricFamilyByName(mf, "router_engine_messages_sent_total")
+					resolversMaxConcurrentMetrics := findMetricFamilyByName(mf, "router_engine_resolvers_max_concurrent")
+					resolversInflightMetrics := findMetricFamilyByName(mf, "router_engine_resolvers_inflight")
 
 					require.NotNil(t, connectionMetrics)
 					require.NotNil(t, subscriptionMetrics)
 					require.NotNil(t, triggerMetrics)
 					require.NotNil(t, messagesSentCounter)
+					require.NotNil(t, resolversMaxConcurrentMetrics)
+					require.NotNil(t, resolversInflightMetrics)
 
 					// We only provide base attributes here. In the testing scenario we don't have any additional attributes
 					// that can increase the cardinality.
@@ -5631,6 +5691,12 @@ func TestExcludeAttributesWithCustomExporterPrometheus(t *testing.T) {
 					require.Len(t, messagesSentCounter.Metric, 1)
 					require.ElementsMatch(t, baseAttributes, messagesSentCounter.Metric[0].Label)
 
+					require.Len(t, resolversMaxConcurrentMetrics.Metric, 1)
+					require.ElementsMatch(t, baseAttributes, resolversMaxConcurrentMetrics.Metric[0].Label)
+
+					require.Len(t, resolversInflightMetrics.Metric, 1)
+					require.ElementsMatch(t, baseAttributes, resolversInflightMetrics.Metric[0].Label)
+
 					// close the connection
 					require.NoError(t, conn.Close())
 
@@ -5639,7 +5705,7 @@ func TestExcludeAttributesWithCustomExporterPrometheus(t *testing.T) {
 					promMetrics, err = promRegistry.Gather()
 					require.NoError(t, err)
 					mf = findEngineMetrics(promMetrics)
-					require.Len(t, mf, 4)
+					require.Len(t, mf, 6)
 
 					connectionMetrics = findMetricFamilyByName(mf, "router_engine_connections")
 					subscriptionMetrics = findMetricFamilyByName(mf, "router_engine_subscriptions")

router-tests/testenv/sync_reporter.go

@@ -31,18 +31,20 @@ type Event struct {
 // snapshot-based wait semantics local to router-tests while also emitting
 // buffered events for publish-retry helpers.
 type SyncReporter struct {
-	cond   *sync.Cond
-	report statistics.UsageReport
-	events chan Event
+	cond              *sync.Cond
+	report            statistics.UsageReport
+	events            chan Event
+	resolverReporters map[statistics.ResolverConcurrencyReporter]struct{}
 }
 
 var _ statistics.EngineStatistics = (*SyncReporter)(nil)
 
 // NewSyncReporter creates a SyncReporter for router-tests.
 func NewSyncReporter(_ context.Context, _ *zap.Logger) *SyncReporter {
 	return &SyncReporter{
-		cond:   sync.NewCond(&sync.Mutex{}),
-		events: make(chan Event, 256),
+		cond:              sync.NewCond(&sync.Mutex{}),
+		events:            make(chan Event, 256),
+		resolverReporters: make(map[statistics.ResolverConcurrencyReporter]struct{}),
 	}
 }
 
@@ -153,9 +155,31 @@ func (sr *SyncReporter) withReport(update func(report *statistics.UsageReport))
 
 func (sr *SyncReporter) snapshotLocked() *statistics.UsageReport {
 	report := sr.report
+	for r := range sr.resolverReporters {
+		report.ResolverMaxConcurrent += uint64(r.MaxConcurrentResolves())
+		report.ResolverInflight += uint64(r.InflightResolves())
+	}
 	return &report
 }
 
+func (sr *SyncReporter) RegisterResolver(r statistics.ResolverConcurrencyReporter) {
+	if r == nil {
+		return
+	}
+	sr.cond.L.Lock()
+	sr.resolverReporters[r] = struct{}{}
+	sr.cond.L.Unlock()
+}
+
+func (sr *SyncReporter) UnregisterResolver(r statistics.ResolverConcurrencyReporter) {
+	if r == nil {
+		return
+	}
+	sr.cond.L.Lock()
+	delete(sr.resolverReporters, r)
+	sr.cond.L.Unlock()
+}
+
 // WaitForEvent drains events until one matching kind is received, or ctx is cancelled.
 func (sr *SyncReporter) WaitForEvent(ctx context.Context, kind EventKind) bool {
 	for {

router-tests/testenv/testenv.go

@@ -269,6 +269,7 @@ type MetricExclusions struct {
 
 type EngineStatOptions struct {
 	EnableSubscription bool
+	EnableResolver     bool
 }
 
 type MetricsLogExporterOptions struct {
@@ -279,22 +280,24 @@ type MetricsLogExporterOptions struct {
 }
 
 type MetricOptions struct {
-	MetricExclusions                      MetricExclusions
-	EnableRuntimeMetrics                  bool
-	EnableOTLPRouterCache                 bool
-	EnablePrometheusRouterCache           bool
-	OTLPEngineStatsOptions                EngineStatOptions
-	PrometheusEngineStatsOptions          EngineStatOptions
-	PrometheusSchemaFieldUsage            PrometheusSchemaFieldUsage
-	EnableOTLPConnectionMetrics           bool
-	EnableOTLPCircuitBreakerMetrics       bool
-	EnableOTLPStreamMetrics               bool
-	EnablePrometheusConnectionMetrics     bool
-	EnablePrometheusCircuitBreakerMetrics bool
-	EnablePrometheusStreamMetrics         bool
-	LogExporter                           MetricsLogExporterOptions
-	OTLPCostStats                         config.CostStats
-	PrometheusCostStats                   config.CostStats
+	MetricExclusions                        MetricExclusions
+	EnableRuntimeMetrics                    bool
+	EnableOTLPRouterCache                   bool
+	EnablePrometheusRouterCache             bool
+	OTLPEngineStatsOptions                  EngineStatOptions
+	PrometheusEngineStatsOptions            EngineStatOptions
+	PrometheusSchemaFieldUsage              PrometheusSchemaFieldUsage
+	EnableOTLPConnectionMetrics             bool
+	EnableOTLPEnhancedConnectionStats       bool
+	EnableOTLPCircuitBreakerMetrics         bool
+	EnableOTLPStreamMetrics                 bool
+	EnablePrometheusConnectionMetrics       bool
+	EnablePrometheusEnhancedConnectionStats bool
+	EnablePrometheusCircuitBreakerMetrics   bool
+	EnablePrometheusStreamMetrics           bool
+	LogExporter                             MetricsLogExporterOptions
+	OTLPCostStats                           config.CostStats
+	PrometheusCostStats                     config.CostStats
 }
 
 type PrometheusSchemaFieldUsage struct {
@@ -1619,14 +1622,16 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 		}
 
 		prometheusConfig = rmetric.PrometheusConfig{
-			Enabled:         true,
-			ListenAddr:      fmt.Sprintf("localhost:%d", testConfig.PrometheusPort),
-			Path:            "/metrics",
-			TestRegistry:    testConfig.PrometheusRegistry,
-			GraphqlCache:    testConfig.MetricOptions.EnablePrometheusRouterCache,
-			ConnectionStats: testConfig.MetricOptions.EnablePrometheusConnectionMetrics,
+			Enabled:                 true,
+			ListenAddr:              fmt.Sprintf("localhost:%d", testConfig.PrometheusPort),
+			Path:                    "/metrics",
+			TestRegistry:            testConfig.PrometheusRegistry,
+			GraphqlCache:            testConfig.MetricOptions.EnablePrometheusRouterCache,
+			ConnectionStats:         testConfig.MetricOptions.EnablePrometheusConnectionMetrics,
+			EnhancedConnectionStats: testConfig.MetricOptions.EnablePrometheusEnhancedConnectionStats,
 			EngineStats: rmetric.EngineStatsConfig{
 				Subscription: testConfig.MetricOptions.PrometheusEngineStatsOptions.EnableSubscription,
+				Resolver:     testConfig.MetricOptions.PrometheusEngineStatsOptions.EnableResolver,
 			},
 			CircuitBreaker:       testConfig.MetricOptions.EnablePrometheusCircuitBreakerMetrics,
 			CostStats:            testConfig.MetricOptions.PrometheusCostStats,
@@ -1649,13 +1654,15 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 					Enabled: true,
 				},
 				OTLP: config.MetricsOTLP{
-					Enabled:         true,
-					RouterRuntime:   testConfig.MetricOptions.EnableRuntimeMetrics,
-					GraphqlCache:    testConfig.MetricOptions.EnableOTLPRouterCache,
-					Streams:         testConfig.MetricOptions.EnableOTLPStreamMetrics,
-					ConnectionStats: testConfig.MetricOptions.EnableOTLPConnectionMetrics,
+					Enabled:                 true,
+					RouterRuntime:           testConfig.MetricOptions.EnableRuntimeMetrics,
+					GraphqlCache:            testConfig.MetricOptions.EnableOTLPRouterCache,
+					Streams:                 testConfig.MetricOptions.EnableOTLPStreamMetrics,
+					ConnectionStats:         testConfig.MetricOptions.EnableOTLPConnectionMetrics,
+					EnhancedConnectionStats: testConfig.MetricOptions.EnableOTLPEnhancedConnectionStats,
 					EngineStats: config.EngineStats{
 						Subscriptions: testConfig.MetricOptions.OTLPEngineStatsOptions.EnableSubscription,
+						Resolvers:     testConfig.MetricOptions.OTLPEngineStatsOptions.EnableResolver,
 					},
 					CircuitBreaker:      testConfig.MetricOptions.EnableOTLPCircuitBreakerMetrics,
 					CostStats:           testConfig.MetricOptions.OTLPCostStats,

router/core/graph_server.go

@@ -156,9 +156,13 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig
 		return nil, fmt.Errorf(`the compatibility version "%s" is not compatible with this router version`, response.Config.CompatibilityVersion)
 	}
 
-	isConnStoreEnabled := r.metricConfig.OpenTelemetry.ConnectionStats || r.metricConfig.Prometheus.ConnectionStats
+	// Active-connection tracking via TraceDialer is only needed when ConnectionStats is on.
+	// The httptrace-based client trace phases (EnhancedConnectionStats) attach in the RoundTripper
+	// and don't require the dialer.
+	enhancedStatsEnabled := r.metricConfig.OpenTelemetry.EnhancedConnectionStats || r.metricConfig.Prometheus.EnhancedConnectionStats
+	connectionStatsEnabled := enhancedStatsEnabled || r.metricConfig.OpenTelemetry.ConnectionStats || r.metricConfig.Prometheus.ConnectionStats
 	var traceDialer *TraceDialer
-	if isConnStoreEnabled {
+	if connectionStatsEnabled {
 		traceDialer = NewTraceDialer()
 	}
 
@@ -248,7 +252,7 @@ func newGraphServer(routerCtx context.Context, r *Router, response *routerconfig
 		}
 	}
 
-	if isConnStoreEnabled {
+	if connectionStatsEnabled {
 		connStore, err := rmetric.NewConnectionMetricStore(
 			s.logger,
 			nil,
@@ -1049,10 +1053,12 @@ func (s *graphServer) buildGraphMux(
 		// but in this case we use the same metric store
 		otlpOpts := rmetric.MetricOpts{
 			EnableCircuitBreaker: s.metricConfig.OpenTelemetry.Enabled,
+			EngineResolverStats:  s.metricConfig.OpenTelemetry.EngineStats.Resolver,
 			CostStats:            s.metricConfig.OpenTelemetry.CostStats,
 		}
 		promOpts := rmetric.MetricOpts{
 			EnableCircuitBreaker: s.metricConfig.Prometheus.Enabled,
+			EngineResolverStats:  s.metricConfig.Prometheus.EngineStats.Resolver,
 			CostStats:            s.metricConfig.Prometheus.CostStats,
 		}
 
@@ -1454,6 +1460,13 @@ func (s *graphServer) buildGraphMux(
 		return nil, fmt.Errorf("failed to build plan configuration: %w", err)
 	}
 
+	if s.engineStats != nil && executor.Resolver != nil {
+		s.engineStats.RegisterResolver(executor.Resolver)
+		context.AfterFunc(graphMuxCtx, func() {
+			s.engineStats.UnregisterResolver(executor.Resolver)
+		})
+	}
+
 	s.pubSubProviders = providers
 	if pubSubStartupErr := s.startupPubSubProviders(s.graphServerCtx); pubSubStartupErr != nil {
 		return nil, pubSubStartupErr
@@ -1685,6 +1698,7 @@ func (s *graphServer) buildGraphMux(
 		EnableResponseHeaderPropagation: s.headerRules != nil,
 		EnableCostResponseHeaders:       s.securityConfiguration.CostControl != nil && s.securityConfiguration.CostControl.ExposeHeaders,
 		EngineStats:                     s.engineStats,
+		MetricStore:                     gm.metricStore,
 		TracerProvider:                  s.tracerProvider,
 		Authorizer:                      NewCosmoAuthorizer(authorizerOptions),
 		SubgraphErrorPropagation:        s.subgraphErrorPropagation,