diff --git a/internal/daemon/controller/handlers/workers/worker_service_test.go b/internal/daemon/controller/handlers/workers/worker_service_test.go index fccdef7c5..0bc49cffd 100644 --- a/internal/daemon/controller/handlers/workers/worker_service_test.go +++ b/internal/daemon/controller/handlers/workers/worker_service_test.go @@ -869,6 +869,7 @@ func TestUpdate_KMS(t *testing.T) { t.Run(tc.name, func(t *testing.T) { req := proto.Clone(toMerge).(*pbs.UpdateWorkerRequest) proto.Merge(req, tc.req) + req.Item.Version = wkr.GetVersion() got, gErr := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, scope.Global.String()), req) assert.Error(t, gErr) assert.Nil(t, got) @@ -900,7 +901,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.TestKmsWorker(t, conn, wrapper) + wkr := servers.TestPkiWorker(t, conn, wrapper) upTar, err := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), &pbs.UpdateWorkerRequest{ Id: wkr.GetPublicId(), diff --git a/internal/servers/repository_worker.go b/internal/servers/repository_worker.go index e652bc317..4ddd3302b 100644 --- a/internal/servers/repository_worker.go +++ b/internal/servers/repository_worker.go @@ -267,22 +267,22 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { - worker := worker.clone() - worker.PublicId = workerId + workerClone := worker.clone() + workerClone.PublicId = workerId switch { case opts.withKeyId != "": // This case goes first in case we want to relax the restriction // around both name and key ID being supplied to account for // e.g. config processing bugs. In that case, a key ID being // supplied should be a clear indicator that we are working with - // a PKI worker, and a lack of one a clear indication we are - // working with a KMS worker. + // a PKI workerClone, and a lack of one a clear indication we are + // working with a KMS workerClone. // // Note: unlike in the below case, this purposefully leaves out // "description" since we want description changes for PKI-based // workers to come via API only. We can't really guard on this // in the DB so we need to be sure to not include it here. - n, err := w.Update(ctx, worker, []string{"address"}, nil) + n, err := w.Update(ctx, workerClone, []string{"address"}, nil) if err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("unable to update status of pki worker")) } @@ -292,22 +292,25 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt case 1: break default: - return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("multiple records found when updating worker with id %q", worker.GetPublicId())) + return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("multiple records found when updating worker with id %q", workerClone.GetPublicId())) } - case worker.GetName() != "": - worker.Type = KmsWorkerType.String() + case workerClone.GetName() != "": + workerClone.Type = KmsWorkerType.String() workerCreateConflict := &db.OnConflict{ Target: db.Columns{"public_id"}, - Action: append(db.SetColumns([]string{"description", "address"}), + Action: append(db.SetColumns([]string{"address"}), db.SetColumnValues(map[string]interface{}{"last_status_time": "now()"})...), } var withRowsAffected int64 - err := w.Create(ctx, worker, db.WithOnConflict(workerCreateConflict), db.WithReturnRowsAffected(&withRowsAffected), + err := w.Create(ctx, workerClone, db.WithOnConflict(workerCreateConflict), db.WithReturnRowsAffected(&withRowsAffected), // The intent of this WithWhere option is to operate with the OnConflict such that the action // taken by the OnConflict only applies if the conflict is on a row that is returned by this where // statement, otherwise it should error out. db.WithWhere("server_worker.type = 'kms'")) + if err == nil && workerClone.Description != worker.Description { + _, err = w.Update(ctx, workerClone, []string{"description"}, nil) + } switch { case err != nil: return errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker")) @@ -321,12 +324,12 @@ 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 { - if err := setWorkerTags(ctx, w, worker.GetPublicId(), ConfigurationTagSource, worker.inputTags); err != nil { + if err := setWorkerTags(ctx, w, workerClone.GetPublicId(), ConfigurationTagSource, workerClone.inputTags); err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("error setting worker tags")) } } - wAgg := &workerAggregate{PublicId: worker.GetPublicId()} + wAgg := &workerAggregate{PublicId: workerClone.GetPublicId()} if err := reader.LookupById(ctx, wAgg); err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("error looking up worker aggregate")) } @@ -442,10 +445,14 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { worker := worker.clone() - rowsUpdated, err = w.Update(ctx, worker, dbMask, nullFields, db.WithVersion(&version)) + rowsUpdated, err = w.Update(ctx, worker, dbMask, nullFields, db.WithVersion(&version), + db.WithWhere("server_worker.type = 'pki'")) if err != nil { return errors.Wrap(ctx, err, op) } + if worker.Type == KmsWorkerType.String() { + return errors.New(ctx, errors.InvalidParameter, op, "cannot update a KMS worker") + } if rowsUpdated > 1 { // return err, which will result in a rollback of the update return errors.New(ctx, errors.MultipleRecords, op, "more than 1 resource would have been updated") diff --git a/internal/servers/repository_worker_test.go b/internal/servers/repository_worker_test.go index 217a255c8..39bfc3003 100644 --- a/internal/servers/repository_worker_test.go +++ b/internal/servers/repository_worker_test.go @@ -274,15 +274,13 @@ func TestUpsertWorkerStatus(t *testing.T) { // update again and see updated last status time wStatus2 := servers.NewWorker(scope.Global.String(), - servers.WithAddress("new_address"), servers.WithName("config_name1"), - servers.WithDescription("kms_description2")) + servers.WithAddress("new_address"), servers.WithName("config_name1")) worker, err = repo.UpsertWorkerStatus(ctx, wStatus2) require.NoError(t, err) assert.Greater(t, worker.GetLastStatusTime().AsTime(), worker.GetCreateTime().AsTime()) assert.Equal(t, "config_name1", worker.Name) - assert.Equal(t, "kms_description2", worker.Description) - assert.Equal(t, worker.GetLastStatusTime().AsTime(), worker.GetUpdateTime().AsTime()) - assert.Equal(t, uint32(2), worker.Version) + // Version does not change for status updates + assert.Equal(t, uint32(1), worker.Version) assert.Equal(t, "new_address", worker.GetAddress()) }) @@ -1082,20 +1080,48 @@ func TestRepository_UpdateWorker(t *testing.T) { input: func() *servers.Worker { w := servers.TestKmsWorker(t, conn, wrapper) w.Name = "some change" + w.Description = "" + w.Type = "" return w }(), path: []string{"name"}, version: 1, - wantErr: errors.T(errors.NotSpecificIntegrity), + wantErr: errors.T(errors.InvalidParameter), }, { name: "clearing kms name", input: func() *servers.Worker { w := servers.TestKmsWorker(t, conn, wrapper) w.Name = "" + w.Description = "" + w.Type = "" return w }(), path: []string{"name"}, version: 1, - wantErr: errors.T(errors.NotSpecificIntegrity), + wantErr: errors.T(errors.InvalidParameter), + }, + { + name: "changing kms description", + input: func() *servers.Worker { + w := servers.TestKmsWorker(t, conn, wrapper) + w.Description = "some change" + w.Name = "" + w.Type = "" + return w + }(), path: []string{"description"}, + version: 1, + wantErr: errors.T(errors.InvalidParameter), + }, + { + name: "clearing kms description", + input: func() *servers.Worker { + w := servers.TestKmsWorker(t, conn, wrapper) + w.Description = "" + w.Name = "" + w.Type = "" + return w + }(), path: []string{"description"}, + version: 1, + wantErr: errors.T(errors.InvalidParameter), }, { name: "no public id", diff --git a/internal/servers/testing.go b/internal/servers/testing.go index 8e5a6ee05..2e2369428 100644 --- a/internal/servers/testing.go +++ b/internal/servers/testing.go @@ -113,10 +113,12 @@ func TestKmsWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...O } wrk := NewWorker(scope.Global.String(), WithName(name), - WithAddress(address)) + WithAddress(address), + WithDescription(opts.withDescription)) wrk, err = serversRepo.UpsertWorkerStatus(ctx, wrk) require.NoError(t, err) require.NotNil(t, wrk) + require.Equal(t, "kms", wrk.Type) if len(opts.withWorkerTags) > 0 { var tags []interface{} @@ -130,18 +132,7 @@ func TestKmsWorker(t *testing.T, conn *db.DB, wrapper wrapping.Wrapper, opt ...O } require.NoError(t, rw.CreateItems(ctx, tags)) } - var mask []string - if opts.withDescription != "" { - wrk.Description = opts.withDescription - mask = append(mask, "description") - } - if len(mask) > 0 { - var n int - wrk, n, err = serversRepo.UpdateWorker(ctx, wrk, wrk.Version, mask) - require.NoError(t, err) - require.Equal(t, 1, n) - require.NotNil(t, wrk) - } + wrk, err = serversRepo.LookupWorker(ctx, wrk.GetPublicId()) require.NoError(t, err) return wrk