Merge branch 'jc-workhorse-inject-id-username' into 'master'

Inject user_id and username into Gitaly context

See merge request gitlab-org/gitlab!87305

workhorse/gitaly_integration_test.go

@@ -54,7 +54,12 @@ func realGitalyOkBody(t *testing.T) *api.Response {
 }
 func ensureGitalyRepository(t *testing.T, apiResponse *api.Response) error {
-	ctx, namespace, err := gitaly.NewNamespaceClient(context.Background(), apiResponse.GitalyServer)
+	ctx, namespace, err := gitaly.NewNamespaceClient(
+		context.Background(),
+		apiResponse.GitalyServer,
+		gitaly.WithFeatures(apiResponse.GitalyServer.Features),
+	)
+
 	if err != nil {
 		return err
 	}

workhorse/gitaly_test.go

@@ -28,7 +28,6 @@ import (
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/git"
-	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/testhelper"
 )
@@ -40,7 +39,7 @@ func TestFailedCloneNoGitaly(t *testing.T) {
 		GL_ID:       "user-123",
 		GL_USERNAME: "username",
 		// This will create a failure to connect to Gitaly
-		GitalyServer: gitaly.Server{Address: "unix:/nonexistent"},
+		GitalyServer: api.GitalyServer{Address: "unix:/nonexistent"},
 	}
 	// Prepare test server and backend

workhorse/internal/api/api.go

@@ -17,7 +17,6 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/config"
-	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/log"
@@ -150,7 +149,7 @@ type Response struct {
 	// Used to communicate channel session details
 	Channel *ChannelSettings
 	// GitalyServer specifies an address and authentication token for a gitaly server we should connect to.
-	GitalyServer gitaly.Server
+	GitalyServer GitalyServer
 	// Repository object for making gRPC requests to Gitaly.
 	Repository gitalypb.Repository
 	// For git-http, does the requestor have the right to view all refs?
@@ -163,6 +162,12 @@ type Response struct {
 	MaximumSize int64
 }
+type GitalyServer struct {
+	Address  string            `json:"address"`
+	Token    string            `json:"token"`
+	Features map[string]string `json:"features"`
+}
+
 // singleJoiningSlash is taken from reverseproxy.go:singleJoiningSlash
 func singleJoiningSlash(a, b string) string {
 	aslash := strings.HasSuffix(a, "/")

workhorse/internal/git/archive.go

@@ -22,6 +22,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/log"
@@ -33,7 +34,7 @@ type archiveParams struct {
 	ArchivePath       string
 	ArchivePrefix     string
 	CommitId          string
-	GitalyServer      gitaly.Server
+	GitalyServer      api.GitalyServer
 	GitalyRepository  gitalypb.Repository
 	DisableCache      bool
 	GetArchiveRequest []byte
@@ -132,7 +133,12 @@ func (a *archive) Inject(w http.ResponseWriter, r *http.Request, sendData string
 func handleArchiveWithGitaly(r *http.Request, params *archiveParams, format gitalypb.GetArchiveRequest_Format) (io.Reader, error) {
 	var request *gitalypb.GetArchiveRequest
-	ctx, c, err := gitaly.NewRepositoryClient(r.Context(), params.GitalyServer)
+	ctx, c, err := gitaly.NewRepositoryClient(
+		r.Context(),
+		params.GitalyServer,
+		gitaly.WithFeatures(params.GitalyServer.Features),
+	)
+
 	if err != nil {
 		return nil, err
 	}

workhorse/internal/git/blob.go

@@ -6,6 +6,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/senddata"
@@ -13,7 +14,7 @@ import (
 type blob struct{ senddata.Prefix }
 type blobParams struct {
-	GitalyServer   gitaly.Server
+	GitalyServer   api.GitalyServer
 	GetBlobRequest gitalypb.GetBlobRequest
 }
@@ -26,7 +27,12 @@ func (b *blob) Inject(w http.ResponseWriter, r *http.Request, sendData string) {
 		return
 	}
-	ctx, blobClient, err := gitaly.NewBlobClient(r.Context(), params.GitalyServer)
+	ctx, blobClient, err := gitaly.NewBlobClient(
+		r.Context(),
+		params.GitalyServer,
+		gitaly.WithFeatures(params.GitalyServer.Features),
+	)
+
 	if err != nil {
 		helper.Fail500(w, r, fmt.Errorf("blob.GetBlob: %v", err))
 		return

workhorse/internal/git/diff.go

@@ -6,6 +6,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/log"
@@ -14,7 +15,7 @@ import (
 type diff struct{ senddata.Prefix }
 type diffParams struct {
-	GitalyServer   gitaly.Server
+	GitalyServer   api.GitalyServer
 	RawDiffRequest string
 }
@@ -33,7 +34,11 @@ func (d *diff) Inject(w http.ResponseWriter, r *http.Request, sendData string) {
 		return
 	}
-	ctx, diffClient, err := gitaly.NewDiffClient(r.Context(), params.GitalyServer)
+	ctx, diffClient, err := gitaly.NewDiffClient(
+		r.Context(),
+		params.GitalyServer,
+		gitaly.WithFeatures(params.GitalyServer.Features),
+	)
 	if err != nil {
 		helper.Fail500(w, r, fmt.Errorf("diff.RawDiff: %v", err))
 		return

workhorse/internal/git/format-patch.go

@@ -6,6 +6,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/log"
@@ -14,7 +15,7 @@ import (
 type patch struct{ senddata.Prefix }
 type patchParams struct {
-	GitalyServer    gitaly.Server
+	GitalyServer    api.GitalyServer
 	RawPatchRequest string
 }
@@ -33,7 +34,12 @@ func (p *patch) Inject(w http.ResponseWriter, r *http.Request, sendData string)
 		return
 	}
-	ctx, diffClient, err := gitaly.NewDiffClient(r.Context(), params.GitalyServer)
+	ctx, diffClient, err := gitaly.NewDiffClient(
+		r.Context(),
+		params.GitalyServer,
+		gitaly.WithFeatures(params.GitalyServer.Features),
+	)
+
 	if err != nil {
 		helper.Fail500(w, r, fmt.Errorf("diff.RawPatch: %v", err))
 		return

workhorse/internal/git/info-refs.go

@@ -55,7 +55,13 @@ func handleGetInfoRefs(rw http.ResponseWriter, r *http.Request, a *api.Response)
 }
 func handleGetInfoRefsWithGitaly(ctx context.Context, responseWriter *HttpResponseWriter, a *api.Response, rpc, gitProtocol, encoding string) error {
-	ctx, smarthttp, err := gitaly.NewSmartHTTPClient(ctx, a.GitalyServer)
+	ctx, smarthttp, err := gitaly.NewSmartHTTPClient(
+		ctx,
+		a.GitalyServer,
+		gitaly.WithFeatures(a.GitalyServer.Features),
+		gitaly.WithUserID(a.GL_ID),
+		gitaly.WithUsername(a.GL_USERNAME),
+	)
 	if err != nil {
 		return err
 	}

workhorse/internal/git/info-refs_test.go

@@ -11,7 +11,6 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
-	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 )
 type smartHTTPServiceServerWithInfoRefs struct {
@@ -32,7 +31,7 @@ func TestGetInfoRefsHandler(t *testing.T) {
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest("GET", "/?service=git-upload-pack", nil)
-	a := &api.Response{GitalyServer: gitaly.Server{Address: addr}}
+	a := &api.Response{GitalyServer: api.GitalyServer{Address: addr}}
 	handleGetInfoRefs(NewHttpResponseWriter(w), r, a)
 	require.Equal(t, 503, w.Code)

workhorse/internal/git/receive-pack.go

@@ -20,7 +20,13 @@ func handleReceivePack(w *HttpResponseWriter, r *http.Request, a *api.Response)
 	gitProtocol := r.Header.Get("Git-Protocol")
-	ctx, smarthttp, err := gitaly.NewSmartHTTPClient(r.Context(), a.GitalyServer)
+	ctx, smarthttp, err := gitaly.NewSmartHTTPClient(
+		r.Context(),
+		a.GitalyServer,
+		gitaly.WithFeatures(a.GitalyServer.Features),
+		gitaly.WithUserID(a.GL_ID),
+		gitaly.WithUsername(a.GL_USERNAME),
+	)
 	if err != nil {
 		return fmt.Errorf("smarthttp.ReceivePack: %v", err)
 	}

workhorse/internal/git/snapshot.go

@@ -7,6 +7,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/log"
@@ -18,7 +19,7 @@ type snapshot struct {
 }
 type snapshotParams struct {
-	GitalyServer       gitaly.Server
+	GitalyServer       api.GitalyServer
 	GetSnapshotRequest string
 }
@@ -40,7 +41,12 @@ func (s *snapshot) Inject(w http.ResponseWriter, r *http.Request, sendData strin
 		return
 	}
-	ctx, c, err := gitaly.NewRepositoryClient(r.Context(), params.GitalyServer)
+	ctx, c, err := gitaly.NewRepositoryClient(
+		r.Context(),
+		params.GitalyServer,
+		gitaly.WithFeatures(params.GitalyServer.Features),
+	)
+
 	if err != nil {
 		helper.Fail500(w, r, fmt.Errorf("SendSnapshot: gitaly.NewRepositoryClient: %v", err))
 		return

workhorse/internal/git/upload-pack.go

@@ -44,7 +44,13 @@ func handleUploadPack(w *HttpResponseWriter, r *http.Request, a *api.Response) e
 }
 func handleUploadPackWithGitaly(ctx context.Context, a *api.Response, clientRequest io.Reader, clientResponse io.Writer, gitProtocol string) error {
-	ctx, smarthttp, err := gitaly.NewSmartHTTPClient(ctx, a.GitalyServer)
+	ctx, smarthttp, err := gitaly.NewSmartHTTPClient(
+		ctx,
+		a.GitalyServer,
+		gitaly.WithFeatures(a.GitalyServer.Features),
+		gitaly.WithUserID(a.GL_ID),
+		gitaly.WithUsername(a.GL_USERNAME),
+	)
 	if err != nil {
 		return fmt.Errorf("get gitaly client: %w", err)
 	}

workhorse/internal/git/upload-pack_test.go

@@ -20,7 +20,6 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
-	"gitlab.com/gitlab-org/gitlab/workhorse/internal/gitaly"
 	"gitlab.com/gitlab-org/gitlab/workhorse/internal/testhelper"
 )
@@ -67,7 +66,7 @@ func TestUploadPackTimesOut(t *testing.T) {
 	w := httptest.NewRecorder()
 	r := httptest.NewRequest("GET", "/", body)
-	a := &api.Response{GitalyServer: gitaly.Server{Address: addr}}
+	a := &api.Response{GitalyServer: api.GitalyServer{Address: addr}}
 	err := handleUploadPack(NewHttpResponseWriter(w), r, a)
 	require.True(t, errors.Is(err, context.DeadlineExceeded))

workhorse/internal/gitaly/gitaly.go

@@ -19,21 +19,17 @@ import (
 	gitalyclient "gitlab.com/gitlab-org/gitaly/v14/client"
 	"gitlab.com/gitlab-org/gitaly/v14/proto/go/gitalypb"
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
+
 	grpccorrelation "gitlab.com/gitlab-org/labkit/correlation/grpc"
 	grpctracing "gitlab.com/gitlab-org/labkit/tracing/grpc"
 )
-type Server struct {
-	Address  string            `json:"address"`
-	Token    string            `json:"token"`
-	Features map[string]string `json:"features"`
-}
-
 type cacheKey struct {
 	address, token string
 }
-func (server Server) cacheKey() cacheKey {
+func getCacheKey(server api.GitalyServer) cacheKey {
 	return cacheKey{address: server.Address, token: server.Token}
 }
@@ -71,19 +67,42 @@ func InitializeSidechannelRegistry(logger *logrus.Logger) {
 	}
 }
-func withOutgoingMetadata(ctx context.Context, features map[string]string) context.Context {
-	md := metadata.New(nil)
-	for k, v := range features {
-		if !strings.HasPrefix(k, "gitaly-feature-") {
-			continue
+type MetadataFunc func(metadata.MD)
+
+func WithUserID(userID string) MetadataFunc {
+	return func(md metadata.MD) {
+		md.Append("user_id", userID)
+	}
+}
+
+func WithUsername(username string) MetadataFunc {
+	return func(md metadata.MD) {
+		md.Append("username", username)
+	}
+}
+
+func WithFeatures(features map[string]string) MetadataFunc {
+	return func(md metadata.MD) {
+		for k, v := range features {
+			if !strings.HasPrefix(k, "gitaly-feature-") {
+				continue
+			}
+			md.Append(k, v)
 		}
-		md.Append(k, v)
+	}
+}
+
+func withOutgoingMetadata(ctx context.Context, addMetadataFuncs ...MetadataFunc) context.Context {
+	md := metadata.New(nil)
+
+	for _, f := range addMetadataFuncs {
+		f(md)
 	}
 	return metadata.NewOutgoingContext(ctx, md)
 }
-func NewSmartHTTPClient(ctx context.Context, server Server) (context.Context, *SmartHTTPClient, error) {
+func NewSmartHTTPClient(ctx context.Context, server api.GitalyServer, metadataFuncs ...MetadataFunc) (context.Context, *SmartHTTPClient, error) {
 	conn, err := getOrCreateConnection(server)
 	if err != nil {
 		return nil, nil, err
@@ -93,48 +112,52 @@ func NewSmartHTTPClient(ctx context.Context, server Server) (context.Context, *S
 		SmartHTTPServiceClient: grpcClient,
 		sidechannelRegistry:    sidechannelRegistry,
 	}
-	return withOutgoingMetadata(ctx, server.Features), smartHTTPClient, nil
+
+	return withOutgoingMetadata(
+		ctx,
+		metadataFuncs...,
+	), smartHTTPClient, nil
 }
-func NewBlobClient(ctx context.Context, server Server) (context.Context, *BlobClient, error) {
+func NewBlobClient(ctx context.Context, server api.GitalyServer, addMetadataFuncs ...MetadataFunc) (context.Context, *BlobClient, error) {
 	conn, err := getOrCreateConnection(server)
 	if err != nil {
 		return nil, nil, err
 	}
 	grpcClient := gitalypb.NewBlobServiceClient(conn)
-	return withOutgoingMetadata(ctx, server.Features), &BlobClient{grpcClient}, nil
+	return withOutgoingMetadata(ctx, addMetadataFuncs...), &BlobClient{grpcClient}, nil
 }
-func NewRepositoryClient(ctx context.Context, server Server) (context.Context, *RepositoryClient, error) {
+func NewRepositoryClient(ctx context.Context, server api.GitalyServer, addMetadataFuncs ...MetadataFunc) (context.Context, *RepositoryClient, error) {
 	conn, err := getOrCreateConnection(server)
 	if err != nil {
 		return nil, nil, err
 	}
 	grpcClient := gitalypb.NewRepositoryServiceClient(conn)
-	return withOutgoingMetadata(ctx, server.Features), &RepositoryClient{grpcClient}, nil
+	return withOutgoingMetadata(ctx, addMetadataFuncs...), &RepositoryClient{grpcClient}, nil
 }
 // NewNamespaceClient is only used by the Gitaly integration tests at present
-func NewNamespaceClient(ctx context.Context, server Server) (context.Context, *NamespaceClient, error) {
+func NewNamespaceClient(ctx context.Context, server api.GitalyServer, addMetadataFuncs ...MetadataFunc) (context.Context, *NamespaceClient, error) {
 	conn, err := getOrCreateConnection(server)
 	if err != nil {
 		return nil, nil, err
 	}
 	grpcClient := gitalypb.NewNamespaceServiceClient(conn)
-	return withOutgoingMetadata(ctx, server.Features), &NamespaceClient{grpcClient}, nil
+	return withOutgoingMetadata(ctx, addMetadataFuncs...), &NamespaceClient{grpcClient}, nil
 }
-func NewDiffClient(ctx context.Context, server Server) (context.Context, *DiffClient, error) {
+func NewDiffClient(ctx context.Context, server api.GitalyServer, addMetadataFuncs ...MetadataFunc) (context.Context, *DiffClient, error) {
 	conn, err := getOrCreateConnection(server)
 	if err != nil {
 		return nil, nil, err
 	}
 	grpcClient := gitalypb.NewDiffServiceClient(conn)
-	return withOutgoingMetadata(ctx, server.Features), &DiffClient{grpcClient}, nil
+	return withOutgoingMetadata(ctx, addMetadataFuncs...), &DiffClient{grpcClient}, nil
 }
-func getOrCreateConnection(server Server) (*grpc.ClientConn, error) {
-	key := server.cacheKey()
+func getOrCreateConnection(server api.GitalyServer) (*grpc.ClientConn, error) {
+	key := getCacheKey(server)
 	cache.RLock()
 	conn := cache.connections[key]
@@ -170,7 +193,7 @@ func CloseConnections() {
 	}
 }
-func newConnection(server Server) (*grpc.ClientConn, error) {
+func newConnection(server api.GitalyServer) (*grpc.ClientConn, error) {
 	connOpts := append(gitalyclient.DefaultDialOpts,
 		grpc.WithPerRPCCredentials(gitalyauth.RPCCredentialsV2(server.Token)),
 		grpc.WithStreamInterceptor(

workhorse/internal/gitaly/gitaly_test.go

@@ -8,6 +8,8 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/metadata"
+
+	"gitlab.com/gitlab-org/gitlab/workhorse/internal/api"
 )
 func TestMain(m *testing.M) {
@@ -16,32 +18,56 @@ func TestMain(m *testing.M) {
 }
 func TestNewSmartHTTPClient(t *testing.T) {
-	ctx, client, err := NewSmartHTTPClient(context.Background(), serverFixture())
+	ctx, client, err := NewSmartHTTPClient(
+		context.Background(),
+		serverFixture(),
+		WithFeatures(features()),
+		WithUsername("gl_username"),
+		WithUserID("gl_id"),
+	)
 	require.NoError(t, err)
 	testOutgoingMetadata(t, ctx)
+	testOutgoingIDAndUsername(t, ctx)
 	require.NotNil(t, client.sidechannelRegistry)
 }
 func TestNewBlobClient(t *testing.T) {
-	ctx, _, err := NewBlobClient(context.Background(), serverFixture())
+	ctx, _, err := NewBlobClient(
+		context.Background(),
+		serverFixture(),
+		WithFeatures(features()),
+	)
 	require.NoError(t, err)
 	testOutgoingMetadata(t, ctx)
 }
 func TestNewRepositoryClient(t *testing.T) {
-	ctx, _, err := NewRepositoryClient(context.Background(), serverFixture())
+	ctx, _, err := NewRepositoryClient(
+		context.Background(),
+		serverFixture(),
+		WithFeatures(features()),
+	)
+
 	require.NoError(t, err)
 	testOutgoingMetadata(t, ctx)
 }
 func TestNewNamespaceClient(t *testing.T) {
-	ctx, _, err := NewNamespaceClient(context.Background(), serverFixture())
+	ctx, _, err := NewNamespaceClient(
+		context.Background(),
+		serverFixture(),
+		WithFeatures(features()),
+	)
 	require.NoError(t, err)
 	testOutgoingMetadata(t, ctx)
 }
 func TestNewDiffClient(t *testing.T) {
-	ctx, _, err := NewDiffClient(context.Background(), serverFixture())
+	ctx, _, err := NewDiffClient(
+		context.Background(),
+		serverFixture(),
+		WithFeatures(features()),
+	)
 	require.NoError(t, err)
 	testOutgoingMetadata(t, ctx)
 }
@@ -61,16 +87,29 @@ func testOutgoingMetadata(t *testing.T, ctx context.Context) {
 	}
 }
-func serverFixture() Server {
+func testOutgoingIDAndUsername(t *testing.T, ctx context.Context) {
+	md, ok := metadata.FromOutgoingContext(ctx)
+	require.True(t, ok, "get metadata from context")
+
+	require.Equal(t, md["user_id"], []string{"gl_id"})
+	require.Equal(t, md["username"], []string{"gl_username"})
+}
+
+func features() map[string]string {
 	features := make(map[string]string)
 	for k, v := range allowedFeatures() {
 		features[k] = v
 	}
+
 	for k, v := range badFeatureMetadata() {
 		features[k] = v
 	}
-	return Server{Address: "tcp://localhost:123", Features: features}
+	return features
+}
+
+func serverFixture() api.GitalyServer {
+	return api.GitalyServer{Address: "tcp://localhost:123"}
 }
 func allowedFeatures() map[string]string {