diff --git a/core/pkg/service/iservice.go b/core/pkg/service/iservice.go index 67f281195..512179556 100644 --- a/core/pkg/service/iservice.go +++ b/core/pkg/service/iservice.go @@ -36,6 +36,7 @@ type Configuration struct { ContextValues map[string]any HeaderToContextKeyMappings map[string]string StreamDeadline time.Duration + StreamDeadlineGraceful bool MaxRequestHeaderBytes int64 MaxRequestBodyBytes int64 diff --git a/docs/reference/flagd-cli/flagd_start.md b/docs/reference/flagd-cli/flagd_start.md index 3339b8b74..811dff141 100644 --- a/docs/reference/flagd-cli/flagd_start.md +++ b/docs/reference/flagd-cli/flagd_start.md @@ -38,6 +38,7 @@ flagd start [flags] -d, --socket-path string Flagd unix socket path. With grpc the evaluations service will become available on this address. With http(s) the grpc-gateway proxy will use this address internally. -s, --sources string JSON representation of an array of SourceConfig objects. Required fields: uri (string) and provider (string). Optional source-specific fields are also available, see https://flagd.dev/reference/sync-configuration/#source-configuration --stream-deadline duration Set a server-side deadline for flagd sync and event streams (default 0, means no deadline). + --stream-deadline-graceful-close Close streams reaching the server-side deadline with an OK status instead of DEADLINE_EXCEEDED, so providers reconnect without reporting an error. Defaults to false. --sync-http-enabled Serve the flag configuration document over HTTP at /v1/flags on the sync port. Defaults to true. (default true) -g, --sync-port int32 Flag sync service port, serving gRPC, gRPC-Web and Connect (default 8015) -e, --sync-socket-path string Flagd sync service socket path. With grpc the sync service will be available on this address. diff --git a/flagd/cmd/start.go b/flagd/cmd/start.go index 2749a19e4..bbb521e33 100644 --- a/flagd/cmd/start.go +++ b/flagd/cmd/start.go @@ -45,6 +45,7 @@ const ( contextValueFlagName = "context-value" headerToContextKeyFlagName = "context-from-header" streamDeadlineFlagName = "stream-deadline" + streamDeadlineGracefulFlag = "stream-deadline-graceful-close" maxRequestBodyFlagName = "max-request-body" maxRequestHeaderFlagName = "max-request-header" @@ -105,6 +106,7 @@ func init() { flags.StringToStringP(headerToContextKeyFlagName, "H", map[string]string{}, "add key-value pairs to map "+ "header values to context values, where key is Header name, value is context key") flags.Duration(streamDeadlineFlagName, 0, "Set a server-side deadline for flagd sync and event streams (default 0, means no deadline).") + flags.Bool(streamDeadlineGracefulFlag, false, "Close streams reaching the server-side deadline with an OK status instead of DEADLINE_EXCEEDED, so providers reconnect without reporting an error. Defaults to false.") flags.Bool(disableSyncMetadata, false, "Disables the getMetadata endpoint of the sync service. Defaults to false, but will default to true in later versions.") flags.Int64P(maxRequestBodyFlagName, "B", 1_000_000, "Maximum allowed request body size in bytes. Requests exceeding this are rejected with HTTP 413 (OFREP) or 429 (connect). Set to 0 to disable. WARNING: disabling this limit may allow memory exhaustion from oversized requests.") flags.Int64P(maxRequestHeaderFlagName, "R", 1_000_000, "Maximum allowed request header size in bytes. Requests exceeding this are rejected with HTTP 431. Set to 0 to use Go's built-in default (1 MiB). WARNING: setting a very large or zero value may allow memory exhaustion from oversized headers.") @@ -139,6 +141,7 @@ func bindFlags(flags *pflag.FlagSet) { _ = viper.BindPFlag(contextValueFlagName, flags.Lookup(contextValueFlagName)) _ = viper.BindPFlag(headerToContextKeyFlagName, flags.Lookup(headerToContextKeyFlagName)) _ = viper.BindPFlag(streamDeadlineFlagName, flags.Lookup(streamDeadlineFlagName)) + _ = viper.BindPFlag(streamDeadlineGracefulFlag, flags.Lookup(streamDeadlineGracefulFlag)) _ = viper.BindPFlag(disableSyncMetadata, flags.Lookup(disableSyncMetadata)) _ = viper.BindPFlag(maxRequestBodyFlagName, flags.Lookup(maxRequestBodyFlagName)) _ = viper.BindPFlag(maxRequestHeaderFlagName, flags.Lookup(maxRequestHeaderFlagName)) @@ -239,6 +242,7 @@ var startCmd = &cobra.Command{ SyncServiceHTTPEnabled: viper.GetBool(syncHTTPFlagName), SyncServiceSocketPath: viper.GetString(syncSocketPathFlagName), StreamDeadline: viper.GetDuration(streamDeadlineFlagName), + StreamDeadlineGraceful: viper.GetBool(streamDeadlineGracefulFlag), DisableSyncMetadata: viper.GetBool(disableSyncMetadata), //nolint:staticcheck // SA1019 no-ops, read so FromConfig can warn when they are set KeepAliveMinTime: viper.GetDuration(keepAliveMinTimeFlagName), diff --git a/flagd/pkg/runtime/from_config.go b/flagd/pkg/runtime/from_config.go index 931951ff2..fd34710ec 100644 --- a/flagd/pkg/runtime/from_config.go +++ b/flagd/pkg/runtime/from_config.go @@ -44,6 +44,7 @@ type Config struct { SyncServiceHTTPEnabled bool SyncServiceSocketPath string StreamDeadline time.Duration + StreamDeadlineGraceful bool DisableSyncMetadata bool // Deprecated: no-op, retained so existing configurations keep working. See warnOnNoOpConfig. @@ -168,21 +169,22 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime, // flag sync service flagSyncService, err := flagsync.NewSyncService(flagsync.SvcConfigurations{ - Logger: logger.WithFields(zap.String("component", "FlagSyncService")), - Port: config.SyncServicePort, - Sources: sources, - Store: store, - ContextValues: config.ContextValues, - KeyPath: config.ServiceKeyPath, - CertPath: config.ServiceCertPath, - SocketPath: config.SyncServiceSocketPath, - HTTPEnabled: config.SyncServiceHTTPEnabled, - ServiceName: svcName, - CORS: config.CORS, - Options: options, - StreamDeadline: config.StreamDeadline, - DisableSyncMetadata: config.DisableSyncMetadata, - MetricsRecorder: recorder, + Logger: logger.WithFields(zap.String("component", "FlagSyncService")), + Port: config.SyncServicePort, + Sources: sources, + Store: store, + ContextValues: config.ContextValues, + KeyPath: config.ServiceKeyPath, + CertPath: config.ServiceCertPath, + SocketPath: config.SyncServiceSocketPath, + HTTPEnabled: config.SyncServiceHTTPEnabled, + ServiceName: svcName, + CORS: config.CORS, + Options: options, + StreamDeadline: config.StreamDeadline, + StreamDeadlineGraceful: config.StreamDeadlineGraceful, + DisableSyncMetadata: config.DisableSyncMetadata, + MetricsRecorder: recorder, MaxRequestHeaderBytes: config.MaxRequestHeaderBytes, MaxRequestBodyBytes: config.MaxRequestBodyBytes, @@ -219,6 +221,7 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime, ContextValues: config.ContextValues, HeaderToContextKeyMappings: config.HeaderToContextKeyMappings, StreamDeadline: config.StreamDeadline, + StreamDeadlineGraceful: config.StreamDeadlineGraceful, MaxRequestBodyBytes: config.MaxRequestBodyBytes, MaxRequestHeaderBytes: config.MaxRequestHeaderBytes, }, diff --git a/flagd/pkg/service/flag-evaluation/connect_service.go b/flagd/pkg/service/flag-evaluation/connect_service.go index 6e49ae570..21a50110c 100644 --- a/flagd/pkg/service/flag-evaluation/connect_service.go +++ b/flagd/pkg/service/flag-evaluation/connect_service.go @@ -160,6 +160,7 @@ func (s *ConnectService) setupServer(svcConf service.Configuration) (net.Listene svcConf.HeaderToContextKeyMappings, svcConf.StreamDeadline, ) + v1Fes.deadlineGraceful = svcConf.StreamDeadlineGraceful _, v1Handler := evaluationV1.NewServiceHandler(v1Fes, append(svcConf.Options, marshalOpts)...) @@ -173,6 +174,7 @@ func (s *ConnectService) setupServer(svcConf service.Configuration) (net.Listene svcConf.HeaderToContextKeyMappings, svcConf.StreamDeadline, ) + v2Fes.deadlineGraceful = svcConf.StreamDeadlineGraceful _, v2Handler := evaluationV2.NewServiceHandler(v2Fes, append(svcConf.Options, marshalOpts)...) diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator.go b/flagd/pkg/service/flag-evaluation/flag_evaluator.go index 1f21aced1..805a30fa2 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator.go +++ b/flagd/pkg/service/flag-evaluation/flag_evaluator.go @@ -358,6 +358,17 @@ func (s *OldFlagEvaluationService) ResolveObject( return res, err } +// deadlineExceededResult is the result of an event stream reaching its server-side deadline: +// OK when graceful, so providers reconnect as for a completed stream, DEADLINE_EXCEEDED otherwise +func deadlineExceededResult(log *logger.Logger, deadline time.Duration, graceful bool) error { + if graceful { + log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", deadline.String())) + return nil + } + log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", deadline.String())) + return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("%s", "stream closed due to server-side timeout")) +} + func mergeContexts(reqCtx, configFlagsCtx map[string]any, headers http.Header, headerToContextKeyMappings map[string]string) map[string]any { return MergeContextsAndHeaders(reqCtx, configFlagsCtx, headers, headerToContextKeyMappings) } diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go index d7e45797c..3835a8696 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go +++ b/flagd/pkg/service/flag-evaluation/flag_evaluator_test.go @@ -6,6 +6,7 @@ import ( "context" "errors" "testing" + "time" schemaV1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/schema/v1" "connectrpc.com/connect" @@ -1112,3 +1113,17 @@ func assertInvalidSelectorConnect(t *testing.T, cases []invalidSelectorCase) { }) } } + +func TestDeadlineExceededResult(t *testing.T) { + log := logger.NewLogger(nil, false) + + t.Run("graceful closes the stream with OK", func(t *testing.T) { + require.NoError(t, deadlineExceededResult(log, time.Minute, true)) + }) + + t.Run("non-graceful returns DEADLINE_EXCEEDED", func(t *testing.T) { + err := deadlineExceededResult(log, time.Minute, false) + require.Error(t, err) + require.Equal(t, connect.CodeDeadlineExceeded, connect.CodeOf(err)) + }) +} diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go index a1391670c..795d41a01 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go +++ b/flagd/pkg/service/flag-evaluation/flag_evaluator_v1.go @@ -30,6 +30,7 @@ type FlagEvaluationService struct { contextValues map[string]any headerToContextKeyMappings map[string]string deadline time.Duration + deadlineGraceful bool } // NewFlagEvaluationService creates a FlagEvaluationService with provided parameters @@ -203,8 +204,7 @@ func (s *FlagEvaluationService) EventStream( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { - s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String())) - return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("%s", "stream closed due to server-side timeout")) + return deadlineExceededResult(s.logger, s.deadline, s.deadlineGraceful) } return nil } diff --git a/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go b/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go index 1a3d68bd3..794b77c2f 100644 --- a/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go +++ b/flagd/pkg/service/flag-evaluation/flag_evaluator_v2.go @@ -32,6 +32,7 @@ type FlagEvaluationServiceV2 struct { contextValues map[string]any headerToContextKeyMappings map[string]string deadline time.Duration + deadlineGraceful bool } // NewFlagEvaluationServiceV2 creates a FlagEvaluationServiceV2 with provided parameters @@ -125,8 +126,7 @@ func (s *FlagEvaluationServiceV2) EventStream( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { - s.logger.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String())) - return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("%s", "stream closed due to server-side timeout")) + return deadlineExceededResult(s.logger, s.deadline, s.deadlineGraceful) } return nil } diff --git a/flagd/pkg/service/flag-sync/handler.go b/flagd/pkg/service/flag-sync/handler.go index e6d976747..774862dde 100644 --- a/flagd/pkg/service/flag-sync/handler.go +++ b/flagd/pkg/service/flag-sync/handler.go @@ -26,6 +26,7 @@ type syncHandler struct { log *logger.Logger contextValues map[string]any deadline time.Duration + deadlineGraceful bool disableSyncMetadata bool metricsRecorder telemetry.IMetricsRecorder } @@ -115,8 +116,13 @@ func (s syncHandler) syncFlags( } case <-ctx.Done(): if errors.Is(ctx.Err(), context.DeadlineExceeded) { - s.log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String())) exitReason = "deadline_exceeded" + if s.deadlineGraceful { + // end the stream with OK, providers treat a completed stream as a routine reconnect + s.log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, closing stream gracefully", s.deadline.String())) + return nil + } + s.log.Debug(fmt.Sprintf("server-side deadline of %s exceeded, exiting stream request with grpc error code 4", s.deadline.String())) return connect.NewError(connect.CodeDeadlineExceeded, errors.New("stream closed due to server-side timeout")) } s.log.Debug("context complete and exiting stream request") diff --git a/flagd/pkg/service/flag-sync/sync_service.go b/flagd/pkg/service/flag-sync/sync_service.go index 666851287..a11aa6c37 100644 --- a/flagd/pkg/service/flag-sync/sync_service.go +++ b/flagd/pkg/service/flag-sync/sync_service.go @@ -30,17 +30,18 @@ type ISyncService interface { } type SvcConfigurations struct { - Logger *logger.Logger - Port uint16 - Sources []string - Store store.IStore - ContextValues map[string]any - CertPath string - KeyPath string - SocketPath string - StreamDeadline time.Duration - DisableSyncMetadata bool - MetricsRecorder telemetry.IMetricsRecorder + Logger *logger.Logger + Port uint16 + Sources []string + Store store.IStore + ContextValues map[string]any + CertPath string + KeyPath string + SocketPath string + StreamDeadline time.Duration + StreamDeadlineGraceful bool + DisableSyncMetadata bool + MetricsRecorder telemetry.IMetricsRecorder HTTPEnabled bool ServiceName string @@ -96,6 +97,7 @@ func NewSyncService(cfg SvcConfigurations) (*Service, error) { log: l, contextValues: cfg.ContextValues, deadline: cfg.StreamDeadline, + deadlineGraceful: cfg.StreamDeadlineGraceful, disableSyncMetadata: cfg.DisableSyncMetadata, metricsRecorder: cfg.MetricsRecorder, }, flagsHandler) diff --git a/flagd/pkg/service/flag-sync/sync_service_test.go b/flagd/pkg/service/flag-sync/sync_service_test.go index c9f58e068..40663421c 100644 --- a/flagd/pkg/service/flag-sync/sync_service_test.go +++ b/flagd/pkg/service/flag-sync/sync_service_test.go @@ -2,7 +2,9 @@ package sync import ( "context" + "errors" "fmt" + "io" "log" "testing" "time" @@ -55,6 +57,7 @@ func TestSyncServiceEndToEnd(t *testing.T) { tc.socketPath, ctx, 0, + false, disableSyncMetadata, ) @@ -166,9 +169,11 @@ func TestSyncServiceDeadlineEndToEnd(t *testing.T) { testCases := []struct { title string deadline time.Duration + graceful bool }{ {title: "without deadline", deadline: 0}, {title: "with deadline", deadline: 2 * time.Second}, + {title: "with graceful deadline", deadline: 2 * time.Second, graceful: true}, } for _, tc := range testCases { @@ -184,7 +189,7 @@ func TestSyncServiceDeadlineEndToEnd(t *testing.T) { ctx, cancelFunc := context.WithCancel(context.Background()) defer cancelFunc() - _, doneChan, err := createAndStartSyncService(port, sources, flagStore, certPath, keyPath, socketPath, ctx, tc.deadline, false) + _, doneChan, err := createAndStartSyncService(port, sources, flagStore, certPath, keyPath, socketPath, ctx, tc.deadline, tc.graceful, false) if err != nil { t.Fatal("error creating sync service") } @@ -232,10 +237,17 @@ func TestSyncServiceDeadlineEndToEnd(t *testing.T) { // received data, continuing.. break case err := <-errorChan: + if errors.Is(err, io.EOF) { + if !tc.graceful { + t.Fatal("stream closed with OK status even though graceful deadline was not configured.") + } + // expected graceful close due to deadline + return + } st, _ := status.FromError(err) if st.Code() == codes.DeadlineExceeded { - if tc.deadline == 0 { - t.Fatal("ran into deadline exceeded error even though no deadline was configured.") + if tc.deadline == 0 || tc.graceful { + t.Fatal("ran into deadline exceeded error even though no (non-graceful) deadline was configured.") } // expected error due to deadline return @@ -261,18 +273,20 @@ func createAndStartSyncService( socketPath string, ctx context.Context, deadline time.Duration, + deadlineGraceful bool, disableSyncMetadata bool, ) (*Service, chan interface{}, error) { service, err := NewSyncService(SvcConfigurations{ - Logger: logger.NewLogger(nil, false), - Port: uint16(port), - Sources: sources, - Store: store, - CertPath: certPath, - KeyPath: keyPath, - SocketPath: socketPath, - StreamDeadline: deadline, - DisableSyncMetadata: disableSyncMetadata, + Logger: logger.NewLogger(nil, false), + Port: uint16(port), + Sources: sources, + Store: store, + CertPath: certPath, + KeyPath: keyPath, + SocketPath: socketPath, + StreamDeadline: deadline, + StreamDeadlineGraceful: deadlineGraceful, + DisableSyncMetadata: disableSyncMetadata, }) if err != nil { return nil, nil, err