feat(flagd): add eventing with graceperiod for inprocess resolver (open-feature#744)

Signed-off-by: Simon Schrottner <[email protected]>

Author: aepfli

.gitmodules

@@ -4,3 +4,4 @@
 [submodule "providers/flagd/flagd-testbed"]
 	path = providers/flagd/flagd-testbed
 	url = https://github.com/open-feature/flagd-testbed.git
+	branch = v2.11.1

providers/flagd/e2e/inprocess_test.go

@@ -25,8 +25,8 @@ func TestInProcessProviderE2E(t *testing.T) {
 		"./",
 	}
 
-	// Run tests with in-process specific tags - exclude connection/event issues we won't tackle
-	tags := "@in-process && ~@unixsocket && ~@metadata && ~@grace && ~@customCert && ~@reconnect && ~@contextEnrichment && ~@sync-payload && ~@events"
+	// Run tests with in-process specific tags
+	tags := "@in-process && ~@unixsocket && ~@metadata && ~@customCert && ~@contextEnrichment && ~@sync-payload"
 
 	if err := runner.RunGherkinTestsWithSubtests(t, featurePaths, tags); err != nil {
 		t.Fatalf("Gherkin tests failed: %v", err)

providers/flagd/pkg/configuration.go

@@ -3,15 +3,14 @@ package flagd
 import (
 	"errors"
 	"fmt"
-	"os"
-	"strconv"
-	"strings"
-
 	"github.com/go-logr/logr"
 	"github.com/open-feature/flagd/core/pkg/sync"
 	"github.com/open-feature/go-sdk-contrib/providers/flagd/internal/cache"
 	"github.com/open-feature/go-sdk-contrib/providers/flagd/internal/logger"
 	"google.golang.org/grpc"
+	"os"
+	"strconv"
+	"strings"
 )
 
 type ResolverType string
@@ -26,6 +25,7 @@ const (
 	defaultCache                        = cache.LRUValue
 	defaultHost                         = "localhost"
 	defaultResolver                     = rpc
+	defaultGracePeriod                  = 5
 
 	rpc       ResolverType = "rpc"
 	inProcess ResolverType = "in-process"
@@ -44,6 +44,7 @@ const (
 	flagdSourceSelectorEnvironmentVariableName        = "FLAGD_SOURCE_SELECTOR"
 	flagdOfflinePathEnvironmentVariableName           = "FLAGD_OFFLINE_FLAG_SOURCE_PATH"
 	flagdTargetUriEnvironmentVariableName             = "FLAGD_TARGET_URI"
+	flagdGracePeriodVariableName                      = "FLAGD_RETRY_GRACE_PERIOD"
 )
 
 type ProviderConfiguration struct {
@@ -64,6 +65,7 @@ type ProviderConfiguration struct {
 	CustomSyncProvider               sync.ISync
 	CustomSyncProviderUri            string
 	GrpcDialOptionsOverride          []grpc.DialOption
+	RetryGracePeriod                 int
 
 	log logr.Logger
 }
@@ -77,6 +79,7 @@ func newDefaultConfiguration(log logr.Logger) *ProviderConfiguration {
 		MaxCacheSize:                     defaultMaxCacheSize,
 		Resolver:                         defaultResolver,
 		Tls:                              defaultTLS,
+		RetryGracePeriod:                 defaultGracePeriod,
 	}
 
 	p.updateFromEnvVar()
@@ -224,6 +227,14 @@ 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
+		} else {
+			// Handle parsing error
+			cfg.log.Error(err, fmt.Sprintf("invalid grace period '%s'", gracePeriod))
+		}
+	}
 
 }
 
@@ -397,3 +408,10 @@ func WithGrpcDialOptionsOverride(grpcDialOptionsOverride []grpc.DialOption) Prov
 		p.GrpcDialOptionsOverride = grpcDialOptionsOverride
 	}
 }
+
+// WithRetryGracePeriod allows to set a time window for the transition from stale to error state
+func WithRetryGracePeriod(gracePeriod int) ProviderOption {
+	return func(p *ProviderConfiguration) {
+		p.RetryGracePeriod = gracePeriod
+	}
+}

providers/flagd/pkg/provider.go

@@ -73,6 +73,7 @@ func NewProvider(opts ...ProviderOption) (*Provider, error) {
 			CustomSyncProvider:      provider.providerConfiguration.CustomSyncProvider,
 			CustomSyncProviderUri:   provider.providerConfiguration.CustomSyncProviderUri,
 			GrpcDialOptionsOverride: provider.providerConfiguration.GrpcDialOptionsOverride,
+			RetryGracePeriod:        provider.providerConfiguration.RetryGracePeriod,
 		})
 	default:
 		service = process.NewInProcessService(process.Configuration{