feat: add timings per client fetch for GraphQL http

router-tests/structured_logging_test.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -2867,6 +2868,121 @@ func TestFlakyAccessLogs(t *testing.T) {
 	})
 
 	t.Run("verify subgraph expressions", func(t *testing.T) {
+		t.Parallel()
+
+		t.Run("verify data source fetch duration value is attached", func(t *testing.T) {
+			t.Parallel()
+
+			testenv.Run(t, &testenv.Config{
+				SubgraphAccessLogsEnabled: true,
+				SubgraphAccessLogFields: []config.CustomAttribute{
+					{
+						Key: "data_source_fetch_duration",
+						ValueFrom: &config.CustomDynamicAttribute{
+							Expression: "subgraph.request.clientTrace.dataSourceFetchDuration",
+						},
+					},
+				},
+				LogObservation: testenv.LogObservationConfig{
+					Enabled:  true,
+					LogLevel: zapcore.InfoLevel,
+				},
+			}, func(t *testing.T, xEnv *testenv.Environment) {
+				xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+					Query: `query myQuery { employees { id } }`,
+				})
+				requestLog := xEnv.Observer().FilterMessage("/graphql")
+				requestLogAll := requestLog.All()
+				requestContextMap := requestLogAll[0].ContextMap()
+
+				dataSourceFetchDuration, ok := requestContextMap["data_source_fetch_duration"].(time.Duration)
+				require.True(t, ok)
+				require.Greater(t, int(dataSourceFetchDuration), 0)
+			})
+		})
+
+		t.Run("verify data source fetch duration value is attached for multiple subgraph calls", func(t *testing.T) {
+			t.Parallel()
+
+			testenv.Run(t, &testenv.Config{
+				SubgraphAccessLogsEnabled: true,
+				SubgraphAccessLogFields: []config.CustomAttribute{
+					{
+						Key: "data_source_fetch_duration",
+						ValueFrom: &config.CustomDynamicAttribute{
+							Expression: "subgraph.request.clientTrace.dataSourceFetchDuration",
+						},
+					},
+				},
+				LogObservation: testenv.LogObservationConfig{
+					Enabled:  true,
+					LogLevel: zapcore.InfoLevel,
+				},
+			}, func(t *testing.T, xEnv *testenv.Environment) {
+				xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+					Query: `query myQuery { employees { id isAvailable } }`,
+				})
+				requestLog := xEnv.Observer().FilterMessage("/graphql")
+				requestLogAll := requestLog.All()
+
+				employeeSubgraphLogs := requestLogAll[0]
+				dataSourceFetchDuration1, ok := employeeSubgraphLogs.ContextMap()["data_source_fetch_duration"].(time.Duration)
+				require.True(t, ok)
+				require.Greater(t, int(dataSourceFetchDuration1), 0)
+
+				availabilitySubgraphLogs := requestLogAll[1]
+				dataSourceFetchDuration2, ok := availabilitySubgraphLogs.ContextMap()["data_source_fetch_duration"].(time.Duration)
+				require.True(t, ok)
+				require.Greater(t, int(dataSourceFetchDuration2), 0)
+			})
+		})
+
+		t.Run("verify data source fetch duration in conditional expression", func(t *testing.T) {
+			t.Parallel()
+
+			testenv.Run(t, &testenv.Config{
+				SubgraphAccessLogsEnabled: true,
+				SubgraphAccessLogFields: []config.CustomAttribute{
+					{
+						Key: "data_source_fetch_duration",
+						ValueFrom: &config.CustomDynamicAttribute{
+							Expression: "subgraph.request.error != nil ? subgraph.request.clientTrace.dataSourceFetchDuration : ''",
+						},
+					},
+				},
+				LogObservation: testenv.LogObservationConfig{
+					Enabled:  true,
+					LogLevel: zapcore.InfoLevel,
+				},
+				Subgraphs: testenv.SubgraphsConfig{
+					Availability: testenv.SubgraphConfig{
+						Middleware: func(_ http.Handler) http.Handler {
+							return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+								w.Header().Set("Content-Type", "application/json")
+								w.WriteHeader(http.StatusForbidden)
+								_, _ = w.Write([]byte(`{"errors":[{"message":"Unauthorized","extensions":{"code":"UNAUTHORIZED"}}]}`))
+							})
+						},
+					},
+				},
+			}, func(t *testing.T, xEnv *testenv.Environment) {
+				xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+					Query: `query myQuery { employees { id isAvailable } }`,
+				})
+				requestLog := xEnv.Observer().FilterMessage("/graphql")
+				requestLogAll := requestLog.All()
+
+				employeeSubgraphLogs := requestLogAll[0]
+				_, ok := employeeSubgraphLogs.ContextMap()["data_source_fetch_duration"]
+				require.False(t, ok)
+
+				availabilitySubgraphLogs := requestLogAll[1]
+				dataSourceFetchDuration2, ok := availabilitySubgraphLogs.ContextMap()["data_source_fetch_duration"].(float64)
+				require.True(t, ok)
+				require.Greater(t, int(dataSourceFetchDuration2), 0)
+			})
+		})
+
 		t.Run("verify connAcquireDuration value is attached", func(t *testing.T) {
 			t.Parallel()
 

router-tests/telemetry/telemetry_test.go

@@ -9795,7 +9795,7 @@ func TestFlakyTelemetry(t *testing.T) {
 					{
 						Key: "custom.subgraph",
 						ValueFrom: &config.CustomDynamicAttribute{
-							Expression: "string(subgraph.request.clientTrace.connAcquireDuration)",
+							Expression: "string(subgraph.request.clientTrace.connAcquireDuration.Seconds())",
 						},
 					},
 				},
@@ -9854,6 +9854,58 @@ func TestFlakyTelemetry(t *testing.T) {
 				require.False(t, ok)
 			})
 		})
