Add flagd-selector header support to in-process service with backward compatibility

Co-authored-by: aepfli <[email protected]>

Author: Copilot

providers/flagd/README.md

@@ -189,6 +189,25 @@ The flagd provider currently support following flag evaluation metadata,
 | `scope`      | string | "selector" set for the associated source in flagd   |
 | `providerID` | string | "providerID" set for the associated source in flagd |
 
+## Selector Handling
+
+When using the in-process resolver with a gRPC sync source, the provider supports filtering flag configurations using a selector. The selector can be configured using the `WithSelector` option or the `FLAGD_SOURCE_SELECTOR` environment variable.
+
+### Header-based Selector (Recommended)
+
+The provider now sends the selector as a `flagd-selector` gRPC metadata header when communicating with flagd sync services. This approach is consistent with how selectors are handled across all flagd services (sync, evaluation, and OFREP).
+
+```go
+provider, err := flagd.NewProvider(
+    flagd.WithInProcessResolver(),
+    flagd.WithSelector("source=database,app=myapp"),
+)
+```
+
+### Backward Compatibility
+
+For backward compatibility with older flagd versions, the provider continues to include the selector in the gRPC request fields alongside the header. This dual approach ensures compatibility during the migration period until all flagd instances are updated.
+
 ## Logging
 
 If not configured, logging falls back to the standard Go log package at error level only.

providers/flagd/pkg/service/in_process/service.go

