fix: improve router.http.request.duration_milliseconds

router-tests/prometheus_parallel_subgraph_metrics_test.go

@@ -0,0 +1,74 @@
+package integration
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"github.com/wundergraph/cosmo/router-tests/testenv"
+	"go.opentelemetry.io/otel/sdk/metric"
+)
+
+func TestPrometheusParallelSubgraphRequestDurationMetrics(t *testing.T) {
+	t.Parallel()
+
+	const productsDelay = 1200 * time.Millisecond
+
+	metricReader := metric.NewManualReader()
+	promRegistry := prometheus.NewRegistry()
+
+	testenv.Run(t, &testenv.Config{
+		MetricReader:       metricReader,
+		PrometheusRegistry: promRegistry,
+		Subgraphs: testenv.SubgraphsConfig{
+			Products: testenv.SubgraphConfig{
+				Delay: productsDelay,
+			},
+		},
+	}, func(t *testing.T, xEnv *testenv.Environment) {
+		res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+			Query: `query TestParallel2 {
+				employee(id: 1) {
+					id
+				}
+				productTypes {
+					... on Consultancy {
+						name
+					}
+					... on Cosmo {
+						name
+					}
+				}
+			}`,
+		})
+
+		require.Contains(t, res.Body, `"employee":{"id":1}`)
+		require.Contains(t, res.Body, `"productTypes"`)
+
+		metricFamily, err := promRegistry.Gather()
+		require.NoError(t, err)
+
+		requestDuration := findMetricFamilyByName(metricFamily, "router_http_request_duration_milliseconds")
+		require.NotNil(t, requestDuration)
+
+		employeesDurationMetrics := findMetricsByLabel(requestDuration, "wg_subgraph_name", "employees")
+		require.Len(t, employeesDurationMetrics, 1)
+
+		productsDurationMetrics := findMetricsByLabel(requestDuration, "wg_subgraph_name", "products")
+		require.Len(t, productsDurationMetrics, 1)
+
+		employeesHistogram := employeesDurationMetrics[0].GetHistogram()
+		productsHistogram := productsDurationMetrics[0].GetHistogram()
+		require.NotNil(t, employeesHistogram)
+		require.NotNil(t, productsHistogram)
+		require.EqualValues(t, 1, employeesHistogram.GetSampleCount())
+		require.EqualValues(t, 1, productsHistogram.GetSampleCount())
+
+		employeesDurationMs := employeesHistogram.GetSampleSum()
+		productsDurationMs := productsHistogram.GetSampleSum()
+
+		require.GreaterOrEqual(t, productsDurationMs, float64(productsDelay.Milliseconds()))
+		require.Less(t, employeesDurationMs, float64(productsDelay.Milliseconds()))
+	})
+}

router/core/engine_loader_hooks.go

@@ -52,10 +52,6 @@ type engineLoaderHooks struct {
 	headerPropagation         *HeaderPropagation
 }
 
-type engineLoaderHooksRequestContext struct {
-	startTime time.Time
-}
-
 func NewEngineRequestHooks(
 	metricStore metric.Store,
 	logger *requestlogger.SubgraphAccessLogger,
@@ -97,8 +93,6 @@ func (f *engineLoaderHooks) OnLoad(ctx context.Context, ds resolve.DataSourceInf
 		return ctx
 	}
 
-	start := time.Now()
-
 	ctx = context.WithValue(ctx, rcontext.CurrentSubgraphContextKey{}, ds.Name)
 
 	duration := atomic.Int64{}
@@ -116,9 +110,7 @@ func (f *engineLoaderHooks) OnLoad(ctx context.Context, ds resolve.DataSourceInf
 		}...),
 	)
 
-	return context.WithValue(ctx, rcontext.EngineLoaderHooksContextKey, &engineLoaderHooksRequestContext{
-		startTime: start,
-	})
+	return ctx
 }
 
 func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourceInfo, responseInfo *resolve.ResponseInfo) {
@@ -148,13 +140,6 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 		return
 	}
 
-	hookCtx, ok := ctx.Value(rcontext.EngineLoaderHooksContextKey).(*engineLoaderHooksRequestContext)
-	if !ok {
-		return
-	}
-
-	latency := time.Since(hookCtx.startTime)
-
 	span := trace.SpanFromContext(ctx)
 	defer span.End()
 
@@ -175,9 +160,11 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 	exprCtx.Subgraph.Name = ds.Name
 	exprCtx.Subgraph.Request.Error = WrapExprError(responseInfo.Err)
 
+	var fetchLatency time.Duration
 	if value := ctx.Value(rcontext.FetchTimingKey); value != nil {
 		if fetchTiming, ok := value.(*atomic.Int64); ok {
-			exprCtx.Subgraph.Request.ClientTrace.FetchDuration = time.Duration(fetchTiming.Load())
+			fetchLatency = time.Duration(fetchTiming.Load())
+			exprCtx.Subgraph.Request.ClientTrace.FetchDuration = fetchLatency
 		}
 	}
 
@@ -228,7 +215,7 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 			zap.String("subgraph_name", ds.Name),
 			zap.String("subgraph_id", ds.ID),
 			zap.Int("status", responseInfo.StatusCode),
-			zap.Duration("latency", latency),
+			zap.Duration("latency", fetchLatency),
 		}
 		path := ds.Name
 		if responseInfo.Request != nil {
@@ -299,10 +286,10 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 
 		attrOpt := otelmetric.WithAttributeSet(attribute.NewSet(metricAttrs...))
 		f.metricStore.MeasureRequestCount(ctx, metricSliceAttrs, attrOpt)
-		f.metricStore.MeasureLatency(ctx, latency, metricSliceAttrs, attrOpt)
+		f.metricStore.MeasureLatency(ctx, fetchLatency, metricSliceAttrs, attrOpt)
 	} else {
 		f.metricStore.MeasureRequestCount(ctx, reqContext.telemetry.metricSliceAttrs, metricAddOpt)
-		f.metricStore.MeasureLatency(ctx, latency, reqContext.telemetry.metricSliceAttrs, metricAddOpt)
+		f.metricStore.MeasureLatency(ctx, fetchLatency, reqContext.telemetry.metricSliceAttrs, metricAddOpt)
 	}
 
 	span.SetAttributes(traceAttrs...)