+
+		t.Run("verify data source fetch duration value is attached for multiple subgraph calls", func(t *testing.T) {
+			t.Parallel()
+
+			exporter := tracetest.NewInMemoryExporter(t)
+			metricReader := metric.NewManualReader()
+			testenv.Run(t, &testenv.Config{
+				TraceExporter: exporter,
+				MetricReader:  metricReader,
+				CustomTelemetryAttributes: []config.CustomAttribute{
+					{
+						Key: "data_source_fetch_duration.subgraph",
+						ValueFrom: &config.CustomDynamicAttribute{
+							Expression: "string(subgraph.request.clientTrace.dataSourceFetchDuration.Seconds())",
+						},
+					},
+				},
+			}, func(t *testing.T, xEnv *testenv.Environment) {
+				xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+					Query: `query myQuery { employees { id isAvailable } }`,
+				})
+
+				sn := exporter.GetSpans().Snapshots()
+				require.Len(t, sn, 11)
+
+				var attributesDetected int
+
+				for i := 0; i < len(sn); i++ {
+					attributes := sn[i].Attributes()
+
+					if slices.Contains([]string{"Engine - Fetch"}, sn[i].Name()) {
+						for _, attributeEntry := range attributes {
+							if attributeEntry.Key == "data_source_fetch_duration.subgraph" {
+								attributesDetected++
+								valueString := attributeEntry.Value.AsString()
+								floatValue, err := strconv.ParseFloat(valueString, 64)
+								require.NoError(t, err)
+								require.Greater(t, floatValue, 0.0)
+							}
+						}
+					} else {
+						for _, attributeEntry := range attributes {
+							if attributeEntry.Key == "data_source_fetch_duration.subgraph" {
+								require.Fail(t, "data_source_fetch_duration.subgraph should not be present on non engine fetch spans")
+							}
+						}
+					}
+				}
+
+				require.Equal(t, 2, attributesDetected)
+			})
+		})
 	})
 
 }

router/core/engine_loader_hooks.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"sync/atomic"
 	"time"
 
 	rcontext "github.com/wundergraph/cosmo/router/internal/context"
