diff --git a/internal/cmd/commands/server/worker_tags_reload_test.go b/internal/cmd/commands/server/worker_tags_reload_test.go index 74bc29f01..0fbd9a13f 100644 --- a/internal/cmd/commands/server/worker_tags_reload_test.go +++ b/internal/cmd/commands/server/worker_tags_reload_test.go @@ -106,7 +106,7 @@ func TestServer_ReloadWorkerTags(t *testing.T) { t.Helper() serversRepo, err := testController.Controller().ServersRepoFn() require.NoError(err) - w, err := serversRepo.LookupWorkerByWorkerReportedName(testController.Context(), name) + w, err := serversRepo.LookupWorkerByName(testController.Context(), name) require.NoError(err) require.NotNil(w) v, ok := w.CanonicalTags()[key] diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index d1badb5a3..3b2810754 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -65,7 +65,9 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques } switch { case wStat.GetName() == "" && wStat.GetKeyId() == "": - return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Name and keyId are not set in the request; at least one is required.") + return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Name and keyId are not set in the request; one is required.") + case wStat.GetName() != "" && wStat.GetKeyId() != "": + return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Name and keyId are both set in the request; only one is allowed.") case wStat.GetAddress() == "": return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Address is not set but is required.") } @@ -88,17 +90,17 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques }) } - wKeyId := wStat.GetKeyId() - - wConf := servers.NewWorkerForStatus(scope.Global.String(), + wConf := servers.NewWorker(scope.Global.String(), servers.WithName(wStat.GetName()), servers.WithAddress(wStat.GetAddress()), - servers.WithWorkerTags(workerTags...), - servers.WithKeyId(wKeyId)) + servers.WithWorkerTags(workerTags...)) opts := []servers.Option{servers.WithUpdateTags(req.GetUpdateTags())} if wStat.GetPublicId() != "" { opts = append(opts, servers.WithPublicId(wStat.GetPublicId())) } + if wStat.GetKeyId() != "" { + opts = append(opts, servers.WithKeyId(wStat.GetKeyId())) + } wrk, err := serverRepo.UpsertWorkerStatus(ctx, wConf, opts...) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status")) @@ -229,7 +231,7 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku } var w *servers.Worker if req.WorkerKeyId != "" { - workerId, err := serversRepo.LookupWorkerIdByWorkerReportedKeyId(ctx, req.WorkerKeyId) + workerId, err := serversRepo.LookupWorkerIdByKeyId(ctx, req.WorkerKeyId) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up worker by keyId", "keyId", req.WorkerKeyId)) return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up tags for server: %v", err) @@ -240,7 +242,7 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up tags for server: %v", err) } } else { - w, err = serversRepo.LookupWorkerByWorkerReportedName(ctx, req.GetServerId()) + w, err = serversRepo.LookupWorkerByName(ctx, req.GetServerId()) if err != nil { 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) @@ -255,12 +257,8 @@ 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": name, + "name": w.GetName(), "tags": tagMap, } ok, err := eval.Evaluate(filterInput) @@ -391,7 +389,7 @@ func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs if err != nil { return nil, status.Errorf(codes.Internal, "error getting servers repo: %v", err) } - w, err := serversRepo.LookupWorkerByWorkerReportedName(ctx, req.GetWorkerId()) + w, err := serversRepo.LookupWorkerByName(ctx, req.GetWorkerId()) if err != nil { return nil, status.Errorf(codes.Internal, "error looking up worker: %v", err) } diff --git a/internal/daemon/cluster/handlers/worker_service_status_test.go b/internal/daemon/cluster/handlers/worker_service_status_test.go index 144a753b2..1500eab42 100644 --- a/internal/daemon/cluster/handlers/worker_service_status_test.go +++ b/internal/daemon/cluster/handlers/worker_service_status_test.go @@ -71,7 +71,7 @@ func TestStatus(t *testing.T) { target.WithSessionConnectionLimit(-1), ) - worker1 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, @@ -257,7 +257,7 @@ func TestStatusSessionClosed(t *testing.T) { target.WithSessionConnectionLimit(-1), ) - worker1 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, @@ -408,7 +408,7 @@ func TestStatusDeadConnection(t *testing.T) { Address: "127.0.0.1", }) - worker1 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) serversRepoFn := func() (*servers.Repository, error) { return serverRepo, nil @@ -578,7 +578,7 @@ func TestStatusWorkerWithKeyId(t *testing.T) { target.WithSessionConnectionLimit(-1), ) - worker1 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) rootStorage, err := servers.NewRepositoryStorage(ctx, rw, rw, kms) require.NoError(t, err) diff --git a/internal/daemon/controller/handlers/targets/target_service.go b/internal/daemon/controller/handlers/targets/target_service.go index ad2c33956..787e50cd7 100644 --- a/internal/daemon/controller/handlers/targets/target_service.go +++ b/internal/daemon/controller/handlers/targets/target_service.go @@ -2159,15 +2159,8 @@ func (w workerList) workerInfos() []*pb.WorkerInfo { func (w workerList) filtered(eval *bexpr.Evaluator) (workerList, error) { var ret []*servers.Worker for _, worker := range w { - - // TODO: Decide how we are actually going to filter by name now that - // we are separating reported status name and the resource name. - name := worker.GetName() - if name == "" { - name = worker.GetWorkerReportedName() - } filterInput := map[string]interface{}{ - "name": name, + "name": worker.GetName(), "tags": worker.CanonicalTags(), } ok, err := eval.Evaluate(filterInput) diff --git a/internal/daemon/controller/handlers/targets/tcp/target_service_test.go b/internal/daemon/controller/handlers/targets/tcp/target_service_test.go index 940963743..cb0f2a8a1 100644 --- a/internal/daemon/controller/handlers/targets/tcp/target_service_test.go +++ b/internal/daemon/controller/handlers/targets/tcp/target_service_test.go @@ -2784,7 +2784,7 @@ func TestAuthorizeSession(t *testing.T) { require.NoError(t, err) // Tell our DB that there is a worker ready to serve the data - servers.TestWorker(t, conn, wrapper) + servers.TestKmsWorker(t, conn, wrapper) asRes1, err := s.AuthorizeSession(ctx, &pbs.AuthorizeSessionRequest{ Id: tar.GetPublicId(), @@ -3062,7 +3062,7 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) { require.NoError(t, err) // Tell our DB that there is a worker ready to serve the data - servers.TestWorker(t, conn, wrapper) + servers.TestKmsWorker(t, conn, wrapper) asRes, err := s.AuthorizeSession(ctx, &pbs.AuthorizeSessionRequest{ Id: tar.GetPublicId(), diff --git a/internal/daemon/controller/handlers/workers/worker_service.go b/internal/daemon/controller/handlers/workers/worker_service.go index 52ea12c3b..a98abf7a7 100644 --- a/internal/daemon/controller/handlers/workers/worker_service.go +++ b/internal/daemon/controller/handlers/workers/worker_service.go @@ -457,7 +457,7 @@ func toProto(ctx context.Context, in *servers.Worker, opt ...handlers.Option) (* out.AuthorizedActions = opts.WithAuthorizedActions if in.Type == KmsWorkerType { // KMS workers cannot be updated through the API - out.AuthorizedActions = strutil.StrListDelete(action.Update.String()) + out.AuthorizedActions = strutil.StrListDelete(out.AuthorizedActions, action.Update.String()) } } if outputFields.Has(globals.AddressField) && in.GetAddress() != "" { diff --git a/internal/daemon/controller/handlers/workers/worker_service_test.go b/internal/daemon/controller/handlers/workers/worker_service_test.go index a09c0c530..f384e97c7 100644 --- a/internal/daemon/controller/handlers/workers/worker_service_test.go +++ b/internal/daemon/controller/handlers/workers/worker_service_test.go @@ -63,7 +63,7 @@ func TestGet(t *testing.T) { return repo, nil } - worker := servers.TestWorker(t, conn, wrap, + worker := servers.TestKmsWorker(t, conn, wrap, servers.WithName("test worker names"), servers.WithDescription("test worker description"), servers.WithAddress("test worker address"), @@ -174,7 +174,7 @@ func TestList(t *testing.T) { var wantWorkers []*pb.Worker for i := 0; i < 10; i++ { - w := servers.TestWorker(t, conn, wrap, servers.WithName(fmt.Sprintf("worker%d", i))) + w := servers.TestKmsWorker(t, conn, wrap, servers.WithName(fmt.Sprintf("worker%d", i))) wantWorkers = append(wantWorkers, &pb.Worker{ Id: w.GetPublicId(), ScopeId: w.GetScopeId(), @@ -285,7 +285,7 @@ func TestDelete(t *testing.T) { s, err := NewService(ctx, repoFn, iamRepoFn) require.NoError(t, err, "Error when getting new worker service.") - w := servers.TestWorker(t, conn, wrap) + w := servers.TestKmsWorker(t, conn, wrap) cases := []struct { name string @@ -349,7 +349,7 @@ func TestUpdate(t *testing.T) { return repo, nil } - wkr := servers.TestWorker(t, conn, wrapper, + wkr := servers.TestKmsWorker(t, conn, wrapper, servers.WithName("default"), servers.WithDescription("default"), servers.WithAddress("default")) @@ -806,7 +806,7 @@ func TestUpdate_BadVersion(t *testing.T) { workerService, err := NewService(ctx, repoFn, iamRepoFn) require.NoError(t, err, "Failed to create a new host set service.") - wkr := servers.TestWorker(t, conn, wrapper) + wkr := servers.TestKmsWorker(t, conn, wrapper) upTar, err := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), &pbs.UpdateWorkerRequest{ Id: wkr.GetPublicId(), 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 c0ebbfea6..e67230adb 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 @@ -120,6 +120,7 @@ begin table = tg_table_name, column = 'name'; end if; + return new; end; $$ language plpgsql; comment on function immutable_kms_name is @@ -134,8 +135,8 @@ as $$ begin if new.type = 'kms' then new.last_status_time = now(); - return new; end if; + return new; end; $$ language plpgsql; comment on function update_kms_server_worker_update_last_status_time_column is @@ -150,8 +151,8 @@ as $$ begin if new.type = 'pki' then new.last_status_time = now(); - return new; end if; + return new; end; $$ language plpgsql; comment on function update_pki_server_worker_update_last_status_time_column is @@ -166,8 +167,8 @@ as $$ begin if new.type = 'kms' then new.last_status_time = now(); - return new; end if; + return new; end; $$ language plpgsql; comment on function insert_kms_server_worker_update_last_status_time_column is @@ -176,26 +177,6 @@ comment on function insert_kms_server_worker_update_last_status_time_column is create trigger insert_server_worker_last_update_time_column before insert on server_worker for each row execute procedure insert_kms_server_worker_update_last_status_time_column(); -create function insert_server_worker_type_column() - returns trigger -as $$ -begin - perform from worker_auth_authorized where worker_id = new.public_id; - case - when new.type = 'pki' and not found then - raise exception 'invalid type: pki worker is not authorized in worker_auth_authorized table'; - when new.type = 'kms' and found then - raise exception 'invalid type: kms worker cannot authorized in worker_auth_authorized table'; - end case; - return new; -end; -$$ language plpgsql; -comment on function insert_server_worker_type_column is - 'function used to constraint the type column in server_worker'; - -create trigger insert_server_worker_type_column before insert on server_worker - for each row execute procedure insert_server_worker_type_column(); - -- Create table worker tag create table server_worker_tag_enm ( source text primary key diff --git a/internal/db/schema/migrations/oss/postgres/99/03_worker_authentication.up.sql b/internal/db/schema/migrations/oss/postgres/99/03_worker_authentication.up.sql index 33d71f06e..b6e39a48e 100644 --- a/internal/db/schema/migrations/oss/postgres/99/03_worker_authentication.up.sql +++ b/internal/db/schema/migrations/oss/postgres/99/03_worker_authentication.up.sql @@ -112,6 +112,8 @@ create table worker_auth_authorized( on update cascade, nonce bytea -- encrypted value ); +comment on table worker_auth_authorized is + 'worker_auth_authorized is a table where each row represents key and cert data associated with an authorized worker.'; create trigger immutable_columns @@ -120,8 +122,22 @@ create trigger for each row execute procedure immutable_columns('worker_key_identifier', 'worker_id', 'worker_signing_pub_key', 'worker_encryption_pub_key', 'controller_encryption_priv_key', 'key_id', 'nonce'); -comment on table worker_auth_authorized is - 'worker_auth_authorized is a table where each row represents key and cert data associated with an authorized worker.'; +create function insert_worker_auth_authorized_type_check() + returns trigger +as $$ +begin + perform from server_worker where type = 'pki' and public_id = new.worker_id; + if not found then + raise exception 'invalid type: non pki worker cannot be authorized in worker_auth_authorized table'; + end if; + return new; +end; +$$ language plpgsql; +comment on function insert_worker_auth_authorized_type_check is + 'function used to only allow workers of type pki to be authorized'; + +create trigger insert_worker_auth_authorized_type_check before insert on worker_auth_authorized + for each row execute procedure insert_worker_auth_authorized_type_check(); create table worker_auth_certificate_bundle( root_certificate_public_key bytea not null diff --git a/internal/db/schema/migrations/oss/postgres/99/04_views.up.sql b/internal/db/schema/migrations/oss/postgres/99/04_views.up.sql index a323a08fc..c67d3ee4a 100644 --- a/internal/db/schema/migrations/oss/postgres/99/04_views.up.sql +++ b/internal/db/schema/migrations/oss/postgres/99/04_views.up.sql @@ -30,6 +30,7 @@ select w.update_time, w.version, w.last_status_time, + w.type, cc.count as active_connection_count, -- keys and tags can be any lowercase printable character so use uppercase characters as delimitors. wt.tags as api_tags, @@ -44,7 +45,6 @@ from server_worker w comment on view server_worker_aggregate is 'server_worker_aggregate contains the worker resource with its worker provided config values and its configuration and api provided tags.'; - -- Replaces the view created in 9/01. -- Remove the worker id from this view. In actuality this is almost a no-op -- because no server information was ever getting populated here due to a bug diff --git a/internal/servers/public_ids.go b/internal/servers/public_ids.go index 16d9a7829..0658c36f2 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 } + +// newWorkerIdFromScopeAndName generates a predictable public id based on the +// scope and the worker name. This should only be used on kms workers at +// upsert time. +func newWorkerIdFromScopeAndName(ctx context.Context, scope, name string) (string, error) { + const op = "servers.newWorkerIdFromScopeAndName" + id, err := db.NewPublicId(WorkerPrefix, db.WithPrngValues([]string{scope, name})) + if err != nil { + return "", errors.Wrap(ctx, err, "servers.newWorkerId") + } + return id, nil +} diff --git a/internal/servers/repository_worker.go b/internal/servers/repository_worker.go index 836509f3f..0105de633 100644 --- a/internal/servers/repository_worker.go +++ b/internal/servers/repository_worker.go @@ -57,19 +57,32 @@ 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" +// LookupWorkerByName returns the worker with the provided name. In the event +// that no worker is found that matches then nil, nil will be returned. +func (r *Repository) LookupWorkerByName(ctx context.Context, name string) (*Worker, error) { + const op = "servers.(Repository).LookupWorkerByName" switch { case name == "": return nil, errors.New(ctx, errors.InvalidParameter, op, "name is empty") } + w, err := lookupWorkerByName(ctx, r.reader, name) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + return w, nil +} + +func lookupWorkerByName(ctx context.Context, reader db.Reader, name string) (*Worker, error) { + const op = "servers.lookupWorkerByName" + switch { + case isNil(reader): + return nil, errors.New(ctx, errors.InvalidParameter, op, "reader is nil") + case name == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "name is empty") + } wAgg := &workerAggregate{} - err := r.reader.LookupWhere(ctx, &wAgg, "worker_reported_name = ?", []interface{}{name}) + err := reader.LookupWhere(ctx, &wAgg, "name = ?", []interface{}{name}) if err != nil { if errors.IsNotFoundError(err) { return nil, nil @@ -83,8 +96,8 @@ func (r *Repository) LookupWorkerByWorkerReportedName(ctx context.Context, name return w, nil } -func (r *Repository) LookupWorkerIdByWorkerReportedKeyId(ctx context.Context, keyId string) (string, error) { - const op = "servers.(Repository).LookupWorkerIdByWorkerReportedKeyId" +func (r *Repository) LookupWorkerIdByKeyId(ctx context.Context, keyId string) (string, error) { + const op = "servers.(Repository).LookupWorkerIdByKeyId" switch { case keyId == "": return "", errors.New(ctx, errors.InvalidParameter, op, "keyId is empty") @@ -200,62 +213,51 @@ func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ... return workers, nil } -// 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 WithPublicId and WithUpdateTags options are the only ones used. All others are ignored. +// UpsertWorkerStatus will update the address and last status time for a worker. +// If the worker is a kms worker that hasn't been seen yet, it'll attempt to +// create a new one, but will return an error if another worker (kms or other) +// has the same name. This returns the Worker object with the changes applied. +// The WithPublicId, WithKeyId, and WithUpdateTags options are +// the only ones used. All others are ignored. // Workers are intentionally not oplogged. func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt ...Option) (*Worker, error) { const op = "servers.UpsertWorkerStatus" + opts := getOpts(opt...) switch { case worker == nil: return nil, errors.New(ctx, errors.InvalidParameter, op, "worker is nil") - case worker.GetWorkerReportedAddress() == "": + case worker.GetAddress() == "": 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() == "" && worker.WorkerReportedKeyId == "": - return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both empty; at least one is required") - case len(worker.apiTags) > 0: - return nil, errors.New(ctx, errors.InvalidParameter, op, "api tags is not empty") + case worker.GetName() == "" && opts.withKeyId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both empty; one is required") + case worker.GetName() != "" && opts.withKeyId != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "worker keyId and reported name are both set; no more than one is allowed") } - // Only retain the worker reported fields. - worker = NewWorkerForStatus(worker.GetScopeId(), - WithName(worker.GetWorkerReportedName()), - WithAddress(worker.GetWorkerReportedAddress()), - WithWorkerTags(worker.configTags...), - WithKeyId(worker.WorkerReportedKeyId)) - - opts := getOpts(opt...) - + var workerId string var err error switch { case opts.withPublicId != "": - worker.PublicId = opts.withPublicId - case worker.WorkerReportedKeyId != "": - workerId, err := r.LookupWorkerIdByWorkerReportedKeyId(ctx, worker.WorkerReportedKeyId) + workerId = opts.withPublicId + case opts.withKeyId != "": + workerId, err = r.LookupWorkerIdByKeyId(ctx, opts.withKeyId) if err != nil || workerId == "" { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error finding worker by keyId")) } - worker.PublicId = workerId default: - // Attempt to find workerId by name. If not found, generate an id - w, err := r.LookupWorkerByWorkerReportedName(ctx, worker.GetWorkerReportedName()) - if err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error finding worker by name")) - } - if w != nil { - worker.PublicId = w.PublicId - } else { - workerId, err := newWorkerId(ctx) - if err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error error generating worker id")) - } - worker.PublicId = workerId + // generating the worker id based off of the scope and name ensures + // that if 2 kms workers are making requests with the same name they + // are treated as the same worker. Allowing this only for kms workers + // also ensures that we maintain the unique name constraint between pki + // workers and kms workers. + workerId, err = newWorkerIdFromScopeAndName(ctx, worker.GetScopeId(), worker.GetName()) + if err != nil || workerId == "" { + return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker id")) } } @@ -266,13 +268,31 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { worker := worker.clone() - workerCreateConflict := &db.OnConflict{ - Target: db.Columns{"public_id"}, - 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")) + worker.PublicId = workerId + switch { + case worker.GetName() != "": + worker.Type = KmsWorkerType.String() + workerCreateConflict := &db.OnConflict{ + Target: db.Columns{"public_id"}, + Action: append(db.SetColumns([]string{"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")) + } + case opts.withKeyId != "": + n, err := w.Update(ctx, worker, []string{"address"}, nil) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update status of pki worker")) + } + switch n { + case 0: + return errors.New(ctx, errors.RecordNotFound, op, fmt.Sprintf("failed to find worker with key id %q", opts.withKeyId)) + case 1: + break + default: + return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("multiple records found when updating worker with id %q", worker.GetPublicId())) + } } // If we've been told to update tags, we need to clean out old @@ -280,7 +300,7 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt // delete all tags for the given worker, then add the new ones // we've been sent. if opts.withUpdateTags { - setWorkerTags(ctx, w, worker.GetPublicId(), ConfigurationTagSource, worker.configTags) + setWorkerTags(ctx, w, worker.GetPublicId(), ConfigurationTagSource, worker.inputTags) } wAgg := &workerAggregate{PublicId: worker.GetPublicId()} @@ -355,9 +375,8 @@ func setWorkerTags(ctx context.Context, w db.Writer, id string, ts TagSource, ta // the path then an error is returned. func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version uint32, fieldMaskPaths []string, opt ...Option) (*Worker, int, error) { const ( - nameField = "name" - descField = "description" - addressField = "address" + nameField = "name" + descField = "description" ) const op = "servers.(Repository).UpdateWorker" switch { @@ -373,7 +392,6 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u switch { case strings.EqualFold(nameField, f): case strings.EqualFold(descField, f): - case strings.EqualFold(addressField, f): default: return nil, db.NoRowsAffected, errors.New(ctx, errors.InvalidFieldMask, op, fmt.Sprintf("invalid field mask: %s", f)) } @@ -382,9 +400,8 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u var dbMask, nullFields []string dbMask, nullFields = dbcommon.BuildUpdatePaths( map[string]interface{}{ - nameField: worker.Name, - descField: worker.Description, - addressField: worker.Address, + nameField: worker.Name, + descField: worker.Description, }, fieldMaskPaths, nil, @@ -448,10 +465,8 @@ 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.Address != "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "address is not empty") case worker.LastStatusTime != nil: return nil, errors.New(ctx, errors.InvalidParameter, op, "last status time is not nil") } @@ -476,7 +491,6 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op if err != nil { return nil, errors.Wrap(ctx, err, op) } - } var returnedWorker *Worker @@ -486,6 +500,7 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op db.ExpBackoff{}, func(read db.Reader, w db.Writer) error { returnedWorker = worker.clone() + returnedWorker.Type = PkiWorkerType.String() if err := w.Create( ctx, returnedWorker, diff --git a/internal/servers/repository_worker_test.go b/internal/servers/repository_worker_test.go index 73df90403..d5848006b 100644 --- a/internal/servers/repository_worker_test.go +++ b/internal/servers/repository_worker_test.go @@ -54,7 +54,7 @@ func TestDeleteWorker(t *testing.T) { { name: "valid", args: args{ - worker: servers.TestWorker(t, conn, wrapper), + worker: servers.TestKmsWorker(t, conn, wrapper), }, wantRowsDeleted: 1, wantErr: false, @@ -108,7 +108,7 @@ func TestDeleteWorker(t *testing.T) { } } -func TestLookupWorkerByWorkerReportedName(t *testing.T) { +func TestLookupWorkerByName(t *testing.T) { conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) wrapper := db.TestWrapper(t) @@ -117,14 +117,14 @@ func TestLookupWorkerByWorkerReportedName(t *testing.T) { require.NoError(t, err) ctx := context.Background() - w := servers.TestWorker(t, conn, wrapper) + w := servers.TestKmsWorker(t, conn, wrapper) t.Run("success", func(t *testing.T) { - got, err := repo.LookupWorkerByWorkerReportedName(ctx, w.GetWorkerReportedName()) + got, err := repo.LookupWorkerByName(ctx, w.GetName()) 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") + got, err := repo.LookupWorkerByName(ctx, "unknown_name") require.NoError(t, err) assert.Nil(t, got) }) @@ -134,14 +134,14 @@ func TestLookupWorkerByWorkerReportedName(t *testing.T) { 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()) + got, err := r.LookupWorkerByName(ctx, w.GetName()) assert.NoError(t, mock.ExpectationsWereMet()) - assert.Truef(t, errors.Match(errors.T(errors.Op("servers.(Repository).LookupWorkerByWorkerReportedName")), err), "got error %v", err) + assert.Truef(t, errors.Match(errors.T(errors.Op("servers.(Repository).LookupWorkerByName")), err), "got error %v", err) assert.Nil(t, got) }) } -func TestLookupWorkerByWorkerReportedKeyId(t *testing.T) { +func TestLookupWorkerIdByKeyId(t *testing.T) { conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) wrapper := db.TestWrapper(t) @@ -153,7 +153,7 @@ func TestLookupWorkerByWorkerReportedKeyId(t *testing.T) { err = kmsCache.CreateKeys(context.Background(), scope.Global.String(), kms.WithRandomReader(rand.Reader)) require.NoError(t, err) - w := servers.TestWorker(t, conn, wrapper) + w := servers.TestPkiWorker(t, conn, wrapper) rootStorage, err := servers.NewRepositoryStorage(ctx, rw, rw, kmsCache) require.NoError(t, err) @@ -176,12 +176,12 @@ func TestLookupWorkerByWorkerReportedKeyId(t *testing.T) { registeredNode, err := registration.AuthorizeNode(ctx, rootStorage, fetchReq, nodeenrollment.WithState(state)) require.NoError(t, err) t.Run("success", func(t *testing.T) { - got, err := repo.LookupWorkerIdByWorkerReportedKeyId(ctx, registeredNode.Id) + got, err := repo.LookupWorkerIdByKeyId(ctx, registeredNode.Id) require.NoError(t, err) assert.Equal(t, w.PublicId, got) }) t.Run("not found", func(t *testing.T) { - got, err := repo.LookupWorkerIdByWorkerReportedKeyId(ctx, "unknown_key") + got, err := repo.LookupWorkerIdByKeyId(ctx, "unknown_key") require.NoError(t, err) assert.Empty(t, got) }) @@ -191,9 +191,9 @@ func TestLookupWorkerByWorkerReportedKeyId(t *testing.T) { mock.ExpectQuery(`SELECT`).WillReturnError(errors.New(context.Background(), errors.Internal, "test", "lookup-error")) r, err := servers.NewRepository(rw, rw, kmsCache) require.NoError(t, err) - got, err := r.LookupWorkerIdByWorkerReportedKeyId(ctx, w.GetWorkerReportedName()) + got, err := r.LookupWorkerIdByKeyId(ctx, "somekey") assert.NoError(t, mock.ExpectationsWereMet()) - assert.Truef(t, errors.Match(errors.T(errors.Op("servers.(Repository).LookupWorkerIdByWorkerReportedKeyId")), err), "got error %v", err) + assert.Truef(t, errors.Match(errors.T(errors.Op("servers.(Repository).LookupWorkerIdByKeyId")), err), "got error %v", err) assert.Empty(t, got) }) } @@ -207,15 +207,15 @@ func TestLookupWorker(t *testing.T) { require.NoError(t, err) ctx := context.Background() - w := servers.TestWorker(t, conn, wrapper, + w := servers.TestKmsWorker(t, conn, wrapper, servers.WithName("name"), servers.WithDescription("description"), servers.WithAddress("address"), servers.WithWorkerTags(&servers.Tag{"key", "val"})) w, err = repo.UpsertWorkerStatus(context.Background(), servers.NewWorkerForStatus(w.GetScopeId(), - servers.WithName(w.GetWorkerReportedName()), - servers.WithAddress(w.GetWorkerReportedAddress()), + servers.WithName(w.GetName()), + servers.WithAddress(w.GetAddress()), servers.WithWorkerTags(&servers.Tag{ Key: "config", Value: "test", @@ -301,14 +301,18 @@ func TestUpsertWorkerStatus(t *testing.T) { require.NoError(t, err) { assert.True(t, strings.HasPrefix(worker.GetPublicId(), "w_")) - assert.Equal(t, wStatus1.GetWorkerReportedAddress(), worker.CanonicalAddress()) - assert.Empty(t, worker.Name) + assert.Equal(t, wStatus1.GetAddress(), worker.CanonicalAddress()) + assert.Equal(t, "config_name1", worker.Name) assert.Equal(t, worker.GetLastStatusTime(), worker.UpdateTime) assert.Equal(t, uint32(1), worker.Version) - assert.Empty(t, worker.Address) + assert.Equal(t, "address", worker.Address) assert.Empty(t, worker.Description) } + // TODO: Add tests that attempt to upsert a KMS worker where there already + // exists a PKI worker with the same name. It should just error with an + // easy to identify error. + failureCases := []struct { name string repo *servers.Repository @@ -370,21 +374,6 @@ func TestUpsertWorkerStatus(t *testing.T) { 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 { @@ -443,10 +432,10 @@ func TestTagUpdatingListing(t *testing.T) { require.NoError(err) ctx := context.Background() - worker1 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) wStatus := servers.NewWorkerForStatus(scope.Global.String(), - servers.WithName(worker1.GetWorkerReportedName()), - servers.WithAddress("127.0.0.1"), + servers.WithName(worker1.GetName()), + servers.WithAddress("somethingnew"), servers.WithWorkerTags( &servers.Tag{ Key: "tag1", @@ -465,8 +454,8 @@ func TestTagUpdatingListing(t *testing.T) { // Update without saying to update tags wStatus = servers.NewWorkerForStatus(scope.Global.String(), - servers.WithName(worker1.GetWorkerReportedName()), - servers.WithAddress(worker1.GetWorkerReportedAddress()), + servers.WithName(worker1.GetName()), + servers.WithAddress(worker1.GetAddress()), servers.WithWorkerTags( &servers.Tag{ Key: "tag22", @@ -542,7 +531,7 @@ func TestListWorkers(t *testing.T) { w := servers.NewWorker(scope.Global.String()) db.TestDeleteWhere(t, conn, w, "true") for i := 0; i < tt.createCnt; i++ { - servers.TestWorker(t, conn, wrapper) + servers.TestKmsWorker(t, conn, wrapper) } // the purpose of these tests isn't to check liveness, so disable // liveness checking. @@ -569,18 +558,20 @@ func TestListWorkers_WithLiveness(t *testing.T) { require.NoError(err) ctx := context.Background() - worker1 := servers.TestWorker(t, conn, wrapper) - worker2 := servers.TestWorker(t, conn, wrapper) - worker3 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) + worker2 := servers.TestKmsWorker(t, conn, wrapper) + worker3 := servers.TestKmsWorker(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. - _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), - servers.WithName(worker1.GetWorkerReportedName()), - servers.WithAddress(worker1.GetWorkerReportedAddress()))) + _, err = serversRepo.UpsertWorkerStatus(ctx, + servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker1.GetName()), + servers.WithAddress(worker1.GetAddress())), + servers.WithPublicId(worker1.GetPublicId())) require.NoError(err) requireIds := func(expected []string, actual []*servers.Worker) { @@ -605,9 +596,11 @@ func TestListWorkers_WithLiveness(t *testing.T) { requireIds([]string{worker1.GetPublicId()}, result) // Upsert second server. - _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), - servers.WithName(worker2.GetWorkerReportedName()), - servers.WithAddress(worker1.GetWorkerReportedAddress()))) + _, err = serversRepo.UpsertWorkerStatus(ctx, + servers.NewWorkerForStatus(scope.Global.String(), + servers.WithName(worker2.GetName()), + servers.WithAddress(worker2.GetAddress())), + servers.WithPublicId(worker2.GetPublicId())) require.NoError(err) // Static liveness. Should get two, so long as this did not take @@ -739,25 +732,13 @@ func TestRepository_CreateWorker(t *testing.T) { name: "no worker reported address", setup: func() *servers.Worker { w := servers.NewWorker(scope.Global.String()) - w.WorkerReportedAddress = "foo" + w.Address = "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", + wantErrContains: "address is not empty", }, { name: "no last status update", @@ -819,6 +800,7 @@ func TestRepository_CreateWorker(t *testing.T) { name: "success-with-fetch-node-req", setup: func() *servers.Worker { w := servers.NewWorker(scope.Global.String()) + w.Name = "success-with-fetch-node-req" return w }, reader: rw, @@ -893,28 +875,13 @@ func TestRepository_UpdateWorker(t *testing.T) { repo, err := servers.NewRepository(rw, rw, kmsCache) require.NoError(t, err) - setCases := []struct { + pkiCases := []struct { name string modifyWorker func(*testing.T, *servers.Worker) path []string assertGot func(*testing.T, *servers.Worker) wantErr bool }{ - { - name: "update address", - modifyWorker: func(t *testing.T, w *servers.Worker) { - t.Helper() - w.Address = "foo" - }, - path: []string{"address"}, - assertGot: func(t *testing.T, w *servers.Worker) { - t.Helper() - assert.Equal(t, "foo", w.GetAddress()) - assert.Equal(t, uint32(2), w.GetVersion()) - assert.Equal(t, w.GetLastStatusTime().AsTime(), w.GetCreateTime().AsTime()) - assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) - }, - }, { name: "update name", modifyWorker: func(t *testing.T, w *servers.Worker) { @@ -926,7 +893,7 @@ func TestRepository_UpdateWorker(t *testing.T) { t.Helper() assert.Equal(t, "foo", w.GetName()) assert.Equal(t, uint32(2), w.GetVersion()) - assert.Equal(t, w.GetLastStatusTime().AsTime(), w.GetCreateTime().AsTime()) + assert.Nil(t, w.GetLastStatusTime()) assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) }, }, @@ -941,39 +908,18 @@ func TestRepository_UpdateWorker(t *testing.T) { t.Helper() assert.Equal(t, "foo", w.GetDescription()) assert.Equal(t, uint32(2), w.GetVersion()) - assert.Equal(t, w.GetLastStatusTime().AsTime(), w.GetCreateTime().AsTime()) + assert.Nil(t, w.GetLastStatusTime()) assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) }, }, - { - name: "update worker reported name", - modifyWorker: func(t *testing.T, w *servers.Worker) { - t.Helper() - w.WorkerReportedName = "foo" - }, - path: []string{"WorkerReportedName"}, - wantErr: true, - }, - { - name: "update worker reported name and name", - modifyWorker: func(t *testing.T, w *servers.Worker) { - t.Helper() - w.WorkerReportedName = "foo" - w.Name = "another" - }, - path: []string{"WorkerReportedName", "Name"}, - wantErr: true, - }, - { - name: "Clear worker reported name", - modifyWorker: func(t *testing.T, w *servers.Worker) {}, - path: []string{"WorkerReportedName"}, - wantErr: true, - }, } - for _, tt := range setCases { + for _, tt := range pkiCases { t.Run(tt.name, func(t *testing.T) { - wkr := servers.TestWorker(t, conn, wrapper) + wkr := servers.TestPkiWorker(t, conn, wrapper) + defer func() { + _, err := repo.DeleteWorker(ctx, wkr.GetPublicId()) + require.NoError(t, err) + }() tt.modifyWorker(t, wkr) got, _, err := repo.UpdateWorker(ctx, wkr, 1, tt.path) if tt.wantErr { @@ -994,20 +940,6 @@ func TestRepository_UpdateWorker(t *testing.T) { assertGot func(*testing.T, *servers.Worker) wantErr bool }{ - { - name: "clear address", - modifyWorker: func(t *testing.T, w *servers.Worker) { - t.Helper() - w.Address = "" - }, - path: []string{"address"}, - assertGot: func(t *testing.T, w *servers.Worker) { - t.Helper() - assert.Empty(t, w.GetAddress()) - assert.Equal(t, w.GetLastStatusTime().AsTime(), w.GetCreateTime().AsTime()) - assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) - }, - }, { name: "clear name", modifyWorker: func(t *testing.T, w *servers.Worker) { @@ -1018,7 +950,7 @@ func TestRepository_UpdateWorker(t *testing.T) { assertGot: func(t *testing.T, w *servers.Worker) { t.Helper() assert.Empty(t, w.GetName()) - assert.Equal(t, w.GetLastStatusTime().AsTime(), w.GetCreateTime().AsTime()) + assert.Nil(t, w.GetLastStatusTime()) assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) }, }, @@ -1032,18 +964,17 @@ func TestRepository_UpdateWorker(t *testing.T) { assertGot: func(t *testing.T, w *servers.Worker) { t.Helper() assert.Empty(t, w.GetDescription()) - assert.Equal(t, w.GetLastStatusTime().AsTime(), w.GetCreateTime().AsTime()) + assert.Nil(t, w.GetLastStatusTime()) assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) }, }, } for _, tt := range clearCases { t.Run(tt.name, func(t *testing.T) { - wkr := servers.TestWorker(t, conn, wrapper) + wkr := servers.TestPkiWorker(t, conn, wrapper) wkr.Name = tt.name wkr.Description = tt.name - wkr.Address = tt.name - wkr, _, err := repo.UpdateWorker(ctx, wkr, 1, []string{"name", "description", "address"}) + wkr, _, err := repo.UpdateWorker(ctx, wkr, 1, []string{"name", "description"}) require.NoError(t, err) tt.modifyWorker(t, wkr) @@ -1059,9 +990,9 @@ func TestRepository_UpdateWorker(t *testing.T) { } t.Run("version is wrong", func(t *testing.T) { - wkr := servers.TestWorker(t, conn, wrapper) - wkr.Address = "version is wrong" - result, numUpdated, err := repo.UpdateWorker(ctx, wkr, 2, []string{"address"}) + wkr := servers.TestPkiWorker(t, conn, wrapper) + wkr.Name = "version is wrong" + result, numUpdated, err := repo.UpdateWorker(ctx, wkr, 2, []string{"name"}) require.NoError(t, err) assert.Zero(t, numUpdated) assert.Equal(t, wkr.GetUpdateTime().AsTime(), result.GetUpdateTime().AsTime()) @@ -1082,21 +1013,41 @@ func TestRepository_UpdateWorker(t *testing.T) { }, { name: "empty path", - input: servers.TestWorker(t, conn, wrapper), + input: servers.TestPkiWorker(t, conn, wrapper), version: 1, wantErr: errors.T(errors.EmptyFieldMask), }, { name: "0 version", - input: servers.TestWorker(t, conn, wrapper), + input: servers.TestPkiWorker(t, conn, wrapper), path: []string{"name"}, version: 0, wantErr: errors.T(errors.InvalidParameter), }, + { + name: "changing kms name", + input: func() *servers.Worker { + w := servers.TestKmsWorker(t, conn, wrapper) + w.Name = "some change" + return w + }(), path: []string{"name"}, + version: 1, + wantErr: errors.T(errors.NotSpecificIntegrity), + }, + { + name: "clearing kms name", + input: func() *servers.Worker { + w := servers.TestKmsWorker(t, conn, wrapper) + w.Name = "" + return w + }(), path: []string{"name"}, + version: 1, + wantErr: errors.T(errors.NotSpecificIntegrity), + }, { name: "no public id", input: func() *servers.Worker { - w := servers.TestWorker(t, conn, wrapper) + w := servers.TestPkiWorker(t, conn, wrapper) w.PublicId = "" return w }(), @@ -1106,7 +1057,7 @@ func TestRepository_UpdateWorker(t *testing.T) { }, { name: "unrecognized path", - input: servers.TestWorker(t, conn, wrapper), + input: servers.TestPkiWorker(t, conn, wrapper), path: []string{"UnrecognizedField"}, version: 1, wantErr: errors.T(errors.InvalidFieldMask), @@ -1114,7 +1065,7 @@ func TestRepository_UpdateWorker(t *testing.T) { { name: "not found worker", input: func() *servers.Worker { - w := servers.TestWorker(t, conn, wrapper) + w := servers.TestPkiWorker(t, conn, wrapper) w.PublicId = "w_notfoundworker" return w }(), @@ -1125,11 +1076,11 @@ func TestRepository_UpdateWorker(t *testing.T) { { name: "duplicate name", input: func() *servers.Worker { - w1 := servers.TestWorker(t, conn, wrapper) + w1 := servers.TestPkiWorker(t, conn, wrapper) w1.Name = "somenamethatijustmadeup" w1, _, err := repo.UpdateWorker(ctx, w1, w1.Version, []string{"name"}, nil) require.NoError(t, err) - w2 := servers.TestWorker(t, conn, wrapper) + w2 := servers.TestPkiWorker(t, conn, wrapper) w2.Name = w1.Name return w2 }(), diff --git a/internal/servers/repository_workerauth_test.go b/internal/servers/repository_workerauth_test.go index 95b85853d..162eca4b9 100644 --- a/internal/servers/repository_workerauth_test.go +++ b/internal/servers/repository_workerauth_test.go @@ -146,7 +146,7 @@ func TestStoreWorkerAuth(t *testing.T) { require.NoError(err) require.NotNil(wrapper) - worker := TestWorker(t, conn, wrapper) + worker := TestPkiWorker(t, conn, wrapper) rw := db.New(conn) rootStorage, err := NewRepositoryStorage(ctx, rw, rw, kmsCache) @@ -266,7 +266,7 @@ func TestStoreNodeInformationTx(t *testing.T) { // Create struct to pass in with workerId that will be passed along to // storage - testWorker := TestWorker(t, conn, testWrapper) + testWorker := TestPkiWorker(t, conn, testWrapper) testState, err := AttachWorkerIdToState(testCtx, testWorker.PublicId) require.NoError(t, err) diff --git a/internal/servers/testing.go b/internal/servers/testing.go index 9314cc51e..fb0f16dc5 100644 --- a/internal/servers/testing.go +++ b/internal/servers/testing.go @@ -81,30 +81,37 @@ func TestWorkerAuth(ctx context.Context, t *testing.T, conn *db.DB, worker *Work return workerAuth } -// TestWorker inserts a worker into the db to satisfy foreign key constraints. -// The worker provided fields are auto generated. WithName, WithDescription, -// and WithAddress are applied to the resource name, description and address if +// TestKmsWorker inserts a worker into the db to satisfy foreign key constraints. +// The worker provided fields are auto generated. WithName and WithDescription, +// are applied to the resource name, description and address if // present. -func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *Worker { +func TestKmsWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *Worker { t.Helper() rw := db.New(conn) kms := kms.TestKms(t, conn, wrapper) serversRepo, err := NewRepository(rw, rw, kms) require.NoError(t, err) ctx := context.Background() + opts := getOpts(opt...) namePart, err := newWorkerId(ctx) require.NoError(t, err) name := "test-worker-" + strings.ToLower(namePart) - id, err := newWorkerId(ctx) + if opts.withName != "" { + name = opts.withName + } + address := "127.0.0.1" + if opts.withAddress != "" { + address = opts.withAddress + } + id, err := newWorkerIdFromScopeAndName(ctx, scope.Global.String(), name) require.NoError(t, err) - wrk := NewWorkerForStatus(scope.Global.String(), + wrk := NewWorker(scope.Global.String(), WithName(name), - WithAddress("127.0.0.1")) + WithAddress(address)) wrk, err = serversRepo.UpsertWorkerStatus(ctx, wrk, WithPublicId(id)) require.NoError(t, err) require.NotNil(t, wrk) - opts := getOpts(opt...) if len(opts.withWorkerTags) > 0 { var tags []interface{} @@ -119,18 +126,10 @@ func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Opti require.NoError(t, rw.CreateItems(ctx, tags)) } var mask []string - if opts.withName != "" { - wrk.Name = opts.withName - mask = append(mask, "name") - } if opts.withDescription != "" { wrk.Description = opts.withDescription mask = append(mask, "description") } - if opts.withAddress != "" { - wrk.Address = opts.withAddress - mask = append(mask, "address") - } if len(mask) > 0 { var n int wrk, n, err = serversRepo.UpdateWorker(ctx, wrk, wrk.Version, mask) @@ -141,3 +140,38 @@ func TestWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Opti return wrk } + +// TestKmsWorker inserts a worker into the db to satisfy foreign key constraints. +// The worker provided fields are auto generated. WithName and WithDescription, +// are applied to the resource name, description and address if +// present. +func TestPkiWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...Option) *Worker { + t.Helper() + rw := db.New(conn) + kms := kms.TestKms(t, conn, wrapper) + serversRepo, err := NewRepository(rw, rw, kms) + require.NoError(t, err) + ctx := context.Background() + opts := getOpts(opt...) + + require.NoError(t, err) + wrk := NewWorker(scope.Global.String(), + opt...) + wrk, err = serversRepo.CreateWorker(ctx, wrk, opt...) + require.NoError(t, err) + require.NotNil(t, wrk) + + if len(opts.withWorkerTags) > 0 { + var tags []interface{} + for _, t := range opts.withWorkerTags { + tags = append(tags, &store.WorkerTag{ + WorkerId: wrk.GetPublicId(), + Key: t.Key, + Value: t.Value, + Source: "config", + }) + } + require.NoError(t, rw.CreateItems(ctx, tags)) + } + return wrk +} diff --git a/internal/servers/testing_test.go b/internal/servers/testing_test.go index c12687450..26f84a2f1 100644 --- a/internal/servers/testing_test.go +++ b/internal/servers/testing_test.go @@ -11,7 +11,7 @@ import ( "github.com/stretchr/testify/require" ) -func TestTestWorker(t *testing.T) { +func TestTestKmsWorker(t *testing.T) { conn, _ := db.TestSetup(t, "postgres") wrapper := db.TestWrapper(t) const ( @@ -19,7 +19,7 @@ func TestTestWorker(t *testing.T) { description = "test description" address = "test address" ) - tWorker := TestWorker(t, conn, wrapper, WithName(name), WithDescription(description), WithAddress(address)) + tWorker := TestKmsWorker(t, conn, wrapper, WithName(name), WithDescription(description), WithAddress(address)) assert.NotNil(t, tWorker) assert.True(t, strings.HasPrefix(tWorker.GetPublicId(), WorkerPrefix)) @@ -29,7 +29,30 @@ func TestTestWorker(t *testing.T) { require.NoError(t, rw.LookupById(context.Background(), lkpWorker)) assert.NotNil(t, lkpWorker) assert.NotNil(t, lkpWorker.GetLastStatusTime()) + assert.Equal(t, KmsWorkerType.String(), lkpWorker.GetType()) assert.Equal(t, name, lkpWorker.GetName()) assert.Equal(t, description, lkpWorker.GetDescription()) assert.Equal(t, address, lkpWorker.GetAddress()) } + +func TestTestPkiWorker(t *testing.T) { + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + const ( + name = "test name" + description = "test description" + ) + tWorker := TestPkiWorker(t, conn, wrapper, WithName(name), WithDescription(description)) + assert.NotNil(t, tWorker) + assert.True(t, strings.HasPrefix(tWorker.GetPublicId(), WorkerPrefix)) + + 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) + assert.Equal(t, PkiWorkerType.String(), lkpWorker.GetType()) + assert.Equal(t, name, lkpWorker.GetName()) + assert.Equal(t, description, lkpWorker.GetDescription()) + assert.Nil(t, lkpWorker.GetLastStatusTime()) +} diff --git a/internal/servers/worker.go b/internal/servers/worker.go index a80383508..fb2e7f665 100644 --- a/internal/servers/worker.go +++ b/internal/servers/worker.go @@ -12,6 +12,26 @@ import ( "google.golang.org/protobuf/types/known/structpb" ) +type WorkerType string + +const ( + UnknownWorkerType WorkerType = "unknown" + KmsWorkerType WorkerType = "kms" + PkiWorkerType WorkerType = "pki" +) + +func (t WorkerType) Valid() bool { + switch t { + case KmsWorkerType, PkiWorkerType: + return true + } + return false +} + +func (t WorkerType) String() string { + return string(t) +} + type workerAuthWorkerId struct { WorkerId string `mapstructure:"worker_id"` } @@ -40,6 +60,13 @@ type Worker struct { activeConnectionCount uint32 `gorm:"-"` apiTags []*Tag `gorm:"-"` configTags []*Tag `gorm:"-"` + + // inputTags is not specified to be api or config tags and is not intended + // to be read by clients. Since config tags and api tags are applied in + // mutually exclusive contexts, inputTags is interpreted to be one or the + // other based on the context in which the worker is passed. As such + // inputTags should only be read when performing mutations on the database. + inputTags []*Tag `gorm:"-"` } // NewWorker returns a new Worker. Valid options are WithName, WithDescription @@ -54,7 +81,7 @@ func NewWorker(scopeId string, opt ...Option) *Worker { Description: opts.withDescription, Address: opts.withAddress, }, - apiTags: opts.withWorkerTags, + inputTags: opts.withWorkerTags, } } @@ -63,16 +90,7 @@ func NewWorker(scopeId string, opt ...Option) *Worker { // 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, - WorkerReportedKeyId: opts.withKeyId, - }, - configTags: opts.withWorkerTags, - } + return NewWorker(scopeId, opt...) } // allocWorker will allocate a Worker @@ -100,6 +118,12 @@ func (w *Worker) clone() *Worker { cWorker.configTags = append(cWorker.configTags, &Tag{Key: t.Key, Value: t.Value}) } } + if w.inputTags != nil { + cWorker.inputTags = make([]*Tag, 0, len(w.inputTags)) + for _, t := range w.inputTags { + cWorker.inputTags = append(cWorker.inputTags, &Tag{Key: t.Key, Value: t.Value}) + } + } return cWorker } @@ -109,10 +133,7 @@ func (w *Worker) clone() *Worker { // in its connection status updates. If neither is available, an empty string // is returned. func (w *Worker) CanonicalAddress() string { - if w.GetAddress() != "" { - return w.GetAddress() - } - return w.GetWorkerReportedAddress() + return w.GetAddress() } // ActiveConnectionCount is the current number of sessions this worker is handling @@ -183,32 +204,28 @@ type workerAggregate struct { UpdateTime *timestamp.Timestamp Address string Version uint32 + Type string ApiTags string ActiveConnectionCount uint32 // Config Fields - WorkerReportedName string - WorkerReportedAddress string - LastStatusTime *timestamp.Timestamp - WorkerReportedKeyId string - WorkerConfigTags string + LastStatusTime *timestamp.Timestamp + WorkerConfigTags string } func (a *workerAggregate) toWorker(ctx context.Context) (*Worker, error) { const op = "servers.(workerAggregate).toWorker" worker := &Worker{ Worker: &store.Worker{ - PublicId: a.PublicId, - Name: a.Name, - Description: a.Description, - Address: a.Address, - CreateTime: a.CreateTime, - UpdateTime: a.UpdateTime, - ScopeId: a.ScopeId, - Version: a.Version, - WorkerReportedAddress: a.WorkerReportedAddress, - WorkerReportedName: a.WorkerReportedName, - LastStatusTime: a.LastStatusTime, - WorkerReportedKeyId: a.WorkerReportedKeyId, + PublicId: a.PublicId, + Name: a.Name, + Description: a.Description, + Address: a.Address, + CreateTime: a.CreateTime, + UpdateTime: a.UpdateTime, + ScopeId: a.ScopeId, + Version: a.Version, + LastStatusTime: a.LastStatusTime, + Type: a.Type, }, activeConnectionCount: a.ActiveConnectionCount, } diff --git a/internal/servers/worker_tags_test.go b/internal/servers/worker_tags_test.go index 679832e6a..db371cd43 100644 --- a/internal/servers/worker_tags_test.go +++ b/internal/servers/worker_tags_test.go @@ -15,7 +15,7 @@ func TestWorkerTags_Create(t *testing.T) { wrapper := db.TestWrapper(t) rw := db.New(conn) - worker := TestWorker(t, conn, wrapper) + worker := TestKmsWorker(t, conn, wrapper) tests := []struct { name string diff --git a/internal/servers/worker_test.go b/internal/servers/worker_test.go index 4e6d26a2d..4b514091d 100644 --- a/internal/servers/worker_test.go +++ b/internal/servers/worker_test.go @@ -5,15 +5,18 @@ import ( "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/iam" "github.com/hashicorp/boundary/internal/servers/store" "github.com/hashicorp/boundary/internal/types/scope" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/protobuf/testing/protocmp" + "google.golang.org/protobuf/types/known/timestamppb" ) func TestWorkerCanonicalAddress(t *testing.T) { @@ -24,12 +27,12 @@ func TestWorkerCanonicalAddress(t *testing.T) { } func TestWorkerCanonicalTags(t *testing.T) { - w := NewWorker(scope.Global.String(), - WithWorkerTags( - &Tag{Key: "key", Value: "apis unique"}, - &Tag{Key: "key", Value: "shared"}, - &Tag{Key: "key2", Value: "apis key2 unique"}, - )) + w := NewWorker(scope.Global.String()) + w.apiTags = []*Tag{ + {Key: "key", Value: "apis unique"}, + {Key: "key", Value: "shared"}, + {Key: "key2", Value: "apis key2 unique"}, + } w.configTags = []*Tag{ {Key: "key", Value: "configs unique"}, {Key: "key", Value: "shared"}, @@ -57,33 +60,36 @@ func TestWorkerAggregate(t *testing.T) { return got } - // Worker without a status - { + t.Run("kms worker", func(t *testing.T) { id, err := newWorkerId(ctx) require.NoError(t, err) + id = strings.ToLower(id) w := NewWorker(scope.Global.String(), WithName(id), WithAddress("address")) + w.Type = KmsWorkerType.String() w.PublicId = id require.NoError(t, rw.Create(ctx, w)) got := getAggWorker(id) + assert.Equal(t, KmsWorkerType.String(), got.GetType()) assert.Equal(t, id, got.GetPublicId()) assert.Equal(t, scope.Global.String(), got.GetScopeId()) assert.Equal(t, id, got.GetName()) assert.Equal(t, "address", got.GetAddress()) assert.Equal(t, uint32(1), got.GetVersion()) - assert.Nil(t, got.GetLastStatusTime()) + assert.NotNil(t, got.GetLastStatusTime()) assert.Empty(t, got.CanonicalTags()) - } + }) // Worker with status - { + t.Run("Worker with status", func(t *testing.T) { id, err := newWorkerId(ctx) require.NoError(t, err) w := NewWorkerForStatus(scope.Global.String(), WithAddress("address"), WithName(strings.ToLower(id))) + w.Type = KmsWorkerType.String() w.PublicId = id require.NoError(t, rw.Create(ctx, w)) @@ -91,19 +97,19 @@ func TestWorkerAggregate(t *testing.T) { assert.Equal(t, id, got.GetPublicId()) assert.Equal(t, uint32(1), got.GetVersion()) assert.NotNil(t, got.GetLastStatusTime()) - assert.Equal(t, strings.ToLower(id), got.GetWorkerReportedName()) - assert.Equal(t, "address", got.GetWorkerReportedAddress()) - } + assert.Equal(t, strings.ToLower(id), got.GetName()) + assert.Equal(t, "address", got.GetAddress()) + }) - // Worker with a config tag - { + t.Run("Worker with a config tag", func(t *testing.T) { id, err := newWorkerId(ctx) require.NoError(t, err) - ws := NewWorkerForStatus(scope.Global.String(), + w := NewWorkerForStatus(scope.Global.String(), WithAddress("address"), WithName(strings.ToLower(id))) - ws.PublicId = id - require.NoError(t, rw.Create(ctx, ws)) + w.Type = KmsWorkerType.String() + w.PublicId = id + require.NoError(t, rw.Create(ctx, w)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, @@ -118,17 +124,17 @@ func TestWorkerAggregate(t *testing.T) { 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 - { + t.Run("Worker with many config tag", func(t *testing.T) { id, err := newWorkerId(ctx) require.NoError(t, err) - ws := NewWorkerForStatus(scope.Global.String(), + w := NewWorkerForStatus(scope.Global.String(), WithAddress("address"), WithName(strings.ToLower(id))) - ws.PublicId = id - require.NoError(t, rw.Create(ctx, ws)) + w.Type = KmsWorkerType.String() + w.PublicId = id + require.NoError(t, rw.Create(ctx, w)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, Key: "key", @@ -156,17 +162,17 @@ func TestWorkerAggregate(t *testing.T) { {Key: "key", Value: "val2"}, {Key: "key2", Value: "val2"}, }) - } + }) - // Worker with an api tag - { + t.Run("Worker with an api tag", func(t *testing.T) { id, err := newWorkerId(ctx) require.NoError(t, err) - ws := NewWorkerForStatus(scope.Global.String(), + w := NewWorkerForStatus(scope.Global.String(), WithAddress("address"), WithName(strings.ToLower(id))) - ws.PublicId = id - require.NoError(t, rw.Create(ctx, ws)) + w.Type = KmsWorkerType.String() + w.PublicId = id + require.NoError(t, rw.Create(ctx, w)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, @@ -181,17 +187,18 @@ func TestWorkerAggregate(t *testing.T) { 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 - { + t.Run("Worker with mix of tag sources", func(t *testing.T) { id, err := newWorkerId(ctx) require.NoError(t, err) - ws := NewWorkerForStatus(scope.Global.String(), + w := NewWorkerForStatus(scope.Global.String(), WithAddress("address"), WithName(strings.ToLower(id))) - ws.PublicId = id - require.NoError(t, rw.Create(ctx, ws)) + w.Type = KmsWorkerType.String() + w.PublicId = id + require.NoError(t, rw.Create(ctx, w)) require.NoError(t, rw.Create(ctx, &store.WorkerTag{ WorkerId: id, Key: "key", @@ -220,7 +227,7 @@ func TestWorkerAggregate(t *testing.T) { assert.ElementsMatch(t, got.configTags, []*Tag{ {Key: "key", Value: "val"}, }) - } + }) } func TestWorker_Update(t *testing.T) { @@ -245,260 +252,213 @@ func TestWorker_Update(t *testing.T) { wantUpdateError bool }{ { - name: "base update with base", + name: "kms update address", initial: Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - Description: "base update with base", + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms update address", + Address: "kms update address", }, }, update: Worker{ Worker: &store.Worker{ - Description: "base update with base updated", + Address: "updated kms update address", }, }, - mask: []string{"Description"}, + mask: []string{"Address"}, assert: func(t *testing.T, init, up *Worker) { t.Helper() - assert.Equal(t, "base update with base updated", up.Description) + assert.Equal(t, "kms update address", up.Name) + assert.Equal(t, "updated kms update address", up.Address) + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, up.GetLastStatusTime().AsTime(), up.GetUpdateTime().AsTime()) + }, + }, + { + name: "kms update name", + initial: Worker{ + Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms update name", + Address: "kms update name", + }, + }, + update: Worker{ + Worker: &store.Worker{ + Name: "updated kms update name", + }, + }, + mask: []string{"Name"}, + wantUpdateError: true, + }, + { + name: "kms clear name", + initial: Worker{ + Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms clear name", + Address: "kms clear name", + }, + }, + update: Worker{ + Worker: &store.Worker{}, + }, + nullMask: []string{"Name"}, + wantUpdateError: true, + }, + { + name: "pki update address", + initial: Worker{ + Worker: &store.Worker{ + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update address", + }, + }, + update: Worker{ + Worker: &store.Worker{ + Address: "updated pki update address", + }, + }, + mask: []string{"Address"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "pki update address", up.Name) + assert.Equal(t, "updated pki update address", up.Address) + assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) + assert.Equal(t, up.GetLastStatusTime().AsTime(), up.GetUpdateTime().AsTime()) + }, + }, + { + name: "pki update name", + initial: Worker{ + Worker: &store.Worker{ + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update name", + }, + }, + update: Worker{ + Worker: &store.Worker{ + Name: "updated pki update name", + }, + }, + mask: []string{"Name"}, + assert: func(t *testing.T, init, up *Worker) { + t.Helper() + assert.Equal(t, "updated pki update name", up.Name) 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", + name: "pki update description", initial: Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - Description: "base update with status", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update description", }, }, 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 and keyId", - initial: Worker{ - Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - Description: "base update with status", - }, - }, - update: Worker{ - Worker: &store.Worker{ - WorkerReportedKeyId: "base update with worker reported address and keyId", - WorkerReportedAddress: "base update with worker reported address and keyId", - }, - }, - mask: []string{"WorkerReportedAddress", "WorkerReportedKeyId"}, - assert: func(t *testing.T, init, up *Worker) { - t.Helper() - assert.Equal(t, "base update with worker reported address and keyId", up.WorkerReportedAddress) - assert.Equal(t, "base update with worker reported address and keyId", up.WorkerReportedKeyId) - 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, - }, - { - // If any status fields are set then worker reported address must - // be set. - name: "base update with worker reported keyId", - initial: Worker{ - Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - Description: "base update with status", - }, - }, - update: Worker{ - Worker: &store.Worker{ - WorkerReportedKeyId: "base update with worker reported keyId", - }, - }, - mask: []string{"WorkerReportedKeyId"}, - 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", + Description: "updated pki update description", }, }, mask: []string{"Description"}, assert: func(t *testing.T, init, up *Worker) { t.Helper() - assert.Equal(t, "worker reported update with base", up.Description) + assert.Equal(t, "pki update description", up.Name) + assert.Equal(t, "updated pki update description", 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()) + assert.Nil(t, up.GetLastStatusTime()) }, }, { - name: "worker reported update name to unprintable", + name: "pki clear address", 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", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki clear address", + Address: "pki clear address", + LastStatusTime: ×tamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}, }, }, update: Worker{ Worker: &store.Worker{}, }, - nullMask: []string{"WorkerReportedAddress"}, + nullMask: []string{"Address"}, wantUpdateError: true, }, { - name: "worker reported updating worker reported", + name: "pki clear name", initial: Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - WorkerReportedAddress: "start", - WorkerReportedName: "start", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki clear name", + Address: "pki clear name", + LastStatusTime: ×tamp.Timestamp{Timestamp: timestamppb.New(time.Now().Add(time.Hour))}, }, }, update: Worker{ - Worker: &store.Worker{ - WorkerReportedAddress: "worker reported updating worker reported", - }, + Worker: &store.Worker{}, }, - mask: []string{"WorkerReportedAddress"}, + nullMask: []string{"Name"}, 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.Empty(t, up.Name) assert.Greater(t, up.GetUpdateTime().AsTime(), up.GetCreateTime().AsTime()) - assert.Equal(t, uint32(1), up.Version) + assert.Greater(t, up.GetLastStatusTime().AsTime(), up.GetUpdateTime().AsTime()) + assert.Less(t, up.GetLastStatusTime().AsTime(), time.Now().Add(time.Hour)) }, }, { - name: "worker reported clearing worker reported name", + name: "pki update name to capital", 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", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki update name to capital", }, }, update: Worker{ - Worker: &store.Worker{}, + Worker: &store.Worker{ + Name: "PKI Update Name To Capital", + }, }, - nullMask: []string{"WorkerReportedName"}, + mask: []string{"Name"}, + wantUpdateError: true, + }, + { + name: "pki update name to unprintable", + initial: Worker{ + Worker: &store.Worker{ + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "worker reported update name to unprintable", + }, + }, + update: Worker{ + Worker: &store.Worker{ + Name: "unprintable \u0008 name", + }, + }, + mask: []string{"Name"}, wantUpdateError: true, }, } @@ -544,9 +504,88 @@ func TestWorker_Create(t *testing.T) { want wanted }{ { - name: "base", + name: "kms base", in: Worker{ Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Address: "address", + Name: "kms base", + Description: "kms base", + }, + }, + want: wanted{lastStatusUpdated: true}, + }, + { + name: "kms with no name field", + in: Worker{ + Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Address: "address", + }, + }, + want: wanted{createError: true}, + }, + { + name: "kms with no address field", + in: Worker{ + Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "kms with no address field", + }, + }, + want: wanted{createError: true}, + }, + { + name: "kms with upper case name", + in: Worker{ + Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "With Upper Case Worker Reported Name", + Address: "address", + }, + }, + want: wanted{createError: true}, + }, + { + name: "kms with unprintable worker reported name", + in: Worker{ + Worker: &store.Worker{ + Type: KmsWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "unprintable \u0008 worker reported name", + Address: "address", + }, + }, + want: wanted{createError: true}, + }, + // PKI workers are not created in a way that should set the address + // or the last_status_time. + { + name: "pki base", + in: Worker{ + Worker: &store.Worker{ + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "pki base", + }, + }, + want: wanted{}, + }, + { + name: "pki with no name field", + in: Worker{ + Worker: &store.Worker{ + Type: PkiWorkerType.String(), ScopeId: scope.Global.String(), PublicId: newId(), }, @@ -554,134 +593,42 @@ func TestWorker_Create(t *testing.T) { want: wanted{}, }, { - name: "with non status fields", + name: "pki with address field", 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", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Address: "address", }, }, want: wanted{createError: true}, }, { - name: "with unprintable worker reported name", + name: "pki with upper case name", in: Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - WorkerReportedName: "unprintable \u0008 worker reported name", - WorkerReportedAddress: "address", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "PKI With Upper Case Worker Reported Name", + Address: "address", }, }, want: wanted{createError: true}, }, { - name: "with status fields", + name: "pki with unprintable worker reported name", in: Worker{ Worker: &store.Worker{ - ScopeId: scope.Global.String(), - PublicId: newId(), - WorkerReportedName: "with non status fields", - WorkerReportedAddress: "with non status fields", + Type: PkiWorkerType.String(), + ScopeId: scope.Global.String(), + PublicId: newId(), + Name: "unprintable \u0008 worker reported name", + Address: "address", }, }, - 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, - }, + want: wanted{createError: true}, }, } for _, tc := range cases { @@ -711,7 +658,6 @@ func TestWorker_New(t *testing.T) { t.Parallel() conn, _ := db.TestSetup(t, "postgres") wrapper := db.TestWrapper(t) - rw := db.New(conn) org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) @@ -721,10 +667,9 @@ func TestWorker_New(t *testing.T) { } tests := []struct { - name string - args args - want *Worker - wantCreateErr bool + name string + args args + want *Worker }{ { name: "missing-scope-id", @@ -732,7 +677,6 @@ func TestWorker_New(t *testing.T) { want: &Worker{ Worker: &store.Worker{}, }, - wantCreateErr: true, }, { name: "global-scope-id", @@ -755,7 +699,6 @@ func TestWorker_New(t *testing.T) { ScopeId: org.GetPublicId(), }, }, - wantCreateErr: true, }, { name: "project-scope-id", @@ -767,7 +710,6 @@ func TestWorker_New(t *testing.T) { ScopeId: prj.GetPublicId(), }, }, - wantCreateErr: true, }, { name: "with name", @@ -829,7 +771,7 @@ func TestWorker_New(t *testing.T) { Worker: &store.Worker{ ScopeId: scope.Global.String(), }, - apiTags: []*Tag{ + inputTags: []*Tag{ { Key: "key", Value: "val", @@ -845,20 +787,6 @@ func TestWorker_New(t *testing.T) { assert := assert.New(t) got := NewWorker(tt.args.scopeId, tt.args.opts...) assert.Empty(cmp.Diff(tt.want, got.clone(), protocmp.Transform())) - - id, err := newWorkerId(context.Background()) - assert.NoError(err) - - tt.want.PublicId = id - got.PublicId = id - - err2 := rw.Create(context.Background(), got) - if tt.wantCreateErr { - assert.Error(err2) - } else { - assert.NoError(err2) - assert.Equal(uint32(1), got.Version) - } }) } } diff --git a/internal/servers/workerauth_store_test.go b/internal/servers/workerauth_store_test.go index fdbe415d6..10f8099a3 100644 --- a/internal/servers/workerauth_store_test.go +++ b/internal/servers/workerauth_store_test.go @@ -291,7 +291,7 @@ func TestWorkerAuthStore(t *testing.T) { testKey, err := databaseWrapper.KeyId(ctx) require.NoError(t, err) - worker := TestWorker(t, conn, wrapper) + worker := TestPkiWorker(t, conn, wrapper) wSignPubKey := populateBytes(defaultLength) wEncPubKey := populateBytes(defaultLength) @@ -430,7 +430,7 @@ func TestWorkerCertBundle(t *testing.T) { wrapper := db.TestWrapper(t) testKey := TestKmsKey(ctx, t, conn, wrapper) - worker := TestWorker(t, conn, wrapper) + worker := TestPkiWorker(t, conn, wrapper) workerAuth := TestWorkerAuth(ctx, t, conn, worker, testKey) rootCA := TestRootCertificate(ctx, t, conn, testKey) certBundle := populateBytes(defaultLength) diff --git a/internal/session/job_session_cleanup_test.go b/internal/session/job_session_cleanup_test.go index b261506dc..126d76411 100644 --- a/internal/session/job_session_cleanup_test.go +++ b/internal/session/job_session_cleanup_test.go @@ -45,8 +45,8 @@ func TestSessionConnectionCleanupJob(t *testing.T) { // Create two "workers". One will remain untouched while the other "goes // away and comes back" (worker 2). - worker1 := servers.TestWorker(t, conn, wrapper) - worker2 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) + worker2 := servers.TestKmsWorker(t, conn, wrapper) // Create a few sessions on each, activate, and authorize a connection var connIds []string @@ -110,8 +110,9 @@ func TestSessionConnectionCleanupJob(t *testing.T) { // Push an upsert to the first worker so that its status has been // updated. _, err = serversRepo.UpsertWorkerStatus(ctx, servers.NewWorkerForStatus(scope.Global.String(), - servers.WithName(worker1.GetWorkerReportedName()), - servers.WithAddress(worker1.GetWorkerReportedAddress()))) + servers.WithName(worker1.GetName()), + servers.WithAddress(worker1.GetAddress())), + servers.WithPublicId(worker1.GetPublicId())) require.NoError(err) // Run the job. @@ -198,10 +199,10 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { // firstly, the last worker will have no connections at all, and we will be // closing the others in stages to test multiple servers being closed at // once. - worker1 := servers.TestWorker(t, conn, wrapper) - worker2 := servers.TestWorker(t, conn, wrapper) - worker3 := servers.TestWorker(t, conn, wrapper) - worker4 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) + worker2 := servers.TestKmsWorker(t, conn, wrapper) + worker3 := servers.TestKmsWorker(t, conn, wrapper) + worker4 := servers.TestKmsWorker(t, conn, wrapper) // Create sessions on the first three, activate, and authorize connections var worker1ConnIds, worker2ConnIds, worker3ConnIds []string @@ -292,11 +293,9 @@ func TestCloseConnectionsForDeadWorkers(t *testing.T) { // the most up-to-date fields. updateServer := func(t *testing.T, w *servers.Worker) *servers.Worker { t.Helper() - // The worker provided name will derive the WorkerId every time - // UpsertWorkerStatus is called. Setting the Worker Id on the caller - // side is an error. + pubId := w.GetPublicId() w.PublicId = "" - wkr, err := serversRepo.UpsertWorkerStatus(ctx, w) + wkr, err := serversRepo.UpsertWorkerStatus(ctx, w, servers.WithPublicId(pubId)) require.NoError(err) return wkr } @@ -465,7 +464,7 @@ func TestCloseWorkerlessConnections(t *testing.T) { } // Setup deleted worker connections - deletedWorker := servers.TestWorker(t, conn, wrapper) + deletedWorker := servers.TestKmsWorker(t, conn, wrapper) dActiveConn := createConnection(deletedWorker.GetPublicId()) dClosedConn := createConnection(deletedWorker.GetPublicId()) connRepo.closeConnections(ctx, []CloseWith{{ @@ -478,7 +477,7 @@ func TestCloseWorkerlessConnections(t *testing.T) { require.NoError(err) // Non deleted worker case - activeWorker := servers.TestWorker(t, conn, wrapper) + activeWorker := servers.TestKmsWorker(t, conn, wrapper) activeConn := createConnection(activeWorker.GetPublicId()) closedConn := createConnection(activeWorker.GetPublicId()) connRepo.closeConnections(ctx, []CloseWith{{ diff --git a/internal/session/repository_connection_test.go b/internal/session/repository_connection_test.go index 200df8272..df43a59c9 100644 --- a/internal/session/repository_connection_test.go +++ b/internal/session/repository_connection_test.go @@ -359,8 +359,8 @@ func TestRepository_orphanedConnections(t *testing.T) { // Create two "workers". One will remain untouched while the other "goes // away and comes back" (worker 2). - worker1 := servers.TestWorker(t, conn, wrapper) - worker2 := servers.TestWorker(t, conn, wrapper) + worker1 := servers.TestKmsWorker(t, conn, wrapper) + worker2 := servers.TestKmsWorker(t, conn, wrapper) // Create a few sessions on each, activate, and authorize a connection var connIds []string diff --git a/internal/session/service_authorize_connection_test.go b/internal/session/service_authorize_connection_test.go index 6261bfb06..e0d3cf739 100644 --- a/internal/session/service_authorize_connection_test.go +++ b/internal/session/service_authorize_connection_test.go @@ -36,7 +36,7 @@ func TestService_AuthorizeConnection(t *testing.T) { composedOf.ExpirationTime = exp } s := TestSession(t, conn, wrapper, composedOf) - srv := servers.TestWorker(t, conn, wrapper) + srv := servers.TestKmsWorker(t, conn, wrapper) testServer = srv.PublicId tofu := TestTofu(t) _, _, err := repo.ActivateSession(context.Background(), s.PublicId, s.Version, tofu) diff --git a/internal/session/service_worker_status_report_test.go b/internal/session/service_worker_status_report_test.go index 706c2db56..d9956ccfd 100644 --- a/internal/session/service_worker_status_report_test.go +++ b/internal/session/service_worker_status_report_test.go @@ -66,7 +66,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "No Sessions", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) return testCase{ worker: worker, req: []session.StateReport{}, @@ -77,7 +77,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "No Sessions already canceled", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -109,7 +109,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "Still Active", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -143,7 +143,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "SessionClosed", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -183,7 +183,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "MultipleSessionsClosed", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -250,7 +250,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "OrphanedConnection", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -301,7 +301,7 @@ func TestWorkerStatusReport(t *testing.T) { { name: "MultipleSessionsAndOrphanedConnections", caseFn: func(t *testing.T) testCase { - worker := servers.TestWorker(t, conn, wrapper) + worker := servers.TestKmsWorker(t, conn, wrapper) sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), diff --git a/internal/session/testing_test.go b/internal/session/testing_test.go index f344ae896..1403a3296 100644 --- a/internal/session/testing_test.go +++ b/internal/session/testing_test.go @@ -74,7 +74,7 @@ func Test_TestWorker(t *testing.T) { require := require.New(t) conn, _ := db.TestSetup(t, "postgres") wrapper := db.TestWrapper(t) - w := servers.TestWorker(t, conn, wrapper) + w := servers.TestKmsWorker(t, conn, wrapper) require.NotNil(w) }