feat: add timings per client fetch for GraphQL http

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.FetchDuration = time.Duration(fetchTiming.Load())
+		}
+	}
+
 	if f.storeSubgraphResponseBody {
 		exprCtx.Subgraph.Response.Body.Raw = responseInfo.GetResponseBody()
 	}

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"`
+	FetchDuration             time.Duration `expr:"fetchDuration"`
+	ConnectionAcquireDuration float64       `expr:"connAcquireDuration"`
 }
 
 // Subgraph Related

router/pkg/grpcconnector/grpccommon/grpc_plugin_client.go

@@ -3,6 +3,7 @@ package grpccommon
 import (
 	"context"
 	"errors"
+	rcontext "github.com/wundergraph/cosmo/router/internal/context"
 	"go.opentelemetry.io/otel/trace"
 	"io"
 
@@ -147,9 +148,21 @@ func (g *GRPCPluginClient) SetClients(pluginClient *plugin.Client, clientConn gr
 // Invoke implements grpc.ClientConnInterface.
 func (g *GRPCPluginClient) Invoke(ctx context.Context, method string, args any, reply any, opts ...grpc.CallOption) error {
 	spanName, traceAttributes := g.getTraceAttributes(ctx)
+
 	ctx, span := g.tracer.Start(ctx, spanName, traceAttributes)
 	defer span.End()
 
+	startTime := time.Now()
+	defer func() {
+		// In case of GRPC Clients there will be multiple invocations
+		// so adding is required
+		if value := ctx.Value(rcontext.FetchTimingKey); value != nil {
+			if fetchTiming, ok := value.(*atomic.Int64); ok {
+				fetchTiming.Add(int64(time.Since(startTime)))
+			}
+		}
+	}()
+
 	if g.IsPluginProcessExited() {
 		if err := g.waitForPluginToBeActive(); err != nil {
 			span.RecordError(err)

router/pkg/trace/transport.go

@@ -1,9 +1,11 @@
 package trace
 
 import (
-	"net/http"
-
+	"github.com/wundergraph/cosmo/router/internal/context"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"net/http"
+	"sync/atomic"
+	"time"
 )
 
 type TransportOption func(svr *transport)
@@ -36,8 +38,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