feat(flagd): Add flagd-selector gRPC metadata header to in-process service. (#790)

Signed-off-by: Simon Schrottner <simon.schrottner@dynatrace.com>
Signed-off-by: Todd Baert <todd.baert@dynatrace.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: aepfli <9987394+aepfli@users.noreply.github.com>
Co-authored-by: Simon Schrottner <simon.schrottner@dynatrace.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: Todd Baert <todd.baert@dynatrace.com>

Author: 4 people

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/configuration.go

@@ -18,6 +18,10 @@ type ResolverType string
 
 // Naming and defaults must comply with flagd environment variables
 const (
+	// DefaultRetryBackoffMs is the default initial backoff duration for stream retry
+	DefaultRetryBackoffMs = 1000
+	// DefaultRetryBackoffMaxMs is the default maximum backoff duration for stream retry
+	DefaultRetryBackoffMaxMs            = 120000
 	defaultMaxCacheSize          int    = 1000
 	defaultRpcPort               uint16 = 8013
 	defaultInProcessPort         uint16 = 8015
@@ -27,8 +31,6 @@ const (
 	defaultHost                         = "localhost"
 	defaultResolver                     = rpc
 	defaultGracePeriod                  = 5
-	defaultRetryBackoffMs               = 1000
-	defaultRetryBackoffMaxMs            = 120000
 	defaultFatalStatusCodes             = ""
 
 	rpc       ResolverType = "rpc"
@@ -90,8 +92,8 @@ func newDefaultConfiguration(log logr.Logger) *ProviderConfiguration {
 		Resolver:                         defaultResolver,
 		Tls:                              defaultTLS,
 		RetryGracePeriod:                 defaultGracePeriod,
-		RetryBackoffMs:                   defaultRetryBackoffMs,
-		RetryBackoffMaxMs:                defaultRetryBackoffMaxMs,
+		RetryBackoffMs:                   DefaultRetryBackoffMs,
+		RetryBackoffMaxMs:                DefaultRetryBackoffMaxMs,
 	}
 
 	p.updateFromEnvVar()
@@ -222,8 +224,8 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 	}
 
 	cfg.RetryGracePeriod = getIntFromEnvVarOrDefault(flagdGracePeriodVariableName, defaultGracePeriod, cfg.log)
-	cfg.RetryBackoffMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMsVariableName, defaultRetryBackoffMs, cfg.log)
-	cfg.RetryBackoffMaxMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMaxMsVariableName, defaultRetryBackoffMaxMs, cfg.log)
+	cfg.RetryBackoffMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMsVariableName, DefaultRetryBackoffMs, cfg.log)
+	cfg.RetryBackoffMaxMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMaxMsVariableName, DefaultRetryBackoffMaxMs, cfg.log)
 
 	var fatalStatusCodes string
 	if envVal := os.Getenv(flagdFatalStatusCodesVariableName); envVal != "" {

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

@@ -3,14 +3,18 @@ package process
 import (
 	"encoding/json"
 	"fmt"
-	"google.golang.org/grpc/codes"
 	"time"
+
+	"google.golang.org/grpc/codes"
 )
 
 const (
-	// Default timeouts and retry intervals
+	// Default timeouts for keepalive settings
 	defaultKeepaliveTime    = 30 * time.Second
 	defaultKeepaliveTimeout = 5 * time.Second
+	// Default retry intervals per https://flagd.dev/reference/specifications/providers/#configuration
+	DefaultRetryBackoffMs    = 1000   // 1 second
+	DefaultRetryBackoffMaxMs = 120000 // 120 seconds
 )
 
 type RetryPolicy struct {
@@ -22,6 +26,25 @@ type RetryPolicy struct {
 }
 
 func (g *Sync) buildRetryPolicy() string {
+	// Use default values if not configured (per https://flagd.dev/reference/specifications/providers/#configuration)
+	initialBackoffMs := g.RetryBackOffMs
+	if initialBackoffMs <= 0 {
+		initialBackoffMs = DefaultRetryBackoffMs
+	}
+
+	maxBackoffMs := g.RetryBackOffMaxMs
+	if maxBackoffMs <= 0 {
+		maxBackoffMs = DefaultRetryBackoffMaxMs
+	}
+
+	// Format durations for gRPC service config (requires seconds-only format like "1s", "0.1s", "120s")
+	initialDur := time.Duration(initialBackoffMs) * time.Millisecond
+	maxDur := time.Duration(maxBackoffMs) * time.Millisecond
+
+	// Convert to seconds and format as gRPC expects (no compound formats like "2m0s")
+	initialBackoff := fmt.Sprint(initialDur.Seconds()) + "s"
+	maxBackoff := fmt.Sprint(maxDur.Seconds()) + "s"
+
 	var policy = map[string]interface{}{
 		"methodConfig": []map[string]interface{}{
 			{
@@ -30,8 +53,8 @@ func (g *Sync) buildRetryPolicy() string {
 				},
 				"retryPolicy": RetryPolicy{
 					MaxAttempts:          3,
-					InitialBackoff:       (time.Duration(g.RetryBackOffMs) * time.Millisecond).String(),
-					MaxBackoff:           (time.Duration(g.RetryBackOffMaxMs) * time.Millisecond).String(),
+					InitialBackoff:       initialBackoff,
+					MaxBackoff:           maxBackoff,
 					BackoffMultiplier:    2.0,
 					RetryableStatusCodes: []string{"UNKNOWN", "UNAVAILABLE"},
 				},

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

@@ -2,11 +2,12 @@ package process
 
 import (
 	"encoding/json"
+	"strings"
+	"testing"
+
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
-	"strings"
-	"testing"
 )
 
 func TestBuildRetryPolicy(t *testing.T) {
@@ -37,11 +38,11 @@ func TestBuildRetryPolicy(t *testing.T) {
 	if retryPolicy["MaxAttempts"].(float64) != 3 {
 		t.Errorf("MaxAttempts = %v; want 3", retryPolicy["MaxAttempts"])
 	}
-	if retryPolicy["InitialBackoff"].(string) != "100ms" {
-		t.Errorf("InitialBackoff = %v; want 100ms", retryPolicy["InitialBackoff"])
+	if retryPolicy["InitialBackoff"].(string) != "0.1s" {
+		t.Errorf("InitialBackoff = %v; want 0.1s", retryPolicy["InitialBackoff"])
 	}
-	if retryPolicy["MaxBackoff"].(string) != "500ms" {
-		t.Errorf("MaxBackoff = %v; want 500ms", retryPolicy["MaxBackoff"])
+	if retryPolicy["MaxBackoff"].(string) != "0.5s" {
+		t.Errorf("MaxBackoff = %v; want 0.5s", retryPolicy["MaxBackoff"])
 	}
 	if retryPolicy["BackoffMultiplier"].(float64) != 2.0 {
 		t.Errorf("BackoffMultiplier = %v; want 2.0", retryPolicy["BackoffMultiplier"])
@@ -58,17 +59,87 @@ func TestBuildRetryPolicy(t *testing.T) {
 	if !strings.Contains(result, `"MaxAttempts":3`) {
 		t.Error("Result does not contain MaxAttempts")
 	}
-	if !strings.Contains(result, `"InitialBackoff":"100ms"`) {
+	if !strings.Contains(result, `"InitialBackoff":"0.1s"`) {
 		t.Error("Result does not contain InitialBackoff")
 	}
-	if !strings.Contains(result, `"MaxBackoff":"500ms"`) {
+	if !strings.Contains(result, `"MaxBackoff":"0.5s"`) {
 		t.Error("Result does not contain MaxBackoff")
 	}
 	if !strings.Contains(result, `"RetryableStatusCodes":["UNKNOWN","UNAVAILABLE"]`) {
 		t.Error("Result does not contain RetryableStatusCodes")
 	}
 }
 
+// TestBuildRetryPolicyDefaults verifies that default values are applied per spec
+func TestBuildRetryPolicyDefaults(t *testing.T) {
+	g := &Sync{
+		RetryBackOffMs:    0, // Should use default 1000ms
+		RetryBackOffMaxMs: 0, // Should use default 120000ms
+	}
+
+	result := g.buildRetryPolicy()
+
+	// Unmarshal to check structure
+	var policy map[string]interface{}
+	if err := json.Unmarshal([]byte(result), &policy); err != nil {
+		t.Fatalf("Failed to unmarshal result: %v", err)
+	}
+
+	methodConfig, ok := policy["methodConfig"].([]interface{})
+	if !ok || len(methodConfig) == 0 {
+		t.Fatalf("methodConfig missing or empty")
+	}
+
+	config := methodConfig[0].(map[string]interface{})
+	retryPolicy, ok := config["retryPolicy"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("retryPolicy missing")
+	}
+
+	// time.Duration.String() formats milliseconds as "1s", "120s", etc.
+	if retryPolicy["InitialBackoff"].(string) != "1s" {
+		t.Errorf("InitialBackoff = %v; want 1s (default)", retryPolicy["InitialBackoff"])
+	}
+	if retryPolicy["MaxBackoff"].(string) != "120s" {
+		t.Errorf("MaxBackoff = %v; want 120s (gRPC format for 120000ms)", retryPolicy["MaxBackoff"])
+	}
+}
+
+// TestBuildRetryPolicyFractionalSeconds verifies fractional second durations are formatted correctly
+func TestBuildRetryPolicyFractionalSeconds(t *testing.T) {
+	g := &Sync{
+		RetryBackOffMs:    1500, // 1.5 seconds
+		RetryBackOffMaxMs: 2500, // 2.5 seconds
+	}
+
+	result := g.buildRetryPolicy()
+
+	// Unmarshal to check structure
+	var policy map[string]interface{}
+	if err := json.Unmarshal([]byte(result), &policy); err != nil {
+		t.Fatalf("Failed to unmarshal result: %v", err)
+	}
+
+	methodConfig, ok := policy["methodConfig"].([]interface{})
+	if !ok || len(methodConfig) == 0 {
+		t.Fatalf("methodConfig missing or empty")
+	}
+
+	config := methodConfig[0].(map[string]interface{})
+	retryPolicy, ok := config["retryPolicy"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("retryPolicy missing")
+	}
+
+	// Verify fractional seconds are preserved in gRPC format
+	if retryPolicy["InitialBackoff"].(string) != "1.5s" {
+		t.Errorf("InitialBackoff = %v; want 1.5s", retryPolicy["InitialBackoff"])
+	}
+	if retryPolicy["MaxBackoff"].(string) != "2.5s" {
+		t.Errorf("MaxBackoff = %v; want 2.5s", retryPolicy["MaxBackoff"])
+	}
+}
+
 func TestSync_initNonRetryableStatusCodesSet(t *testing.T) {
 	tests := []struct {
 		name             string

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

@@ -0,0 +1,44 @@
+package process
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+)
+
+const SelectorMetadataKey = "flagd-selector"
+
+// selectorUnaryInterceptor adds the flagd-selector metadata header to unary gRPC calls
+func selectorUnaryInterceptor(selector string) grpc.UnaryClientInterceptor {
+	return func(
+		ctx context.Context,
+		method string,
+		req, reply interface{},
+		cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		opts ...grpc.CallOption,
+	) error {
+		if selector != "" {
+			ctx = metadata.AppendToOutgoingContext(ctx, SelectorMetadataKey, selector)
+		}
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+}
+
+// selectorStreamInterceptor adds the flagd-selector metadata header to streaming gRPC calls
+func selectorStreamInterceptor(selector string) grpc.StreamClientInterceptor {
+	return func(
+		ctx context.Context,
+		desc *grpc.StreamDesc,
+		cc *grpc.ClientConn,
+		method string,
+		streamer grpc.Streamer,
+		opts ...grpc.CallOption,
+	) (grpc.ClientStream, error) {
+		if selector != "" {
+			ctx = metadata.AppendToOutgoingContext(ctx, SelectorMetadataKey, selector)
+		}
+		return streamer(ctx, desc, cc, method, opts...)
+	}
+}

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

@@ -81,16 +81,29 @@ func (g *Sync) Init(ctx context.Context) error {
 
 // createConnection creates and configures the gRPC connection
 func (g *Sync) createConnection() (*grpc.ClientConn, error) {
+
+	var grpcInterceptorDialOptions []grpc.DialOption
+	if g.Selector != "" {
+		grpcInterceptorDialOptions = append(grpcInterceptorDialOptions,
+			grpc.WithChainUnaryInterceptor(selectorUnaryInterceptor(g.Selector)),
+			grpc.WithChainStreamInterceptor(selectorStreamInterceptor(g.Selector)),
+		)
+	}
+
 	if len(g.GrpcDialOptionsOverride) > 0 {
 		g.Logger.Debug("using provided gRPC DialOptions override")
-		return grpc.NewClient(g.URI, g.GrpcDialOptionsOverride...)
+		dialOptions := make([]grpc.DialOption, 0, len(g.GrpcDialOptionsOverride)+len(grpcInterceptorDialOptions))
+		dialOptions = append(dialOptions, g.GrpcDialOptionsOverride...)
+		dialOptions = append(dialOptions, grpcInterceptorDialOptions...)
+		return grpc.NewClient(g.URI, dialOptions...)
 	}
 
 	// Build standard dial options
 	dialOptions, err := g.buildDialOptions()
 	if err != nil {
 		return nil, fmt.Errorf("failed to build dial options: %w", err)
 	}
+	dialOptions = append(dialOptions, grpcInterceptorDialOptions...)
 
 	return grpc.NewClient(g.URI, dialOptions...)
 }

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

@@ -1,16 +1,17 @@
 package process
 
 import (
-	"buf.build/gen/go/open-feature/flagd/grpc/go/flagd/sync/v1/syncv1grpc"
-	v1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/flagd/sync/v1"
 	"context"
 	"fmt"
-	"github.com/open-feature/go-sdk/openfeature"
-	"google.golang.org/grpc"
 	"log"
 	"net"
 	"testing"
 	"time"
+
+	"buf.build/gen/go/open-feature/flagd/grpc/go/flagd/sync/v1/syncv1grpc"
+	v1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/flagd/sync/v1"
+	"github.com/open-feature/go-sdk/openfeature"
+	"google.golang.org/grpc"
 )
 
 // shared flag for tests