requestinteraction: add X-Sourcegraph-Interaction-ID propagation (#58016)

This change allows clients to provide an interaction ID, essentially a trace but maybe simpler for some scenarios, and have that be propagated throughout Sourcegraph systems implicitly. We can automatically add this to events (telemetry V2 and Cody Gateway). The implementation closely follows the patterns we have for `internal/requestclient` and `internal/actor`

If we land this mechanism, I'll follow up with an change that automatically injects these IDs into telemetry events within a context as well. If clients send a bunch of completion requests to the backend with this header, the interaction ID will automatically be added to the completion events recorded by the backend, like [these ones](https://sourcegraph.com/search?q=context:global+repo:sourcegraph/sourcegraph+.Record(...,+%22cody.completions%22,+...)&patternType=structural&sm=1&groupBy=repo), and any future events. Similarly, if clients set this header on outbound requests when recording events in the new telemetry (go/telemetry-v2), they'll get added as well - we can also consider adding an interaction ID as an explicit argument in the `recordEvent` GraphQL mutation to take precedence over the context interaction ID.

cmd/cody-gateway/shared/BUILD.bazel

@@ -34,6 +34,7 @@ go_library(
         "//internal/rcache",
         "//internal/redispool",
         "//internal/requestclient",
+        "//internal/requestinteraction",
         "//internal/service",
         "//internal/trace",
         "//internal/trace/policy",

cmd/cody-gateway/shared/main.go

@@ -32,6 +32,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/rcache"
 	"github.com/sourcegraph/sourcegraph/internal/redispool"
 	"github.com/sourcegraph/sourcegraph/internal/requestclient"
+	"github.com/sourcegraph/sourcegraph/internal/requestinteraction"
 	"github.com/sourcegraph/sourcegraph/internal/service"
 	"github.com/sourcegraph/sourcegraph/internal/trace"
 	"github.com/sourcegraph/sourcegraph/internal/version"
@@ -179,6 +180,7 @@ func Main(ctx context.Context, obctx *observation.Context, ready service.ReadyFu
 	// Cloudflare in from of Cody Gateway. This comes first.
 	hasCloudflare := !config.InsecureDev
 	handler = requestclient.ExternalHTTPMiddleware(handler, hasCloudflare)
+	handler = requestinteraction.HTTPMiddleware(handler)
 
 	// Initialize our server
 	address := fmt.Sprintf(":%d", config.Port)

cmd/frontend/internal/cli/BUILD.bazel

@@ -69,6 +69,7 @@ go_library(
         "//internal/oobmigration/migrations/register",
         "//internal/redispool",
         "//internal/requestclient",
+        "//internal/requestinteraction",
         "//internal/service",
         "//internal/session",
         "//internal/symbols",

cmd/frontend/internal/cli/http.go

@@ -30,6 +30,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/featureflag"
 	"github.com/sourcegraph/sourcegraph/internal/instrumentation"
 	"github.com/sourcegraph/sourcegraph/internal/requestclient"
+	"github.com/sourcegraph/sourcegraph/internal/requestinteraction"
 	"github.com/sourcegraph/sourcegraph/internal/session"
 	tracepkg "github.com/sourcegraph/sourcegraph/internal/trace"
 	"github.com/sourcegraph/sourcegraph/internal/version"
@@ -69,6 +70,7 @@ func newExternalHTTPHandler(
 	apiHandler = session.CookieMiddlewareWithCSRFSafety(logger, db, apiHandler, corsAllowHeader, isTrustedOrigin) // API accepts cookies with special header
 	apiHandler = httpapi.AccessTokenAuthMiddleware(db, logger, apiHandler)                                        // API accepts access tokens
 	apiHandler = requestclient.ExternalHTTPMiddleware(apiHandler, envvar.SourcegraphDotComMode())
+	apiHandler = requestinteraction.HTTPMiddleware(apiHandler)
 	apiHandler = gziphandler.GzipHandler(apiHandler)
 	if envvar.SourcegraphDotComMode() {
 		apiHandler = deviceid.Middleware(apiHandler)
@@ -92,6 +94,7 @@ func newExternalHTTPHandler(
 	appHandler = session.CookieMiddleware(logger, db, appHandler)          // app accepts cookies
 	appHandler = httpapi.AccessTokenAuthMiddleware(db, logger, appHandler) // app accepts access tokens
 	appHandler = requestclient.ExternalHTTPMiddleware(appHandler, envvar.SourcegraphDotComMode())
+	appHandler = requestinteraction.HTTPMiddleware(appHandler)
 	if envvar.SourcegraphDotComMode() {
 		appHandler = deviceid.Middleware(appHandler)
 	}

cmd/gitserver/shared/BUILD.bazel

@@ -42,6 +42,7 @@ go_library(
         "//internal/observation",
         "//internal/ratelimit",
         "//internal/requestclient",
+        "//internal/requestinteraction",
         "//internal/service",
         "//internal/trace",
         "//internal/wrexec",

cmd/gitserver/shared/shared.go

@@ -45,6 +45,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/observation"
 	"github.com/sourcegraph/sourcegraph/internal/ratelimit"
 	"github.com/sourcegraph/sourcegraph/internal/requestclient"
+	"github.com/sourcegraph/sourcegraph/internal/requestinteraction"
 	"github.com/sourcegraph/sourcegraph/internal/service"
 	"github.com/sourcegraph/sourcegraph/internal/trace"
 	"github.com/sourcegraph/sourcegraph/internal/wrexec"
@@ -138,6 +139,7 @@ func Main(ctx context.Context, observationCtx *observation.Context, ready servic
 	handler := gitserver.Handler()
 	handler = actor.HTTPMiddleware(logger, handler)
 	handler = requestclient.InternalHTTPMiddleware(handler)
+	handler = requestinteraction.HTTPMiddleware(handler)
 	handler = trace.HTTPMiddleware(logger, handler, conf.DefaultClient())
 	handler = instrumentation.HTTPMiddleware("", handler)
 	handler = internalgrpc.MultiplexHandlers(makeGRPCServer(logger, &gitserver), handler)

internal/actor/http.go

@@ -60,6 +60,8 @@ var (
 //
 // 🚨 SECURITY: Wherever possible, prefer to act in the context of a specific user rather
 // than as an internal actor, which can grant a lot of access in some cases.
+//
+// TODO(@bobheadxi): Migrate to httpcli.Doer and httpcli.Middleware
 type HTTPTransport struct {
 	RoundTripper http.RoundTripper
 }
@@ -72,6 +74,10 @@ func (t *HTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 		t.RoundTripper = http.DefaultTransport
 	}
 
+	// RoundTripper should not modify original request. All the code paths
+	// below set a header, so we clone the request immediately.
+	req = req.Clone(req.Context())
+
 	actor := FromContext(req.Context())
 	path := getCondensedURLPath(req.URL.Path)
 	switch {

internal/grpc/defaults/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "//internal/grpc/messagesize",
         "//internal/grpc/propagator",
         "//internal/requestclient",
+        "//internal/requestinteraction",
         "//internal/trace/policy",
         "//internal/ttlcache",
         "//lib/errors",

internal/grpc/defaults/defaults.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/sourcegraph/sourcegraph/internal/grpc/contextconv"
 	"github.com/sourcegraph/sourcegraph/internal/grpc/messagesize"
+	"github.com/sourcegraph/sourcegraph/internal/requestinteraction"
 
 	"github.com/sourcegraph/sourcegraph/internal/actor"
 	internalgrpc "github.com/sourcegraph/sourcegraph/internal/grpc"
@@ -80,6 +81,7 @@ func defaultDialOptions(logger log.Logger, creds credentials.TransportCredential
 			propagator.StreamClientPropagator(actor.ActorPropagator{}),
 			propagator.StreamClientPropagator(policy.ShouldTracePropagator{}),
 			propagator.StreamClientPropagator(requestclient.Propagator{}),
+			propagator.StreamClientPropagator(requestinteraction.Propagator{}),
 			otelgrpc.StreamClientInterceptor(),
 			internalerrs.PrometheusStreamClientInterceptor,
 			internalerrs.LoggingStreamClientInterceptor(logger),
@@ -91,6 +93,7 @@ func defaultDialOptions(logger log.Logger, creds credentials.TransportCredential
 			propagator.UnaryClientPropagator(actor.ActorPropagator{}),
 			propagator.UnaryClientPropagator(policy.ShouldTracePropagator{}),
 			propagator.UnaryClientPropagator(requestclient.Propagator{}),
+			propagator.UnaryClientPropagator(requestinteraction.Propagator{}),
 			otelgrpc.UnaryClientInterceptor(),
 			internalerrs.PrometheusUnaryClientInterceptor,
 			internalerrs.LoggingUnaryClientInterceptor(logger),
@@ -137,6 +140,7 @@ func ServerOptions(logger log.Logger, additionalOptions ...grpc.ServerOption) []
 			metrics.StreamServerInterceptor(),
 			messagesize.StreamServerInterceptor,
 			propagator.StreamServerPropagator(requestclient.Propagator{}),
+			propagator.StreamServerPropagator(requestinteraction.Propagator{}),
 			propagator.StreamServerPropagator(actor.ActorPropagator{}),
 			propagator.StreamServerPropagator(policy.ShouldTracePropagator{}),
 			otelgrpc.StreamServerInterceptor(),
@@ -148,6 +152,7 @@ func ServerOptions(logger log.Logger, additionalOptions ...grpc.ServerOption) []
 			metrics.UnaryServerInterceptor(),
 			messagesize.UnaryServerInterceptor,
 			propagator.UnaryServerPropagator(requestclient.Propagator{}),
+			propagator.UnaryServerPropagator(requestinteraction.Propagator{}),
 			propagator.UnaryServerPropagator(actor.ActorPropagator{}),
 			propagator.UnaryServerPropagator(policy.ShouldTracePropagator{}),
 			otelgrpc.UnaryServerInterceptor(),

internal/httpcli/BUILD.bazel

@@ -22,6 +22,7 @@ go_library(
         "//internal/metrics",
         "//internal/rcache",
         "//internal/requestclient",
+        "//internal/requestinteraction",
         "//internal/trace",
         "//internal/trace/policy",
         "//internal/types",

internal/httpcli/client.go

@@ -29,6 +29,7 @@ import (
 	"github.com/sourcegraph/sourcegraph/internal/metrics"
 	"github.com/sourcegraph/sourcegraph/internal/rcache"
 	"github.com/sourcegraph/sourcegraph/internal/requestclient"
+	"github.com/sourcegraph/sourcegraph/internal/requestinteraction"
 	"github.com/sourcegraph/sourcegraph/internal/trace"
 	"github.com/sourcegraph/sourcegraph/internal/trace/policy"
 	"github.com/sourcegraph/sourcegraph/lib/errors"
@@ -144,6 +145,7 @@ func newExternalClientFactory(cache bool, middleware ...Middleware) *Factory {
 			NewRetryPolicy(MaxRetries(externalRetryMaxAttempts), externalRetryAfterMaxDuration),
 			ExpJitterDelayOrRetryAfterDelay(externalRetryDelayBase, externalRetryDelayMax),
 		),
+		RequestInteractionTransportOpt,
 		TracedTransportOpt,
 	}
 	if cache {
@@ -212,6 +214,7 @@ func NewInternalClientFactory(subsystem string, middleware ...Middleware) *Facto
 		MeteredTransportOpt(subsystem),
 		ActorTransportOpt,
 		RequestClientTransportOpt,
+		RequestInteractionTransportOpt,
 		TracedTransportOpt,
 	)
 }
@@ -840,6 +843,19 @@ func RequestClientTransportOpt(cli *http.Client) error {
 	return nil
 }
 
+func RequestInteractionTransportOpt(cli *http.Client) error {
+	if cli.Transport == nil {
+		cli.Transport = http.DefaultTransport
+	}
+
+	cli.Transport = &wrappedTransport{
+		RoundTripper: &requestinteraction.HTTPTransport{RoundTripper: cli.Transport},
+		Wrapped:      cli.Transport,
+	}
+
+	return nil
+}
+
 // IsRiskyHeader returns true if the request or response header is likely to contain private data.
 func IsRiskyHeader(name string, values []string) bool {
 	return isRiskyHeaderName(name) || containsRiskyHeaderValue(values)

internal/requestclient/http.go

@@ -20,6 +20,8 @@ const (
 // HTTPTransport is a roundtripper that sets client IP information within request context as
 // headers on outgoing requests. The attached headers can be picked up and attached to
 // incoming request contexts with client.HTTPMiddleware.
+//
+// TODO(@bobheadxi): Migrate to httpcli.Doer and httpcli.Middleware
 type HTTPTransport struct {
 	RoundTripper http.RoundTripper
 }
@@ -33,6 +35,7 @@ func (t *HTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 
 	client := FromContext(req.Context())
 	if client != nil {
+		req = req.Clone(req.Context()) // RoundTripper should not modify original request
 		req.Header.Set(headerKeyClientIP, client.IP)
 		req.Header.Set(headerKeyForwardedFor, client.ForwardedFor)
 	}

internal/requestinteraction/BUILD.bazel

@@ -0,0 +1,17 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "requestinteraction",
+    srcs = [
+        "client.go",
+        "grpc.go",
+        "http.go",
+    ],
+    importpath = "github.com/sourcegraph/sourcegraph/internal/requestinteraction",
+    visibility = ["//:__subpackages__"],
+    deps = [
+        "//internal/grpc/propagator",
+        "@com_github_sourcegraph_log//:log",
+        "@org_golang_google_grpc//metadata",
+    ],
+)

internal/requestinteraction/client.go

@@ -0,0 +1,38 @@
+package requestinteraction
+
+import (
+	"context"
+
+	"github.com/sourcegraph/log"
+)
+
+type requestInteractionKey struct{}
+
+// Interaction carries information about the interaction associated with a
+// request - a sort of manually instrumented trace.
+type Interaction struct {
+	// ID identifies the interaction
+	ID string
+}
+
+func FromContext(ctx context.Context) *Interaction {
+	ip, ok := ctx.Value(requestInteractionKey{}).(*Interaction)
+	if !ok || ip == nil {
+		return nil
+	}
+	return ip
+}
+
+// WithClient adds client IP information to context for propagation.
+func WithClient(ctx context.Context, client *Interaction) context.Context {
+	return context.WithValue(ctx, requestInteractionKey{}, client)
+}
+
+func (c *Interaction) LogFields() []log.Field {
+	if c == nil {
+		return []log.Field{log.String("requestInteraction", "<nil>")}
+	}
+	return []log.Field{
+		log.String("requestInteraction.id", c.ID),
+	}
+}

internal/requestinteraction/grpc.go

@@ -0,0 +1,35 @@
+package requestinteraction
+
+import (
+	"context"
+
+	"google.golang.org/grpc/metadata"
+
+	internalgrpc "github.com/sourcegraph/sourcegraph/internal/grpc/propagator"
+)
+
+type Propagator struct{}
+
+func (Propagator) FromContext(ctx context.Context) metadata.MD {
+	interaction := FromContext(ctx)
+	if interaction == nil {
+		return metadata.New(nil)
+	}
+
+	return metadata.Pairs(
+		headerKeyInteractionID, interaction.ID,
+	)
+}
+
+func (Propagator) InjectContext(ctx context.Context, md metadata.MD) context.Context {
+	if vals := md.Get(headerKeyInteractionID); len(vals) > 0 {
+		id := vals[0]
+		return WithClient(ctx, &Interaction{
+			ID: id,
+		})
+	}
+
+	return ctx
+}
+
+var _ internalgrpc.Propagator = Propagator{}

internal/requestinteraction/http.go

@@ -0,0 +1,48 @@
+package requestinteraction
+
+import (
+	"net/http"
+)
+
+const (
+	// Sourcegraph-specific header key for propagating an interaction ID.
+	headerKeyInteractionID = "X-Sourcegraph-Interaction-ID"
+)
+
+// TODO(@bobheadxi): Migrate to httpcli.Doer and httpcli.Middleware
+type HTTPTransport struct {
+	RoundTripper http.RoundTripper
+}
+
+var _ http.RoundTripper = &HTTPTransport{}
+
+func (t *HTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if t.RoundTripper == nil {
+		t.RoundTripper = http.DefaultTransport
+	}
+
+	interaction := FromContext(req.Context())
+	if interaction != nil {
+		req = req.Clone(req.Context()) // RoundTripper should not modify original request
+		req.Header.Set(headerKeyInteractionID, interaction.ID)
+	}
+
+	return t.RoundTripper.RoundTrip(req)
+}
+
+func HTTPMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		interactionID := req.Header.Get(headerKeyInteractionID)
+
+		// If empty, nothing to do, just pass through
+		if interactionID == "" {
+			next.ServeHTTP(rw, req)
+			return
+		}
+
+		ctxWithClient := WithClient(req.Context(), &Interaction{
+			ID: interactionID,
+		})
+		next.ServeHTTP(rw, req.WithContext(ctxWithClient))
+	})
+}

internal/telemetrygateway/v1/telemetrygateway.proto

@@ -204,4 +204,6 @@ message EventMarketingTracking {
 message EventInteraction {
   // OpenTelemetry trace ID representing the interaction associated with the event.
   optional string trace_id = 1;
+  // Reserve entry for client-provided interaction ID in follow-up change.
+  reserved 2;
 }