From 0c96c6ff6c37ae289c57d4d4160a659fa9cf338b Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Thu, 10 Nov 2022 13:14:57 -0500 Subject: [PATCH] Split grace period into multiple config values (#2578) (#2603) Co-authored-by: Todd --- internal/cmd/base/servers.go | 66 ------- internal/cmd/commands/dev/dev.go | 4 - internal/cmd/commands/server/server.go | 4 - internal/cmd/config/config.go | 131 ++++++++++-- internal/cmd/config/config_load_test.go | 187 +++++++++--------- .../daemon/cluster/handlers/worker_service.go | 30 +-- .../handlers/worker_service_status_test.go | 13 +- .../cluster/handlers/worker_service_test.go | 5 +- internal/daemon/controller/controller.go | 23 ++- internal/daemon/controller/handler.go | 3 +- .../handlers/targets/target_service.go | 43 ++-- .../targets/tcp/target_service_test.go | 17 +- .../daemon/controller/rpc_registration.go | 4 +- internal/daemon/controller/testing.go | 48 +++-- internal/daemon/worker/common/common.go | 10 +- internal/daemon/worker/session/manager.go | 8 + internal/daemon/worker/session/session.go | 4 +- internal/daemon/worker/status.go | 13 +- internal/daemon/worker/status_test.go | 12 +- internal/daemon/worker/testing.go | 21 +- internal/daemon/worker/worker.go | 57 ++++-- internal/server/repository.go | 10 +- internal/session/job_session_cleanup.go | 30 ++- internal/session/job_session_cleanup_test.go | 36 +++- internal/session/jobs.go | 8 +- .../tests/cluster/session_cleanup_test.go | 48 ++--- .../tests/cluster/worker_bytesupdown_test.go | 18 +- internal/tests/cluster/worker_proxy_test.go | 18 +- internal/tests/helper/testing_helper.go | 5 +- 29 files changed, 509 insertions(+), 367 deletions(-) diff --git a/internal/cmd/base/servers.go b/internal/cmd/base/servers.go index 205bd524e..3bf765bf1 100644 --- a/internal/cmd/base/servers.go +++ b/internal/cmd/base/servers.go @@ -26,7 +26,6 @@ import ( berrors "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/observability/event" - "github.com/hashicorp/boundary/internal/server" "github.com/hashicorp/boundary/internal/types/scope" kms_plugin_assets "github.com/hashicorp/boundary/plugins/kms" plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin" @@ -48,21 +47,6 @@ import ( ) const ( - // defaultStatusGracePeriod is the default status grace period, or the period - // of time that we will go without a status report before we start - // disconnecting and marking connections as closed. This is tied to the - // server default liveness setting, a related value. See the server package - // for more details. - defaultStatusGracePeriod = server.DefaultLiveness - - // statusGracePeriodEnvVar is the environment variable that can be used to - // configure the status grace period. This setting is provided in seconds, - // and can never be lower than the default status grace period defined above. - // - // TODO: This value is temporary, it will be removed once we have a better - // story/direction on attributes and system defaults. - statusGracePeriodEnvVar = "BOUNDARY_STATUS_GRACE_PERIOD" - // File name to use for storing workerAuth requests WorkerAuthReqFile = "auth_request_token" ) @@ -144,11 +128,6 @@ type Server struct { DevDatabaseCleanupFunc func() error Database *db.DB - - // StatusGracePeriodDuration represents the period of time (as a - // duration) that the controller will wait before marking - // connections from a disconnected worker as invalid. - StatusGracePeriodDuration time.Duration } // NewServer creates a new Server. @@ -807,48 +786,3 @@ func MakeSighupCh() chan struct{} { }() return resultCh } - -// SetStatusGracePeriodDuration sets the value for -// StatusGracePeriodDuration. -// -// The grace period is the length of time we allow connections to run -// on a worker in the event of an error sending status updates. The -// period is defined the length of time since the last successful -// update. -// -// The setting is derived from one of the following, in order: -// -// - Via the supplied value if non-zero. -// - BOUNDARY_STATUS_GRACE_PERIOD, if defined, can be set to an -// integer value to define the setting. -// - If either of these is missing, the default is used. See the -// defaultStatusGracePeriod value for the default value. -// -// The minimum setting for this value is the default setting. Values -// below this will be reset to the default. -func (s *Server) SetStatusGracePeriodDuration(value time.Duration) { - const op = "base.(Server).SetStatusGracePeriodDuration" - ctx := context.TODO() - var result time.Duration - switch { - case value > 0: - result = value - case os.Getenv(statusGracePeriodEnvVar) != "": - // TODO: See the description of the constant for more details on - // this env var - v := os.Getenv(statusGracePeriodEnvVar) - n, err := strconv.Atoi(v) - if err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("could not read status grace period setting", "envvar", statusGracePeriodEnvVar, "value", v)) - break - } - - result = time.Second * time.Duration(n) - } - - if result < defaultStatusGracePeriod { - result = defaultStatusGracePeriod - } - - s.StatusGracePeriodDuration = result -} diff --git a/internal/cmd/commands/dev/dev.go b/internal/cmd/commands/dev/dev.go index fad91112d..839758dee 100644 --- a/internal/cmd/commands/dev/dev.go +++ b/internal/cmd/commands/dev/dev.go @@ -565,10 +565,6 @@ func (c *Command) Run(args []string) int { return base.CommandUserError } - // Initialize status grace period (0 denotes using env or default - // here) - c.SetStatusGracePeriodDuration(0) - base.StartMemProfiler(c.Context) if err := c.SetupEventing( diff --git a/internal/cmd/commands/server/server.go b/internal/cmd/commands/server/server.go index cd0064f28..36cd6df87 100644 --- a/internal/cmd/commands/server/server.go +++ b/internal/cmd/commands/server/server.go @@ -173,10 +173,6 @@ func (c *Command) Run(args []string) int { return base.CommandUserError } - // Initialize status grace period (0 denotes using env or default - // here) - c.SetStatusGracePeriodDuration(0) - base.StartMemProfiler(c.Context) // Note: the checks directly after this must remain where they are because diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index 1e1eff9b6..40f434d74 100644 --- a/internal/cmd/config/config.go +++ b/internal/cmd/config/config.go @@ -19,6 +19,7 @@ import ( "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/boundary/internal/util" kms_plugin_assets "github.com/hashicorp/boundary/plugins/kms" "github.com/hashicorp/boundary/sdk/wrapper" "github.com/hashicorp/go-hclog" @@ -145,26 +146,45 @@ type Controller struct { Scheduler *Scheduler `hcl:"scheduler"` // AuthTokenTimeToLive is the total valid lifetime of a token denoted by time.Duration - AuthTokenTimeToLive any `hcl:"auth_token_time_to_live"` - AuthTokenTimeToLiveDuration time.Duration + AuthTokenTimeToLive any `hcl:"auth_token_time_to_live"` + AuthTokenTimeToLiveDuration time.Duration `hcl:"-"` // AuthTokenTimeToStale is the total time a token can go unused before becoming invalid // denoted by time.Duration - AuthTokenTimeToStale any `hcl:"auth_token_time_to_stale"` - AuthTokenTimeToStaleDuration time.Duration + AuthTokenTimeToStale any `hcl:"auth_token_time_to_stale"` + AuthTokenTimeToStaleDuration time.Duration `hcl:"-"` // GracefulShutdownWait is the amount of time that we'll wait before actually // starting the Controller shutdown. This allows the health endpoint to // return a status code to indicate that the instance is shutting down. - GracefulShutdownWait any `hcl:"graceful_shutdown_wait_duration"` - GracefulShutdownWaitDuration time.Duration + GracefulShutdownWait any `hcl:"graceful_shutdown_wait_duration"` + GracefulShutdownWaitDuration time.Duration `hcl:"-"` - // StatusGracePeriod represents the period of time (as a duration) that the - // controller will wait before marking connections from a disconnected worker - // as invalid. + // WorkerStatusGracePeriod represents the period of time (as a duration) + // that the controller will wait before deciding a worker is disconnected + // and marking connections from it as canceling + // + // TODO: This isn't documented (on purpose) because the right place for this + // is central configuration so you can't drift across controllers, but we + // don't have that yet. + WorkerStatusGracePeriod interface{} `hcl:"worker_status_grace_period"` + WorkerStatusGracePeriodDuration time.Duration `hcl:"-"` + + // LivenessTimeToStale represents the period of time (as a duration) after + // which it will consider other controllers to be no longer accessible, + // based on time since their last status update in the database + // + // TODO: This isn't documented (on purpose) because the right place for this + // is central configuration so you can't drift across controllers, but we + // don't have that yet. + LivenessTimeToStale interface{} `hcl:"liveness_time_to_stale"` + LivenessTimeToStaleDuration time.Duration `hcl:"-"` + + // SchedulerRunJobInterval is the time interval between waking up the + // scheduler to run pending jobs. // // TODO: This field is currently internal. - StatusGracePeriodDuration time.Duration `hcl:"-"` + SchedulerRunJobInterval time.Duration `hcl:"-"` } func (c *Controller) InitNameIfEmpty() error { @@ -206,12 +226,22 @@ type Worker struct { Tags map[string][]string `hcl:"-"` TagsRaw any `hcl:"tags"` - // StatusGracePeriod represents the period of time (as a duration) that the - // worker will wait before disconnecting connections if it cannot make a - // status report to a controller. + // StatusCallTimeout represents the period of time (as a duration) that + // the worker will allow a status RPC call to attempt to finish before + // canceling it to try again. // - // TODO: This field is currently internal. - StatusGracePeriodDuration time.Duration `hcl:"-"` + // TODO: This is currently not documented and considered internal. + StatusCallTimeout interface{} `hcl:"status_call_timeout"` + StatusCallTimeoutDuration time.Duration `hcl:"-"` + + // SuccessfulStatusGracePeriod represents the period of time (as a duration) + // that the worker will wait before disconnecting connections if it cannot + // successfully complete a status report to a controller. This cannot be + // less than StatusCallTimeout. + // + // TODO: This is currently not documented and considered internal. + SuccessfulStatusGracePeriod interface{} `hcl:"successful_status_grace_period"` + SuccessfulStatusGracePeriodDuration time.Duration `hcl:"-"` // AuthStoragePath represents the location a worker stores its node credentials, if set AuthStoragePath string `hcl:"auth_storage_path"` @@ -498,6 +528,36 @@ func Parse(d string) (*Config, error) { } } + workerStatusGracePeriod := result.Controller.WorkerStatusGracePeriod + if util.IsNil(workerStatusGracePeriod) { + workerStatusGracePeriod = os.Getenv("BOUNDARY_CONTROLLER_WORKER_STATUS_GRACE_PERIOD") + } + if workerStatusGracePeriod != nil { + t, err := parseutil.ParseDurationSecond(workerStatusGracePeriod) + if err != nil { + return result, err + } + result.Controller.WorkerStatusGracePeriodDuration = t + } + if result.Controller.WorkerStatusGracePeriodDuration < 0 { + return nil, errors.New("Controller worker status grace period value is negative") + } + + livenessTimeToStale := result.Controller.LivenessTimeToStale + if util.IsNil(livenessTimeToStale) { + livenessTimeToStale = os.Getenv("BOUNDARY_CONTROLLER_LIVENESS_TIME_TO_STALE") + } + if livenessTimeToStale != nil { + t, err := parseutil.ParseDurationSecond(livenessTimeToStale) + if err != nil { + return result, err + } + result.Controller.LivenessTimeToStaleDuration = t + } + if result.Controller.LivenessTimeToStaleDuration < 0 { + return nil, errors.New("Controller liveness time to stale value is negative") + } + if result.Controller.Database != nil { if result.Controller.Database.MaxOpenConnectionsRaw != nil { switch t := result.Controller.Database.MaxOpenConnectionsRaw.(type) { @@ -583,6 +643,47 @@ func Parse(d string) (*Config, error) { return nil, fmt.Errorf("Error parsing worker activation token: %w", err) } + statusCallTimeoutDuration := result.Worker.StatusCallTimeout + if util.IsNil(statusCallTimeoutDuration) { + statusCallTimeoutDuration = os.Getenv("BOUNDARY_WORKER_STATUS_CALL_TIMEOUT") + } + if statusCallTimeoutDuration != nil { + t, err := parseutil.ParseDurationSecond(statusCallTimeoutDuration) + if err != nil { + return result, err + } + result.Worker.StatusCallTimeoutDuration = t + } + if result.Worker.StatusCallTimeoutDuration < 0 { + return nil, errors.New("Status call timeout value is negative") + } + + successfulStatusGracePeriod := result.Worker.SuccessfulStatusGracePeriod + if util.IsNil(successfulStatusGracePeriod) { + successfulStatusGracePeriod = os.Getenv("BOUNDARY_WORKER_SUCCESSFUL_STATUS_GRACE_PERIOD") + } + if successfulStatusGracePeriod != nil { + t, err := parseutil.ParseDurationSecond(successfulStatusGracePeriod) + if err != nil { + return result, err + } + result.Worker.SuccessfulStatusGracePeriodDuration = t + } + if result.Worker.SuccessfulStatusGracePeriodDuration < 0 { + return nil, errors.New("Successful status grace period value is negative") + } + + switch { + case result.Worker.StatusCallTimeoutDuration == 0 && result.Worker.SuccessfulStatusGracePeriodDuration == 0: + // Nothing + case result.Worker.StatusCallTimeoutDuration != 0 && result.Worker.SuccessfulStatusGracePeriodDuration != 0: + if result.Worker.StatusCallTimeoutDuration > result.Worker.SuccessfulStatusGracePeriodDuration { + return nil, fmt.Errorf("Worker setting for status call timeout duration must be less than or equal to successful status grace period duration") + } + default: + return nil, fmt.Errorf("Worker settings for status call timeout duration and successful status grace period duration must either both be set or both be empty") + } + if result.Worker.TagsRaw != nil { switch t := result.Worker.TagsRaw.(type) { // We allow `tags` to be a simple string containing a URL with schema. diff --git a/internal/cmd/config/config_load_test.go b/internal/cmd/config/config_load_test.go index 9546db6a6..1c98ac79b 100644 --- a/internal/cmd/config/config_load_test.go +++ b/internal/cmd/config/config_load_test.go @@ -347,17 +347,18 @@ func TestLoad(t *testing.T) { ClusterName: "", }, Worker: &config.Worker{ - Name: "worker0", - Description: "A worker", - PublicAddr: "", - InitialUpstreams: []string{"boundary:9201"}, - InitialUpstreamsRaw: []any{"boundary:9201"}, - ControllersRaw: nil, - Tags: nil, - TagsRaw: nil, - StatusGracePeriodDuration: 0, - AuthStoragePath: "", - ControllerGeneratedActivationToken: "", + Name: "worker0", + Description: "A worker", + PublicAddr: "", + InitialUpstreams: []string{"boundary:9201"}, + InitialUpstreamsRaw: []any{"boundary:9201"}, + ControllersRaw: nil, + Tags: nil, + TagsRaw: nil, + SuccessfulStatusGracePeriodDuration: 0, + StatusCallTimeoutDuration: 0, + AuthStoragePath: "", + ControllerGeneratedActivationToken: "", }, Controller: &config.Controller{ Name: "controller0", @@ -373,15 +374,16 @@ func TestLoad(t *testing.T) { ConnMaxIdleTimeDuration: nil, SkipSharedLockAcquisition: false, }, - PublicClusterAddr: "", - Scheduler: nil, - AuthTokenTimeToLive: nil, - AuthTokenTimeToLiveDuration: 0, - AuthTokenTimeToStale: nil, - AuthTokenTimeToStaleDuration: 0, - GracefulShutdownWait: nil, - GracefulShutdownWaitDuration: 0, - StatusGracePeriodDuration: 0, + PublicClusterAddr: "", + Scheduler: nil, + AuthTokenTimeToLive: nil, + AuthTokenTimeToLiveDuration: 0, + AuthTokenTimeToStale: nil, + AuthTokenTimeToStaleDuration: 0, + GracefulShutdownWait: nil, + GracefulShutdownWaitDuration: 0, + WorkerStatusGracePeriodDuration: 0, + LivenessTimeToStaleDuration: 0, }, DevController: false, DevUiPassthroughDir: "", @@ -751,17 +753,18 @@ func TestLoad(t *testing.T) { ClusterName: "", }, Worker: &config.Worker{ - Name: "worker0", - Description: "A worker", - PublicAddr: "", - InitialUpstreams: []string{"boundary:9201"}, - InitialUpstreamsRaw: []any{"boundary:9201"}, - ControllersRaw: nil, - Tags: nil, - TagsRaw: nil, - StatusGracePeriodDuration: 0, - AuthStoragePath: "", - ControllerGeneratedActivationToken: "", + Name: "worker0", + Description: "A worker", + PublicAddr: "", + InitialUpstreams: []string{"boundary:9201"}, + InitialUpstreamsRaw: []any{"boundary:9201"}, + ControllersRaw: nil, + Tags: nil, + TagsRaw: nil, + SuccessfulStatusGracePeriodDuration: 0, + StatusCallTimeoutDuration: 0, + AuthStoragePath: "", + ControllerGeneratedActivationToken: "", }, Controller: &config.Controller{ Name: "controller0", @@ -777,15 +780,16 @@ func TestLoad(t *testing.T) { ConnMaxIdleTimeDuration: nil, SkipSharedLockAcquisition: false, }, - PublicClusterAddr: "", - Scheduler: nil, - AuthTokenTimeToLive: nil, - AuthTokenTimeToLiveDuration: 0, - AuthTokenTimeToStale: nil, - AuthTokenTimeToStaleDuration: 0, - GracefulShutdownWait: nil, - GracefulShutdownWaitDuration: 0, - StatusGracePeriodDuration: 0, + PublicClusterAddr: "", + Scheduler: nil, + AuthTokenTimeToLive: nil, + AuthTokenTimeToLiveDuration: 0, + AuthTokenTimeToStale: nil, + AuthTokenTimeToStaleDuration: 0, + GracefulShutdownWait: nil, + GracefulShutdownWaitDuration: 0, + WorkerStatusGracePeriodDuration: 0, + LivenessTimeToStaleDuration: 0, }, DevController: false, DevUiPassthroughDir: "", @@ -1155,17 +1159,18 @@ func TestLoad(t *testing.T) { ClusterName: "", }, Worker: &config.Worker{ - Name: "worker0", - Description: "A worker", - PublicAddr: "", - InitialUpstreams: []string{"boundary:9201"}, - InitialUpstreamsRaw: []any{"boundary:9201"}, - ControllersRaw: nil, - Tags: nil, - TagsRaw: nil, - StatusGracePeriodDuration: 0, - AuthStoragePath: "", - ControllerGeneratedActivationToken: "", + Name: "worker0", + Description: "A worker", + PublicAddr: "", + InitialUpstreams: []string{"boundary:9201"}, + InitialUpstreamsRaw: []any{"boundary:9201"}, + ControllersRaw: nil, + Tags: nil, + TagsRaw: nil, + SuccessfulStatusGracePeriodDuration: 0, + StatusCallTimeoutDuration: 0, + AuthStoragePath: "", + ControllerGeneratedActivationToken: "", }, Controller: &config.Controller{ Name: "controller0", @@ -1181,15 +1186,16 @@ func TestLoad(t *testing.T) { ConnMaxIdleTimeDuration: nil, SkipSharedLockAcquisition: false, }, - PublicClusterAddr: "", - Scheduler: nil, - AuthTokenTimeToLive: nil, - AuthTokenTimeToLiveDuration: 0, - AuthTokenTimeToStale: nil, - AuthTokenTimeToStaleDuration: 0, - GracefulShutdownWait: nil, - GracefulShutdownWaitDuration: 0, - StatusGracePeriodDuration: 0, + PublicClusterAddr: "", + Scheduler: nil, + AuthTokenTimeToLive: nil, + AuthTokenTimeToLiveDuration: 0, + AuthTokenTimeToStale: nil, + AuthTokenTimeToStaleDuration: 0, + GracefulShutdownWait: nil, + GracefulShutdownWaitDuration: 0, + WorkerStatusGracePeriodDuration: 0, + LivenessTimeToStaleDuration: 0, }, DevController: false, DevUiPassthroughDir: "", @@ -1559,17 +1565,18 @@ func TestLoad(t *testing.T) { ClusterName: "", }, Worker: &config.Worker{ - Name: "worker0", - Description: "A worker", - PublicAddr: "", - InitialUpstreams: []string{"boundary:9201"}, - InitialUpstreamsRaw: []any{"boundary:9201"}, - ControllersRaw: nil, - Tags: nil, - TagsRaw: nil, - StatusGracePeriodDuration: 0, - AuthStoragePath: "", - ControllerGeneratedActivationToken: "", + Name: "worker0", + Description: "A worker", + PublicAddr: "", + InitialUpstreams: []string{"boundary:9201"}, + InitialUpstreamsRaw: []any{"boundary:9201"}, + ControllersRaw: nil, + Tags: nil, + TagsRaw: nil, + SuccessfulStatusGracePeriodDuration: 0, + StatusCallTimeoutDuration: 0, + AuthStoragePath: "", + ControllerGeneratedActivationToken: "", }, Controller: &config.Controller{ Name: "controller0", @@ -1585,15 +1592,16 @@ func TestLoad(t *testing.T) { ConnMaxIdleTimeDuration: nil, SkipSharedLockAcquisition: false, }, - PublicClusterAddr: "", - Scheduler: nil, - AuthTokenTimeToLive: nil, - AuthTokenTimeToLiveDuration: 0, - AuthTokenTimeToStale: nil, - AuthTokenTimeToStaleDuration: 0, - GracefulShutdownWait: nil, - GracefulShutdownWaitDuration: 0, - StatusGracePeriodDuration: 0, + PublicClusterAddr: "", + Scheduler: nil, + AuthTokenTimeToLive: nil, + AuthTokenTimeToLiveDuration: 0, + AuthTokenTimeToStale: nil, + AuthTokenTimeToStaleDuration: 0, + GracefulShutdownWait: nil, + GracefulShutdownWaitDuration: 0, + WorkerStatusGracePeriodDuration: 0, + LivenessTimeToStaleDuration: 0, }, DevController: false, DevUiPassthroughDir: "", @@ -1669,15 +1677,16 @@ func TestLoad(t *testing.T) { ConnMaxIdleTimeDuration: nil, SkipSharedLockAcquisition: false, }, - PublicClusterAddr: "", - Scheduler: nil, - AuthTokenTimeToLive: nil, - AuthTokenTimeToLiveDuration: 0, - AuthTokenTimeToStale: nil, - AuthTokenTimeToStaleDuration: 0, - GracefulShutdownWait: nil, - GracefulShutdownWaitDuration: 0, - StatusGracePeriodDuration: 0, + PublicClusterAddr: "", + Scheduler: nil, + AuthTokenTimeToLive: nil, + AuthTokenTimeToLiveDuration: 0, + AuthTokenTimeToStale: nil, + AuthTokenTimeToStaleDuration: 0, + GracefulShutdownWait: nil, + GracefulShutdownWaitDuration: 0, + WorkerStatusGracePeriodDuration: 0, + LivenessTimeToStaleDuration: 0, }, DevController: false, DevUiPassthroughDir: "", diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index 6aceba5cb..5ff6d9f15 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "time" "github.com/hashicorp/boundary/internal/daemon/controller/common" @@ -28,12 +29,13 @@ type workerServiceServer struct { pbs.UnsafeServerCoordinationServiceServer pbs.UnsafeSessionServiceServer - serversRepoFn common.ServersRepoFactory - workerAuthRepoFn common.WorkerAuthRepoStorageFactory - sessionRepoFn session.RepositoryFactory - connectionRepoFn common.ConnectionRepoFactory - updateTimes *sync.Map - kms *kms.Kms + serversRepoFn common.ServersRepoFactory + workerAuthRepoFn common.WorkerAuthRepoStorageFactory + sessionRepoFn session.RepositoryFactory + connectionRepoFn common.ConnectionRepoFactory + updateTimes *sync.Map + kms *kms.Kms + livenessTimeToStale *atomic.Int64 } var ( @@ -48,14 +50,16 @@ func NewWorkerServiceServer( connectionRepoFn common.ConnectionRepoFactory, updateTimes *sync.Map, kms *kms.Kms, + livenessTimeToStale *atomic.Int64, ) *workerServiceServer { return &workerServiceServer{ - serversRepoFn: serversRepoFn, - workerAuthRepoFn: workerAuthRepoFn, - sessionRepoFn: sessionRepoFn, - connectionRepoFn: connectionRepoFn, - updateTimes: updateTimes, - kms: kms, + serversRepoFn: serversRepoFn, + workerAuthRepoFn: workerAuthRepoFn, + sessionRepoFn: sessionRepoFn, + connectionRepoFn: connectionRepoFn, + updateTimes: updateTimes, + kms: kms, + livenessTimeToStale: livenessTimeToStale, } } @@ -124,7 +128,7 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err) } - controllers, err := serverRepo.ListControllers(ctx) + controllers, err := serverRepo.ListControllers(ctx, server.WithLiveness(time.Duration(ws.livenessTimeToStale.Load()))) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error getting current controllers")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error getting current controllers: %v", err) diff --git a/internal/daemon/cluster/handlers/worker_service_status_test.go b/internal/daemon/cluster/handlers/worker_service_status_test.go index 249c7bccc..7cf7c69f9 100644 --- a/internal/daemon/cluster/handlers/worker_service_status_test.go +++ b/internal/daemon/cluster/handlers/worker_service_status_test.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "sort" "sync" + "sync/atomic" "testing" "github.com/google/go-cmp/cmp" @@ -95,7 +96,7 @@ func TestStatus(t *testing.T) { require.NoError(t, err) require.NoError(t, err) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64)) require.NotNil(t, s) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId) @@ -299,7 +300,7 @@ func TestStatusSessionClosed(t *testing.T) { sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, tofu2) require.NoError(t, err) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64)) require.NotNil(t, s) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId) @@ -483,7 +484,7 @@ func TestStatusDeadConnection(t *testing.T) { sess2, _, err = repo.ActivateSession(ctx, sess2.PublicId, sess2.Version, tofu2) require.NoError(t, err) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64)) require.NotNil(t, s) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId) @@ -641,7 +642,7 @@ func TestStatusWorkerWithKeyId(t *testing.T) { require.NoError(t, err) require.NoError(t, err) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64)) require.NotNil(t, s) connection, _, err := connRepo.AuthorizeConnection(ctx, sess.PublicId, worker1.PublicId) @@ -782,7 +783,7 @@ func TestStatusAuthorizedWorkers(t *testing.T) { _ = server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w1KeyId)) _ = server.TestPkiWorker(t, conn, wrapper, server.WithTestPkiWorkerAuthorizedKeyId(&w2KeyId)) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kmsCache) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kmsCache, new(atomic.Int64)) require.NotNil(t, s) cases := []struct { @@ -923,7 +924,7 @@ func TestWorkerOperationalStatus(t *testing.T) { worker1 := server.TestKmsWorker(t, conn, wrapper) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms, new(atomic.Int64)) require.NotNil(t, s) cases := []struct { diff --git a/internal/daemon/cluster/handlers/worker_service_test.go b/internal/daemon/cluster/handlers/worker_service_test.go index a02f99af9..61a3eac2d 100644 --- a/internal/daemon/cluster/handlers/worker_service_test.go +++ b/internal/daemon/cluster/handlers/worker_service_test.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "fmt" "sync" + "sync/atomic" "testing" "github.com/google/go-cmp/cmp" @@ -125,7 +126,7 @@ func TestLookupSession(t *testing.T) { err = repo.AddSessionCredentials(ctx, sessWithCreds.ProjectId, sessWithCreds.GetPublicId(), workerCreds) require.NoError(t, err) - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms, new(atomic.Int64)) require.NotNil(t, s) cases := []struct { @@ -237,7 +238,7 @@ func TestHcpbWorkers(t *testing.T) { server.TestPkiWorker(t, conn, wrapper, opt...) } - s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kmsCache) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kmsCache, new(atomic.Int64)) require.NotNil(t, s) res, err := s.ListHcpbWorkers(ctx, &pbs.ListHcpbWorkersRequest{}) diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index 8b342a6ac..9e6eeaf6c 100644 --- a/internal/daemon/controller/controller.go +++ b/internal/daemon/controller/controller.go @@ -6,6 +6,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "github.com/hashicorp/boundary/internal/auth/oidc" "github.com/hashicorp/boundary/internal/auth/password" @@ -104,6 +105,11 @@ type Controller struct { // Used for testing and tracking worker health workerStatusUpdateTimes *sync.Map + // Timing variables. These are atomics for SIGHUP support, and are int64 + // because they are casted to time.Duration. + workerStatusGracePeriod *atomic.Int64 + livenessTimeToStale *atomic.Int64 + apiGrpcServer *grpc.Server apiGrpcServerListener grpcServerListener apiGrpcGatewayTicket string @@ -150,6 +156,8 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { enabledPlugins: conf.Server.EnabledPlugins, apiListeners: make([]*base.ServerListener, 0), pkiConnManager: cluster.NewDownstreamManager(), + workerStatusGracePeriod: new(atomic.Int64), + livenessTimeToStale: new(atomic.Int64), } if downstreamRouterFactory != nil { @@ -187,6 +195,19 @@ func New(ctx context.Context, conf *Config) (*Controller, error) { } } + switch conf.RawConfig.Controller.WorkerStatusGracePeriodDuration { + case 0: + c.workerStatusGracePeriod.Store(int64(server.DefaultLiveness)) + default: + c.workerStatusGracePeriod.Store(int64(conf.RawConfig.Controller.WorkerStatusGracePeriodDuration)) + } + switch conf.RawConfig.Controller.LivenessTimeToStaleDuration { + case 0: + c.livenessTimeToStale.Store(int64(server.DefaultLiveness)) + default: + c.livenessTimeToStale.Store(int64(conf.RawConfig.Controller.LivenessTimeToStaleDuration)) + } + clusterListeners := make([]*base.ServerListener, 0) for i := range conf.Listeners { l := conf.Listeners[i] @@ -487,7 +508,7 @@ func (c *Controller) registerJobs() error { if err := pluginhost.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.conf.HostPlugins); err != nil { return err } - if err := session.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.conf.StatusGracePeriodDuration); err != nil { + if err := session.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms, c.workerStatusGracePeriod); err != nil { return err } if err := serversjob.RegisterJobs(c.baseContext, c.scheduler, rw, rw, c.kms); err != nil { diff --git a/internal/daemon/controller/handler.go b/internal/daemon/controller/handler.go index cb0a4b87e..45f649170 100644 --- a/internal/daemon/controller/handler.go +++ b/internal/daemon/controller/handler.go @@ -177,7 +177,8 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error { c.PluginHostRepoFn, c.StaticHostRepoFn, c.VaultCredentialRepoFn, - c.StaticCredentialRepoFn) + c.StaticCredentialRepoFn, + c.workerStatusGracePeriod) if err != nil { return fmt.Errorf("failed to create target handler service: %w", err) } diff --git a/internal/daemon/controller/handlers/targets/target_service.go b/internal/daemon/controller/handlers/targets/target_service.go index d46fb985c..721fa100a 100644 --- a/internal/daemon/controller/handlers/targets/target_service.go +++ b/internal/daemon/controller/handlers/targets/target_service.go @@ -9,6 +9,8 @@ import ( "net/url" "strconv" "strings" + "sync/atomic" + "time" "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/credential" @@ -95,15 +97,16 @@ var ( type Service struct { pbs.UnsafeTargetServiceServer - repoFn target.RepositoryFactory - iamRepoFn common.IamRepoFactory - serversRepoFn common.ServersRepoFactory - sessionRepoFn session.RepositoryFactory - pluginHostRepoFn common.PluginHostRepoFactory - staticHostRepoFn common.StaticRepoFactory - vaultCredRepoFn common.VaultCredentialRepoFactory - staticCredRepoFn common.StaticCredentialRepoFactory - kmsCache *kms.Kms + repoFn target.RepositoryFactory + iamRepoFn common.IamRepoFactory + serversRepoFn common.ServersRepoFactory + sessionRepoFn session.RepositoryFactory + pluginHostRepoFn common.PluginHostRepoFactory + staticHostRepoFn common.StaticRepoFactory + vaultCredRepoFn common.VaultCredentialRepoFactory + staticCredRepoFn common.StaticCredentialRepoFactory + kmsCache *kms.Kms + workerStatusGracePeriod *atomic.Int64 } var _ pbs.TargetServiceServer = (*Service)(nil) @@ -120,6 +123,7 @@ func NewService( staticHostRepoFn common.StaticRepoFactory, vaultCredRepoFn common.VaultCredentialRepoFactory, staticCredRepoFn common.StaticCredentialRepoFactory, + workerStatusGracePeriod *atomic.Int64, ) (Service, error) { const op = "targets.NewService" if repoFn == nil { @@ -147,15 +151,16 @@ func NewService( return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing static credential repository") } return Service{ - repoFn: repoFn, - iamRepoFn: iamRepoFn, - serversRepoFn: serversRepoFn, - sessionRepoFn: sessionRepoFn, - pluginHostRepoFn: pluginHostRepoFn, - staticHostRepoFn: staticHostRepoFn, - vaultCredRepoFn: vaultCredRepoFn, - staticCredRepoFn: staticCredRepoFn, - kmsCache: kmsCache, + repoFn: repoFn, + iamRepoFn: iamRepoFn, + serversRepoFn: serversRepoFn, + sessionRepoFn: sessionRepoFn, + pluginHostRepoFn: pluginHostRepoFn, + staticHostRepoFn: staticHostRepoFn, + vaultCredRepoFn: vaultCredRepoFn, + staticCredRepoFn: staticCredRepoFn, + kmsCache: kmsCache, + workerStatusGracePeriod: workerStatusGracePeriod, }, nil } @@ -681,7 +686,7 @@ func (s Service) AuthorizeSession(ctx context.Context, req *pbs.AuthorizeSession // worker IDs below is used to contain their IDs in the same order. This is // used to fetch tags for filtering. But we avoid allocation unless we // actually need it. - selectedWorkers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true)) + selectedWorkers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithLiveness(time.Duration(s.workerStatusGracePeriod.Load()))) if err != nil { return nil, err } diff --git a/internal/daemon/controller/handlers/targets/tcp/target_service_test.go b/internal/daemon/controller/handlers/targets/tcp/target_service_test.go index 3941e77ea..7186b885f 100644 --- a/internal/daemon/controller/handlers/targets/tcp/target_service_test.go +++ b/internal/daemon/controller/handlers/targets/tcp/target_service_test.go @@ -10,6 +10,7 @@ import ( "fmt" "path" "strings" + "sync/atomic" "testing" "github.com/google/go-cmp/cmp" @@ -78,6 +79,8 @@ var testAuthorizedActions = []string{ func testService(t *testing.T, ctx context.Context, conn *db.DB, kms *kms.Kms, wrapper wrapping.Wrapper) (targets.Service, error) { rw := db.New(conn) sche := scheduler.TestScheduler(t, conn, wrapper) + statusGracePeriod := new(atomic.Int64) + statusGracePeriod.Store(int64(server.DefaultLiveness)) repoFn := func(o ...target.Option) (*target.Repository, error) { return target.NewRepository(ctx, rw, rw, kms, o...) } @@ -102,7 +105,7 @@ func testService(t *testing.T, ctx context.Context, conn *db.DB, kms *kms.Kms, w staticCredRepoFn := func() (*credstatic.Repository, error) { return credstatic.NewRepository(context.Background(), rw, rw, kms) } - return targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn) + return targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod) } func TestGet(t *testing.T) { @@ -2406,7 +2409,9 @@ func TestAuthorizeSession(t *testing.T) { }, } - s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn) + statusGracePeriod := new(atomic.Int64) + statusGracePeriod.Store(int64(server.DefaultLiveness)) + s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod) require.NoError(t, err) for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -2571,7 +2576,9 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) { _ = iam.TestUserRole(t, conn, r.GetPublicId(), at.GetIamUserId()) _ = iam.TestRoleGrant(t, conn, r.GetPublicId(), "id=*;type=*;actions=*") - s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn) + statusGracePeriod := new(atomic.Int64) + statusGracePeriod.Store(int64(server.DefaultLiveness)) + s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod) require.NoError(t, err) hc := static.TestCatalogs(t, conn, proj.GetPublicId(), 1)[0] @@ -3134,7 +3141,9 @@ func TestAuthorizeSession_Errors(t *testing.T) { } org, proj := iam.TestScopes(t, iamRepo) - s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn) + statusGracePeriod := new(atomic.Int64) + statusGracePeriod.Store(int64(server.DefaultLiveness)) + s, err := targets.NewService(ctx, kms, repoFn, iamRepoFn, serversRepoFn, sessionRepoFn, pluginHostRepoFn, staticHostRepoFn, vaultCredRepoFn, staticCredRepoFn, statusGracePeriod) require.NoError(t, err) // Authorized user gets full permissions diff --git a/internal/daemon/controller/rpc_registration.go b/internal/daemon/controller/rpc_registration.go index 0910fdbd1..a60ab09a0 100644 --- a/internal/daemon/controller/rpc_registration.go +++ b/internal/daemon/controller/rpc_registration.go @@ -34,7 +34,7 @@ func registerControllerServerCoordinationService(ctx context.Context, c *Control } workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.WorkerAuthRepoStorageFn, - c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms) + c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale) pbs.RegisterServerCoordinationServiceServer(server, workerService) return nil } @@ -52,7 +52,7 @@ func registerControllerSessionService(ctx context.Context, c *Controller, server } workerService := handlers.NewWorkerServiceServer(c.ServersRepoFn, c.WorkerAuthRepoStorageFn, - c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms) + c.SessionRepoFn, c.ConnectionRepoFn, c.workerStatusUpdateTimes, c.kms, c.livenessTimeToStale) pbs.RegisterSessionServiceServer(server, workerService) return nil } diff --git a/internal/daemon/controller/testing.go b/internal/daemon/controller/testing.go index f370d44a2..34f5de30f 100644 --- a/internal/daemon/controller/testing.go +++ b/internal/daemon/controller/testing.go @@ -423,9 +423,14 @@ type TestControllerOpts struct { // (overrides address provided in config, if any) PublicClusterAddr string - // The amount of time to wait before marking connections as closed when a + // The amount of time to wait before marking connections as canceling when a // worker has not reported in - StatusGracePeriodDuration time.Duration + WorkerStatusGracePeriodDuration time.Duration + + // The period of time after which it will consider other controllers to be + // no longer accessible, based on time since their last status update in the + // database + LivenessTimeToStaleDuration time.Duration // The amount of time between the scheduler waking up to run it's registered // jobs. @@ -612,8 +617,12 @@ func TestControllerConfig(t testing.TB, ctx context.Context, tc *TestController, } } - // Initialize status grace period - tc.b.SetStatusGracePeriodDuration(opts.StatusGracePeriodDuration) + if opts.WorkerStatusGracePeriodDuration != 0 { + opts.Config.Controller.WorkerStatusGracePeriodDuration = opts.WorkerStatusGracePeriodDuration + } + if opts.LivenessTimeToStaleDuration != 0 { + opts.Config.Controller.LivenessTimeToStaleDuration = opts.LivenessTimeToStaleDuration + } tc.name = opts.Config.Controller.Name @@ -738,20 +747,21 @@ func (tc *TestController) AddClusterControllerMember(t testing.TB, opts *TestCon opts = new(TestControllerOpts) } nextOpts := &TestControllerOpts{ - DatabaseUrl: tc.c.conf.DatabaseUrl, - DefaultPasswordAuthMethodId: tc.c.conf.DevPasswordAuthMethodId, - DefaultOidcAuthMethodId: tc.c.conf.DevOidcAuthMethodId, - RootKms: tc.c.conf.RootKms, - WorkerAuthKms: tc.c.conf.WorkerAuthKms, - RecoveryKms: tc.c.conf.RecoveryKms, - Name: opts.Name, - Logger: tc.c.conf.Logger, - DefaultLoginName: tc.b.DevLoginName, - DefaultPassword: tc.b.DevPassword, - DisableKmsKeyCreation: true, - DisableAuthMethodCreation: true, - PublicClusterAddr: opts.PublicClusterAddr, - StatusGracePeriodDuration: opts.StatusGracePeriodDuration, + DatabaseUrl: tc.c.conf.DatabaseUrl, + DefaultPasswordAuthMethodId: tc.c.conf.DevPasswordAuthMethodId, + DefaultOidcAuthMethodId: tc.c.conf.DevOidcAuthMethodId, + RootKms: tc.c.conf.RootKms, + WorkerAuthKms: tc.c.conf.WorkerAuthKms, + RecoveryKms: tc.c.conf.RecoveryKms, + Name: opts.Name, + Logger: tc.c.conf.Logger, + DefaultLoginName: tc.b.DevLoginName, + DefaultPassword: tc.b.DevPassword, + DisableKmsKeyCreation: true, + DisableAuthMethodCreation: true, + PublicClusterAddr: opts.PublicClusterAddr, + WorkerStatusGracePeriodDuration: opts.WorkerStatusGracePeriodDuration, + LivenessTimeToStaleDuration: opts.LivenessTimeToStaleDuration, } if opts.Logger != nil { nextOpts.Logger = opts.Logger @@ -775,7 +785,7 @@ func (tc *TestController) WaitForNextWorkerStatusUpdate(workerStatusName string) ctx := context.TODO() event.WriteSysEvent(ctx, op, "waiting for next status report from worker", "worker", workerStatusName) waitStatusStart := time.Now() - ctx, cancel := context.WithTimeout(tc.ctx, tc.b.StatusGracePeriodDuration) + ctx, cancel := context.WithTimeout(tc.ctx, time.Duration(tc.c.workerStatusGracePeriod.Load())) defer cancel() var err error for { diff --git a/internal/daemon/worker/common/common.go b/internal/daemon/worker/common/common.go index 8b568252c..eb091ab0f 100644 --- a/internal/daemon/worker/common/common.go +++ b/internal/daemon/worker/common/common.go @@ -1,6 +1,10 @@ package common -import "time" +import ( + "time" + + "github.com/hashicorp/boundary/internal/server" +) // In the future we could make this configurable const ( @@ -8,7 +12,7 @@ const ( // during the worker status report StatusInterval = 2 * time.Second - // StatusTimeout is the timeout duration on status calls to the controller from + // DefaultStatusTimeout is the timeout duration on status calls to the controller from // the worker - StatusTimeout = 5 * time.Second + DefaultStatusTimeout = server.DefaultLiveness / 3 ) diff --git a/internal/daemon/worker/session/manager.go b/internal/daemon/worker/session/manager.go index ceb379ec2..ee1335174 100644 --- a/internal/daemon/worker/session/manager.go +++ b/internal/daemon/worker/session/manager.go @@ -5,11 +5,19 @@ import ( "fmt" "reflect" "sync" + "sync/atomic" "github.com/hashicorp/boundary/internal/errors" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" ) +// FIXME: This is really ugly -- but not as ugly as plumbing this value into the +// interface. We should figure out something better. For now, it will at least +// keep sync with any changes to the value once it's initialized as the worker +// will set it in its new function; it can also be overridden in tests if +// desired with normal global variable caveats. +var CloseCallTimeout = new(atomic.Int64) + // Manager stores session information locally and exposes ways to operate on the // set of sessions locally in batch. // This is thread-safe. diff --git a/internal/daemon/worker/session/session.go b/internal/daemon/worker/session/session.go index 3d2180323..8ba56fa41 100644 --- a/internal/daemon/worker/session/session.go +++ b/internal/daemon/worker/session/session.go @@ -9,7 +9,6 @@ import ( "sync" "time" - "github.com/hashicorp/boundary/internal/daemon/worker/common" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/session" @@ -503,13 +502,14 @@ func closeConnections(ctx context.Context, sessClient pbs.SessionServiceClient, // bit of formalization in terms of how we handle timeouts. For now, this // just ensures consistency with the same status call in that it times out // within an adequate period of time. - closeConnCtx, closeConnCancel := context.WithTimeout(ctx, common.StatusTimeout) + closeConnCtx, closeConnCancel := context.WithTimeout(ctx, time.Duration(CloseCallTimeout.Load())) defer closeConnCancel() response, err := closeConnection(closeConnCtx, sessClient, makeCloseConnectionRequest(closeInfo)) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connections closed", "warning", "error contacting controller, connections will be closed only on worker", "session_and_connection_ids", fmt.Sprintf("%#v", closeInfo), + "timeout", time.Duration(CloseCallTimeout.Load()).String(), )) // Since we could not reach the controller, we have to make a "fake" response set. diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 97ed25470..09c1f1b62 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -86,7 +86,7 @@ func (w *Worker) LastStatusSuccess() *LastStatusInformation { func (w *Worker) WaitForNextSuccessfulStatusUpdate() error { const op = "worker.(Worker).WaitForNextSuccessfulStatusUpdate" waitStatusStart := time.Now() - ctx, cancel := context.WithTimeout(w.baseContext, w.conf.StatusGracePeriodDuration) + ctx, cancel := context.WithTimeout(w.baseContext, time.Duration(w.successfulStatusGracePeriod.Load())) defer cancel() event.WriteSysEvent(ctx, op, "waiting for next status report to controller") for { @@ -122,8 +122,9 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess var jobInfo pbs.SessionJobInfo status := s.GetStatus() sessionId := s.GetId() - connections := make([]*pbs.Connection, 0, len(s.GetLocalConnections())) - for k, v := range s.GetLocalConnections() { + localConnections := s.GetLocalConnections() + connections := make([]*pbs.Connection, 0, len(localConnections)) + for k, v := range localConnections { connections = append(connections, &pbs.Connection{ ConnectionId: k, Status: v.Status, @@ -155,7 +156,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess if w.updateTags.Load() { tags = w.tags.Load().([]*pb.TagPair) } - statusCtx, statusCancel := context.WithTimeout(cancelCtx, common.StatusTimeout) + statusCtx, statusCancel := context.WithTimeout(cancelCtx, time.Duration(w.statusCallTimeoutDuration.Load())) defer statusCancel() keyId := w.WorkerAuthCurrentKeyId.Load() @@ -229,7 +230,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess } } else if w.conf.RawConfig.HcpbClusterId != "" && len(w.conf.RawConfig.Worker.InitialUpstreams) == 0 { // This is a worker that is one hop away from managed workers, so attempt to get that list - hcpbWorkersCtx, hcpbWorkersCancel := context.WithTimeout(cancelCtx, common.StatusTimeout) + hcpbWorkersCtx, hcpbWorkersCancel := context.WithTimeout(cancelCtx, time.Duration(w.statusCallTimeoutDuration.Load())) defer hcpbWorkersCancel() workersResp, err := client.ListHcpbWorkers(hcpbWorkersCtx, &pbs.ListHcpbWorkersRequest{}) if err != nil { @@ -409,7 +410,7 @@ func (w *Worker) lastSuccessfulStatusTime() time.Time { func (w *Worker) isPastGrace() (bool, time.Time, time.Duration) { t := w.lastSuccessfulStatusTime() - u := w.conf.StatusGracePeriodDuration + u := time.Duration(w.successfulStatusGracePeriod.Load()) v := time.Since(t) return v > u, t, u } diff --git a/internal/daemon/worker/status_test.go b/internal/daemon/worker/status_test.go index e5ad90ebd..32adf6299 100644 --- a/internal/daemon/worker/status_test.go +++ b/internal/daemon/worker/status_test.go @@ -27,18 +27,18 @@ func TestWorkerWaitForNextSuccessfulStatusUpdate(t *testing.T) { // As-needed initialization of a mock worker w := &Worker{ - logger: hclog.New(nil), - lastStatusSuccess: new(atomic.Value), - baseContext: context.Background(), + logger: hclog.New(nil), + lastStatusSuccess: new(atomic.Value), + baseContext: context.Background(), + successfulStatusGracePeriod: new(atomic.Int64), conf: &Config{ - Server: &base.Server{ - StatusGracePeriodDuration: time.Second * 2, - }, + Server: &base.Server{}, }, } // This is present in New() w.lastStatusSuccess.Store((*LastStatusInformation)(nil)) + w.successfulStatusGracePeriod.Store(int64(time.Second * 2)) var wg sync.WaitGroup var err error diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index 601f5f141..7cd687037 100644 --- a/internal/daemon/worker/testing.go +++ b/internal/daemon/worker/testing.go @@ -204,7 +204,7 @@ type TestWorkerOpts struct { // The amount of time to wait before marking connections as closed when a // connection cannot be made back to the controller - StatusGracePeriodDuration time.Duration + SuccessfulStatusGracePeriodDuration time.Duration // Overrides worker's nonceFn, for cases where we want to have control // over the nonce we send to the Controller @@ -263,9 +263,6 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker { tw.b.PrometheusRegisterer = opts.PrometheusRegisterer - // Initialize status grace period - tw.b.SetStatusGracePeriodDuration(opts.StatusGracePeriodDuration) - if opts.Config.Worker == nil { opts.Config.Worker = &config.Worker{ Name: opts.Name, @@ -277,6 +274,10 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker { } tw.name = opts.Config.Worker.Name + if opts.SuccessfulStatusGracePeriodDuration != 0 { + opts.Config.Worker.SuccessfulStatusGracePeriod = opts.SuccessfulStatusGracePeriodDuration + } + serverName, err := os.Hostname() if err != nil { t.Fatal(err) @@ -357,12 +358,12 @@ func (tw *TestWorker) AddClusterWorkerMember(t testing.TB, opts *TestWorkerOpts) opts = new(TestWorkerOpts) } nextOpts := &TestWorkerOpts{ - WorkerAuthKms: tw.w.conf.WorkerAuthKms, - WorkerAuthStorageKms: tw.w.conf.WorkerAuthStorageKms, - Name: opts.Name, - InitialUpstreams: tw.UpstreamAddrs(), - Logger: tw.w.conf.Logger, - StatusGracePeriodDuration: opts.StatusGracePeriodDuration, + WorkerAuthKms: tw.w.conf.WorkerAuthKms, + WorkerAuthStorageKms: tw.w.conf.WorkerAuthStorageKms, + Name: opts.Name, + InitialUpstreams: tw.UpstreamAddrs(), + Logger: tw.w.conf.Logger, + SuccessfulStatusGracePeriodDuration: opts.SuccessfulStatusGracePeriodDuration, } if nextOpts.Name == "" { var err error diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index effa043bc..1d3174249 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -17,6 +17,7 @@ import ( "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/cmd/config" "github.com/hashicorp/boundary/internal/daemon/cluster" + "github.com/hashicorp/boundary/internal/daemon/worker/common" "github.com/hashicorp/boundary/internal/daemon/worker/internal/metric" "github.com/hashicorp/boundary/internal/daemon/worker/session" "github.com/hashicorp/boundary/internal/errors" @@ -128,6 +129,11 @@ type Worker struct { downstreamWorkers downstreamers downstreamRoutes downstreamRouter + // Timing variables. These are atomics for SIGHUP support, and are int64 + // because they are casted to time.Duration. + successfulStatusGracePeriod *atomic.Int64 + statusCallTimeoutDuration *atomic.Int64 + // Test-specific options (and possibly hidden dev-mode flags) TestOverrideX509VerifyDnsName string TestOverrideX509VerifyCertPool *x509.CertPool @@ -146,19 +152,21 @@ func New(conf *Config) (*Worker, error) { initializeReverseGrpcClientCollectors(conf.PrometheusRegisterer) w := &Worker{ - conf: conf, - logger: conf.Logger.Named("worker"), - started: ua.NewBool(false), - controllerStatusConn: new(atomic.Value), - everAuthenticated: ua.NewUint32(authenticationStatusNeverAuthenticated), - lastStatusSuccess: new(atomic.Value), - controllerMultihopConn: new(atomic.Value), - tags: new(atomic.Value), - updateTags: ua.NewBool(false), - nonceFn: base62.Random, - WorkerAuthCurrentKeyId: new(ua.String), - operationalState: new(atomic.Value), - pkiConnManager: cluster.NewDownstreamManager(), + conf: conf, + logger: conf.Logger.Named("worker"), + started: ua.NewBool(false), + controllerStatusConn: new(atomic.Value), + everAuthenticated: ua.NewUint32(authenticationStatusNeverAuthenticated), + lastStatusSuccess: new(atomic.Value), + controllerMultihopConn: new(atomic.Value), + tags: new(atomic.Value), + updateTags: ua.NewBool(false), + nonceFn: base62.Random, + WorkerAuthCurrentKeyId: new(ua.String), + operationalState: new(atomic.Value), + pkiConnManager: cluster.NewDownstreamManager(), + successfulStatusGracePeriod: new(atomic.Int64), + statusCallTimeoutDuration: new(atomic.Int64), } if downstreamRouterFactory != nil { @@ -196,6 +204,21 @@ func New(conf *Config) (*Worker, error) { } } + switch conf.RawConfig.Worker.SuccessfulStatusGracePeriodDuration { + case 0: + w.successfulStatusGracePeriod.Store(int64(server.DefaultLiveness)) + default: + w.successfulStatusGracePeriod.Store(int64(conf.RawConfig.Worker.SuccessfulStatusGracePeriodDuration)) + } + switch conf.RawConfig.Worker.StatusCallTimeoutDuration { + case 0: + w.statusCallTimeoutDuration.Store(int64(common.DefaultStatusTimeout)) + default: + w.statusCallTimeoutDuration.Store(int64(conf.RawConfig.Worker.StatusCallTimeoutDuration)) + } + // FIXME: This is really ugly, but works. + session.CloseCallTimeout = w.statusCallTimeoutDuration + var listenerCount int for i := range conf.Listeners { l := conf.Listeners[i] @@ -502,11 +525,11 @@ func (w *Worker) Shutdown() error { // Shut down all connections. w.cleanupConnections(w.baseContext, true, w.sessionManager) - // Wait for next status request to succeed. Don't wait too long; - // wrap the base context in a timeout equal to our status grace - // period. + // Wait for next status request to succeed. Don't wait too long; time it out + // at our default liveness value, which is also our default status grace + // period timeout waitStatusStart := time.Now() - nextStatusCtx, nextStatusCancel := context.WithTimeout(w.baseContext, w.conf.StatusGracePeriodDuration) + nextStatusCtx, nextStatusCancel := context.WithTimeout(w.baseContext, server.DefaultLiveness) defer nextStatusCancel() for { if err := nextStatusCtx.Err(); err != nil { diff --git a/internal/server/repository.go b/internal/server/repository.go index bd85cbda8..b22d6daec 100644 --- a/internal/server/repository.go +++ b/internal/server/repository.go @@ -10,14 +10,8 @@ import ( ) const ( - // DefaultLiveness is the setting that controls the server "liveness" time, - // or the maximum allowable time that a worker can't send a status update to - // the controller for. After this, the server is considered dead, and it will - // be taken out of the rotation for allowable workers for connections, and - // connections will possibly start to be terminated and marked as closed - // depending on the grace period setting (see - // base.Server.StatusGracePeriodDuration). This value serves as the default - // and minimum allowable setting for the grace period. + // DefaultLiveness is a default used for various timing parameters, such as + // grace period for status updates, server liveness, etc. DefaultLiveness = 15 * time.Second ) diff --git a/internal/session/job_session_cleanup.go b/internal/session/job_session_cleanup.go index 1ba42116d..4132f7c3d 100644 --- a/internal/session/job_session_cleanup.go +++ b/internal/session/job_session_cleanup.go @@ -4,21 +4,15 @@ import ( "context" "database/sql" stderrors "errors" - "fmt" + "sync/atomic" "time" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/scheduler" - "github.com/hashicorp/boundary/internal/server" ) -// deadWorkerConnCloseMinGrace is the minimum allowable setting for -// the closeConnectionsForDeadWorkers method. This is synced with -// the default server liveness setting. -const deadWorkerConnCloseMinGrace = server.DefaultLiveness - type closeConnectionsForDeadWorkersResult struct { WorkerId string LastUpdateTime time.Time @@ -37,10 +31,10 @@ type closeConnectionsForDeadWorkersResult struct { type sessionConnectionCleanupJob struct { writer db.Writer - // The amount of time to give disconnected workers before marking - // their connections as closed. This should be at larger than the liveness - // setting for the worker. - gracePeriod time.Duration + // The amount of time to give disconnected workers before marking their + // connections as closed. This should be larger than the liveness setting + // for the worker. + gracePeriod *atomic.Int64 // The total number of connections closed in the last run. totalClosed int @@ -49,15 +43,16 @@ type sessionConnectionCleanupJob struct { // newSessionConnectionCleanupJob instantiates the session cleanup job. func newSessionConnectionCleanupJob( writer db.Writer, - gracePeriod time.Duration, + gracePeriod *atomic.Int64, ) (*sessionConnectionCleanupJob, error) { const op = "session.newNewSessionConnectionCleanupJob" switch { case writer == nil: return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing db writer") - case gracePeriod < deadWorkerConnCloseMinGrace: - return nil, errors.NewDeprecated( - errors.InvalidParameter, op, fmt.Sprintf("invalid gracePeriod, must be greater than %s", deadWorkerConnCloseMinGrace)) + case gracePeriod == nil: + return nil, errors.NewDeprecated(errors.InvalidParameter, op, "missing grace period") + case gracePeriod.Load() == 0: + return nil, errors.NewDeprecated(errors.InvalidParameter, op, "grace period is zero") } return &sessionConnectionCleanupJob{ @@ -99,7 +94,8 @@ func (j *sessionConnectionCleanupJob) Run(ctx context.Context) error { j.totalClosed = 0 // Run the atomic dead worker cleanup job. - results, err := j.closeConnectionsForDeadWorkers(ctx, j.gracePeriod) + gracePeriod := time.Duration(j.gracePeriod.Load()) + results, err := j.closeConnectionsForDeadWorkers(ctx, gracePeriod) if err != nil { return errors.Wrap(ctx, err, op) } @@ -109,7 +105,7 @@ func (j *sessionConnectionCleanupJob) Run(ctx context.Context) error { event.WithInfo( "public_id", result.WorkerId, "update_time", result.LastUpdateTime, - "grace_period_seconds", j.gracePeriod.Seconds(), + "grace_period_seconds", gracePeriod.Seconds(), "number_connections_closed", result.NumberConnectionsClosed, )) diff --git a/internal/session/job_session_cleanup_test.go b/internal/session/job_session_cleanup_test.go index c65b45443..037280fa5 100644 --- a/internal/session/job_session_cleanup_test.go +++ b/internal/session/job_session_cleanup_test.go @@ -3,6 +3,7 @@ package session import ( "context" "fmt" + "sync/atomic" "testing" "time" @@ -27,7 +28,8 @@ func TestSessionConnectionCleanupJob(t *testing.T) { t.Parallel() ctx := context.Background() - const gracePeriod = 1 * time.Second + gracePeriod := new(atomic.Int64) + gracePeriod.Store(int64(time.Second)) require, assert := require.New(t), assert.New(t) conn, _ := db.TestSetup(t, "postgres") @@ -100,12 +102,12 @@ func TestSessionConnectionCleanupJob(t *testing.T) { } // Create the job. - job, err := newSessionConnectionCleanupJob(rw, deadWorkerConnCloseMinGrace) + job, err := newSessionConnectionCleanupJob(rw, gracePeriod) job.gracePeriod = gracePeriod // by-pass factory assert so we dont have to wait so long require.NoError(err) // sleep the status grace period. - time.Sleep(gracePeriod) + time.Sleep(time.Duration(gracePeriod.Load())) // Push an upsert to the first worker so that its status has been // updated. @@ -149,7 +151,10 @@ func TestSessionConnectionCleanupJobNewJobErr(t *testing.T) { const op = "session.newNewSessionConnectionCleanupJob" require := require.New(t) - job, err := newSessionConnectionCleanupJob(nil, 0) + grace := new(atomic.Int64) + grace.Store(1000000) + + job, err := newSessionConnectionCleanupJob(nil, grace) require.Equal(err, errors.E( ctx, errors.WithCode(errors.InvalidParameter), @@ -161,12 +166,21 @@ func TestSessionConnectionCleanupJobNewJobErr(t *testing.T) { conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) - job, err = newSessionConnectionCleanupJob(rw, 0) + job, err = newSessionConnectionCleanupJob(rw, nil) require.Equal(err, errors.E( ctx, errors.WithCode(errors.InvalidParameter), errors.WithOp(op), - errors.WithMsg(fmt.Sprintf("invalid gracePeriod, must be greater than %s", deadWorkerConnCloseMinGrace)), + errors.WithMsg(fmt.Sprintf("missing grace period")), + )) + require.Nil(job) + + job, err = newSessionConnectionCleanupJob(rw, new(atomic.Int64)) + require.Equal(err, errors.E( + ctx, + errors.WithCode(errors.InvalidParameter), + errors.WithOp(op), + errors.WithMsg(fmt.Sprintf("grace period is zero")), )) require.Nil(job) } @@ -188,7 +202,10 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { serversRepo, err := server.NewRepository(rw, rw, kms) require.NoError(err) - job, err := newSessionConnectionCleanupJob(rw, deadWorkerConnCloseMinGrace) + defaultLiveness := new(atomic.Int64) + defaultLiveness.Store(int64(server.DefaultLiveness)) + + job, err := newSessionConnectionCleanupJob(rw, defaultLiveness) require.NoError(err) // connection count = 6 * states(authorized, connected, closed = 3) * servers_with_open_connections(3) @@ -448,7 +465,10 @@ func TestCloseWorkerlessConnections(t *testing.T) { connRepo, err := NewConnectionRepository(ctx, rw, rw, kms) require.NoError(err) - job, err := newSessionConnectionCleanupJob(rw, time.Hour) + hourDuration := new(atomic.Int64) + hourDuration.Store(int64(time.Hour)) + + job, err := newSessionConnectionCleanupJob(rw, hourDuration) require.NoError(err) createConnection := func(workerId string) *Connection { diff --git a/internal/session/jobs.go b/internal/session/jobs.go index cb4b38246..d0fc6dc5e 100644 --- a/internal/session/jobs.go +++ b/internal/session/jobs.go @@ -3,9 +3,11 @@ package session import ( "context" "fmt" + "sync/atomic" "time" "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/scheduler" ) @@ -13,9 +15,13 @@ import ( const deleteTerminatedThreshold = time.Hour // RegisterJobs registers session related jobs with the provided scheduler. -func RegisterJobs(ctx context.Context, scheduler *scheduler.Scheduler, w db.Writer, r db.Reader, k *kms.Kms, gracePeriod time.Duration) error { +func RegisterJobs(ctx context.Context, scheduler *scheduler.Scheduler, w db.Writer, r db.Reader, k *kms.Kms, gracePeriod *atomic.Int64) error { const op = "session.RegisterJobs" + if gracePeriod == nil { + return errors.New(ctx, errors.InvalidParameter, op, "nil grace period") + } + sessionConnectionCleanupJob, err := newSessionConnectionCleanupJob(w, gracePeriod) if err != nil { return fmt.Errorf("error creating session cleanup job: %w", err) diff --git a/internal/tests/cluster/session_cleanup_test.go b/internal/tests/cluster/session_cleanup_test.go index ad6739430..447ef16e1 100644 --- a/internal/tests/cluster/session_cleanup_test.go +++ b/internal/tests/cluster/session_cleanup_test.go @@ -46,14 +46,14 @@ var timeoutBurdenCases = []timeoutBurdenType{timeoutBurdenTypeDefault, timeoutBu func controllerGracePeriod(ty timeoutBurdenType) time.Duration { if ty == timeoutBurdenTypeWorker { - return helper.DefaultGracePeriod * 10 + return helper.DefaultWorkerStatusGracePeriod * 10 } - return helper.DefaultGracePeriod + return helper.DefaultWorkerStatusGracePeriod } func workerGracePeriod(ty timeoutBurdenType) time.Duration { - return helper.DefaultGracePeriod + return helper.DefaultWorkerStatusGracePeriod } // TestSessionCleanup is the main test for session cleanup, and @@ -86,11 +86,11 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin pl, err := net.Listen("tcp", "localhost:0") require.NoError(err) c1 := controller.NewTestController(t, &controller.TestControllerOpts{ - Config: conf, - InitialResourcesSuffix: "1234567890", - Logger: logger.Named("c1"), - PublicClusterAddr: pl.Addr().String(), - StatusGracePeriodDuration: controllerGracePeriod(burdenCase), + Config: conf, + InitialResourcesSuffix: "1234567890", + Logger: logger.Named("c1"), + PublicClusterAddr: pl.Addr().String(), + WorkerStatusGracePeriodDuration: controllerGracePeriod(burdenCase), }) defer c1.Shutdown() @@ -108,10 +108,10 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin require.NotEmpty(t, proxy.ListenerAddr()) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialUpstreams: []string{proxy.ListenerAddr()}, - Logger: logger.Named("w1"), - StatusGracePeriodDuration: workerGracePeriod(burdenCase), + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: []string{proxy.ListenerAddr()}, + Logger: logger.Named("w1"), + SuccessfulStatusGracePeriodDuration: workerGracePeriod(burdenCase), }) defer w1.Shutdown() @@ -216,11 +216,11 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing pl1, err := net.Listen("tcp", "localhost:0") require.NoError(err) c1 := controller.NewTestController(t, &controller.TestControllerOpts{ - Config: conf1, - InitialResourcesSuffix: "1234567890", - Logger: logger.Named("c1"), - PublicClusterAddr: pl1.Addr().String(), - StatusGracePeriodDuration: controllerGracePeriod(burdenCase), + Config: conf1, + InitialResourcesSuffix: "1234567890", + Logger: logger.Named("c1"), + PublicClusterAddr: pl1.Addr().String(), + WorkerStatusGracePeriodDuration: controllerGracePeriod(burdenCase), }) defer c1.Shutdown() @@ -230,9 +230,9 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing pl2, err := net.Listen("tcp", "localhost:0") require.NoError(err) c2 := c1.AddClusterControllerMember(t, &controller.TestControllerOpts{ - Logger: logger.Named("c2"), - PublicClusterAddr: pl2.Addr().String(), - StatusGracePeriodDuration: controllerGracePeriod(burdenCase), + Logger: logger.Named("c2"), + PublicClusterAddr: pl2.Addr().String(), + WorkerStatusGracePeriodDuration: controllerGracePeriod(burdenCase), }) defer c2.Shutdown() expectWorkers(t, c1) @@ -268,10 +268,10 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing // ** Worker ** // ************ w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialUpstreams: []string{p1.ListenerAddr(), p2.ListenerAddr()}, - Logger: logger.Named("w1"), - StatusGracePeriodDuration: workerGracePeriod(burdenCase), + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: []string{p1.ListenerAddr(), p2.ListenerAddr()}, + Logger: logger.Named("w1"), + SuccessfulStatusGracePeriodDuration: workerGracePeriod(burdenCase), }) defer w1.Shutdown() diff --git a/internal/tests/cluster/worker_bytesupdown_test.go b/internal/tests/cluster/worker_bytesupdown_test.go index 13907877b..66f43e8b4 100644 --- a/internal/tests/cluster/worker_bytesupdown_test.go +++ b/internal/tests/cluster/worker_bytesupdown_test.go @@ -31,11 +31,11 @@ func TestWorkerBytesUpDown(t *testing.T) { pl, err := net.Listen("tcp", "localhost:0") require.NoError(err) c1 := controller.NewTestController(t, &controller.TestControllerOpts{ - Config: conf, - InitialResourcesSuffix: "1234567890", - Logger: logger.Named("c1"), - PublicClusterAddr: pl.Addr().String(), - StatusGracePeriodDuration: helper.DefaultGracePeriod, + Config: conf, + InitialResourcesSuffix: "1234567890", + Logger: logger.Named("c1"), + PublicClusterAddr: pl.Addr().String(), + WorkerStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, }) t.Cleanup(c1.Shutdown) @@ -53,10 +53,10 @@ func TestWorkerBytesUpDown(t *testing.T) { t.Cleanup(func() { _ = proxy.Close() }) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialUpstreams: []string{proxy.ListenerAddr()}, - Logger: logger.Named("w1"), - StatusGracePeriodDuration: helper.DefaultGracePeriod, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: []string{proxy.ListenerAddr()}, + Logger: logger.Named("w1"), + SuccessfulStatusGracePeriodDuration: helper.DefaultSuccessfulStatusGracePeriod, }) t.Cleanup(w1.Shutdown) diff --git a/internal/tests/cluster/worker_proxy_test.go b/internal/tests/cluster/worker_proxy_test.go index 6ded71440..062046dac 100644 --- a/internal/tests/cluster/worker_proxy_test.go +++ b/internal/tests/cluster/worker_proxy_test.go @@ -31,11 +31,11 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) { pl, err := net.Listen("tcp", "localhost:0") require.NoError(err) c1 := controller.NewTestController(t, &controller.TestControllerOpts{ - Config: conf, - InitialResourcesSuffix: "1234567890", - Logger: logger.Named("c1"), - PublicClusterAddr: pl.Addr().String(), - StatusGracePeriodDuration: helper.DefaultGracePeriod, + Config: conf, + InitialResourcesSuffix: "1234567890", + Logger: logger.Named("c1"), + PublicClusterAddr: pl.Addr().String(), + WorkerStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, }) defer c1.Shutdown() @@ -53,10 +53,10 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) { require.NotEmpty(t, proxy.ListenerAddr()) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ - WorkerAuthKms: c1.Config().WorkerAuthKms, - InitialUpstreams: []string{proxy.ListenerAddr()}, - Logger: logger.Named("w1"), - StatusGracePeriodDuration: helper.DefaultGracePeriod, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: []string{proxy.ListenerAddr()}, + Logger: logger.Named("w1"), + SuccessfulStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, }) defer w1.Shutdown() diff --git a/internal/tests/helper/testing_helper.go b/internal/tests/helper/testing_helper.go index 29b0563ea..16fdee72c 100644 --- a/internal/tests/helper/testing_helper.go +++ b/internal/tests/helper/testing_helper.go @@ -33,9 +33,10 @@ import ( ) const ( - DefaultGracePeriod = time.Second * 15 + DefaultWorkerStatusGracePeriod = time.Second * 15 + DefaultSuccessfulStatusGracePeriod = time.Second * 15 expectConnectionStateOnControllerTimeout = time.Minute * 2 - expectConnectionStateOnWorkerTimeout = DefaultGracePeriod * 3 + expectConnectionStateOnWorkerTimeout = DefaultWorkerStatusGracePeriod * 3 // This is the interval that we check states on in the worker. It // needs to be particularly granular to ensure that we allow for