From fc0ead73e6a2e4e6e27085159ab94957d7ef3f06 Mon Sep 17 00:00:00 2001 From: Irena Rindos Date: Mon, 12 Sep 2022 16:10:03 -0400 Subject: [PATCH] feat(workers): worker operational state (#2427) * feat(workers): worker operational state --- .../daemon/cluster/handlers/worker_service.go | 13 +- .../handlers/worker_service_status_test.go | 105 ++++++++ .../handlers/targets/target_service.go | 2 +- internal/daemon/worker/status.go | 17 +- internal/daemon/worker/worker.go | 20 +- .../52/01_worker_operational_state.up.sql | 76 ++++++ .../migrations/oss/postgres_52_01_test.go | 101 ++++++++ internal/gen/controller/servers/servers.pb.go | 26 +- .../proto/controller/servers/v1/servers.proto | 3 + .../storage/servers/store/v1/worker.proto | 4 + internal/server/options.go | 19 +- internal/server/options_test.go | 14 ++ internal/server/repository_worker.go | 15 +- internal/server/repository_worker_test.go | 103 +++++++- internal/server/store/worker.pb.go | 38 ++- internal/server/worker.go | 71 ++++-- internal/server/worker_test.go | 235 ++++++++++-------- 17 files changed, 696 insertions(+), 166 deletions(-) create mode 100644 internal/db/schema/migrations/oss/postgres/52/01_worker_operational_state.up.sql create mode 100644 internal/db/schema/migrations/oss/postgres_52_01_test.go diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index 3b2ae393b..9825daada 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -92,12 +92,23 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques }) } + if wStat.OperationalState == "" { + // If this is an older worker (pre 0.11), it will not have ReleaseVersion and we'll default to active. + // Otherwise, default to Uknown. + if wStat.ReleaseVersion == "" { + wStat.OperationalState = server.ActiveOperationalState.String() + } else { + wStat.OperationalState = server.UnknownOperationalState.String() + } + } + wConf := server.NewWorker(scope.Global.String(), server.WithName(wStat.GetName()), server.WithDescription(wStat.GetDescription()), server.WithAddress(wStat.GetAddress()), server.WithWorkerTags(workerTags...), - server.WithReleaseVersion(wStat.ReleaseVersion)) + server.WithReleaseVersion(wStat.ReleaseVersion), + server.WithOperationalState(wStat.OperationalState)) opts := []server.Option{server.WithUpdateTags(req.GetUpdateTags())} if wStat.GetPublicId() != "" { opts = append(opts, server.WithPublicId(wStat.GetPublicId())) diff --git a/internal/daemon/cluster/handlers/worker_service_status_test.go b/internal/daemon/cluster/handlers/worker_service_status_test.go index 39f422b0d..72c97172c 100644 --- a/internal/daemon/cluster/handlers/worker_service_status_test.go +++ b/internal/daemon/cluster/handlers/worker_service_status_test.go @@ -723,3 +723,108 @@ func TestStatusWorkerWithKeyId(t *testing.T) { }) } } + +func TestWorkerOperationalStatus(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + + serverRepo, _ := server.NewRepository(rw, rw, kms) + serverRepo.UpsertController(ctx, &store.Controller{ + PrivateId: "test_controller1", + Address: "127.0.0.1", + }) + serversRepoFn := func() (*server.Repository, error) { + return serverRepo, nil + } + sessionRepoFn := func(opt ...session.Option) (*session.Repository, error) { + return session.NewRepository(ctx, rw, rw, kms) + } + connRepoFn := func() (*session.ConnectionRepository, error) { + return session.NewConnectionRepository(ctx, rw, rw, kms) + } + + worker1 := server.TestKmsWorker(t, conn, wrapper) + + s := handlers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connRepoFn, new(sync.Map), kms) + require.NotNil(t, s) + + cases := []struct { + name string + wantErr bool + wantErrMsg string + req *pbs.StatusRequest + wantState string + }{ + { + name: "Active worker", + wantErr: false, + req: &pbs.StatusRequest{ + WorkerStatus: &pb.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetName(), + Address: worker1.GetAddress(), + OperationalState: "active", + }, + }, + wantState: "active", + }, + { + name: "Worker in shutdown", + wantErr: false, + req: &pbs.StatusRequest{ + WorkerStatus: &pb.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetName(), + Address: worker1.GetAddress(), + OperationalState: "shutdown", + }, + }, + wantState: "shutdown", + }, + { + name: "No operational state- default to unknown", + wantErr: false, + req: &pbs.StatusRequest{ + WorkerStatus: &pb.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetName(), + Address: worker1.GetAddress(), + ReleaseVersion: "Boundary v0.11.0", + }, + }, + wantState: "unknown", + }, + { + name: "Old worker (empty release version) and no operational state- default to active", + wantErr: false, + req: &pbs.StatusRequest{ + WorkerStatus: &pb.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetName(), + Address: worker1.GetAddress(), + }, + }, + wantState: "active", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + got, err := s.Status(ctx, tc.req) + if tc.wantErr { + require.Error(err) + assert.Equal(tc.wantErrMsg, err.Error()) + return + } + require.NoError(err) + require.NotNil(got) + repoWorker, err := serverRepo.LookupWorkerByName(ctx, worker1.Name) + require.NoError(err) + assert.Equal(tc.wantState, repoWorker.OperationalState) + }) + } +} diff --git a/internal/daemon/controller/handlers/targets/target_service.go b/internal/daemon/controller/handlers/targets/target_service.go index f2b603e83..bed1bc948 100644 --- a/internal/daemon/controller/handlers/targets/target_service.go +++ b/internal/daemon/controller/handlers/targets/target_service.go @@ -813,7 +813,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()}) + selectedWorkers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) if err != nil { return nil, err } diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 3598f3d78..b7fcc43f0 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -7,13 +7,13 @@ import ( "math/rand" "time" - "github.com/hashicorp/boundary/version" - "github.com/hashicorp/boundary/internal/daemon/worker/common" "github.com/hashicorp/boundary/internal/daemon/worker/session" pb "github.com/hashicorp/boundary/internal/gen/controller/servers" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/boundary/internal/server" + "github.com/hashicorp/boundary/version" "github.com/hashicorp/go-secure-stdlib/strutil" ) @@ -165,12 +165,13 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess result, err := client.Status(statusCtx, &pbs.StatusRequest{ Jobs: activeJobs, WorkerStatus: &pb.ServerWorkerStatus{ - Name: w.conf.RawConfig.Worker.Name, - Description: w.conf.RawConfig.Worker.Description, - Address: w.conf.RawConfig.Worker.PublicAddr, - Tags: tags, - KeyId: keyId, - ReleaseVersion: versionInfo.FullVersionNumber(false), + Name: w.conf.RawConfig.Worker.Name, + Description: w.conf.RawConfig.Worker.Description, + Address: w.conf.RawConfig.Worker.PublicAddr, + Tags: tags, + KeyId: keyId, + ReleaseVersion: versionInfo.FullVersionNumber(false), + OperationalState: w.operationalState.Load().(server.OperationalState).String(), }, UpdateTags: w.updateTags.Load(), }) diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 531ec53b6..4fb779dbc 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -13,28 +13,28 @@ import ( "sync/atomic" "time" - "github.com/hashicorp/boundary/internal/errors" - pb "github.com/hashicorp/boundary/internal/gen/controller/servers" - "github.com/hashicorp/nodeenrollment" - nodeenet "github.com/hashicorp/nodeenrollment/net" - "github.com/mr-tron/base58" - "google.golang.org/grpc/resolver/manual" - "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/base" "github.com/hashicorp/boundary/internal/cmd/config" "github.com/hashicorp/boundary/internal/daemon/worker/internal/metric" "github.com/hashicorp/boundary/internal/daemon/worker/session" + "github.com/hashicorp/boundary/internal/errors" + pb "github.com/hashicorp/boundary/internal/gen/controller/servers" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/boundary/internal/server" "github.com/hashicorp/go-hclog" "github.com/hashicorp/go-secure-stdlib/base62" "github.com/hashicorp/go-secure-stdlib/mlock" "github.com/hashicorp/go-secure-stdlib/strutil" + "github.com/hashicorp/nodeenrollment" + nodeenet "github.com/hashicorp/nodeenrollment/net" nodeefile "github.com/hashicorp/nodeenrollment/storage/file" "github.com/hashicorp/nodeenrollment/types" + "github.com/mr-tron/base58" ua "go.uber.org/atomic" "google.golang.org/grpc" + "google.golang.org/grpc/resolver/manual" "google.golang.org/protobuf/proto" ) @@ -90,6 +90,7 @@ type Worker struct { everAuthenticated *ua.Uint32 lastStatusSuccess *atomic.Value workerStartTime time.Time + operationalState *atomic.Value controllerMultihopConn *atomic.Value @@ -142,6 +143,7 @@ func New(conf *Config) (*Worker, error) { updateTags: ua.NewBool(false), nonceFn: base62.Random, WorkerAuthCurrentKeyId: new(ua.String), + operationalState: new(atomic.Value), } if downstreamRouterFactory != nil { @@ -382,6 +384,8 @@ func (w *Worker) Start() error { return errors.Wrap(w.baseContext, err, op, errors.WithMsg("error starting worker listeners")) } + w.operationalState.Store(server.ActiveOperationalState) + // Rather than deal with some of the potential error conditions for Add on // the waitgroup vs. Done (in case a function exits immediately), we will // always start rotation and simply exit early if we're using KMS @@ -430,6 +434,8 @@ func (w *Worker) Shutdown() error { return nil } + // Set state to shutdown + w.operationalState.Store(server.ShutdownOperationalState) // Stop listeners first to prevent new connections to the // controller. defer w.started.Store(false) diff --git a/internal/db/schema/migrations/oss/postgres/52/01_worker_operational_state.up.sql b/internal/db/schema/migrations/oss/postgres/52/01_worker_operational_state.up.sql new file mode 100644 index 000000000..67206f1a5 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/52/01_worker_operational_state.up.sql @@ -0,0 +1,76 @@ +begin; + +create table server_worker_operational_state_enm ( + state text primary key + constraint only_predefined_operational_states_allowed + check ( + state in ( + 'active', + 'shutdown', + 'unknown' + ) + ) +); +comment on table server_worker_operational_state_enm is + 'server_worker_operational_state_enm is an enumeration table for worker operational states.'; + +insert into server_worker_operational_state_enm (state) values + ('active'), + ('shutdown'), + ('unknown'); + +alter table server_worker + add column operational_state text not null default 'active' + constraint server_worker_operational_state_enm_fkey + references server_worker_operational_state_enm (state) + on delete restrict + on update cascade; + +drop view server_worker_aggregate; +-- Updates view created in 51/01_server_worker_release_version.up.sql to add the worker operational state +create view server_worker_aggregate as +with worker_config_tags(worker_id, source, tags) as ( + select + ct.worker_id, + ct.source, + -- keys and tags can be any lowercase printable character so use uppercase characters as delimitors. + string_agg(distinct concat_ws('Y', ct.key, ct.value), 'Z') as tags + from server_worker_tag ct + group by ct.worker_id, ct.source +), +connection_count (worker_id, count) as ( + select + worker_id, + count(1) as count + from session_connection + where closed_reason is null + group by worker_id +) +select + w.public_id, + w.scope_id, + w.description, + w.name, + w.address, + w.create_time, + w.update_time, + w.version, + w.last_status_time, + w.type, + w.release_version, + w.operational_state, + cc.count as active_connection_count, + -- keys and tags can be any lowercase printable character so use uppercase characters as delimitors. + wt.tags as api_tags, + ct.tags as worker_config_tags +from server_worker w + left join worker_config_tags wt on + w.public_id = wt.worker_id and wt.source = 'api' + left join worker_config_tags ct on + w.public_id = ct.worker_id and ct.source = 'configuration' + left join connection_count as cc on + w.public_id = cc.worker_id; +comment on view server_worker_aggregate is + 'server_worker_aggregate contains the worker resource with its worker provided config values and its configuration and api provided tags.'; + +commit; \ No newline at end of file diff --git a/internal/db/schema/migrations/oss/postgres_52_01_test.go b/internal/db/schema/migrations/oss/postgres_52_01_test.go new file mode 100644 index 000000000..41f8a28e3 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres_52_01_test.go @@ -0,0 +1,101 @@ +package oss_test + +import ( + "context" + "testing" + + "github.com/hashicorp/boundary/internal/db/common" + "github.com/hashicorp/boundary/internal/db/schema" + "github.com/hashicorp/boundary/testing/dbtest" + "github.com/stretchr/testify/require" +) + +const ( + insertWorkerQuery = "insert into server_worker (public_id, scope_id, type) values ($1, 'global', 'pki')" + selectWorkerQuery = "select public_id, operational_state from server_worker where public_id = $1" +) + +type testWorker struct { + OperationalState string + PublicId string +} + +func Test_WorkerOperationalStateChanges(t *testing.T) { + t.Parallel() + require := require.New(t) + const priorMigration = 51001 + const serverEnumMigration = 52001 + dialect := dbtest.Postgres + ctx := context.Background() + + c, u, _, err := dbtest.StartUsingTemplate(dialect, dbtest.WithTemplate(dbtest.Template1)) + require.NoError(err) + t.Cleanup(func() { + require.NoError(c()) + }) + d, err := common.SqlOpen(dialect, u) + require.NoError(err) + + // migration to the prior migration (before the one we want to test) + m, err := schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions( + schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": priorMigration}), + )) + require.NoError(err) + + _, err = m.ApplyMigrations(ctx) + require.NoError(err) + state, err := m.CurrentState(ctx) + require.NoError(err) + want := &schema.State{ + Initialized: true, + Editions: []schema.EditionState{ + { + Name: "oss", + BinarySchemaVersion: priorMigration, + DatabaseSchemaVersion: priorMigration, + DatabaseSchemaState: schema.Equal, + }, + }, + } + require.Equal(want, state) + + // Seed the data + execResult, err := d.ExecContext(ctx, insertWorkerQuery, "test-worker") + require.NoError(err) + rowsAffected, err := execResult.RowsAffected() + require.NoError(err) + require.Equal(int64(1), rowsAffected) + + // now we're ready for the migration we want to test. + m, err = schema.NewManager(ctx, schema.Dialect(dialect), d, schema.WithEditions( + schema.TestCreatePartialEditions(schema.Dialect(dialect), schema.PartialEditions{"oss": serverEnumMigration}), + )) + require.NoError(err) + + _, err = m.ApplyMigrations(ctx) + require.NoError(err) + state, err = m.CurrentState(ctx) + require.NoError(err) + + want = &schema.State{ + Initialized: true, + Editions: []schema.EditionState{ + { + Name: "oss", + BinarySchemaVersion: serverEnumMigration, + DatabaseSchemaVersion: serverEnumMigration, + DatabaseSchemaState: schema.Equal, + }, + }, + } + require.Equal(want, state) + + // Assert worker has been migrated and received default state of 'active' + actualWorker := new(testWorker) + row := d.QueryRowContext(ctx, selectWorkerQuery, "test-worker") + require.NoError(row.Scan( + &actualWorker.PublicId, + &actualWorker.OperationalState, + )) + require.Equal("active", actualWorker.OperationalState) +} diff --git a/internal/gen/controller/servers/servers.pb.go b/internal/gen/controller/servers/servers.pb.go index 1e0c13261..195142b66 100644 --- a/internal/gen/controller/servers/servers.pb.go +++ b/internal/gen/controller/servers/servers.pb.go @@ -96,6 +96,8 @@ type ServerWorkerStatus struct { KeyId string `protobuf:"bytes,50,opt,name=key_id,json=keyId,proto3" json:"key_id,omitempty" class:"public"` // @gotags: `class:"public"` // The version of Boundary the worker binary is running ReleaseVersion string `protobuf:"bytes,60,opt,name=release_version,proto3" json:"release_version,omitempty" class:"public"` // @gotags: `class:"public"` + // The state of the worker, to indicate if the worker is active or in shutdown. + OperationalState string `protobuf:"bytes,70,opt,name=operational_state,json=operationalState,proto3" json:"operational_state,omitempty" class:"public"` // @gotags: `class:"public"` } func (x *ServerWorkerStatus) Reset() { @@ -179,6 +181,13 @@ func (x *ServerWorkerStatus) GetReleaseVersion() string { return "" } +func (x *ServerWorkerStatus) GetOperationalState() string { + if x != nil { + return x.OperationalState + } + return "" +} + var File_controller_servers_v1_servers_proto protoreflect.FileDescriptor var file_controller_servers_v1_servers_proto_rawDesc = []byte{ @@ -189,7 +198,7 @@ var file_controller_servers_v1_servers_proto_rawDesc = []byte{ 0x54, 0x61, 0x67, 0x50, 0x61, 0x69, 0x72, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, - 0xf6, 0x01, 0x0a, 0x12, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, + 0xa3, 0x02, 0x0a, 0x12, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, @@ -204,12 +213,15 @@ var file_controller_servers_v1_servers_proto_rawDesc = []byte{ 0x64, 0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x6b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x28, 0x0a, 0x0f, 0x72, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x3c, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0f, 0x72, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, - 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x42, 0x47, 0x5a, 0x45, 0x67, 0x69, 0x74, 0x68, - 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, - 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, - 0x61, 0x6c, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, - 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x2b, 0x0a, 0x11, 0x6f, 0x70, 0x65, 0x72, + 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x61, 0x6c, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0x46, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x10, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x61, 0x6c, + 0x53, 0x74, 0x61, 0x74, 0x65, 0x42, 0x47, 0x5a, 0x45, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, + 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, + 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, + 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, + 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x62, 0x06, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/internal/proto/controller/servers/v1/servers.proto b/internal/proto/controller/servers/v1/servers.proto index eb1afd312..0ffa55cd5 100644 --- a/internal/proto/controller/servers/v1/servers.proto +++ b/internal/proto/controller/servers/v1/servers.proto @@ -32,4 +32,7 @@ message ServerWorkerStatus { // The version of Boundary the worker binary is running string release_version = 60 [json_name = "release_version"]; // @gotags: `class:"public"` + + // The state of the worker, to indicate if the worker is active or in shutdown. + string operational_state = 70; // @gotags: `class:"public"` } diff --git a/internal/proto/controller/storage/servers/store/v1/worker.proto b/internal/proto/controller/storage/servers/store/v1/worker.proto index bc582c5f1..c91154438 100644 --- a/internal/proto/controller/storage/servers/store/v1/worker.proto +++ b/internal/proto/controller/storage/servers/store/v1/worker.proto @@ -58,6 +58,10 @@ message Worker { // The version of Boundary the worker binary is running // @inject_tag: `gorm:"default:null"` string release_version = 140; + + // The state of the worker, to indicate if the worker is active or in shutdown. + // @inject_tag: `gorm:"not_null"` + string operational_state = 150; } // WorkerTag is a tag for a worker. The primary key is comprised of the diff --git a/internal/server/options.go b/internal/server/options.go index 18ca8127b..d44ef7fe8 100644 --- a/internal/server/options.go +++ b/internal/server/options.go @@ -43,11 +43,14 @@ type options struct { withStopAfter uint WithCreateControllerLedActivationToken bool withReleaseVersion string + withOperationalState string + withExcludeShutdownWorkers bool } func getDefaultOptions() options { return options{ - withNewIdFunc: newWorkerId, + withNewIdFunc: newWorkerId, + withOperationalState: ActiveOperationalState.String(), } } @@ -204,3 +207,17 @@ func WithReleaseVersion(version string) Option { o.withReleaseVersion = version } } + +// WithOperationalState provides an optional operational state. +func WithOperationalState(state string) Option { + return func(o *options) { + o.withOperationalState = state + } +} + +// WithExcludeShutdownWorkers provides an optional to filter out workers in shutdown +func WithExcludeShutdownWorkers(exclude bool) Option { + return func(o *options) { + o.withExcludeShutdownWorkers = exclude + } +} diff --git a/internal/server/options_test.go b/internal/server/options_test.go index 131daa1c4..8f7af2320 100644 --- a/internal/server/options_test.go +++ b/internal/server/options_test.go @@ -188,4 +188,18 @@ func Test_GetOpts(t *testing.T) { opts = GetOpts(WithReleaseVersion("version")) assert.Equal(t, "version", opts.withReleaseVersion) }) + t.Run("WithOperationalState", func(t *testing.T) { + opts := GetOpts(WithOperationalState("test state")) + testOpts := getDefaultOptions() + testOpts.withOperationalState = "test state" + opts.withNewIdFunc = nil + testOpts.withNewIdFunc = nil + assert.Equal(t, opts, testOpts) + }) + t.Run("WithExcludeShutdown", func(t *testing.T) { + opts := getDefaultOptions() + assert.Empty(t, opts.withExcludeShutdownWorkers) + opts = GetOpts(WithExcludeShutdownWorkers(true)) + assert.Equal(t, true, opts.withExcludeShutdownWorkers) + }) } diff --git a/internal/server/repository_worker.go b/internal/server/repository_worker.go index e5dbce65a..500937fd0 100644 --- a/internal/server/repository_worker.go +++ b/internal/server/repository_worker.go @@ -164,7 +164,7 @@ func lookupWorker(ctx context.Context, reader db.Reader, id string) (*Worker, er // then the last status update time is ignored. // If WithLimit < 0, then unlimited results are returned. If WithLimit == 0, then // default limits are used for results. -// Also supports: WithWorkerType +// Also supports: WithWorkerType, WithExcludeShutdownWorkers func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...Option) ([]*Worker, error) { const op = "server.(Repository).ListWorkers" switch { @@ -197,6 +197,11 @@ func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ... return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown worker type %v", opts.withWorkerType)) } + if opts.withExcludeShutdownWorkers { + where = append(where, "operational_state not in (?)") + whereArgs = append(whereArgs, ShutdownOperationalState.String()) + } + limit := r.defaultLimit if opts.withLimit != 0 { // non-zero signals an override of the default limit for the repo. @@ -249,6 +254,8 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both empty; one is required") case worker.GetName() != "" && opts.withKeyId != "": return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both set; no more than one is allowed") + case worker.OperationalState == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker operational state is empty") } var workerId string @@ -294,7 +301,7 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt // "description" since we want description changes for PKI-based // workers to come via API only. We can't really guard on this // in the DB so we need to be sure to not include it here. - n, err := w.Update(ctx, workerClone, []string{"address", "ReleaseVersion"}, nil) + n, err := w.Update(ctx, workerClone, []string{"address", "ReleaseVersion", "OperationalState"}, nil) if err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update status of pki worker")) } @@ -311,7 +318,7 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt workerClone.Type = KmsWorkerType.String() workerCreateConflict := &db.OnConflict{ Target: db.Columns{"public_id"}, - Action: append(db.SetColumns([]string{"address", "release_version"}), + Action: append(db.SetColumns([]string{"address", "release_version", "operational_state"}), db.SetColumnValues(map[string]interface{}{"last_status_time": "now()"})...), } var withRowsAffected int64 @@ -520,6 +527,8 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op return nil, errors.New(ctx, errors.InvalidParameter, op, "fetch node credentials request and controller led activation token option cannot both be set") } + worker.OperationalState = UnknownOperationalState.String() + var err error if worker.PublicId, err = opts.withNewIdFunc(ctx); err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to generate worker id")) diff --git a/internal/server/repository_worker_test.go b/internal/server/repository_worker_test.go index b88727144..d9de5447e 100644 --- a/internal/server/repository_worker_test.go +++ b/internal/server/repository_worker_test.go @@ -288,6 +288,27 @@ func TestUpsertWorkerStatus(t *testing.T) { assert.Equal(t, uint32(1), worker.Version) assert.Equal(t, "test-version", wStatus2.ReleaseVersion) assert.Equal(t, "new_address", worker.GetAddress()) + + // Expect this worker to be returned as it is active + workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) + require.NoError(t, err) + assert.Len(t, workers, 1) + + // update again with a shutdown state + wStatus3 := server.NewWorker(scope.Global.String(), + server.WithAddress("new_address"), server.WithName("config_name1"), + server.WithOperationalState("shutdown")) + worker, err = repo.UpsertWorkerStatus(ctx, wStatus3) + require.NoError(t, err) + assert.Greater(t, worker.GetLastStatusTime().AsTime(), worker.GetCreateTime().AsTime()) + // Version does not change for status updates + assert.Equal(t, uint32(1), worker.Version) + assert.Equal(t, "shutdown", worker.GetOperationalState()) + + // Should no longer see this worker in listing if we exclude shutdown workers + workers, err = repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) + require.NoError(t, err) + assert.Len(t, workers, 0) }) // Setup and use a pki worker @@ -424,7 +445,7 @@ func TestUpsertWorkerStatus(t *testing.T) { assert.Error(t, err) tc.errAssert(t, err) - // Still only the original worker exists. + // Still only the original PKI and KMS workers exist. workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}) require.NoError(t, err) assert.Len(t, workers, 2) @@ -442,6 +463,20 @@ func TestUpsertWorkerStatus(t *testing.T) { require.NoError(t, err) assert.Len(t, workers, 3) }) + + t.Run("send shutdown status", func(t *testing.T) { + anotherStatus := server.NewWorker(scope.Global.String(), + server.WithName("another_test_worker"), + server.WithAddress("address"), + server.WithOperationalState("shutdown")) + _, err = repo.UpsertWorkerStatus(ctx, anotherStatus) + require.NoError(t, err) + + // Filtering out shutdown workers will remove the shutdown KMS and this shutdown worker, resulting in 1 + workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) + require.NoError(t, err) + assert.Len(t, workers, 1) + }) } func TestTagUpdatingListing(t *testing.T) { @@ -608,6 +643,62 @@ func TestListWorkers(t *testing.T) { } } +func TestListWorkers_WithExcludeShutdown(t *testing.T) { + t.Parallel() + require := require.New(t) + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + serversRepo, err := server.NewRepository(rw, rw, kms) + require.NoError(err) + ctx := context.Background() + + worker1 := server.TestKmsWorker(t, conn, wrapper) + worker2 := server.TestKmsWorker(t, conn, wrapper) + worker3 := server.TestKmsWorker(t, conn, wrapper) + + result, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}) + require.NoError(err) + require.Len(result, 3) + + // Push an upsert to the first worker so that its status has been + // updated. + _, err = serversRepo.UpsertWorkerStatus(ctx, + server.NewWorker(scope.Global.String(), + server.WithName(worker1.GetName()), + server.WithAddress(worker1.GetAddress()), + server.WithOperationalState(server.ShutdownOperationalState.String()), + server.WithPublicId(worker1.GetPublicId()))) + require.NoError(err) + + result, err = serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) + require.NoError(err) + require.Len(result, 2) + + _, err = serversRepo.UpsertWorkerStatus(ctx, + server.NewWorker(scope.Global.String(), + server.WithName(worker2.GetName()), + server.WithAddress(worker2.GetAddress()), + server.WithOperationalState(server.ShutdownOperationalState.String())), + server.WithPublicId(worker2.GetPublicId())) + require.NoError(err) + result, err = serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) + require.NoError(err) + require.Len(result, 1) + + _, err = serversRepo.UpsertWorkerStatus(ctx, + server.NewWorker(scope.Global.String(), + server.WithName(worker3.GetName()), + server.WithAddress(worker3.GetAddress()), + server.WithOperationalState(server.ShutdownOperationalState.String())), + server.WithPublicId(worker3.GetPublicId())) + require.NoError(err) + result, err = serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithExcludeShutdownWorkers(true)) + require.NoError(err) + require.Len(result, 0) +} + func TestListWorkers_WithLiveness(t *testing.T) { t.Parallel() require := require.New(t) @@ -895,6 +986,16 @@ func TestRepository_CreateWorker(t *testing.T) { wantErr: true, wantErrIs: errors.NotUnique, }, + { + name: "success-with-state", + setup: func() *server.Worker { + w := server.NewWorker(scope.Global.String()) + return w + }, + reader: rw, + opt: []server.Option{server.WithOperationalState("shutdown")}, + repo: testRepo, + }, { name: "success-with-fetch-node-req", setup: func() *server.Worker { diff --git a/internal/server/store/worker.pb.go b/internal/server/store/worker.pb.go index b86edcc8e..b3b0eeeb8 100644 --- a/internal/server/store/worker.pb.go +++ b/internal/server/store/worker.pb.go @@ -61,6 +61,9 @@ type Worker struct { // The version of Boundary the worker binary is running // @inject_tag: `gorm:"default:null"` ReleaseVersion string `protobuf:"bytes,140,opt,name=release_version,json=releaseVersion,proto3" json:"release_version,omitempty" gorm:"default:null"` + // The state of the worker, to indicate if the worker is active or in shutdown. + // @inject_tag: `gorm:"not_null"` + OperationalState string `protobuf:"bytes,150,opt,name=operational_state,json=operationalState,proto3" json:"operational_state,omitempty" gorm:"not_null"` } func (x *Worker) Reset() { @@ -172,6 +175,13 @@ func (x *Worker) GetReleaseVersion() string { return "" } +func (x *Worker) GetOperationalState() string { + if x != nil { + return x.OperationalState + } + return "" +} + // WorkerTag is a tag for a worker. The primary key is comprised of the // worker_id, key, value, and source. type WorkerTag struct { @@ -267,7 +277,7 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ 0x6f, 0x74, 0x6f, 0x1a, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x8b, 0x04, 0x0a, 0x06, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, + 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xb9, 0x04, 0x0a, 0x06, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x24, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x42, 0x10, 0xc2, 0xdd, 0x29, 0x0c, @@ -300,18 +310,20 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x28, 0x0a, 0x0f, 0x72, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x8c, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x72, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, 0x56, 0x65, 0x72, 0x73, 0x69, - 0x6f, 0x6e, 0x22, 0x68, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, - 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, - 0x6b, 0x65, 0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, - 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, - 0x61, 0x6c, 0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x28, - 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x3b, 0x5a, 0x39, - 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, - 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, - 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2f, 0x73, 0x74, - 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, - 0x33, + 0x6f, 0x6e, 0x12, 0x2c, 0x0a, 0x11, 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x61, + 0x6c, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x65, 0x18, 0x96, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, + 0x6f, 0x70, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x61, 0x6c, 0x53, 0x74, 0x61, 0x74, 0x65, + 0x22, 0x68, 0x0a, 0x09, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x54, 0x61, 0x67, 0x12, 0x1b, 0x0a, + 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, + 0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, + 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, + 0x75, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x28, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x3b, 0x5a, 0x39, 0x67, 0x69, + 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, + 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, + 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, + 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/internal/server/worker.go b/internal/server/worker.go index 5c30fe5e4..ff961fa8c 100644 --- a/internal/server/worker.go +++ b/internal/server/worker.go @@ -12,12 +12,18 @@ import ( "google.golang.org/protobuf/types/known/structpb" ) -type WorkerType string +type ( + WorkerType string + OperationalState string +) const ( - UnknownWorkerType WorkerType = "unknown" - KmsWorkerType WorkerType = "kms" - PkiWorkerType WorkerType = "pki" + UnknownWorkerType WorkerType = "unknown" + KmsWorkerType WorkerType = "kms" + PkiWorkerType WorkerType = "pki" + ActiveOperationalState OperationalState = "active" + ShutdownOperationalState OperationalState = "shutdown" + UnknownOperationalState OperationalState = "unknown" ) func (t WorkerType) Valid() bool { @@ -29,13 +35,33 @@ func (t WorkerType) Valid() bool { } func (t WorkerType) String() string { - return string(t) + switch t { + case KmsWorkerType, PkiWorkerType: + return string(t) + } + return string(UnknownWorkerType) } type workerAuthWorkerId struct { WorkerId string `mapstructure:"worker_id"` } +func ValidOperationalState(s string) bool { + switch s { + case ActiveOperationalState.String(), ShutdownOperationalState.String(): + return true + } + return false +} + +func (t OperationalState) String() string { + switch t { + case ActiveOperationalState, ShutdownOperationalState: + return string(t) + } + return string(UnknownOperationalState) +} + // AttachWorkerIdToState accepts a workerId and creates a struct for use with the Nodeenrollment lib // This is intended for use in worker authorization; AuthorizeNode in the lib accepts the option WithState // so that the workerId is passed through to storage and associated with a WorkerAuth record @@ -79,11 +105,12 @@ func NewWorker(scopeId string, opt ...Option) *Worker { opts := GetOpts(opt...) return &Worker{ Worker: &store.Worker{ - ScopeId: scopeId, - Name: opts.withName, - Description: opts.withDescription, - Address: opts.withAddress, - ReleaseVersion: opts.withReleaseVersion, + ScopeId: scopeId, + Name: opts.withName, + Description: opts.withDescription, + Address: opts.withAddress, + ReleaseVersion: opts.withReleaseVersion, + OperationalState: opts.withOperationalState, }, inputTags: opts.withWorkerTags, } @@ -196,6 +223,7 @@ type workerAggregate struct { ReleaseVersion string ApiTags string ActiveConnectionCount uint32 + OperationalState string // Config Fields LastStatusTime *timestamp.Timestamp WorkerConfigTags string @@ -205,17 +233,18 @@ func (a *workerAggregate) toWorker(ctx context.Context) (*Worker, error) { const op = "server.(workerAggregate).toWorker" worker := &Worker{ Worker: &store.Worker{ - PublicId: a.PublicId, - Name: a.Name, - Description: a.Description, - Address: a.Address, - CreateTime: a.CreateTime, - UpdateTime: a.UpdateTime, - ScopeId: a.ScopeId, - Version: a.Version, - LastStatusTime: a.LastStatusTime, - Type: a.Type, - ReleaseVersion: a.ReleaseVersion, + PublicId: a.PublicId, + Name: a.Name, + Description: a.Description, + Address: a.Address, + CreateTime: a.CreateTime, + UpdateTime: a.UpdateTime, + ScopeId: a.ScopeId, + Version: a.Version, + LastStatusTime: a.LastStatusTime, + Type: a.Type, + ReleaseVersion: a.ReleaseVersion, + OperationalState: a.OperationalState, }, activeConnectionCount: a.ActiveConnectionCount, } diff --git a/internal/server/worker_test.go b/internal/server/worker_test.go index daf5a9979..a88730898 100644 --- a/internal/server/worker_test.go +++ b/internal/server/worker_test.go @@ -251,11 +251,12 @@ func TestWorker_Update(t *testing.T) { name: "kms update address", initial: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "kms update address", - Address: "kms update address", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms update address", + Address: "kms update address", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -276,11 +277,12 @@ func TestWorker_Update(t *testing.T) { name: "kms update name", initial: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "kms update name", - Address: "kms update name", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms update name", + Address: "kms update name", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -295,11 +297,12 @@ func TestWorker_Update(t *testing.T) { name: "kms clear name", initial: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "kms clear name", - Address: "kms clear name", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms clear name", + Address: "kms clear name", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -312,10 +315,11 @@ func TestWorker_Update(t *testing.T) { name: "pki update address", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki update address", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update address", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -336,10 +340,11 @@ func TestWorker_Update(t *testing.T) { name: "pki update name", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki update name", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update name", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -359,10 +364,11 @@ func TestWorker_Update(t *testing.T) { name: "pki update description", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki update description", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update description", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -383,12 +389,13 @@ func TestWorker_Update(t *testing.T) { name: "pki clear address", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki clear address", - Address: "pki clear address", - LastStatusTime: ×tamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}, + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki clear address", + Address: "pki clear address", + LastStatusTime: ×tamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}, + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -401,12 +408,13 @@ func TestWorker_Update(t *testing.T) { name: "pki clear name", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki clear name", - Address: "pki clear name", - LastStatusTime: ×tamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}, + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki clear name", + Address: "pki clear name", + LastStatusTime: ×tamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}, + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -425,10 +433,11 @@ func TestWorker_Update(t *testing.T) { name: "pki update name to capital", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki update name to capital", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update name to capital", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -443,10 +452,11 @@ func TestWorker_Update(t *testing.T) { name: "pki update name to unprintable", initial: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "worker reported update name to unprintable", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "worker reported update name to unprintable", + OperationalState: ActiveOperationalState.String(), }, }, update: Worker{ @@ -503,12 +513,13 @@ func TestWorker_Create(t *testing.T) { name: "kms base", in: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Address: "address", - Name: "kms base", - Description: "kms base", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Address: "address", + Name: "kms base", + Description: "kms base", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{lastStatusUpdated: true}, @@ -517,10 +528,11 @@ func TestWorker_Create(t *testing.T) { name: "kms with no name field", in: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Address: "address", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Address: "address", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -529,10 +541,11 @@ func TestWorker_Create(t *testing.T) { name: "kms with no address field", in: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "kms with no address field", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms with no address field", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -541,11 +554,12 @@ func TestWorker_Create(t *testing.T) { name: "kms with upper case name", in: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "With Upper Case Worker Reported Name", - Address: "address", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "With Upper Case Worker Reported Name", + Address: "address", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -554,11 +568,12 @@ func TestWorker_Create(t *testing.T) { name: "kms with unprintable worker reported name", in: Worker{ Worker: &store.Worker{ - Type: KmsWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "unprintable \u0008 worker reported name", - Address: "address", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "unprintable \u0008 worker reported name", + Address: "address", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -569,10 +584,11 @@ func TestWorker_Create(t *testing.T) { name: "pki base", in: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "pki base", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki base", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{}, @@ -581,9 +597,10 @@ func TestWorker_Create(t *testing.T) { name: "pki with no name field", in: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{}, @@ -592,10 +609,11 @@ func TestWorker_Create(t *testing.T) { name: "pki with address field", in: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Address: "address", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Address: "address", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -604,11 +622,12 @@ func TestWorker_Create(t *testing.T) { name: "pki with upper case name", in: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "PKI With Upper Case Worker Reported Name", - Address: "address", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "PKI With Upper Case Worker Reported Name", + Address: "address", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -617,11 +636,12 @@ func TestWorker_Create(t *testing.T) { name: "pki with unprintable worker reported name", in: Worker{ Worker: &store.Worker{ - Type: PkiWorkerType.String(), - ScopeId: scope.Global.String(), - PublicId: newId(), - Name: "unprintable \u0008 worker reported name", - Address: "address", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "unprintable \u0008 worker reported name", + Address: "address", + OperationalState: ActiveOperationalState.String(), }, }, want: wanted{createError: true}, @@ -671,7 +691,9 @@ func TestWorker_New(t *testing.T) { name: "missing-scope-id", args: args{}, want: &Worker{ - Worker: &store.Worker{}, + Worker: &store.Worker{ + OperationalState: ActiveOperationalState.String(), + }, }, }, { @@ -681,7 +703,8 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), + ScopeId: scope.Global.String(), + OperationalState: ActiveOperationalState.String(), }, }, }, @@ -692,7 +715,8 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: org.GetPublicId(), + ScopeId: org.GetPublicId(), + OperationalState: ActiveOperationalState.String(), }, }, }, @@ -703,7 +727,8 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: prj.GetPublicId(), + ScopeId: prj.GetPublicId(), + OperationalState: ActiveOperationalState.String(), }, }, }, @@ -717,8 +742,9 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - Name: "foo", + ScopeId: scope.Global.String(), + Name: "foo", + OperationalState: ActiveOperationalState.String(), }, }, }, @@ -732,8 +758,9 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - Description: "foo", + ScopeId: scope.Global.String(), + Description: "foo", + OperationalState: ActiveOperationalState.String(), }, }, }, @@ -747,8 +774,9 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - Address: "foo", + ScopeId: scope.Global.String(), + Address: "foo", + OperationalState: ActiveOperationalState.String(), }, }, }, @@ -765,7 +793,8 @@ func TestWorker_New(t *testing.T) { }, want: &Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), + ScopeId: scope.Global.String(), + OperationalState: ActiveOperationalState.String(), }, inputTags: []*Tag{ {