@@ -93,6 +94,9 @@ func (f *engineLoaderHooks) OnLoad(ctx context.Context, ds resolve.DataSourceInf
 
 	ctx = context.WithValue(ctx, rcontext.CurrentSubgraphContextKey{}, ds.Name)
 
+	duration := atomic.Int64{}
+	ctx = context.WithValue(ctx, rcontext.FetchTimingKey, &duration)
+
 	reqContext := getRequestContext(ctx)
 	if reqContext == nil {
 		return ctx
@@ -153,6 +157,12 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 	exprCtx.Subgraph.Name = ds.Name
 	exprCtx.Subgraph.Request.Error = WrapExprError(responseInfo.Err)
 
+	if value := ctx.Value(rcontext.FetchTimingKey); value != nil {
+		if fetchTiming, ok := value.(*atomic.Int64); ok {
+			exprCtx.Subgraph.Request.ClientTrace.DataSourceFetchDuration = time.Duration(fetchTiming.Load())
+		}
+	}
+
 	if f.storeSubgraphResponseBody {
 		exprCtx.Subgraph.Response.Body.Raw = responseInfo.GetResponseBody()
 	}

router/demo.config.yaml

@@ -19,4 +19,29 @@ events:
     redis:
       - id: my-redis
         urls:
-        - "redis://localhost:6379/2"
+        - "redis://localhost:6379/2"
+
+telemetry:
+  tracing:
+    enabled: true
+    attributes:
+      - key: conn_duration
+        value_from:
+          expression: "string(subgraph.request.clientTrace.connAcquireDuration)"
+    exporters:
+      - endpoint: "http://localhost:4319"
+        exporter: http
+
+access_logs:
+  subgraphs:
+    enabled: true
+    fields:
+      - key: fetch_duration
+        value_from:
+          expression: "subgraph.request.clientTrace.fetchDuration"
+      - key: conn_acquire_duration
+        value_from:
+          expression: "subgraph.request.clientTrace.connAcquireDuration"
+      - key: there
+        value_from:
+          expression: "subgraph.request.clientTrace.connAcquireDuration.String()"

router/internal/context/keys.go

@@ -12,4 +12,5 @@ const (
 	RequestContextKey ContextKey = iota
 	SubgraphResolverContextKey
 	EngineLoaderHooksContextKey
+	FetchTimingKey
 )

router/internal/expr/expr.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/expr-lang/expr/file"
 	"github.com/wundergraph/cosmo/router/pkg/authentication"
@@ -132,7 +133,8 @@ type SubgraphResponse struct {
 }
 
 type ClientTrace struct {
-	ConnectionAcquireDuration float64 `expr:"connAcquireDuration"`
+	DataSourceFetchDuration   time.Duration `expr:"dataSourceFetchDuration"`
+	ConnectionAcquireDuration time.Duration `expr:"connAcquireDuration"`
 }
 
 // Subgraph Related

router/internal/traceclient/traceclient.go

@@ -2,12 +2,13 @@ package traceclient
 
 import (
 	"context"
-	rcontext "github.com/wundergraph/cosmo/router/internal/context"
-	"github.com/wundergraph/cosmo/router/internal/expr"
 	"net/http"
 	"net/http/httptrace"
 	"time"
 
+	rcontext "github.com/wundergraph/cosmo/router/internal/context"
+	"github.com/wundergraph/cosmo/router/internal/expr"
+
 	"github.com/wundergraph/cosmo/router/pkg/metric"
 	rotel "github.com/wundergraph/cosmo/router/pkg/otel"
 )
@@ -113,9 +114,7 @@ func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Contex
 
 	if trace.ConnectionGet != nil && trace.ConnectionAcquired != nil {
 		duration := trace.ConnectionAcquired.Time.Sub(trace.ConnectionGet.Time)
-		connAcquireTime := float64(duration) / float64(time.Millisecond)
-
-		exprContext.Subgraph.Request.ClientTrace.ConnectionAcquireDuration = connAcquireTime
+		exprContext.Subgraph.Request.ClientTrace.ConnectionAcquireDuration = duration
 
 		serverAttributes := rotel.GetServerAttributes(trace.ConnectionGet.HostPort)
 		serverAttributes = append(
@@ -124,6 +123,7 @@ func (t *TraceInjectingRoundTripper) processConnectionMetrics(ctx context.Contex
 			rotel.WgSubgraphName.String(subgraph),
 		)
 
+		connAcquireTime := float64(duration) / float64(time.Millisecond)
 		t.connectionMetricStore.MeasureConnectionAcquireDuration(ctx,
 			connAcquireTime,
 			serverAttributes...)

router/pkg/trace/transport.go

@@ -2,7 +2,10 @@ package trace
 
 import (
 	"net/http"
+	"sync/atomic"
+	"time"
 
+	"github.com/wundergraph/cosmo/router/internal/context"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
@@ -36,8 +39,16 @@ func (t *transport) RoundTrip(r *http.Request) (*http.Response, error) {
 		t.handler(r)
 	}
 
+	startTime := time.Now()
+
 	res, err := t.rt.RoundTrip(r)
 
+	if value := r.Context().Value(context.FetchTimingKey); value != nil {
+		if fetchTiming, ok := value.(*atomic.Int64); ok {
+			fetchTiming.Add(int64(time.Since(startTime)))
+		}
+	}
+
 	// In case of a roundtrip error the span status is set to error by the otelhttp.RoundTrip function.
 	// Also, status code >= 500 is considered an error