@@ -9,6 +9,7 @@ import (
 
 	"go.uber.org/zap"
 	googlegrpc "google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 
 	"github.com/open-feature/flagd/core/pkg/evaluator"
 	"github.com/open-feature/flagd/core/pkg/logger"
@@ -320,14 +321,47 @@ func makeSyncProvider(cfg Configuration, log *logger.Logger) (sync.ISync, string
 
 	log.Info("operating in in-process mode with flags sourced from " + uri)
 
+	// Prepare gRPC dial options with selector interceptors
+	var dialOptions []googlegrpc.DialOption
+	
+	if len(cfg.GrpcDialOptionsOverride) > 0 {
+		// User has provided custom dial options
+		dialOptions = cfg.GrpcDialOptionsOverride
+		
+		// Add interceptors to inject flagd-selector header if selector is configured
+		if cfg.Selector != "" {
+			dialOptions = append(dialOptions,
+				googlegrpc.WithChainUnaryInterceptor(selectorUnaryInterceptor(cfg.Selector)),
+				googlegrpc.WithChainStreamInterceptor(selectorStreamInterceptor(cfg.Selector)),
+			)
+		}
+	} else if cfg.Selector != "" {
+		// No custom dial options, but we need to add selector interceptors
+		// Build the dial options including credentials and interceptors
+		credBuilder := &credentials.CredentialBuilder{}
+		tCredentials, err := credBuilder.Build(cfg.TLSEnabled, cfg.CertificatePath)
+		if err != nil {
+			log.Error("error building transport credentials, falling back to using Selector field only", zap.Error(err))
+			// Fall back to not using GrpcDialOptionsOverride, which means only the Selector field will be used
+			dialOptions = nil
+		} else {
+			dialOptions = []googlegrpc.DialOption{
+				googlegrpc.WithTransportCredentials(tCredentials),
+				googlegrpc.WithChainUnaryInterceptor(selectorUnaryInterceptor(cfg.Selector)),
+				googlegrpc.WithChainStreamInterceptor(selectorStreamInterceptor(cfg.Selector)),
+			}
+		}
+	}
+	// else: No selector and no custom options, let grpc.Sync handle everything
+
 	return &grpc.Sync{
 		CredentialBuilder:       &credentials.CredentialBuilder{},
-		GrpcDialOptionsOverride: cfg.GrpcDialOptionsOverride,
+		GrpcDialOptionsOverride: dialOptions,
 		Logger:                  log,
 		Secure:                  cfg.TLSEnabled,
 		CertPath:                cfg.CertificatePath,
 		ProviderID:              cfg.ProviderID,
-		Selector:                cfg.Selector,
+		Selector:                cfg.Selector, // Keep for backward compatibility
 		URI:                     uri,
 	}, uri
 }
@@ -352,3 +386,37 @@ func isValidTargetScheme(targetUri string) bool {
 	regx := regexp.MustCompile("^" + grpc.SupportedScheme)
 	return regx.Match([]byte(targetUri))
 }
+
+// selectorUnaryInterceptor adds the flagd-selector metadata header to unary gRPC calls
+func selectorUnaryInterceptor(selector string) googlegrpc.UnaryClientInterceptor {
+	return func(
+		ctx context.Context,
+		method string,
+		req, reply interface{},
+		cc *googlegrpc.ClientConn,
+		invoker googlegrpc.UnaryInvoker,
+		opts ...googlegrpc.CallOption,
+	) error {
+		if selector != "" {
+			ctx = metadata.AppendToOutgoingContext(ctx, "flagd-selector", selector)
+		}
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}
+
+// selectorStreamInterceptor adds the flagd-selector metadata header to streaming gRPC calls
+func selectorStreamInterceptor(selector string) googlegrpc.StreamClientInterceptor {
+	return func(
+		ctx context.Context,
+		desc *googlegrpc.StreamDesc,
+		cc *googlegrpc.ClientConn,
+		method string,
+		streamer googlegrpc.Streamer,
+		opts ...googlegrpc.CallOption,
+	) (googlegrpc.ClientStream, error) {
+		if selector != "" {
+			ctx = metadata.AppendToOutgoingContext(ctx, "flagd-selector", selector)
+		}
+		return streamer(ctx, desc, cc, method, opts...)
+	}
+}

providers/flagd/pkg/service/in_process/service_selector_header_test.go

@@ -0,0 +1,125 @@
+package process
+
+import (
+v1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/flagd/sync/v1"
+"buf.build/gen/go/open-feature/flagd/grpc/go/flagd/sync/v1/syncv1grpc"
+"context"
+"fmt"
+"google.golang.org/grpc"
+"google.golang.org/grpc/metadata"
+"net"
+"testing"
+"time"
+)
+
+// Test that the flagd-selector header is sent in gRPC metadata
+func TestSelectorHeaderIsSent(t *testing.T) {
+// given
+host := "localhost"
+port := 8091
+selector := "source=test,app=selector-test"
+headerReceived := make(chan string, 1)
+
+listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", host, port))
+if err != nil {
+t.Fatal(err)
+}
+
+// Mock server that captures the flagd-selector header
+mockServer := &selectorHeaderCapturingServer{
+listener:       listen,
+headerReceived: headerReceived,
+mockResponse: &v1.SyncFlagsResponse{
+FlagConfiguration: flagRsp,
+},
+}
+
+inProcessService := NewInProcessService(Configuration{
+Host:       host,
+Port:       port,
+Selector:   selector,
+TLSEnabled: false,
+})
+
+// when
+go func() {
+server := grpc.NewServer()
+syncv1grpc.RegisterFlagSyncServiceServer(server, mockServer)
+if err := server.Serve(mockServer.listener); err != nil {
+t.Logf("Server exited with error: %v", err)
+}
+}()
+
+// Initialize service
+err = inProcessService.Init()
+if err != nil {
+t.Fatal(err)
+}
+
+// then - verify that the flagd-selector header was sent
+select {
+case receivedSelector := <-headerReceived:
+if receivedSelector != selector {
+t.Fatalf("Expected selector header to be %q, but got %q", selector, receivedSelector)
+}
+case <-time.After(3 * time.Second):
+t.Fatal("Timeout waiting for flagd-selector header to be received")
+}
+
+inProcessService.Shutdown()
+}
+
+// Mock server that captures the flagd-selector header from incoming requests
+type selectorHeaderCapturingServer struct {
+listener       net.Listener
+headerReceived chan string
+mockResponse   *v1.SyncFlagsResponse
+}
+
+func (s *selectorHeaderCapturingServer) SyncFlags(req *v1.SyncFlagsRequest, stream syncv1grpc.FlagSyncService_SyncFlagsServer) error {
+// Extract metadata from context
+md, ok := metadata.FromIncomingContext(stream.Context())
+if ok {
+// Check for flagd-selector header
+if values := md.Get("flagd-selector"); len(values) > 0 {
+s.headerReceived <- values[0]
+} else {
+s.headerReceived <- ""
+}
+} else {
+s.headerReceived <- ""
+}
+
+// Send mock response
+err := stream.Send(s.mockResponse)
+if err != nil {
+return err
+}
+
+// Keep stream open for a bit
+time.Sleep(1 * time.Second)
+return nil
+}
+
+func (s *selectorHeaderCapturingServer) FetchAllFlags(ctx context.Context, req *v1.FetchAllFlagsRequest) (*v1.FetchAllFlagsResponse, error) {
+// Extract metadata from context
+md, ok := metadata.FromIncomingContext(ctx)
+if ok {
+// Check for flagd-selector header
+if values := md.Get("flagd-selector"); len(values) > 0 {
+s.headerReceived <- values[0]
+} else {
+s.headerReceived <- ""
+}
+} else {
+s.headerReceived <- ""
+}
+
+return &v1.FetchAllFlagsResponse{
+FlagConfiguration: flagRsp,
+}, nil
+}
+
+func (s *selectorHeaderCapturingServer) GetMetadata(ctx context.Context, req *v1.GetMetadataRequest) (*v1.GetMetadataResponse, error) {
+return &v1.GetMetadataResponse{}, nil
+}