fix(flagd): use provider params to configure grpc retry policy

Signed-off-by: Alexandra Oberaigner <[email protected]>

Author: alexandraoberaigner

providers/flagd/flagd-testbed

@@ -26,6 +26,9 @@ const (
 	defaultHost                         = "localhost"
 	defaultResolver                     = rpc
 	defaultGracePeriod                  = 5
+	defaultRetryBackoffMs               = 1000
+	defaultRetryBackoffMaxMs            = 120000
+	defaultFatalStatusCodes             = "UNAUTHENTICATED,PERMISSION_DENIED"
 
 	rpc       ResolverType = "rpc"
 	inProcess ResolverType = "in-process"
@@ -45,6 +48,9 @@ const (
 	flagdOfflinePathEnvironmentVariableName           = "FLAGD_OFFLINE_FLAG_SOURCE_PATH"
 	flagdTargetUriEnvironmentVariableName             = "FLAGD_TARGET_URI"
 	flagdGracePeriodVariableName                      = "FLAGD_RETRY_GRACE_PERIOD"
+	flagdRetryBackoffMsVariableName                   = "FLAGD_RETRY_BACKOFF_MS"
+	flagdRetryBackoffMaxMsVariableName                = "FLAGD_RETRY_BACKOFF_MAX_MS"
+	flagdFatalStatusCodesVariableName                 = "FLAGD_FATAL_STATUS_CODES"
 )
 
 type ProviderConfiguration struct {
@@ -66,6 +72,9 @@ type ProviderConfiguration struct {
 	CustomSyncProviderUri            string
 	GrpcDialOptionsOverride          []grpc.DialOption
 	RetryGracePeriod                 int
+	RetryBackoffMs                   int
+	RetryBackoffMaxMs                int
+	FatalStatusCodes                 []string
 
 	log logr.Logger
 }
@@ -80,6 +89,9 @@ func newDefaultConfiguration(log logr.Logger) *ProviderConfiguration {
 		Resolver:                         defaultResolver,
 		Tls:                              defaultTLS,
 		RetryGracePeriod:                 defaultGracePeriod,
+		RetryBackoffMs:                   defaultRetryBackoffMs,
+		RetryBackoffMaxMs:                defaultRetryBackoffMaxMs,
+		FatalStatusCodes:                 strings.Split(defaultFatalStatusCodes, ","),
 	}
 
 	p.updateFromEnvVar()
@@ -130,6 +142,7 @@ func validateProviderConfiguration(p *ProviderConfiguration) error {
 
 // updateFromEnvVar is a utility to update configurations based on current environment variables
 func (cfg *ProviderConfiguration) updateFromEnvVar() {
+
 	portS := os.Getenv(flagdPortEnvironmentVariableName)
 	if portS != "" {
 		port, err := strconv.Atoi(portS)
@@ -159,17 +172,7 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 		cfg.CertPath = certificatePath
 	}
 
-	if maxCacheSizeS := os.Getenv(flagdMaxCacheSizeEnvironmentVariableName); maxCacheSizeS != "" {
-		maxCacheSizeFromEnv, err := strconv.Atoi(maxCacheSizeS)
-		if err != nil {
-			cfg.log.Error(err,
-				fmt.Sprintf("invalid env config for %s provided, using default value: %d",
-					flagdMaxCacheSizeEnvironmentVariableName, defaultMaxCacheSize,
-				))
-		} else {
-			cfg.MaxCacheSize = maxCacheSizeFromEnv
-		}
-	}
+	cfg.MaxCacheSize = getIntFromEnvVarOrDefault(flagdMaxCacheSizeEnvironmentVariableName, defaultMaxCacheSize, cfg.log)
 
 	if cacheValue := os.Getenv(flagdCacheEnvironmentVariableName); cacheValue != "" {
 		switch cache.Type(cacheValue) {
@@ -185,18 +188,8 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 		}
 	}
 
-	if maxEventStreamRetriesS := os.Getenv(
-		flagdMaxEventStreamRetriesEnvironmentVariableName); maxEventStreamRetriesS != "" {
-
-		maxEventStreamRetries, err := strconv.Atoi(maxEventStreamRetriesS)
-		if err != nil {
-			cfg.log.Error(err,
-				fmt.Sprintf("invalid env config for %s provided, using default value: %d",
-					flagdMaxEventStreamRetriesEnvironmentVariableName, defaultMaxEventStreamRetries))
-		} else {
-			cfg.EventStreamConnectionMaxAttempts = maxEventStreamRetries
-		}
-	}
+	cfg.EventStreamConnectionMaxAttempts = getIntFromEnvVarOrDefault(
+		flagdMaxEventStreamRetriesEnvironmentVariableName, defaultMaxEventStreamRetries, cfg.log)
 
 	if resolver := os.Getenv(flagdResolverEnvironmentVariableName); resolver != "" {
 		switch strings.ToLower(resolver) {
@@ -227,17 +220,34 @@ func (cfg *ProviderConfiguration) updateFromEnvVar() {
 	if targetUri := os.Getenv(flagdTargetUriEnvironmentVariableName); targetUri != "" {
 		cfg.TargetUri = targetUri
 	}
-	if gracePeriod := os.Getenv(flagdGracePeriodVariableName); gracePeriod != "" {
-		if seconds, err := strconv.Atoi(gracePeriod); err == nil {
-			cfg.RetryGracePeriod = seconds
+
+	cfg.RetryGracePeriod = getIntFromEnvVarOrDefault(flagdGracePeriodVariableName, defaultGracePeriod, cfg.log)
+	cfg.RetryBackoffMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMsVariableName, defaultRetryBackoffMs, cfg.log)
+	cfg.RetryBackoffMaxMs = getIntFromEnvVarOrDefault(flagdRetryBackoffMaxMsVariableName, defaultRetryBackoffMaxMs, cfg.log)
+
+	if fatalStatusCodes := os.Getenv(flagdFatalStatusCodesVariableName); fatalStatusCodes != "" {
+		cfg.FatalStatusCodes = strings.Split(fatalStatusCodes, ",")
+	}
+}
+
+// Helper
+
+func getIntFromEnvVarOrDefault(envVarName string, defaultValue int, log logr.Logger) int {
+	if valueFromEnv := os.Getenv(envVarName); valueFromEnv != "" {
+		intValue, err := strconv.Atoi(valueFromEnv)
+		if err != nil {
+			log.Error(err,
+				fmt.Sprintf("invalid env config for %s provided, using default value: %d",
+					envVarName, defaultValue,
+				))
 		} else {
-			// Handle parsing error
-			cfg.log.Error(err, fmt.Sprintf("invalid grace period '%s'", gracePeriod))
+			return intValue
 		}
 	}
-
+	return defaultValue
 }
 
+
 // ProviderOptions
 
 type ProviderOption func(*ProviderConfiguration)
@@ -415,3 +425,25 @@ func WithRetryGracePeriod(gracePeriod int) ProviderOption {
 		p.RetryGracePeriod = gracePeriod
 	}
 }
+
+// WithRetryBackoffMs sets the initial backoff duration (in milliseconds) for retrying failed connections
+func WithRetryBackoffMs(retryBackoffMs int) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.RetryBackoffMs = retryBackoffMs
+	}
+}
+
+// WithRetryBackoffMaxMs sets the maximum backoff duration (in milliseconds) for retrying failed connections
+func WithRetryBackoffMaxMs(retryBackoffMaxMs int) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.RetryBackoffMaxMs = retryBackoffMaxMs
+	}
+}
+
+// WithFatalStatusCodes allows to set a list of gRPC status codes, which will cause streams to give up
+// and put the provider in a PROVIDER_FATAL state
+func WithFatalStatusCodes(fatalStatusCodes []string) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.FatalStatusCodes = fatalStatusCodes
+	}
+}

