diff --git a/internal/cmd/commands/server/worker_shutdown_reload_test.go b/internal/cmd/commands/server/worker_shutdown_reload_test.go index 0afc6efb4..bf77ad451 100644 --- a/internal/cmd/commands/server/worker_shutdown_reload_test.go +++ b/internal/cmd/commands/server/worker_shutdown_reload_test.go @@ -77,7 +77,6 @@ func TestServer_ShutdownWorker(t *testing.T) { tcl := targets.NewClient(client) tgtL, err := tcl.List(ctx, scope.Global.String(), targets.WithRecursive(true)) - // tgt, err := tcl.Read(ctx, "ttcp_1234567890") require.NoError(err) require.Len(tgtL.Items, 1) tgt := tgtL.Items[0] diff --git a/internal/cmd/commands/server/worker_tags_reload_test.go b/internal/cmd/commands/server/worker_tags_reload_test.go index 5229e8cd7..5dce67d62 100644 --- a/internal/cmd/commands/server/worker_tags_reload_test.go +++ b/internal/cmd/commands/server/worker_tags_reload_test.go @@ -44,7 +44,7 @@ listener "tcp" { const tag1Config = ` worker { - name = "w_1234567890" + name = "test" description = "A default worker created in dev mode" controllers = ["%s"] tags { @@ -55,7 +55,7 @@ worker { const tag2Config = ` worker { - name = "w_1234567890" + name = "test" description = "A default worker created in dev mode" controllers = ["%s"] tags { @@ -98,18 +98,17 @@ func TestServer_ReloadWorkerTags(t *testing.T) { t.Helper() serversRepo, err := testController.Controller().ServersRepoFn() require.NoError(err) - tags, err := serversRepo.ListTagsForWorkers(testController.Context(), []string{name}) + w, err := serversRepo.LookupWorkerByWorkerReportedName(testController.Context(), name) require.NoError(err) - require.Len(tags[name], 2) - require.Equal(key, tags[name][0].Key) - require.Equal(values[0], tags[name][0].Value) - require.Equal(key, tags[name][1].Key) - require.Equal(values[1], tags[name][1].Value) + require.NotNil(w) + v, ok := w.CanonicalTags()[key] + require.True(ok) + require.ElementsMatch(values, v) } // Give time to populate up to the controller time.Sleep(10 * time.Second) - fetchWorkerTags("w_1234567890", "type", []string{"dev", "local"}) + fetchWorkerTags("test", "type", []string{"dev", "local"}) cmd.presetConfig.Store(fmt.Sprintf(workerBaseConfig+tag2Config, key, testController.ClusterAddrs()[0])) @@ -121,7 +120,7 @@ func TestServer_ReloadWorkerTags(t *testing.T) { } time.Sleep(10 * time.Second) - fetchWorkerTags("w_1234567890", "foo", []string{"bar", "baz"}) + fetchWorkerTags("test", "foo", []string{"bar", "baz"}) close(cmd.ShutdownCh) diff --git a/internal/cmd/config/config.go b/internal/cmd/config/config.go index 0c753f5f9..6d1b85fba 100644 --- a/internal/cmd/config/config.go +++ b/internal/cmd/config/config.go @@ -208,6 +208,7 @@ func initNameIfEmpty(name *string) error { if *name, err = db.NewPublicId("w"); err != nil { return err } + *name = strings.ToLower(*name) } return nil } diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index 5940f74c6..1ddda90c5 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -14,6 +14,7 @@ import ( "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/boundary/internal/servers" "github.com/hashicorp/boundary/internal/session" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets" "github.com/hashicorp/go-bexpr" "google.golang.org/grpc/codes" @@ -63,12 +64,6 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to store worker status: %v", err) } - sessRepo, err := ws.sessionRepoFn() - connectionRepo, err := ws.connectionRepoFn() - if err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error getting sessions repo")) - return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to query session status: %v", err) - } reqServer := req.GetWorker() // Convert API tags to storage tags @@ -82,16 +77,20 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques } } - wConf := servers.NewWorkerStatus(reqServer.PrivateId, - // TODO: Change the name to its own field to make this explicit. + wConf := servers.NewWorkerForStatus(scope.Global.String(), servers.WithName(reqServer.PrivateId), servers.WithAddress(reqServer.Address), servers.WithWorkerTags(workerTags...)) - controllers, _, err := serverRepo.UpsertWorkerStatus(ctx, wConf, servers.WithUpdateTags(req.GetUpdateTags())) + wrk, err := serverRepo.UpsertWorkerStatus(ctx, wConf, servers.WithUpdateTags(req.GetUpdateTags())) if err != nil { 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) + 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) + } responseControllers := []*servers.Server{} for _, c := range controllers { @@ -139,9 +138,22 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques } } - notActive, err := session.WorkerStatusReport(ctx, sessRepo, connectionRepo, req.Worker.PrivateId, stateReport) + sessRepo, err := ws.sessionRepoFn() if err != nil { - return nil, status.Errorf(codes.Internal, "Error comparing state of sessions for worker: %s: %v", req.Worker.PrivateId, err) + event.WriteError(ctx, op, err, event.WithInfoMsg("error getting sessions repo")) + return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to query session status: %v", err) + } + connectionRepo, err := ws.connectionRepoFn() + if err != nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error getting connection repo")) + return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to query session status: %v", err) + } + + notActive, err := session.WorkerStatusReport(ctx, sessRepo, connectionRepo, wrk.GetPublicId(), stateReport) + if err != nil { + return nil, status.Errorf(codes.Internal, + "Error comparing state of sessions for worker %q with public id %q: %v", + req.Worker.PrivateId, wrk.GetPublicId(), err) } for _, na := range notActive { var connChanges []*pbs.Connection @@ -198,20 +210,13 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo")) return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error acquiring server repo when looking up session: %v", err) } - workerTags, err := serversRepo.ListTagsForWorkers(ctx, []string{req.ServerId}) + w, err := serversRepo.LookupWorkerByWorkerReportedName(ctx, req.GetServerId()) if err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up tags for server", "server_id", req.ServerId)) + event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up worker by name", "name", req.ServerId)) return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up tags for server: %v", err) } // Build the map for filtering. - tagMap := make(map[string][]string) - for _, tags := range workerTags { - for _, tag := range tags { - tagMap[tag.Key] = append(tagMap[tag.Key], tag.Value) - // We don't need to reinsert after the fact because maps are - // reference types, so we don't need to re-insert into tagMap - } - } + tagMap := w.CanonicalTags() // Create the evaluator eval, err := bexpr.CreateEvaluator(sessionInfo.WorkerFilter) @@ -219,8 +224,12 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku event.WriteError(ctx, op, err, event.WithInfoMsg("error creating worker filter evaluator", "server_id", req.ServerId)) return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error creating worker filter evaluator: %v", err) } + name := w.GetName() + if name == "" { + name = w.GetWorkerReportedName() + } filterInput := map[string]interface{}{ - "name": req.ServerId, + "name": name, "tags": tagMap, } ok, err := eval.Evaluate(filterInput) @@ -347,7 +356,19 @@ func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err) } - connectionInfo, connStates, authzSummary, err := session.AuthorizeConnection(ctx, sessionRepo, connectionRepo, req.GetSessionId(), req.GetWorkerId()) + serversRepo, err := ws.serversRepoFn() + if err != nil { + return nil, status.Errorf(codes.Internal, "error getting servers repo: %v", err) + } + w, err := serversRepo.LookupWorkerByWorkerReportedName(ctx, req.GetWorkerId()) + if err != nil { + return nil, status.Errorf(codes.Internal, "error looking up worker: %v", err) + } + if w == nil { + return nil, status.Errorf(codes.NotFound, "worker not found with name %q", req.GetWorkerId()) + } + + connectionInfo, connStates, authzSummary, err := session.AuthorizeConnection(ctx, sessionRepo, connectionRepo, req.GetSessionId(), w.GetPublicId()) if err != nil { return nil, err } diff --git a/internal/daemon/cluster/handlers/worker_service_status_test.go b/internal/daemon/cluster/handlers/worker_service_status_test.go index 5d3fefb72..67112c63d 100644 --- a/internal/daemon/cluster/handlers/worker_service_status_test.go +++ b/internal/daemon/cluster/handlers/worker_service_status_test.go @@ -36,10 +36,6 @@ func TestStatus(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus( - "test_worker1", - servers.WithAddress("127.0.0.1"))) - serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil } @@ -103,7 +99,7 @@ func TestStatus(t *testing.T) { wantErr: false, req: &pbs.StatusRequest{ Worker: &servers.Server{ - PrivateId: worker1.PublicId, + PrivateId: worker1.GetWorkerReportedName(), Address: worker1.CanonicalAddress(), CreateTime: worker1.CreateTime, UpdateTime: worker1.UpdateTime, @@ -123,7 +119,7 @@ func TestStatus(t *testing.T) { wantErr: false, req: &pbs.StatusRequest{ Worker: &servers.Server{ - PrivateId: worker1.PublicId, + PrivateId: worker1.GetWorkerReportedName(), Address: worker1.CanonicalAddress(), CreateTime: worker1.CreateTime, UpdateTime: worker1.UpdateTime, @@ -203,10 +199,6 @@ func TestStatusSessionClosed(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus( - "test_worker1", - servers.WithAddress("127.0.0.1"))) - serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil } @@ -287,7 +279,7 @@ func TestStatusSessionClosed(t *testing.T) { }, req: &pbs.StatusRequest{ Worker: &servers.Server{ - PrivateId: worker1.PublicId, + PrivateId: worker1.GetWorkerReportedName(), Address: worker1.CanonicalAddress(), CreateTime: worker1.CreateTime, UpdateTime: worker1.UpdateTime, @@ -384,9 +376,8 @@ func TestStatusDeadConnection(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus( - "test_worker1", - servers.WithAddress("127.0.0.1"))) + + worker1 := servers.TestWorker(t, conn, wrapper) serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil @@ -416,8 +407,6 @@ func TestStatusDeadConnection(t *testing.T) { target.WithSessionConnectionLimit(-1), ) - worker1 := servers.TestWorker(t, conn, wrapper) - sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -456,10 +445,8 @@ func TestStatusDeadConnection(t *testing.T) { req := &pbs.StatusRequest{ Worker: &servers.Server{ - PrivateId: worker1.PublicId, - Address: worker1.CanonicalAddress(), - CreateTime: worker1.CreateTime, - UpdateTime: worker1.UpdateTime, + PrivateId: worker1.GetWorkerReportedName(), + Address: worker1.GetWorkerReportedAddress(), }, Jobs: []*pbs.JobStatus{ { diff --git a/internal/daemon/controller/handlers/targets/target_service.go b/internal/daemon/controller/handlers/targets/target_service.go index 251962b32..25dfea85a 100644 --- a/internal/daemon/controller/handlers/targets/target_service.go +++ b/internal/daemon/controller/handlers/targets/target_service.go @@ -2153,7 +2153,7 @@ func (w workerList) filtered(eval *bexpr.Evaluator) (workerList, error) { // we are separating reported status name and the resource name. name := worker.GetName() if name == "" { - name = worker.ReportedStatus.GetName() + name = worker.GetWorkerReportedName() } filterInput := map[string]interface{}{ "name": name, diff --git a/internal/daemon/controller/handlers/targets/target_service_test.go b/internal/daemon/controller/handlers/targets/target_service_test.go index 2644b4d78..675946ac4 100644 --- a/internal/daemon/controller/handlers/targets/target_service_test.go +++ b/internal/daemon/controller/handlers/targets/target_service_test.go @@ -34,20 +34,22 @@ func TestWorkerList_Addresses(t *testing.T) { func TestWorkerList_Filter(t *testing.T) { var workers []*servers.Worker for i := 0; i < 5; i++ { - w := servers.NewWorker(scope.Global.String(), - servers.WithName(fmt.Sprintf("test_worker_%d", i)), - servers.WithWorkerTags(&servers.Tag{ - Key: fmt.Sprintf("key%d", i), - Value: fmt.Sprintf("value%d", i), - })) - w.ReportedStatus = servers.NewWorkerStatus("", - servers.WithName(fmt.Sprintf("config%d", i)), - servers.WithWorkerTags(&servers.Tag{ - Key: "key", - // Note that the value is either 0 or 1 - Value: fmt.Sprintf("configvalue%d", i%2), - })) - workers = append(workers, w) + switch { + case i%2 == 0: + workers = append(workers, servers.NewWorker(scope.Global.String(), + servers.WithName(fmt.Sprintf("test_worker_%d", i)), + servers.WithWorkerTags(&servers.Tag{ + Key: fmt.Sprintf("key%d", i), + Value: fmt.Sprintf("value%d", i), + }))) + default: + workers = append(workers, servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(fmt.Sprintf("test_worker_%d", i)), + servers.WithWorkerTags(&servers.Tag{ + Key: "key", + Value: "configvalue", + }))) + } } { @@ -58,17 +60,17 @@ func TestWorkerList_Filter(t *testing.T) { assert.Len(t, got, 0) } { - ev, err := bexpr.CreateEvaluator(`"/name" matches "test_worker_[12]" and "configvalue0" in "/tags/key"`) + ev, err := bexpr.CreateEvaluator(`"/name" matches "test_worker_[12]" and "configvalue" in "/tags/key"`) require.NoError(t, err) got, err := workerList(workers).filtered(ev) require.NoError(t, err) assert.Len(t, got, 1) } { - ev, err := bexpr.CreateEvaluator(`"configvalue0" in "/tags/key"`) + ev, err := bexpr.CreateEvaluator(`"configvalue" in "/tags/key"`) require.NoError(t, err) got, err := workerList(workers).filtered(ev) require.NoError(t, err) - assert.Len(t, got, 3) + assert.Len(t, got, 2) } } diff --git a/internal/daemon/controller/testing.go b/internal/daemon/controller/testing.go index 39a3663fd..ecbf36466 100644 --- a/internal/daemon/controller/testing.go +++ b/internal/daemon/controller/testing.go @@ -705,10 +705,10 @@ func (tc *TestController) AddClusterControllerMember(t testing.TB, opts *TestCon // WaitForNextWorkerStatusUpdate waits for the next status check from a worker to // come in. If it does not come in within the default status grace // period, this function returns an error. -func (tc *TestController) WaitForNextWorkerStatusUpdate(workerId string) error { +func (tc *TestController) WaitForNextWorkerStatusUpdate(workerStatusName string) error { const op = "controller.(TestController).WaitForNextWorkerStatusUpdate" ctx := context.TODO() - event.WriteSysEvent(ctx, op, "waiting for next status report from worker", "worker", workerId) + 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) defer cancel() @@ -747,7 +747,7 @@ func (tc *TestController) WaitForNextWorkerStatusUpdate(workerId string) error { return false } - if workerStatusUpdateId == workerId { + if workerStatusUpdateId == workerStatusName { waitStatusCurrent = workerStatusUpdateTime return false } @@ -765,10 +765,10 @@ func (tc *TestController) WaitForNextWorkerStatusUpdate(workerId string) error { } if err != nil { - event.WriteError(ctx, op, err, event.WithInfoMsg("error waiting for next status report from worker", "worker", workerId)) + event.WriteError(ctx, op, err, event.WithInfoMsg("error waiting for next status report from worker", "worker", workerStatusName)) return err } - event.WriteSysEvent(ctx, op, "waiting for next status report from worker received successfully", "worker", workerId) + event.WriteSysEvent(ctx, op, "waiting for next status report from worker received successfully", "worker", workerStatusName) return nil } diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index a394e72d1..d73145cc9 100644 --- a/internal/daemon/worker/testing.go +++ b/internal/daemon/worker/testing.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net" + "strings" "testing" "time" @@ -320,6 +321,7 @@ func (tw *TestWorker) AddClusterWorkerMember(t testing.TB, opts *TestWorkerOpts) if err != nil { t.Fatal(err) } + nextOpts.Name = strings.ToLower(nextOpts.Name) event.WriteSysEvent(context.TODO(), op, "worker name generated", "name", nextOpts.Name) } return NewTestWorker(t, nextOpts) diff --git a/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql b/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql index ef48224ea..a9a70096d 100644 --- a/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql +++ b/internal/db/schema/migrations/oss/postgres/99/02_worker_controller_tables.up.sql @@ -40,8 +40,37 @@ create table server_worker ( create_time wt_timestamp, update_time wt_timestamp, version wt_version, + last_status_time timestamp with time zone, + constraint last_status_time_not_before_create_time + check (last_status_time >= create_time), + -- This is the calculated address that the worker reports it is reachable on. + -- This must be set if the worker has ever received a status update. + worker_reported_address wt_network_address + constraint worker_reported_address_must_be_set_by_status + check ( + (last_status_time is null and worker_reported_address is null) + or + (last_status_time is not null and worker_reported_address is not null) + ), + -- This is the name that the worker reports in it's status updates. + worker_reported_name wt_name + -- TODO: When we are recording the worker node's key id allow that as an + -- alternative value that must be set besides worker_reported_name if + -- last_status_time is set. + constraint worker_reported_name_must_be_set_by_status + check ( + (last_status_time is null and worker_reported_name is null) + or + (last_status_time is not null and worker_reported_name is not null) + ) + constraint worker_reported_name_must_be_lowercase + check (lower(trim(worker_reported_name)) = worker_reported_name) + constraint worker_reported_name_only_has_printable_characters + check (worker_reported_name !~ '[^[:print:]]'), constraint server_worker_scope_id_name_uq - unique(scope_id, name) + unique(scope_id, name), + constraint server_worker_scope_id_worker_reported_name_uq + unique(scope_id, worker_reported_name) ); comment on table server_worker is 'server_worker is a table where each row represents a Boundary worker.'; @@ -58,32 +87,38 @@ create trigger worker_insert_time_column before insert on server_worker create trigger worker_update_time_column before update on server_worker for each row execute procedure update_time_column(); -create table server_worker_status ( - worker_id wt_public_id primary key - constraint server_worker_fkey - references server_worker(public_id) - on delete cascade - on update cascade, - create_time wt_timestamp, - update_time wt_timestamp, - -- This is the calculated address that the worker reports it is reachable on. - address wt_network_address not null, - name wt_name -); -comment on table server_worker_status is - 'server_worker_status is a table where each row represents values that a Boundary worker reports to a controller.'; +create trigger update_version_column after update of version, description, name, address on server_worker + for each row execute procedure update_version_column(); -create trigger immutable_columns before update on server_worker_status - for each row execute procedure immutable_columns('worker_id', 'create_time'); +create function update_server_worker_update_last_status_time_column() + returns trigger +as $$ +begin + new.last_status_time = now(); + return new; +end; +$$ language plpgsql; +comment on function update_server_worker_update_last_status_time_column is + 'function used to update the last_status_time column in server_worker to now'; -create trigger default_create_time_column before insert on server_worker_status - for each row execute procedure default_create_time(); +create trigger update_server_worker_last_status_time_column before update of worker_reported_address, worker_reported_name on server_worker + for each row execute procedure update_server_worker_update_last_status_time_column(); -create trigger worker_insert_time_column before insert on server_worker_status - for each row execute procedure update_time_column(); +create function insert_server_worker_update_last_status_time_column() + returns trigger +as $$ +begin + if new.worker_reported_address is not null or new.worker_reported_name is not null then + new.last_status_time = now(); + end if; + return new; +end; +$$ language plpgsql; +comment on function insert_server_worker_update_last_status_time_column is + 'function used to update the last_status_time column in server_worker to now'; -create trigger worker_update_time_column before update on server_worker_status - for each row execute procedure update_time_column(); +create trigger insert_server_worker_last_update_time_column before insert on server_worker + for each row execute procedure insert_server_worker_update_last_status_time_column(); -- Create table worker tag create table server_worker_tag_enm ( @@ -136,16 +171,13 @@ select w.create_time, w.update_time, w.version, - ws.name as worker_status_name, - ws.address as worker_status_address, - ws.update_time as worker_status_update_time, - ws.create_time as worker_status_create_time, + w.worker_reported_name, + w.worker_reported_address, + w.last_status_time, -- 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 server_worker_status ws on - w.public_id = ws.worker_id left join worker_config_tags wt on w.public_id = wt.worker_id and wt.source = 'api' left join worker_config_tags ct on diff --git a/internal/db/timestamp/timestamp.go b/internal/db/timestamp/timestamp.go index 4910d738b..774a79076 100644 --- a/internal/db/timestamp/timestamp.go +++ b/internal/db/timestamp/timestamp.go @@ -22,5 +22,5 @@ func Now() *Timestamp { // AsTime converts x to a time.Time. func (x *Timestamp) AsTime() time.Time { - return x.Timestamp.AsTime() + return x.GetTimestamp().AsTime() } diff --git a/internal/proto/controller/storage/servers/store/v1/worker.proto b/internal/proto/controller/storage/servers/store/v1/worker.proto index 281c9497a..e6c5dff3f 100644 --- a/internal/proto/controller/storage/servers/store/v1/worker.proto +++ b/internal/proto/controller/storage/servers/store/v1/worker.proto @@ -39,6 +39,19 @@ message Worker { // version allows optimistic locking of the resource. // @inject_tag: `gorm:"default:null"` uint32 version = 80; + + // Name of the resource (optional) + // @inject_tag: `gorm:"default:null"` + string worker_reported_name = 90; + + // Address for the worker. This must be set when the worker has received + // at least 1 status update fromt he worker daemon. + // @inject_tag: `gorm:"default:null"` + string worker_reported_address = 100; + + // The last_status_time is the last time a worker daemon sent a status update. + // @inject_tag: `gorm:"default:null"` + timestamp.v1.Timestamp last_status_time = 110; } // WorkerTag is a tag for a worker. The primary key is comprised of the @@ -60,27 +73,3 @@ message WorkerTag { // @inject_tag: `gorm:"default:not_null"` string source = 40; } - -// WorkerStatus contains the fields that the worker reports to the controller -// in its status updates. -message WorkerStatus { - // worker_id is the public key that key of the worker this tag is for. - // @inject_tag: `gorm:"primary_key"` - string worker_id = 10; - - // The create_time is set by the database. - // @inject_tag: `gorm:"default:current_timestamp"` - timestamp.v1.Timestamp create_time = 20; - - // The update_time is set by the database. - // @inject_tag: `gorm:"default:current_timestamp"` - timestamp.v1.Timestamp update_time = 30; - - // Name of the resource (optional) - // @inject_tag: `gorm:"default:null"` - string name = 40; - - // Address for the worker. This must be set. - // @inject_tag: `gorm:"default:null"` - string address = 50; -} diff --git a/internal/servers/options.go b/internal/servers/options.go index cee422b4b..9fb7f6f2e 100644 --- a/internal/servers/options.go +++ b/internal/servers/options.go @@ -22,7 +22,6 @@ type options struct { withName string withDescription string withAddress string - withPublicId string withLimit int withLiveness time.Duration withUpdateTags bool @@ -62,13 +61,6 @@ func WithAddress(address string) Option { } } -// WithPublicId provides an optional public id -func WithPublicId(id string) Option { - return func(o *options) { - o.withPublicId = id - } -} - // WithLimit provides an option to provide a limit. Intentionally allowing // negative integers. If WithLimit < 0, then unlimited results are returned. // If WithLimit == 0, then default limits are used for results. diff --git a/internal/servers/options_test.go b/internal/servers/options_test.go index 08d05c3d7..3a34e4a7d 100644 --- a/internal/servers/options_test.go +++ b/internal/servers/options_test.go @@ -45,14 +45,6 @@ func Test_GetOpts(t *testing.T) { testOpts.withNewIdFunc = nil assert.Equal(t, opts, testOpts) }) - t.Run("WithPublicId", func(t *testing.T) { - opts := getOpts(WithPublicId("test")) - testOpts := getDefaultOptions() - testOpts.withPublicId = "test" - opts.withNewIdFunc = nil - testOpts.withNewIdFunc = nil - assert.Equal(t, opts, testOpts) - }) t.Run("WithLiveness", func(t *testing.T) { opts := getOpts(WithLiveness(time.Hour)) testOpts := getDefaultOptions() diff --git a/internal/servers/public_ids.go b/internal/servers/public_ids.go index 16d9a7829..25974e32c 100644 --- a/internal/servers/public_ids.go +++ b/internal/servers/public_ids.go @@ -19,3 +19,15 @@ func newWorkerId(ctx context.Context) (string, error) { } return id, nil } + +func newWorkerIdFromName(ctx context.Context, name string) (string, error) { + const op = "servers.newWorkerIdFromName" + if name == "" { + return "", errors.New(ctx, errors.InvalidParameter, op, "name is empty") + } + id, err := db.NewPublicId(WorkerPrefix, db.WithPrngValues([]string{name})) + if err != nil { + return "", errors.Wrap(ctx, err, op) + } + return id, nil +} diff --git a/internal/servers/repository_worker.go b/internal/servers/repository_worker.go index 58a254d17..f32fa5a65 100644 --- a/internal/servers/repository_worker.go +++ b/internal/servers/repository_worker.go @@ -50,6 +50,68 @@ func (r *Repository) DeleteWorker(ctx context.Context, publicId string, _ ...Opt return rowsDeleted, nil } +// LookupWorkerByWorkerReportedName returns the worker which has had a status update +// where the worker has reported this name. This is different from the name +// on the resource itself which is set over the API. In the event that no +// worker is found that matches then nil, nil will be returned. +func (r *Repository) LookupWorkerByWorkerReportedName(ctx context.Context, name string) (*Worker, error) { + const op = "servers.(Repository).LookupWorkerByWorkerReportedName" + switch { + case name == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "name is empty") + } + // we derive the id instead of doing a query to be consistant with the + // UpsertWorkerStatus flow which uses this function to upsert a worker. + id, err := newWorkerIdFromName(ctx, name) + if err != nil { + return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error while calculating the worker's id")) + } + w, err := lookupWorker(ctx, r.reader, id) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + return w, nil +} + +// LookupWorker returns the worker for the provided publicId. This returns +// nil nil in the situation where no worker can be found with that public id. +func (r *Repository) LookupWorker(ctx context.Context, publicId string, _ ...Option) (*Worker, error) { + const op = "servers.(Repository).LookupWorker" + switch { + case publicId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "publicId is empty") + } + w, err := lookupWorker(ctx, r.reader, publicId) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + return w, nil +} + +// lookupWorker returns the worker for the provided id. This returns +// nil nil in the situation where no worker can be found with that public id. +func lookupWorker(ctx context.Context, reader db.Reader, id string) (*Worker, error) { + const op = "servers.lookupWorker" + switch { + case id == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "id is empty") + } + wAgg := &workerAggregate{} + wAgg.PublicId = id + err := reader.LookupById(ctx, wAgg) + if err != nil { + if errors.IsNotFoundError(err) { + return nil, nil + } + return nil, errors.Wrap(ctx, err, op) + } + w, err := wAgg.toWorker(ctx) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + return w, nil +} + // ListWorkers is a passthrough to listWorkersWithReader that uses the repo's normal reader. func (r *Repository) ListWorkers(ctx context.Context, opt ...Option) ([]*Worker, error) { return r.listWorkersWithReader(ctx, r.reader, opt...) @@ -73,7 +135,7 @@ func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader var where string if liveness > 0 { - where = fmt.Sprintf("worker_status_update_time > now() - interval '%d seconds'", uint32(liveness.Seconds())) + where = fmt.Sprintf("last_status_time > now() - interval '%d seconds'", uint32(liveness.Seconds())) } var wAggs []*workerAggregate @@ -98,94 +160,84 @@ func (r *Repository) listWorkersWithReader(ctx context.Context, reader db.Reader return workers, nil } -// ListTagsForWorkers returns a map from the worker's id to the list of Tags -// that are for that worker. All options are ignored. -func (r *Repository) ListTagsForWorkers(ctx context.Context, workerIds []string, _ ...Option) (map[string][]*Tag, error) { - const op = "servers.ListTagsForWorkers" - var workerTags []*store.WorkerTag - if err := r.reader.SearchWhere( - ctx, - &workerTags, - "worker_id in (?)", - []interface{}{workerIds}, - db.WithLimit(-1), - ); err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("worker IDs %v", workerIds))) - } - - ret := make(map[string][]*Tag, len(workerIds)) - for _, t := range workerTags { - ret[t.WorkerId] = append(ret[t.WorkerId], &Tag{Key: t.Key, Value: t.Value}) - } - return ret, nil -} - -// UpsertWorkerStatus creates a new worker if one with the provided public id doesn't -// already exist. If it does, UpsertWorkerStatus updates the worker. The -// WithUpdateTags option is the only one used. All others are ignored. +// UpsertWorkerStatus creates a new worker if one with the provided public id +// doesn't already exist. If it does, UpsertWorkerStatus updates the worker +// status. This returns the Worker object with the updated WorkerStatus applied. +// The WithUpdateTags option is the only ones used. All others are ignored. // Workers are intentionally not oplogged. -func (r *Repository) UpsertWorkerStatus(ctx context.Context, wStatus *WorkerStatus, opt ...Option) ([]*store.Controller, int, error) { +func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt ...Option) (*Worker, error) { const op = "servers.UpsertWorkerStatus" switch { - case wStatus == nil: - return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "wStatus is nil") - case wStatus.Address == "": - return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "address is empty") - case wStatus.WorkerId == "": - return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidParameter, op, "worker id is empty") + case worker == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker is nil") + case worker.GetWorkerReportedAddress() == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker reported address is empty") + case worker.ScopeId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "scope id is empty") + case worker.PublicId != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker id is not empty") + case worker.GetWorkerReportedName() == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker reported name is empty") + case len(worker.apiTags) > 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "api tags is not empty") } + // Only retain the worker reported fields. + worker = NewWorkerForStatus(worker.GetScopeId(), + WithName(worker.GetWorkerReportedName()), + WithAddress(worker.GetWorkerReportedAddress()), + WithWorkerTags(worker.configTags...)) + opts := getOpts(opt...) - var rowsUpdated int64 - var controllers []*store.Controller - _, err := r.writer.DoTx( + var err error + worker.PublicId, err = newWorkerIdFromName(ctx, worker.GetWorkerReportedName()) + if err != nil { + return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error generating worker id")) + } + + var ret *Worker + _, err = r.writer.DoTx( ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { - worker := NewWorker(scope.Global.String(), WithPublicId(wStatus.GetWorkerId())) + worker := worker.clone() workerCreateConflict := &db.OnConflict{ Target: db.Columns{"public_id"}, - Action: db.DoNothing(true), + Action: append(db.SetColumns([]string{"worker_reported_name", "worker_reported_address"}), + db.SetColumnValues(map[string]interface{}{"last_status_time": "now()"})...), } if err := w.Create(ctx, worker, db.WithOnConflict(workerCreateConflict)); err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker")) } - var err error - onConfigConflict := &db.OnConflict{ - Target: db.Columns{"worker_id"}, - Action: append(db.SetColumns([]string{"name", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...), - } - err = w.Create(ctx, wStatus, db.WithOnConflict(onConfigConflict), db.WithReturnRowsAffected(&rowsUpdated)) - if err != nil { - return errors.Wrap(ctx, err, op+":Upsert") - } - - // Fetch current controllers to feed to the workers - controllers, err = r.listControllersWithReader(ctx, reader) - if err != nil { - return errors.Wrap(ctx, err, op+":ListController") - } - // If we've been told to update tags, we need to clean out old // ones and add new ones. Within the current transaction, simply // delete all tags for the given worker, then add the new ones // we've been sent. if opts.withUpdateTags { - setWorkerTags(ctx, w, wStatus.GetWorkerId(), ConfigurationTagSource, wStatus.Tags) + setWorkerTags(ctx, w, worker.GetPublicId(), ConfigurationTagSource, worker.configTags) + } + + wAgg := &workerAggregate{PublicId: worker.GetPublicId()} + if err := reader.LookupById(ctx, wAgg); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("error looking up worker aggregate")) + } + ret, err = wAgg.toWorker(ctx) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("error converting worker aggregate to worker")) } return nil }, ) if err != nil { - return nil, db.NoRowsAffected, err + return nil, err } - return controllers, int(rowsUpdated), nil + return ret, nil } // setWorkerTags removes all existing tags from the same source and worker id @@ -249,6 +301,12 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op return nil, errors.New(ctx, errors.InvalidParameter, op, "missing scope id") case worker.ScopeId != scope.Global.String(): return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("scope id must be %q", scope.Global.String())) + case worker.WorkerReportedAddress != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker reported address is not empty") + case worker.WorkerReportedName != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker reported name is not empty") + case worker.LastStatusTime != nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "last status time is not nil") } opts := getOpts(opt...) diff --git a/internal/servers/repository_worker_test.go b/internal/servers/repository_worker_test.go index 0d55c72e3..04e8f4f3e 100644 --- a/internal/servers/repository_worker_test.go +++ b/internal/servers/repository_worker_test.go @@ -3,10 +3,13 @@ package servers_test import ( "context" "fmt" + "strings" "testing" "time" + "github.com/google/go-cmp/cmp" "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/db/timestamp" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/iam" "github.com/hashicorp/boundary/internal/kms" @@ -16,6 +19,7 @@ import ( "github.com/hashicorp/go-dbw" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/protobuf/testing/protocmp" ) func TestDeleteWorker(t *testing.T) { @@ -94,6 +98,72 @@ func TestDeleteWorker(t *testing.T) { } } +func TestLookupWorkerByWorkerReportedName(t *testing.T) { + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + repo, err := servers.NewRepository(rw, rw, kms) + require.NoError(t, err) + ctx := context.Background() + + w := servers.TestWorker(t, conn, wrapper) + t.Run("success", func(t *testing.T) { + got, err := repo.LookupWorkerByWorkerReportedName(ctx, w.GetWorkerReportedName()) + require.NoError(t, err) + assert.Empty(t, cmp.Diff(w.Worker, got.Worker, protocmp.Transform())) + }) + t.Run("not found", func(t *testing.T) { + got, err := repo.LookupWorkerByWorkerReportedName(ctx, "unknown_name") + require.NoError(t, err) + assert.Nil(t, got) + }) + t.Run("db error", func(t *testing.T) { + conn, mock := db.TestSetupWithMock(t) + rw := db.New(conn) + mock.ExpectQuery(`SELECT`).WillReturnError(errors.New(context.Background(), errors.Internal, "test", "lookup-error")) + r, err := servers.NewRepository(rw, rw, kms) + require.NoError(t, err) + got, err := r.LookupWorkerByWorkerReportedName(ctx, w.GetWorkerReportedName()) + assert.NoError(t, mock.ExpectationsWereMet()) + assert.Truef(t, errors.Match(errors.T(errors.Op("servers.(Repository).LookupWorkerByWorkerReportedName")), err), "got error %v", err) + assert.Nil(t, got) + }) +} + +func TestLookupWorker(t *testing.T) { + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + repo, err := servers.NewRepository(rw, rw, kms) + require.NoError(t, err) + ctx := context.Background() + + w := servers.TestWorker(t, conn, wrapper) + t.Run("success", func(t *testing.T) { + got, err := repo.LookupWorker(ctx, w.GetPublicId()) + require.NoError(t, err) + assert.Empty(t, cmp.Diff(w.Worker, got.Worker, protocmp.Transform())) + }) + t.Run("not found", func(t *testing.T) { + got, err := repo.LookupWorker(ctx, "w_unknownid") + require.NoError(t, err) + assert.Nil(t, got) + }) + t.Run("db error", func(t *testing.T) { + conn, mock := db.TestSetupWithMock(t) + rw := db.New(conn) + mock.ExpectQuery(`SELECT`).WillReturnError(errors.New(context.Background(), errors.Internal, "test", "lookup-error")) + r, err := servers.NewRepository(rw, rw, kms) + require.NoError(t, err) + got, err := r.LookupWorker(ctx, w.GetPublicId()) + assert.NoError(t, mock.ExpectationsWereMet()) + assert.Truef(t, errors.Match(errors.T(errors.Op("servers.(Repository).LookupWorker")), err), "got error %v", err) + assert.Nil(t, got) + }) +} + func TestUpsertWorkerStatus(t *testing.T) { conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) @@ -103,47 +173,15 @@ func TestUpsertWorkerStatus(t *testing.T) { require.NoError(t, err) ctx := context.Background() - wStatus1 := servers.NewWorkerStatus("test_worker_1", - servers.WithAddress("address"), - servers.WithName("config_name1")) - _, _, err = repo.UpsertWorkerStatus(ctx, wStatus1) + wStatus1 := servers.NewWorkerForStatus(scope.Global.String(), + servers.WithAddress("address"), servers.WithName("config_name1")) + worker, err := repo.UpsertWorkerStatus(ctx, wStatus1) require.NoError(t, err) { - workers, err := repo.ListWorkers(ctx) - require.NoError(t, err) - assert.Len(t, workers, 1) - worker := workers[0] - assert.Equal(t, wStatus1.GetWorkerId(), worker.GetPublicId()) - assert.Equal(t, wStatus1.Address, worker.CanonicalAddress()) + assert.True(t, strings.HasPrefix(worker.GetPublicId(), "w_")) + assert.Equal(t, wStatus1.GetWorkerReportedAddress(), worker.CanonicalAddress()) assert.Empty(t, worker.Name) - assert.Equal(t, worker.ReportedStatus.CreateTime, worker.CreateTime) - assert.Equal(t, worker.LastConnectionUpdate(), worker.UpdateTime) - assert.Equal(t, uint32(1), worker.Version) - assert.Empty(t, worker.Address) - assert.Empty(t, worker.Description) - } - - // Remove the status name, verify that it doesn't remove the resource's name. - updatedStatus1 := servers.NewWorkerStatus(wStatus1.WorkerId, - servers.WithAddress(wStatus1.Address)) - _, _, err = repo.UpsertWorkerStatus(ctx, updatedStatus1) - require.NoError(t, err) - { - workers, err := repo.ListWorkers(ctx) - require.NoError(t, err) - assert.Len(t, workers, 1) - worker := workers[0] - // The resource's name remains untouched - assert.Empty(t, worker.Name) - // The ReportedStatus's name is correctly cleared - assert.Empty(t, worker.ReportedStatus.Name) - // The last connection update time was increased - assert.Greater(t, worker.LastConnectionUpdate().AsTime().UnixMilli(), worker.UpdateTime.AsTime().UnixMilli()) - assert.Equal(t, worker.UpdateTime, worker.CreateTime) - - assert.Equal(t, updatedStatus1.GetWorkerId(), worker.GetPublicId()) - assert.Equal(t, updatedStatus1.Address, worker.CanonicalAddress()) - assert.Equal(t, worker.ReportedStatus.CreateTime, worker.CreateTime) + assert.Equal(t, worker.GetLastStatusTime(), worker.UpdateTime) assert.Equal(t, uint32(1), worker.Version) assert.Empty(t, worker.Address) assert.Empty(t, worker.Description) @@ -151,12 +189,14 @@ func TestUpsertWorkerStatus(t *testing.T) { failureCases := []struct { name string - status *servers.WorkerStatus + repo *servers.Repository + status *servers.Worker errAssert func(*testing.T, error) }{ { name: "no address", - status: servers.NewWorkerStatus("worker_with_no_address", + repo: repo, + status: servers.NewWorkerForStatus(scope.Global.String(), servers.WithName("worker_with_no_address")), errAssert: func(t *testing.T, err error) { t.Helper() @@ -164,20 +204,25 @@ func TestUpsertWorkerStatus(t *testing.T) { }, }, { - name: "invalid id", - status: servers.NewWorkerStatus("short", - servers.WithAddress("someaddress"), - servers.WithName("worker_with_short_id")), + name: "cant specifying public id", + repo: repo, + status: func() *servers.Worker { + w := servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName("worker_with_no_address"), + servers.WithAddress("workeraddress")) + w.PublicId = "w_specified" + return w + }(), errAssert: func(t *testing.T, err error) { t.Helper() - assert.True(t, errors.IsCheckConstraintError(err)) + assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err)) }, }, { - name: "no id", - status: servers.NewWorkerStatus("", - servers.WithAddress("anotheraddress"), - servers.WithName("no_address")), + name: "no name", + repo: repo, + status: servers.NewWorkerForStatus(scope.Global.String(), + servers.WithAddress("no_name_address")), errAssert: func(t *testing.T, err error) { t.Helper() assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err)) @@ -185,16 +230,63 @@ func TestUpsertWorkerStatus(t *testing.T) { }, { name: "no status", + repo: repo, status: nil, errAssert: func(t *testing.T, err error) { t.Helper() assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err)) }, }, + { + name: "empty scope", + repo: repo, + status: servers.NewWorkerForStatus("", + servers.WithAddress("address"), + servers.WithName("config_name1")), + errAssert: func(t *testing.T, err error) { + t.Helper() + assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err)) + }, + }, + { + name: "providing api tags", + repo: repo, + status: func() *servers.Worker { + w := servers.NewWorker(scope.Global.String(), + servers.WithWorkerTags(&servers.Tag{Key: "test", Value: "test"})) + w.WorkerReportedAddress = "some_address" + w.WorkerReportedName = "providing api tags" + return w + }(), + errAssert: func(t *testing.T, err error) { + t.Helper() + assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err)) + }, + }, + { + name: "database failure", + repo: func() *servers.Repository { + conn, mock := db.TestSetupWithMock(t) + rw := db.New(conn) + mock.ExpectBegin() + mock.ExpectQuery(`INSERT`).WillReturnError(errors.New(context.Background(), errors.Internal, "test", "create-error")) + mock.ExpectRollback() + r, err := servers.NewRepository(rw, rw, kms) + require.NoError(t, err) + return r + }(), + status: servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName("database failure"), + servers.WithAddress("address")), + errAssert: func(t *testing.T, err error) { + t.Helper() + assert.Error(t, err) + }, + }, } for _, tc := range failureCases { t.Run(fmt.Sprintf("Failures %s", tc.name), func(t *testing.T) { - _, _, err = repo.UpsertWorkerStatus(ctx, tc.status) + _, err = tc.repo.UpsertWorkerStatus(ctx, tc.status) assert.Error(t, err) tc.errAssert(t, err) @@ -206,9 +298,10 @@ func TestUpsertWorkerStatus(t *testing.T) { } { - anotherStatus := servers.NewWorkerStatus("another_test_worker", + anotherStatus := servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName("another_test_worker"), servers.WithAddress("address")) - _, _, err = repo.UpsertWorkerStatus(ctx, anotherStatus) + _, err = repo.UpsertWorkerStatus(ctx, anotherStatus) require.NoError(t, err) { workers, err := repo.ListWorkers(ctx) @@ -228,7 +321,9 @@ func TestTagUpdatingListing(t *testing.T) { require.NoError(err) ctx := context.Background() - wStatus := servers.NewWorkerStatus("test_worker_1", + worker1 := servers.TestWorker(t, conn, wrapper) + wStatus := servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker1.GetWorkerReportedName()), servers.WithAddress("127.0.0.1"), servers.WithWorkerTags( &servers.Tag{ @@ -240,53 +335,16 @@ func TestTagUpdatingListing(t *testing.T) { Value: "value2", })) - _, _, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true)) + worker1, err = repo.UpsertWorkerStatus(ctx, wStatus, + servers.WithUpdateTags(true)) require.NoError(err) - - wStatus = servers.NewWorkerStatus("test_worker_2", - servers.WithAddress("127.0.0.1"), - servers.WithWorkerTags( - &servers.Tag{ - Key: "tag2", - Value: "value1", - }, - &servers.Tag{ - Key: "tag2", - Value: "value2", - })) - _, _, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true)) - require.NoError(err) - - tags, err := repo.ListTagsForWorkers(ctx, []string{"test_worker_1", "test_worker_2"}) - require.NoError(err) - - // Base case - exp := map[string][]*servers.Tag{ - "test_worker_1": { - { - Key: "tag1", - Value: "value1", - }, { - Key: "tag1", - Value: "value2", - }, - }, - "test_worker_2": { - { - Key: "tag2", - Value: "value1", - }, - { - Key: "tag2", - Value: "value2", - }, - }, - } - require.Equal(exp, tags) + assert.Len(t, worker1.CanonicalTags(), 1) + assert.ElementsMatch(t, []string{"value1", "value2"}, worker1.CanonicalTags()["tag1"]) // Update without saying to update tags - wStatus = servers.NewWorkerStatus("test_worker_2", - servers.WithAddress("192.168.1.1"), + wStatus = servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker1.GetWorkerReportedName()), + servers.WithAddress(worker1.GetWorkerReportedAddress()), servers.WithWorkerTags( &servers.Tag{ Key: "tag22", @@ -296,30 +354,16 @@ func TestTagUpdatingListing(t *testing.T) { Key: "tag22", Value: "value22", })) - _, _, err = repo.UpsertWorkerStatus(ctx, wStatus) + worker1, err = repo.UpsertWorkerStatus(ctx, wStatus) require.NoError(err) - tags, err = repo.ListTagsForWorkers(ctx, []string{"test_worker_1", "test_worker_2"}) - require.NoError(err) - require.Equal(exp, tags) + assert.Len(t, worker1.CanonicalTags(), 1) + assert.ElementsMatch(t, []string{"value1", "value2"}, worker1.CanonicalTags()["tag1"]) // Update tags and test again - _, _, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true)) + worker1, err = repo.UpsertWorkerStatus(ctx, wStatus, servers.WithUpdateTags(true)) require.NoError(err) - tags, err = repo.ListTagsForWorkers(ctx, []string{"test_worker_1", "test_worker_2"}) - require.NoError(err) - require.NotEqual(exp, tags) - // Update and try again - exp["test_worker_2"] = []*servers.Tag{ - { - Key: "tag22", - Value: "value21", - }, - { - Key: "tag22", - Value: "value22", - }, - } - require.Equal(exp, tags) + assert.Len(t, worker1.CanonicalTags(), 1) + assert.ElementsMatch(t, []string{"value21", "value22"}, worker1.CanonicalTags()["tag22"]) } func TestListWorkersWithLiveness(t *testing.T) { @@ -333,28 +377,19 @@ func TestListWorkersWithLiveness(t *testing.T) { require.NoError(err) ctx := context.Background() - newWorkerStatus := func(publicId string) *servers.WorkerStatus { - result := servers.NewWorkerStatus(publicId, - servers.WithAddress("127.0.0.1")) - _, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, result) - require.NoError(err) - require.Equal(1, rowsUpdated) - - return result - } - - workConf1 := newWorkerStatus("test_worker_1") - workConf2 := newWorkerStatus("test_worker_2") - workConf3 := newWorkerStatus("test_worker_3") + worker1 := servers.TestWorker(t, conn, wrapper) + worker2 := servers.TestWorker(t, conn, wrapper) + worker3 := servers.TestWorker(t, conn, wrapper) // Sleep the default liveness time (15sec currently) +1s time.Sleep(time.Second * 16) // Push an upsert to the first worker so that its status has been // updated. - _, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, workConf1) + _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker1.GetWorkerReportedName()), + servers.WithAddress(worker1.GetWorkerReportedAddress()))) require.NoError(err) - require.Equal(1, rowsUpdated) requireIds := func(expected []string, actual []*servers.Worker) { require.Len(expected, len(actual)) @@ -375,25 +410,26 @@ func TestListWorkersWithLiveness(t *testing.T) { result, err := serversRepo.ListWorkers(ctx) require.NoError(err) require.Len(result, 1) - requireIds([]string{workConf1.WorkerId}, result) + requireIds([]string{worker1.GetPublicId()}, result) // Upsert second server. - _, rowsUpdated, err = serversRepo.UpsertWorkerStatus(ctx, workConf2) + _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker2.GetWorkerReportedName()), + servers.WithAddress(worker1.GetWorkerReportedAddress()))) require.NoError(err) - require.Equal(1, rowsUpdated) // Static liveness. Should get two, so long as this did not take // more than 5s to execute. result, err = serversRepo.ListWorkers(ctx, servers.WithLiveness(time.Second*5)) require.NoError(err) require.Len(result, 2) - requireIds([]string{workConf1.WorkerId, workConf2.WorkerId}, result) + requireIds([]string{worker1.GetPublicId(), worker2.GetPublicId()}, result) // Liveness disabled, should get all three workers. result, err = serversRepo.ListWorkers(ctx, servers.WithLiveness(-1)) require.NoError(err) require.Len(result, 3) - requireIds([]string{workConf1.WorkerId, workConf2.WorkerId, workConf3.WorkerId}, result) + requireIds([]string{worker1.GetPublicId(), worker2.GetPublicId(), worker3.GetPublicId()}, result) } func TestRepository_CreateWorker(t *testing.T) { @@ -500,6 +536,42 @@ func TestRepository_CreateWorker(t *testing.T) { wantErr: true, wantErrContains: "unable to create worker", }, + { + name: "no worker reported address", + setup: func() *servers.Worker { + w := servers.NewWorker(scope.Global.String()) + w.WorkerReportedAddress = "foo" + return w + }, + repo: testRepo, + wantErr: true, + wantErrIs: errors.InvalidParameter, + wantErrContains: "worker reported address is not empty", + }, + { + name: "no worker reported name", + setup: func() *servers.Worker { + w := servers.NewWorker(scope.Global.String()) + w.WorkerReportedName = "foo" + return w + }, + repo: testRepo, + wantErr: true, + wantErrIs: errors.InvalidParameter, + wantErrContains: "worker reported name is not empty", + }, + { + name: "no last status update", + setup: func() *servers.Worker { + w := servers.NewWorker(scope.Global.String()) + w.LastStatusTime = timestamp.Now() + return w + }, + repo: testRepo, + wantErr: true, + wantErrIs: errors.InvalidParameter, + wantErrContains: "last status time is not nil", + }, { name: "success", setup: func() *servers.Worker { @@ -538,9 +610,6 @@ func TestRepository_CreateWorker(t *testing.T) { } err = rw.LookupByPublicId(testCtx, found) require.NoError(err) - if found.Tags == nil { - found.Tags = []*servers.Tag{} - } assert.Equal(got, found) }) } diff --git a/internal/servers/store/worker.pb.go b/internal/servers/store/worker.pb.go index fc7a0b554..c92cc11fc 100644 --- a/internal/servers/store/worker.pb.go +++ b/internal/servers/store/worker.pb.go @@ -51,6 +51,16 @@ type Worker struct { // version allows optimistic locking of the resource. // @inject_tag: `gorm:"default:null"` Version uint32 `protobuf:"varint,80,opt,name=version,proto3" json:"version,omitempty" gorm:"default:null"` + // Name of the resource (optional) + // @inject_tag: `gorm:"default:null"` + WorkerReportedName string `protobuf:"bytes,90,opt,name=worker_reported_name,json=workerReportedName,proto3" json:"worker_reported_name,omitempty" gorm:"default:null"` + // Address for the worker. This must be set when the worker has received + // at least 1 status update fromt he worker daemon. + // @inject_tag: `gorm:"default:null"` + WorkerReportedAddress string `protobuf:"bytes,100,opt,name=worker_reported_address,json=workerReportedAddress,proto3" json:"worker_reported_address,omitempty" gorm:"default:null"` + // The last_status_time is the last time a worker daemon sent a status update. + // @inject_tag: `gorm:"default:null"` + LastStatusTime *timestamp.Timestamp `protobuf:"bytes,110,opt,name=last_status_time,json=lastStatusTime,proto3" json:"last_status_time,omitempty" gorm:"default:null"` } func (x *Worker) Reset() { @@ -141,6 +151,27 @@ func (x *Worker) GetVersion() uint32 { return 0 } +func (x *Worker) GetWorkerReportedName() string { + if x != nil { + return x.WorkerReportedName + } + return "" +} + +func (x *Worker) GetWorkerReportedAddress() string { + if x != nil { + return x.WorkerReportedAddress + } + return "" +} + +func (x *Worker) GetLastStatusTime() *timestamp.Timestamp { + if x != nil { + return x.LastStatusTime + } + return nil +} + // WorkerTag is a tag for a worker. The primary key is comprised of the // worker_id, key, value, and source. type WorkerTag struct { @@ -222,97 +253,6 @@ func (x *WorkerTag) GetSource() string { return "" } -// WorkerStatus contains the fields that the worker reports to the controller -// in its status updates. -type WorkerStatus struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache - unknownFields protoimpl.UnknownFields - - // worker_id is the public key that key of the worker this tag is for. - // @inject_tag: `gorm:"primary_key"` - WorkerId string `protobuf:"bytes,10,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty" gorm:"primary_key"` - // The create_time is set by the database. - // @inject_tag: `gorm:"default:current_timestamp"` - CreateTime *timestamp.Timestamp `protobuf:"bytes,20,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:current_timestamp"` - // The update_time is set by the database. - // @inject_tag: `gorm:"default:current_timestamp"` - UpdateTime *timestamp.Timestamp `protobuf:"bytes,30,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:current_timestamp"` - // Name of the resource (optional) - // @inject_tag: `gorm:"default:null"` - Name string `protobuf:"bytes,40,opt,name=name,proto3" json:"name,omitempty" gorm:"default:null"` - // Address for the worker. This must be set. - // @inject_tag: `gorm:"default:null"` - Address string `protobuf:"bytes,50,opt,name=address,proto3" json:"address,omitempty" gorm:"default:null"` -} - -func (x *WorkerStatus) Reset() { - *x = WorkerStatus{} - if protoimpl.UnsafeEnabled { - mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } -} - -func (x *WorkerStatus) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*WorkerStatus) ProtoMessage() {} - -func (x *WorkerStatus) ProtoReflect() protoreflect.Message { - mi := &file_controller_storage_servers_store_v1_worker_proto_msgTypes[2] - if protoimpl.UnsafeEnabled && x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -// Deprecated: Use WorkerStatus.ProtoReflect.Descriptor instead. -func (*WorkerStatus) Descriptor() ([]byte, []int) { - return file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP(), []int{2} -} - -func (x *WorkerStatus) GetWorkerId() string { - if x != nil { - return x.WorkerId - } - return "" -} - -func (x *WorkerStatus) GetCreateTime() *timestamp.Timestamp { - if x != nil { - return x.CreateTime - } - return nil -} - -func (x *WorkerStatus) GetUpdateTime() *timestamp.Timestamp { - if x != nil { - return x.UpdateTime - } - return nil -} - -func (x *WorkerStatus) GetName() string { - if x != nil { - return x.Name - } - return "" -} - -func (x *WorkerStatus) GetAddress() string { - if x != nil { - return x.Address - } - return "" -} - var File_controller_storage_servers_store_v1_worker_proto protoreflect.FileDescriptor var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ @@ -324,7 +264,7 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ 0x74, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31, 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, 0xc4, 0x02, 0x0a, 0x06, 0x57, 0x6f, 0x72, + 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x84, 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, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, @@ -344,34 +284,30 @@ var file_controller_storage_servers_store_v1_worker_proto_rawDesc = []byte{ 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x46, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x63, 0x6f, 0x70, 0x65, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, - 0x18, 0x50, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, + 0x18, 0x50, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, + 0x30, 0x0a, 0x14, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x72, 0x65, 0x70, 0x6f, 0x72, 0x74, + 0x65, 0x64, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x5a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x12, 0x77, + 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x70, 0x6f, 0x72, 0x74, 0x65, 0x64, 0x4e, 0x61, 0x6d, + 0x65, 0x12, 0x36, 0x0a, 0x17, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x72, 0x65, 0x70, 0x6f, + 0x72, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x64, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x15, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x70, 0x6f, 0x72, 0x74, + 0x65, 0x64, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x54, 0x0a, 0x10, 0x6c, 0x61, 0x73, + 0x74, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x6e, 0x20, + 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, + 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, + 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, + 0x0e, 0x6c, 0x61, 0x73, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x54, 0x69, 0x6d, 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, 0x22, 0xf3, 0x01, 0x0a, 0x0c, 0x57, 0x6f, - 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 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, 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, - 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, - 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, - 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, - 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, - 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, - 0x69, 0x6d, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, - 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, - 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, - 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, - 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, - 0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x42, - 0x3c, 0x5a, 0x3a, 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, - 0x73, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x3c, 0x5a, 0x3a, 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, 0x73, 0x2f, 0x73, 0x74, 0x6f, 0x72, + 0x65, 0x3b, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -386,23 +322,21 @@ func file_controller_storage_servers_store_v1_worker_proto_rawDescGZIP() []byte return file_controller_storage_servers_store_v1_worker_proto_rawDescData } -var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_controller_storage_servers_store_v1_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_controller_storage_servers_store_v1_worker_proto_goTypes = []interface{}{ (*Worker)(nil), // 0: controller.storage.servers.store.v1.Worker (*WorkerTag)(nil), // 1: controller.storage.servers.store.v1.WorkerTag - (*WorkerStatus)(nil), // 2: controller.storage.servers.store.v1.WorkerStatus - (*timestamp.Timestamp)(nil), // 3: controller.storage.timestamp.v1.Timestamp + (*timestamp.Timestamp)(nil), // 2: controller.storage.timestamp.v1.Timestamp } var file_controller_storage_servers_store_v1_worker_proto_depIdxs = []int32{ - 3, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp - 3, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp - 3, // 2: controller.storage.servers.store.v1.WorkerStatus.create_time:type_name -> controller.storage.timestamp.v1.Timestamp - 3, // 3: controller.storage.servers.store.v1.WorkerStatus.update_time:type_name -> controller.storage.timestamp.v1.Timestamp - 4, // [4:4] is the sub-list for method output_type - 4, // [4:4] is the sub-list for method input_type - 4, // [4:4] is the sub-list for extension type_name - 4, // [4:4] is the sub-list for extension extendee - 0, // [0:4] is the sub-list for field type_name + 2, // 0: controller.storage.servers.store.v1.Worker.create_time:type_name -> controller.storage.timestamp.v1.Timestamp + 2, // 1: controller.storage.servers.store.v1.Worker.update_time:type_name -> controller.storage.timestamp.v1.Timestamp + 2, // 2: controller.storage.servers.store.v1.Worker.last_status_time:type_name -> controller.storage.timestamp.v1.Timestamp + 3, // [3:3] is the sub-list for method output_type + 3, // [3:3] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name } func init() { file_controller_storage_servers_store_v1_worker_proto_init() } @@ -435,18 +369,6 @@ func file_controller_storage_servers_store_v1_worker_proto_init() { return nil } } - file_controller_storage_servers_store_v1_worker_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*WorkerStatus); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } } type x struct{} out := protoimpl.TypeBuilder{ @@ -454,7 +376,7 @@ func file_controller_storage_servers_store_v1_worker_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_controller_storage_servers_store_v1_worker_proto_rawDesc, NumEnums: 0, - NumMessages: 3, + NumMessages: 2, NumExtensions: 0, NumServices: 0, }, diff --git a/internal/servers/testing.go b/internal/servers/testing.go index 3a4ec7d28..b027d7876 100644 --- a/internal/servers/testing.go +++ b/internal/servers/testing.go @@ -3,6 +3,7 @@ package servers import ( "context" "math/rand" + "strings" "testing" "time" @@ -89,16 +90,13 @@ func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper) *Worker { id, err := newWorkerId(context.Background()) require.NoError(t, err) - name := "test-worker-" + id - wStatus := NewWorkerStatus(id, + name := "test-worker-" + strings.ToLower(id) + wrk := NewWorkerForStatus(scope.Global.String(), WithName(name), WithAddress("127.0.0.1")) - _, _, err = serversRepo.UpsertWorkerStatus(context.Background(), wStatus) + wrk, err = serversRepo.UpsertWorkerStatus(context.Background(), wrk) require.NoError(t, err) - - wrk := NewWorker(scope.Global.String(), WithPublicId(id)) - require.NoError(t, rw.LookupById(context.Background(), wrk)) - wrk.ReportedStatus = wStatus + require.NotNil(t, wrk) return wrk } diff --git a/internal/servers/testing_test.go b/internal/servers/testing_test.go index 0764adf3e..4525dc678 100644 --- a/internal/servers/testing_test.go +++ b/internal/servers/testing_test.go @@ -18,7 +18,8 @@ func TestTestWorker(t *testing.T) { assert.NotNil(t, tWorker) assert.True(t, strings.HasPrefix(tWorker.GetPublicId(), WorkerPrefix)) - lkpWorker := NewWorker(scope.Global.String(), WithPublicId(tWorker.GetPublicId())) + lkpWorker := NewWorker(scope.Global.String()) + lkpWorker.PublicId = tWorker.GetPublicId() rw := db.New(conn) require.NoError(t, rw.LookupById(context.Background(), lkpWorker)) assert.NotNil(t, lkpWorker) diff --git a/internal/servers/worker.go b/internal/servers/worker.go index bfa218d4c..f7978fb0a 100644 --- a/internal/servers/worker.go +++ b/internal/servers/worker.go @@ -16,23 +16,39 @@ import ( type Worker struct { *store.Worker - ReportedStatus *WorkerStatus `gorm:"-"` - Tags []*Tag `gorm:"-"` + apiTags []*Tag `gorm:"-"` + configTags []*Tag `gorm:"-"` } // NewWorker returns a new Worker. Valid options are WithName, WithDescription -// WithAddress, and WithWorkerTags. All other options are ignored. +// WithAddress, and WithWorkerTags. All other options are ignored. This does +// not set any of the worker reported values. func NewWorker(scopeId string, opt ...Option) *Worker { opts := getOpts(opt...) return &Worker{ Worker: &store.Worker{ ScopeId: scopeId, - PublicId: opts.withPublicId, Name: opts.withName, Description: opts.withDescription, Address: opts.withAddress, }, - Tags: opts.withWorkerTags, + apiTags: opts.withWorkerTags, + } +} + +// NewWorkerForStatus returns a new Worker usable for status updates. +// Valid options are WithName, WithAddress, and WithWorkerTags, all of which +// are assigned to the worker reported variations of these fields. +// All other options are ignored. +func NewWorkerForStatus(scopeId string, opt ...Option) *Worker { + opts := getOpts(opt...) + return &Worker{ + Worker: &store.Worker{ + ScopeId: scopeId, + WorkerReportedName: opts.withName, + WorkerReportedAddress: opts.withAddress, + }, + configTags: opts.withWorkerTags, } } @@ -45,17 +61,23 @@ func (w *Worker) clone() *Worker { if w == nil { return nil } - tags := make([]*Tag, 0, len(w.Tags)) - for _, t := range w.Tags { - tags = append(tags, &Tag{Key: t.Key, Value: t.Value}) - } cw := proto.Clone(w.Worker) - crs := w.ReportedStatus.clone() - return &Worker{ - Worker: cw.(*store.Worker), - Tags: tags, - ReportedStatus: crs, + cWorker := &Worker{ + Worker: cw.(*store.Worker), } + if w.apiTags != nil { + cWorker.apiTags = make([]*Tag, 0, len(w.apiTags)) + for _, t := range w.apiTags { + cWorker.apiTags = append(cWorker.apiTags, &Tag{Key: t.Key, Value: t.Value}) + } + } + if w.configTags != nil { + cWorker.configTags = make([]*Tag, 0, len(w.configTags)) + for _, t := range w.configTags { + cWorker.configTags = append(cWorker.configTags, &Tag{Key: t.Key, Value: t.Value}) + } + } + return cWorker } // CanonicalAddress returns the actual address boundary believes should be used @@ -64,27 +86,21 @@ func (w *Worker) clone() *Worker { // in its connection status updates. If neither is available, an empty string // is returned. func (w *Worker) CanonicalAddress() string { - switch { - case w.Address != "": + if w.GetAddress() != "" { return w.GetAddress() - case w.ReportedStatus != nil: - return w.ReportedStatus.GetAddress() - default: - return "" } + return w.GetWorkerReportedAddress() } // CanonicalTags is the deduplicated set of tags contained on both the resource // set over the API as well as the tags reported by the worker itself. func (w *Worker) CanonicalTags() map[string][]string { dedupedTags := make(map[Tag]struct{}) - for _, t := range w.Tags { + for _, t := range w.apiTags { dedupedTags[*t] = struct{}{} } - if w.ReportedStatus != nil { - for _, t := range w.ReportedStatus.Tags { - dedupedTags[*t] = struct{}{} - } + for _, t := range w.configTags { + dedupedTags[*t] = struct{}{} } tags := make(map[string][]string) for t := range dedupedTags { @@ -93,14 +109,30 @@ func (w *Worker) CanonicalTags() map[string][]string { return tags } -// LastConnectionUpdate contains the last time the worker has reported to the +func (w *Worker) GetApiTags() map[string][]string { + tags := make(map[string][]string) + for _, t := range w.apiTags { + tags[t.Key] = append(tags[t.Key], t.Value) + } + return tags +} + +func (w *Worker) GetConfigTags() map[string][]string { + tags := make(map[string][]string) + for _, t := range w.configTags { + tags[t.Key] = append(tags[t.Key], t.Value) + } + return tags +} + +// GetLastStatusTime contains the last time the worker has reported to the // controller its connection status. If the worker has never reported to a // controller then nil is returned. -func (w *Worker) LastConnectionUpdate() *timestamp.Timestamp { - if w.ReportedStatus == nil { +func (w *Worker) GetLastStatusTime() *timestamp.Timestamp { + if w == nil || w.Worker == nil || w.Worker.GetLastStatusTime().AsTime() == timestamp.NegativeInfinityTS { return nil } - return w.ReportedStatus.GetUpdateTime() + return w.Worker.GetLastStatusTime() } // TableName overrides the table name used by Worker to `server_worker` @@ -111,61 +143,50 @@ func (Worker) TableName() string { // workerAggregate contains an aggregated view of the values associated with // a single worker. type workerAggregate struct { - PublicId string `gorm:"primary_key"` - ScopeId string - Name string - Description string - CreateTime *timestamp.Timestamp - UpdateTime *timestamp.Timestamp - Address string - Version uint32 - ApiTags string - WorkerStatusName string - WorkerStatusAddress string - WorkerStatusCreateTime *timestamp.Timestamp - WorkerStatusUpdateTime *timestamp.Timestamp - WorkerConfigTags string + PublicId string `gorm:"primary_key"` + ScopeId string + Name string + Description string + CreateTime *timestamp.Timestamp + UpdateTime *timestamp.Timestamp + Address string + Version uint32 + ApiTags string + WorkerReportedName string + WorkerReportedAddress string + LastStatusTime *timestamp.Timestamp + WorkerConfigTags string } func (a *workerAggregate) toWorker(ctx context.Context) (*Worker, error) { const op = "servers.(workerAggregate).toWorker" - workerOptions := []Option{ - WithPublicId(a.PublicId), - WithName(a.Name), - WithDescription(a.Description), - WithAddress(a.Address), + 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, + WorkerReportedAddress: a.WorkerReportedAddress, + WorkerReportedName: a.WorkerReportedName, + LastStatusTime: a.LastStatusTime, + }, } tags, err := tagsFromAggregatedTagString(ctx, a.ApiTags) if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error parsing config tag string")) } - if len(tags) > 0 { - workerOptions = append(workerOptions, WithWorkerTags(tags...)) - } - worker := NewWorker(a.ScopeId, workerOptions...) - worker.CreateTime = a.CreateTime - worker.UpdateTime = a.UpdateTime - worker.Version = a.Version + worker.apiTags = tags - if a.WorkerStatusCreateTime == nil { - return worker, nil - } - statusOptions := []Option{ - WithName(a.WorkerStatusName), - WithAddress(a.WorkerStatusAddress), - } tags, err = tagsFromAggregatedTagString(ctx, a.WorkerConfigTags) if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error parsing config tag string")) } - if len(tags) > 0 { - statusOptions = append(statusOptions, WithWorkerTags(tags...)) - } - cfg := NewWorkerStatus(a.PublicId, statusOptions...) - cfg.CreateTime = a.WorkerStatusCreateTime - cfg.UpdateTime = a.WorkerStatusUpdateTime + worker.configTags = tags - worker.ReportedStatus = cfg return worker, nil } diff --git a/internal/servers/worker_status.go b/internal/servers/worker_status.go deleted file mode 100644 index c8918618e..000000000 --- a/internal/servers/worker_status.go +++ /dev/null @@ -1,50 +0,0 @@ -package servers - -import ( - "github.com/hashicorp/boundary/internal/servers/store" - "google.golang.org/protobuf/proto" -) - -// A WorkerStatus is a container for the fields that a worker reports to the -// controller about itself in its status updates. While this may include a -// Name, this is different from the Worker's name in that the Worker's name is -// the name of the Worker resource in boundary where the name in the -// WorkerStatus is the name the worker has reported about itself. -type WorkerStatus struct { - *store.WorkerStatus - Tags []*Tag `gorm:"-"` -} - -// NewWorkerStatus returns a new WorkerStatus. Valid options are WithName, -// WithAddress, and WithWorkerTags. All other options are ignored. -func NewWorkerStatus(workerId string, opt ...Option) *WorkerStatus { - opts := getOpts(opt...) - return &WorkerStatus{ - WorkerStatus: &store.WorkerStatus{ - WorkerId: workerId, - Name: opts.withName, - Address: opts.withAddress, - }, - Tags: opts.withWorkerTags, - } -} - -func (w *WorkerStatus) clone() *WorkerStatus { - if w == nil { - return nil - } - tags := make([]*Tag, 0, len(w.Tags)) - for _, t := range w.Tags { - tags = append(tags, &Tag{Key: t.Key, Value: t.Value}) - } - cw := proto.Clone(w.WorkerStatus) - return &WorkerStatus{ - WorkerStatus: cw.(*store.WorkerStatus), - Tags: tags, - } -} - -// TableName overrides the table name used by WorkerStatus to `server_worker_status` -func (WorkerStatus) TableName() string { - return "server_worker_status" -} diff --git a/internal/servers/worker_test.go b/internal/servers/worker_test.go index d01eeeca2..2830aba62 100644 --- a/internal/servers/worker_test.go +++ b/internal/servers/worker_test.go @@ -2,6 +2,7 @@ package servers import ( "context" + "strings" "testing" "github.com/google/go-cmp/cmp" @@ -15,8 +16,7 @@ import ( ) func TestWorkerCanonicalAddress(t *testing.T) { - worker := NewWorker(scope.Global.String()) - worker.ReportedStatus = NewWorkerStatus("foo", WithAddress("status")) + worker := NewWorkerForStatus(scope.Global.String(), WithAddress("status")) assert.Equal(t, "status", worker.CanonicalAddress()) worker.Address = "worker" assert.Equal(t, "worker", worker.CanonicalAddress()) @@ -29,12 +29,11 @@ func TestWorkerCanonicalTags(t *testing.T) { &Tag{Key: "key", Value: "shared"}, &Tag{Key: "key2", Value: "apis key2 unique"}, )) - w.ReportedStatus = NewWorkerStatus("", - WithWorkerTags( - &Tag{Key: "key", Value: "configs unique"}, - &Tag{Key: "key", Value: "shared"}, - &Tag{Key: "key3", Value: "configs key3 unique"}, - )) + w.configTags = []*Tag{ + {Key: "key", Value: "configs unique"}, + {Key: "key", Value: "shared"}, + {Key: "key3", Value: "configs key3 unique"}, + } got := w.CanonicalTags() assert.Len(t, got, 3, "2 keys expected, 'key' and 'key2'") @@ -61,11 +60,11 @@ func TestWorkerAggregate(t *testing.T) { { id, err := newWorkerId(ctx) require.NoError(t, err) - require.NoError(t, rw.Create(ctx, - NewWorker(scope.Global.String(), - WithPublicId(id), - WithName(id), - WithAddress("address")))) + w := NewWorker(scope.Global.String(), + WithName(id), + WithAddress("address")) + w.PublicId = id + require.NoError(t, rw.Create(ctx, w)) got := getAggWorker(id) assert.Equal(t, id, got.GetPublicId()) @@ -73,7 +72,7 @@ func TestWorkerAggregate(t *testing.T) { assert.Equal(t, id, got.GetName()) assert.Equal(t, "address", got.GetAddress()) assert.Equal(t, uint32(1), got.GetVersion()) - assert.Nil(t, got.ReportedStatus) + assert.Nil(t, got.GetLastStatusTime()) assert.Empty(t, got.CanonicalTags()) } @@ -81,32 +80,29 @@ func TestWorkerAggregate(t *testing.T) { { id, err := newWorkerId(ctx) require.NoError(t, err) - require.NoError(t, rw.Create(ctx, - NewWorker(scope.Global.String(), WithPublicId(id)))) - require.NoError(t, rw.Create(ctx, - NewWorkerStatus(id, - WithAddress("address"), - WithName(id)))) + w := NewWorkerForStatus(scope.Global.String(), + WithAddress("address"), + WithName(strings.ToLower(id))) + w.PublicId = id + require.NoError(t, rw.Create(ctx, w)) got := getAggWorker(id) assert.Equal(t, id, got.GetPublicId()) assert.Equal(t, uint32(1), got.GetVersion()) - assert.NotNil(t, got.ReportedStatus) - assert.Equal(t, id, got.ReportedStatus.GetWorkerId()) - assert.Equal(t, id, got.ReportedStatus.GetName()) - assert.NotNil(t, got.ReportedStatus.CreateTime) - assert.NotNil(t, got.ReportedStatus.UpdateTime) - assert.Equal(t, "address", got.ReportedStatus.GetAddress()) + assert.NotNil(t, got.GetLastStatusTime()) + assert.Equal(t, strings.ToLower(id), got.GetWorkerReportedName()) + assert.Equal(t, "address", got.GetWorkerReportedAddress()) } // Worker with a config tag { id, err := newWorkerId(ctx) require.NoError(t, err) - require.NoError(t, rw.Create(ctx, - NewWorker(scope.Global.String(), WithPublicId(id)))) - require.NoError(t, rw.Create(ctx, - NewWorkerStatus(id, WithAddress("address")))) + ws := NewWorkerForStatus(scope.Global.String(), + WithAddress("address"), + WithName(strings.ToLower(id))) + ws.PublicId = id + require.NoError(t, rw.Create(ctx, ws)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, @@ -118,19 +114,20 @@ func TestWorkerAggregate(t *testing.T) { got := getAggWorker(id) assert.Equal(t, id, got.GetPublicId()) assert.Equal(t, uint32(1), got.GetVersion()) - assert.NotNil(t, got.ReportedStatus) - assert.Empty(t, got.Tags) - assert.Equal(t, got.ReportedStatus.Tags, []*Tag{{Key: "key", Value: "val"}}) + assert.NotNil(t, got.GetLastStatusTime()) + assert.Empty(t, got.apiTags) + assert.Equal(t, got.configTags, []*Tag{{Key: "key", Value: "val"}}) } // Worker with many config tag { id, err := newWorkerId(ctx) require.NoError(t, err) - require.NoError(t, rw.Create(ctx, - NewWorker(scope.Global.String(), WithPublicId(id)))) - require.NoError(t, rw.Create(ctx, - NewWorkerStatus(id, WithAddress("address")))) + ws := NewWorkerForStatus(scope.Global.String(), + WithAddress("address"), + WithName(strings.ToLower(id))) + ws.PublicId = id + require.NoError(t, rw.Create(ctx, ws)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, Key: "key", @@ -151,9 +148,9 @@ func TestWorkerAggregate(t *testing.T) { })) got := getAggWorker(id) - require.NotNil(t, got.ReportedStatus) - assert.Empty(t, got.Tags) - assert.ElementsMatch(t, got.ReportedStatus.Tags, []*Tag{ + require.NotNil(t, got.GetLastStatusTime()) + assert.Empty(t, got.apiTags) + assert.ElementsMatch(t, got.configTags, []*Tag{ {Key: "key", Value: "val"}, {Key: "key", Value: "val2"}, {Key: "key2", Value: "val2"}, @@ -164,10 +161,11 @@ func TestWorkerAggregate(t *testing.T) { { id, err := newWorkerId(ctx) require.NoError(t, err) - require.NoError(t, rw.Create(ctx, - NewWorker(scope.Global.String(), WithPublicId(id)))) - require.NoError(t, rw.Create(ctx, - NewWorkerStatus(id, WithAddress("address")))) + ws := NewWorkerForStatus(scope.Global.String(), + WithAddress("address"), + WithName(strings.ToLower(id))) + ws.PublicId = id + require.NoError(t, rw.Create(ctx, ws)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, @@ -179,19 +177,20 @@ func TestWorkerAggregate(t *testing.T) { got := getAggWorker(id) assert.Equal(t, id, got.GetPublicId()) assert.Equal(t, uint32(1), got.GetVersion()) - assert.NotNil(t, got.ReportedStatus) - assert.Empty(t, got.ReportedStatus.Tags) - assert.Equal(t, got.Tags, []*Tag{{Key: "key", Value: "val"}}) + assert.NotNil(t, got.GetLastStatusTime()) + assert.Empty(t, got.GetConfigTags()) + assert.Equal(t, got.apiTags, []*Tag{{Key: "key", Value: "val"}}) } // Worker with mix of tag sources { id, err := newWorkerId(ctx) require.NoError(t, err) - require.NoError(t, rw.Create(ctx, - NewWorker(scope.Global.String(), WithPublicId(id)))) - require.NoError(t, rw.Create(ctx, - NewWorkerStatus(id, WithAddress("address")))) + ws := NewWorkerForStatus(scope.Global.String(), + WithAddress("address"), + WithName(strings.ToLower(id))) + ws.PublicId = id + require.NoError(t, rw.Create(ctx, ws)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, Key: "key", @@ -212,17 +211,455 @@ func TestWorkerAggregate(t *testing.T) { })) got := getAggWorker(id) - require.NotNil(t, got.ReportedStatus) - assert.ElementsMatch(t, got.Tags, []*Tag{ + require.NotNil(t, got.GetLastStatusTime()) + assert.ElementsMatch(t, got.apiTags, []*Tag{ {Key: "key", Value: "val2"}, {Key: "key2", Value: "val2"}, }) - assert.ElementsMatch(t, got.ReportedStatus.Tags, []*Tag{ + assert.ElementsMatch(t, got.configTags, []*Tag{ {Key: "key", Value: "val"}, }) } } +func TestWorker_Update(t *testing.T) { + t.Parallel() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + ctx := context.Background() + + newId := func() string { + id, err := newWorkerId(ctx) + require.NoError(t, err) + return id + } + + cases := []struct { + name string + initial Worker + update Worker + mask []string + nullMask []string + assert func(t *testing.T, init, up *Worker) + wantUpdateError bool + }{ + { + name: "base update with base", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Description: "base update with base", + }, + }, + update: Worker{ + Worker: &store.Worker{ + Description: "base update with base updated", + }, + }, + mask: []string{"Description"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "base update with base updated", up.Description) + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, uint32(2), up.Version) + assert.Nil(t, up.GetLastStatusTime()) + }, + }, + { + name: "base update with worker reported address and name", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Description: "base update with status", + }, + }, + update: Worker{ + Worker: &store.Worker{ + WorkerReportedName: "base update with worker reported address and name", + WorkerReportedAddress: "base update with worker reported address and name", + }, + }, + mask: []string{"WorkerReportedAddress", "WorkerReportedName"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "base update with worker reported address and name", up.WorkerReportedAddress) + assert.Equal(t, "base update with worker reported address and name", up.WorkerReportedName) + assert.Equal(t, uint32(1), up.Version) + assert.NotNil(t, up.GetLastStatusTime()) + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, up.GetLastStatusTime().AsTime(), up.GetUpdateTime().AsTime()) + }, + }, + { + name: "base update with worker reported address", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Description: "base update with status", + }, + }, + update: Worker{ + Worker: &store.Worker{ + WorkerReportedAddress: "base update with worker reported address", + }, + }, + mask: []string{"WorkerReportedAddress"}, + wantUpdateError: true, + }, + { + // If any status fields are set then worker reported address must + // be set. + name: "base update with worker reported name", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Description: "base update with status", + }, + }, + update: Worker{ + Worker: &store.Worker{ + WorkerReportedName: "base update with worker reported name", + }, + }, + mask: []string{"WorkerReportedName"}, + wantUpdateError: true, + }, + { + name: "base update with worker reported fields", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Description: "base update with status", + }, + }, + update: Worker{ + Worker: &store.Worker{ + WorkerReportedName: "base update with worker reported fields", + WorkerReportedAddress: "base update with worker reported fields", + }, + }, + mask: []string{"WorkerReportedName", "WorkerReportedAddress"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "base update with worker reported fields", up.WorkerReportedAddress) + assert.Equal(t, "base update with worker reported fields", up.WorkerReportedName) + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, uint32(1), up.Version) + assert.NotNil(t, up.GetLastStatusTime()) + assert.Equal(t, up.GetLastStatusTime().AsTime(), up.GetUpdateTime().AsTime()) + }, + }, + { + name: "worker reported update with base", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedAddress: "worker reported update with base", + WorkerReportedName: "worker reported update with base", + }, + }, + update: Worker{ + Worker: &store.Worker{ + Description: "worker reported update with base", + }, + }, + mask: []string{"Description"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "worker reported update with base", up.Description) + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, uint32(2), up.Version) + assert.NotNil(t, up.GetLastStatusTime()) + assert.Equal(t, init.GetLastStatusTime(), up.GetLastStatusTime()) + }, + }, + { + name: "worker reported update name to unprintable", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedAddress: "worker reported update name to unprintable", + WorkerReportedName: "worker reported update name to unprintable", + }, + }, + update: Worker{ + Worker: &store.Worker{ + WorkerReportedName: "unprintable \u0008 name", + }, + }, + mask: []string{"WorkerReportedName"}, + wantUpdateError: true, + }, + { + name: "worker reported clearing address", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedAddress: "worker reported clearing address", + WorkerReportedName: "worker reported clearing address", + }, + }, + update: Worker{ + Worker: &store.Worker{}, + }, + nullMask: []string{"WorkerReportedAddress"}, + wantUpdateError: true, + }, + { + name: "worker reported updating worker reported", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedAddress: "start", + WorkerReportedName: "start", + }, + }, + update: Worker{ + Worker: &store.Worker{ + WorkerReportedAddress: "worker reported updating worker reported", + }, + }, + mask: []string{"WorkerReportedAddress"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "worker reported updating worker reported", up.GetWorkerReportedAddress()) + assert.NotNil(t, up.GetLastStatusTime()) + assert.Greater(t, up.GetLastStatusTime().AsTime(), init.GetLastStatusTime().AsTime()) + // We don't modify the worker version while operating only on the worker fields + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, uint32(1), up.Version) + }, + }, + { + name: "worker reported clearing worker reported name", + initial: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedAddress: "worker reported clearing worker reported name", + WorkerReportedName: "worker reported clearing worker reported name", + }, + }, + update: Worker{ + Worker: &store.Worker{}, + }, + nullMask: []string{"WorkerReportedName"}, + wantUpdateError: true, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + init := tc.initial.clone() + require.NoError(t, rw.Create(ctx, init)) + up := tc.update.clone() + up.PublicId = init.PublicId + _, err := rw.Update(ctx, up, tc.mask, tc.nullMask) + if tc.wantUpdateError { + assert.Error(t, err) + return + } else { + require.NoError(t, err) + } + tc.assert(t, init.clone(), up.clone()) + }) + } +} + +func TestWorker_Create(t *testing.T) { + t.Parallel() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + ctx := context.Background() + + newId := func() string { + id, err := newWorkerId(ctx) + require.NoError(t, err) + return id + } + + type wanted struct { + lastStatusUpdated bool + createError bool + } + + cases := []struct { + name string + in Worker + want wanted + }{ + { + name: "base", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + }, + }, + want: wanted{}, + }, + { + name: "with non status fields", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "with non status fields", + Description: "with non status fields", + Address: "address", + }, + }, + want: wanted{}, + }, + { + name: "with upper case worker reported name", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedName: "With Upper Case Worker Reported Name", + WorkerReportedAddress: "address", + }, + }, + want: wanted{createError: true}, + }, + { + name: "with unprintable worker reported name", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedName: "unprintable \u0008 worker reported name", + WorkerReportedAddress: "address", + }, + }, + want: wanted{createError: true}, + }, + { + name: "with status fields", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedName: "with non status fields", + WorkerReportedAddress: "with non status fields", + }, + }, + want: wanted{ + lastStatusUpdated: true, + }, + }, + { + name: "with worker reported address", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedAddress: "with worker reported address", + }, + }, + want: wanted{ + createError: true, + }, + }, + { + // The worker reported address is a required field if any of the + // worker reported fields are set. + name: "with worker reported name", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + WorkerReportedName: "with worker reported name", + }, + }, + want: wanted{ + createError: true, + }, + }, + { + name: "non status fields with worker reported fields", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "non status fields with worker reported fields", + Description: "non status fields with worker reported fields", + Address: "address", + WorkerReportedAddress: "non status fields with worker reported fields", + WorkerReportedName: "non status fields with worker reported fields", + }, + }, + want: wanted{ + lastStatusUpdated: true, + }, + }, + { + // The worker reported address is a required field if any of the + // worker reported fields are set. + name: "non status fields with worker reported name", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "non status fields with worker reported name", + Description: "non status fields with worker reported name", + Address: "address", + WorkerReportedName: "non status fields with worker reported name", + }, + }, + want: wanted{ + createError: true, + }, + }, + { + name: "non status fields with worker reported address", + in: Worker{ + Worker: &store.Worker{ + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "non status fields with worker reported address", + Description: "non status fields with worker reported address", + Address: "address", + WorkerReportedAddress: "non status fields with worker reported address", + }, + }, + want: wanted{ + createError: true, + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + w := tc.in.clone() + err := rw.Create(ctx, w) + if tc.want.createError { + assert.Error(t, err) + return + } else { + assert.NoError(t, err) + } + + if tc.want.lastStatusUpdated { + assert.NotNil(t, w.GetLastStatusTime()) + } else { + assert.Nil(t, w.GetLastStatusTime()) + } + + assert.NotNil(t, w.GetCreateTime()) + assert.Equal(t, w.GetCreateTime(), w.GetUpdateTime()) + }) + } +} + func TestWorker_New(t *testing.T) { t.Parallel() conn, _ := db.TestSetup(t, "postgres") @@ -285,21 +722,6 @@ func TestWorker_New(t *testing.T) { }, wantCreateErr: true, }, - { - name: "with public id", - args: args{ - scopeId: scope.Global.String(), - opts: []Option{ - WithPublicId("w_test_public_id"), - }, - }, - want: &Worker{ - Worker: &store.Worker{ - PublicId: "w_test_public_id", - ScopeId: scope.Global.String(), - }, - }, - }, { name: "with name", args: args{ @@ -360,7 +782,7 @@ func TestWorker_New(t *testing.T) { Worker: &store.Worker{ ScopeId: scope.Global.String(), }, - Tags: []*Tag{ + apiTags: []*Tag{ { Key: "key", Value: "val", diff --git a/internal/session/query.go b/internal/session/query.go index ae68dcece..3e7a3e3b1 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -327,13 +327,10 @@ where dead_workers (worker_id, last_update_time) as ( select w.public_id as worker_id, - ws.update_time as last_update_time + w.last_status_time as last_update_time from server_worker w - left join server_worker_status ws - on w.public_id = ws.worker_id where - ws.update_time is null - or ws.update_time < wt_sub_seconds_from_now(@grace_period_seconds) + w.last_status_time < wt_sub_seconds_from_now(@grace_period_seconds) ), closed_connections (connection_id, worker_id) as ( update session_connection @@ -343,7 +340,7 @@ where returning public_id, worker_id ) select closed_connections.worker_id, - coalesce(dead_workers.last_update_time, current_timestamp) as last_update_time, + dead_workers.last_update_time as last_update_time, count(closed_connections.connection_id) as number_connections_closed from closed_connections join dead_workers diff --git a/internal/session/service_worker_status_report_test.go b/internal/session/service_worker_status_report_test.go index e524a7458..706c2db56 100644 --- a/internal/session/service_worker_status_report_test.go +++ b/internal/session/service_worker_status_report_test.go @@ -14,6 +14,7 @@ import ( "github.com/hashicorp/boundary/internal/session" "github.com/hashicorp/boundary/internal/target" "github.com/hashicorp/boundary/internal/target/tcp" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -31,8 +32,7 @@ func TestWorkerStatusReport(t *testing.T) { PrivateId: "test_controller1", Address: "127.0.0.1", }) - serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus( - "test_worker1", + serverRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), servers.WithAddress("127.0.0.1"))) repo, err := session.NewRepository(rw, rw, kms) diff --git a/internal/session/session_cleanup_job_test.go b/internal/session/session_cleanup_job_test.go index d7bb5887f..b261506dc 100644 --- a/internal/session/session_cleanup_job_test.go +++ b/internal/session/session_cleanup_job_test.go @@ -12,6 +12,7 @@ import ( "github.com/hashicorp/boundary/internal/kms" "github.com/hashicorp/boundary/internal/scheduler" "github.com/hashicorp/boundary/internal/servers" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -108,9 +109,10 @@ func TestSessionConnectionCleanupJob(t *testing.T) { // Push an upsert to the first worker so that its status has been // updated. - _, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, worker1.ReportedStatus) + _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker1.GetWorkerReportedName()), + servers.WithAddress(worker1.GetWorkerReportedAddress()))) require.NoError(err) - require.Equal(1, rowsUpdated) // Run the job. require.NoError(job.Run(ctx)) @@ -168,61 +170,6 @@ func TestSessionConnectionCleanupJobNewJobErr(t *testing.T) { require.Nil(job) } -func TestCloseConnections_ManuallyDeletedConfig(t *testing.T) { - t.Parallel() - ctx := context.Background() - require := require.New(t) - conn, _ := db.TestSetup(t, "postgres") - rw := db.New(conn) - wrapper := db.TestWrapper(t) - iamRepo := iam.TestRepo(t, conn, wrapper) - kms := kms.TestKms(t, conn, wrapper) - repo, err := NewRepository(rw, rw, kms) - require.NoError(err) - connRepo, err := NewConnectionRepository(ctx, rw, rw, kms) - require.NoError(err) - serversRepo, err := servers.NewRepository(rw, rw, kms) - require.NoError(err) - - gracePeriod := 1 * time.Second - job, err := newSessionConnectionCleanupJob(rw, deadWorkerConnCloseMinGrace) - require.NoError(err) - - worker := servers.TestWorker(t, conn, wrapper) - sess := TestDefaultSession(t, conn, wrapper, iamRepo, WithDbOpts(db.WithSkipVetForWrite(true))) - sess, _, err = repo.ActivateSession(ctx, sess.GetPublicId(), sess.Version, []byte("foo")) - require.NoError(err) - c, cs, err := connRepo.AuthorizeConnection(ctx, sess.GetPublicId(), worker.GetPublicId()) - require.NoError(err) - require.Len(cs, 1) - require.Equal(StatusAuthorized, cs[0].Status) - - _, cs, err = connRepo.ConnectConnection(ctx, ConnectWith{ - ConnectionId: c.GetPublicId(), - ClientTcpAddress: "127.0.0.1", - ClientTcpPort: 22, - EndpointTcpAddress: "127.0.0.1", - EndpointTcpPort: 22, - UserClientIp: "127.0.0.1", - }) - - // The single worker's connections aren't cleaned up - _, _, err = serversRepo.UpsertWorkerStatus(ctx, worker.ReportedStatus) - assert.NoError(t, err) - result, err := job.closeConnectionsForDeadWorkers(ctx, gracePeriod) - require.NoError(err) - require.Empty(result) - - // the manually deleted status causes the connections to be cleaned - _, _, err = serversRepo.UpsertWorkerStatus(ctx, worker.ReportedStatus) - assert.NoError(t, err) - _, err = rw.Delete(ctx, worker.ReportedStatus) - assert.NoError(t, err) - result, err = job.closeConnectionsForDeadWorkers(ctx, gracePeriod) - require.NoError(err) - require.Len(result, 1) -} - func TestCloseConnectionsForDeadWorkers(t *testing.T) { t.Parallel() ctx := context.Background() @@ -345,21 +292,13 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { // the most up-to-date fields. updateServer := func(t *testing.T, w *servers.Worker) *servers.Worker { t.Helper() - _, rowsUpdated, err := serversRepo.UpsertWorkerStatus(ctx, w.ReportedStatus) + // The worker provided name will derive the WorkerId every time + // UpsertWorkerStatus is called. Setting the Worker Id on the caller + // side is an error. + w.PublicId = "" + wkr, err := serversRepo.UpsertWorkerStatus(ctx, w) require.NoError(err) - require.Equal(1, rowsUpdated) - servers, err := serversRepo.ListWorkers(ctx) - require.NoError(err) - for _, server := range servers { - if server.PublicId == w.PublicId { - return server - } - } - - require.FailNowf("server %q not found after updating", w.PublicId) - // Looks weird but needed to build, as we fail in testify instead - // of returning an error - return nil + return wkr } // requireConnectionStatus is a helper expecting all connections on a worker @@ -454,7 +393,7 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { require.Equal([]closeConnectionsForDeadWorkersResult{ { WorkerId: worker1.PublicId, - LastUpdateTime: timestampPbAsUTC(t, worker1.LastConnectionUpdate().AsTime()), + LastUpdateTime: timestampPbAsUTC(t, worker1.GetLastStatusTime().AsTime()), NumberConnectionsClosed: 12, // 18 per server, with 6 closed already }, }, result) @@ -478,12 +417,12 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { require.ElementsMatch([]closeConnectionsForDeadWorkersResult{ { WorkerId: worker2.PublicId, - LastUpdateTime: timestampPbAsUTC(t, worker2.LastConnectionUpdate().AsTime()), + LastUpdateTime: timestampPbAsUTC(t, worker2.GetLastStatusTime().AsTime()), NumberConnectionsClosed: 12, // 18 per server, with 6 closed already }, { WorkerId: worker3.PublicId, - LastUpdateTime: timestampPbAsUTC(t, worker3.LastConnectionUpdate().AsTime()), + LastUpdateTime: timestampPbAsUTC(t, worker3.GetLastStatusTime().AsTime()), NumberConnectionsClosed: 12, // 18 per server, with 6 closed already }, }, result) diff --git a/testing/dbtest/session_list_benchmarks_dump_generation_test.go b/testing/dbtest/session_list_benchmarks_dump_generation_test.go index d09986b9e..daccd5945 100644 --- a/testing/dbtest/session_list_benchmarks_dump_generation_test.go +++ b/testing/dbtest/session_list_benchmarks_dump_generation_test.go @@ -22,6 +22,7 @@ import ( "github.com/hashicorp/boundary/internal/session" "github.com/hashicorp/boundary/internal/target" "github.com/hashicorp/boundary/internal/target/tcp" + "github.com/hashicorp/boundary/internal/types/scope" "github.com/hashicorp/boundary/testing/dbtest" "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" @@ -131,8 +132,7 @@ func TestGenerateSessionBenchmarkTemplateDumps(t *testing.T) { require.NoError(err) serversRepo, err := servers.NewRepository(rw, rw, kms) require.NoError(err) - _, _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerStatus( - "test_worker_1", + _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), servers.WithAddress("127.0.0.1"))) require.NoError(err)