providers/flagd/pkg/configuration.go

@@ -74,6 +74,9 @@ func NewProvider(opts ...ProviderOption) (*Provider, error) {
 			CustomSyncProviderUri:   provider.providerConfiguration.CustomSyncProviderUri,
 			GrpcDialOptionsOverride: provider.providerConfiguration.GrpcDialOptionsOverride,
 			RetryGracePeriod:        provider.providerConfiguration.RetryGracePeriod,
+			RetryBackOffMs:          provider.providerConfiguration.RetryBackoffMs,
+			RetryBackOffMaxMs:       provider.providerConfiguration.RetryBackoffMaxMs,
+			FatalStatusCodes:        provider.providerConfiguration.FatalStatusCodes,
 		})
 	default:
 		service = process.NewInProcessService(process.Configuration{

providers/flagd/pkg/provider.go

@@ -0,0 +1,67 @@
+package process
+
+import (
+	"encoding/json"
+	"strings"
+	"time"
+)
+
+const (
+	// Default timeouts and retry intervals
+	defaultKeepaliveTime    = 30 * time.Second
+	defaultKeepaliveTimeout = 5 * time.Second
+)
+
+type RetryPolicy struct {
+	MaxAttempts          int      `json:"MaxAttempts"`
+	InitialBackoff       string   `json:"InitialBackoff"`
+	MaxBackoff           string   `json:"MaxBackoff"`
+	BackoffMultiplier    float64  `json:"BackoffMultiplier"`
+	RetryableStatusCodes []string `json:"RetryableStatusCodes"`
+}
+
+func (g *Sync) buildRetryPolicy() string {
+	var policy = map[string]interface{}{
+		"methodConfig": []map[string]interface{}{
+			{
+				"name": []map[string]string{
+					{"service": "flagd.sync.v1.FlagSyncService"},
+				},
+				"retryPolicy": RetryPolicy{
+					MaxAttempts:          3,
+					InitialBackoff:		  (time.Duration(g.RetryBackOffMs) * time.Millisecond).String(),
+					MaxBackoff: 		  (time.Duration(g.RetryBackOffMaxMs) * time.Millisecond).String(),
+					BackoffMultiplier:    2.0,
+					RetryableStatusCodes: []string{"UNKNOWN","UNAVAILABLE"},
+				},
+			},
+		},
+	}
+	retryPolicyBytes, _ := json.Marshal(policy)
+	retryPolicy := string(retryPolicyBytes)
+
+	return retryPolicy
+}
+
+// Set of non-retryable gRPC status codes for faster lookup
+var nonRetryableCodes map[string]struct{}
+
+// initNonRetryableStatusCodesSet initializes the set of non-retryable gRPC status codes for quick lookup
+func (g *Sync) initNonRetryableStatusCodesSet()  {
+	nonRetryableCodes = make(map[string]struct{})
+	for _, code := range g.FatalStatusCodes {
+		normalized := toCamelCase(code)
+		nonRetryableCodes[normalized] = struct{}{}
+	}
+}
+
+// toCamelCase converts a SNAKE_CASE string to CamelCase
+func toCamelCase(s string) string {
+	parts := strings.Split(strings.ToLower(s), "_")
+	for i, part := range parts {
+		if len(part) > 0 {
+			parts[i] = strings.ToUpper(part[:1]) + part[1:]
+		}
+	}
+	return strings.Join(parts, "")
+}

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

@@ -0,0 +1,125 @@
+package process
+
+import (
+	"encoding/json"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+
+func TestBuildRetryPolicy(t *testing.T) {
+	g := &Sync{
+		RetryBackOffMs:    100,
+		RetryBackOffMaxMs: 500,
+	}
+
+	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")
+	}
+
+	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["MaxBackoff"].(string) != "500ms" {
+		t.Errorf("MaxBackoff = %v; want 500ms", retryPolicy["MaxBackoff"])
+	}
+	if retryPolicy["BackoffMultiplier"].(float64) != 2.0 {
+		t.Errorf("BackoffMultiplier = %v; want 2.0", retryPolicy["BackoffMultiplier"])
+	}
+	codes := retryPolicy["RetryableStatusCodes"].([]interface{})
+	expectedCodes := []string{"UNKNOWN", "UNAVAILABLE"}
+	for i, code := range expectedCodes {
+		if codes[i].(string) != code {
+			t.Errorf("RetryableStatusCodes[%d] = %v; want %v", i, codes[i], code)
+		}
+	}
+
+	// Also check that the result is valid JSON and contains expected substrings
+	if !strings.Contains(result, `"MaxAttempts":3`) {
+		t.Error("Result does not contain MaxAttempts")
+	}
+	if !strings.Contains(result, `"InitialBackoff":"100ms"`) {
+		t.Error("Result does not contain InitialBackoff")
+	}
+	if !strings.Contains(result, `"MaxBackoff":"500ms"`) {
+		t.Error("Result does not contain MaxBackoff")
+	}
+	if !strings.Contains(result, `"RetryableStatusCodes":["UNKNOWN","UNAVAILABLE"]`) {
+		t.Error("Result does not contain RetryableStatusCodes")
+	}
+}
+
+type syncTestCase struct {
+	input    []string
+	expected map[string]struct{}
+}
+
+func TestInitNonRetryableStatusCodesSet(t *testing.T) {
+	testCases := []syncTestCase{
+		{
+			input:    []string{"PERMISSION_DENIED", "UNKNOWN"},
+			expected: map[string]struct{}{"PermissionDenied": {}, "Unknown": {}},
+		},
+		{
+			input:    []string{"ALREADY_EXISTS"},
+			expected: map[string]struct{}{"AlreadyExists": {}},
+		},
+		{
+			input:    []string{},
+			expected: map[string]struct{}{},
+		},
+	}
+
+	for _, tc := range testCases {
+		g := &Sync{FatalStatusCodes: tc.input}
+		nonRetryableCodes = nil // reset global
+		g.initNonRetryableStatusCodesSet()
+		if !reflect.DeepEqual(nonRetryableCodes, tc.expected) {
+			t.Errorf("input: %v, got: %v, want: %v", tc.input, nonRetryableCodes, tc.expected)
+		}
+	}
+}
+
+func TestToCamelCase(t *testing.T) {
+ testCases := []struct {
+  input    string
+  expected string
+ }{
+  {"INVALID_ARGUMENT", "InvalidArgument"},
+  {"NOT_FOUND", "NotFound"},
+  {"ALREADY_EXISTS", "AlreadyExists"},
+  {"UNKNOWN", "Unknown"},
+  {"", ""},
+  {"SINGLE", "Single"},
+  {"MULTI_WORD_EXAMPLE", "MultiWordExample"},
+  {"_LEADING_UNDERSCORE", "LeadingUnderscore"},
+  {"TRAILING_UNDERSCORE_", "TrailingUnderscore"},
+  {"__DOUBLE__UNDERSCORES__", "DoubleUnderscores"},
+ }
+
+ for _, tc := range testCases {
+  got := toCamelCase(tc.input)
+  if got != tc.expected {
+   t.Errorf("toCamelCase(%q) = %q; want %q", tc.input, got, tc.expected)
+  }
+ }
+}