feat(workers): implement worker service add/set/remove api tags

internal/daemon/controller/handlers/workers/worker_service.go

@@ -44,6 +44,9 @@ var (
 		action.Read,
 		action.Update,
 		action.Delete,
+		action.AddWorkerTags,
+		action.SetWorkerTags,
+		action.RemoveWorkerTags,
 	}
 
 	// CollectionActions contains the set of actions that can be performed on
@@ -297,6 +300,114 @@ func (s Service) UpdateWorker(ctx context.Context, req *pbs.UpdateWorkerRequest)
 	return &pbs.UpdateWorkerResponse{Item: item}, nil
 }
 
+// AddWorkerTags implements the interface pbs.WorkerServiceServer.
+func (s Service) AddWorkerTags(ctx context.Context, req *pbs.AddWorkerTagsRequest) (*pbs.AddWorkerTagsResponse, error) {
+	const op = "workers.(Service).AddWorkerTags"
+
+	if err := validateAddTagsRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.AddWorkerTags)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	w, err := s.addTagsInRepo(ctx, req.GetId(), req.GetVersion(), req.GetApiTags())
+	if err != nil {
+		return nil, err
+	}
+
+	outputFields, ok := requests.OutputFields(ctx)
+	if !ok {
+		return nil, errors.New(ctx, errors.Internal, op, "no request context found")
+	}
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, w.GetPublicId(), IdActions).Strings()))
+	}
+	item, err := toProto(ctx, w, outputOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pbs.AddWorkerTagsResponse{Item: item}, nil
+}
+
+// SetWorkerTags implements the interface pbs.WorkerServiceServer.
+func (s Service) SetWorkerTags(ctx context.Context, req *pbs.SetWorkerTagsRequest) (*pbs.SetWorkerTagsResponse, error) {
+	const op = "workers.(Service).SetWorkerTags"
+
+	if err := validateSetTagsRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.SetWorkerTags)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	w, err := s.setTagsInRepo(ctx, req.GetId(), req.GetVersion(), req.GetApiTags())
+	if err != nil {
+		return nil, err
+	}
+
+	outputFields, ok := requests.OutputFields(ctx)
+	if !ok {
+		return nil, errors.New(ctx, errors.Internal, op, "no request context found")
+	}
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, w.GetPublicId(), IdActions).Strings()))
+	}
+	item, err := toProto(ctx, w, outputOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pbs.SetWorkerTagsResponse{Item: item}, nil
+}
+
+// RemoveWorkerTags implements the interface pbs.WorkerServiceServer.
+func (s Service) RemoveWorkerTags(ctx context.Context, req *pbs.RemoveWorkerTagsRequest) (*pbs.RemoveWorkerTagsResponse, error) {
+	const op = "workers.(Service).RemoveWorkerTags"
+
+	if err := validateRemoveTagsRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.RemoveWorkerTags)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	w, err := s.removeTagsInRepo(ctx, req.GetId(), req.GetVersion(), req.GetApiTags())
+	if err != nil {
+		return nil, err
+	}
+
+	outputFields, ok := requests.OutputFields(ctx)
+	if !ok {
+		return nil, errors.New(ctx, errors.Internal, op, "no request context found")
+	}
+	outputOpts := make([]handlers.Option, 0, 3)
+	outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
+	if outputFields.Has(globals.ScopeField) {
+		outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
+	}
+	if outputFields.Has(globals.AuthorizedActionsField) {
+		outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, w.GetPublicId(), IdActions).Strings()))
+	}
+	item, err := toProto(ctx, w, outputOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pbs.RemoveWorkerTagsResponse{Item: item}, nil
+}
+
 func (s Service) listFromRepo(ctx context.Context, scopeIds []string) ([]*server.Worker, error) {
 	repo, err := s.repoFn()
 	if err != nil {
@@ -390,6 +501,87 @@ func (s Service) updateInRepo(ctx context.Context, scopeId, id string, mask []st
 	return out, nil
 }
 
+func (s Service) addTagsInRepo(ctx context.Context, workerId string, workerVersion uint32, addTags map[string]*structpb.ListValue) (*server.Worker, error) {
+	const op = "workers.(Service).addTagsInRepo"
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+
+	tags := make([]*server.Tag, 0, len(addTags))
+	for k, lv := range addTags {
+		for _, v := range lv.GetValues() {
+			tags = append(tags, &server.Tag{Key: k, Value: v.GetStringValue()})
+		}
+	}
+	_, err = repo.AddWorkerTags(ctx, workerId, workerVersion, tags)
+	if err != nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "Unable to add worker tags in repo: %v.", err)
+	}
+	w, err := repo.LookupWorker(ctx, workerId)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to look up worker after adding tags"))
+	}
+	if w == nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "Unable to look up worker after adding tags.")
+	}
+	return w, nil
+}
+
+func (s Service) setTagsInRepo(ctx context.Context, workerId string, workerVersion uint32, setTags map[string]*structpb.ListValue) (*server.Worker, error) {
+	const op = "workers.(Service).setTagsInRepo"
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+
+	tags := make([]*server.Tag, 0, len(setTags))
+	for k, lv := range setTags {
+		for _, v := range lv.GetValues() {
+			tags = append(tags, &server.Tag{Key: k, Value: v.GetStringValue()})
+		}
+	}
+	_, err = repo.SetWorkerTags(ctx, workerId, workerVersion, tags)
+	if err != nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "Unable to set worker tags in repo: %v.", err)
+	}
+	w, err := repo.LookupWorker(ctx, workerId)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to look up worker after setting tags"))
+	}
+	if w == nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "Unable to look up worker after setting tags.")
+	}
+	return w, nil
+}
+
+func (s Service) removeTagsInRepo(ctx context.Context, workerId string, workerVersion uint32, removeTags map[string]*structpb.ListValue) (*server.Worker, error) {
+	const op = "workers.(Service).removeTagsInRepo"
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+
+	tags := make([]*server.Tag, 0, len(removeTags))
+	for k, lv := range removeTags {
+		for _, v := range lv.GetValues() {
+			tags = append(tags, &server.Tag{Key: k, Value: v.GetStringValue()})
+		}
+	}
+	_, err = repo.DeleteWorkerTags(ctx, workerId, workerVersion, tags)
+	if err != nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "Unable to remove worker tags in repo: %v.", err)
+	}
+	w, err := repo.LookupWorker(ctx, workerId)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to look up worker after removing tags"))
+	}
+	if w == nil {
+		return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "Unable to look up worker after removing tags.")
+	}
+	return w, nil
+}
+
 func (s Service) authResult(ctx context.Context, id string, a action.Type) auth.VerifyResults {
 	res := auth.VerifyResults{}
 	repo, err := s.repoFn()
@@ -485,6 +677,13 @@ func toProto(ctx context.Context, in *server.Worker, opt ...handlers.Option) (*p
 			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing config tags proto"))
 		}
 	}
+	if outputFields.Has(globals.ApiTagsField) && len(in.GetApiTags()) > 0 {
+		var err error
+		out.ApiTags, err = tagsToMapProto(in.GetApiTags())
+		if err != nil {
+			return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing api tags proto"))
+		}
+	}
 	if outputFields.Has(globals.CanonicalTagsField) && len(in.CanonicalTags()) > 0 {
 		var err error
 		out.CanonicalTags, err = tagsToMapProto(in.CanonicalTags())
@@ -615,3 +814,124 @@ func validateCreateRequest(req *pbs.CreateWorkerLedRequest) error {
 		return badFields
 	})
 }
+
+// validateStringForDb checks a string is valid for db storage and returns a string for an error message if needed.
+// returns an empty string otherwise.
+func validateStringForDb(str string) string {
+	switch {
+	case len(str) <= 0:
+		return "must be non-empty."
+	case len(str) > 512:
+		return "must be within 512 characters."
+	default:
+		return ""
+	}
+}
+
+func validateAddTagsRequest(req *pbs.AddWorkerTagsRequest) error {
+	badFields := map[string]string{}
+	if !handlers.ValidId(handlers.Id(req.GetId()), server.WorkerPrefix) {
+		badFields[globals.IdField] = "Incorrectly formatted identifier."
+	}
+	if req.GetVersion() == 0 {
+		badFields[globals.VersionField] = "Required field."
+	}
+	if req.GetApiTags() == nil {
+		badFields[globals.ApiTagsField] = "Must be non-empty."
+	}
+	for k, lv := range req.GetApiTags() {
+		if err := validateStringForDb(k); err != "" {
+			badFields[globals.ApiTagsField] = "Tag keys " + err
+			break
+		}
+		if lv.GetValues() == nil {
+			badFields[globals.ApiTagsField] = "Tag values must be non-empty."
+			break
+		}
+		for _, v := range lv.GetValues() {
+			if _, ok := v.GetKind().(*structpb.Value_StringValue); !ok {
+				badFields[globals.ApiTagsField] = "Tag values must be strings."
+				break
+			}
+			if err := validateStringForDb(v.GetStringValue()); err != "" {
+				badFields[globals.ApiTagsField] = "Tag values " + err
+				break
+			}
+		}
+	}
+	if len(badFields) > 0 {
+		return handlers.InvalidArgumentErrorf("Errors in provided fields.", badFields)
+	}
+	return nil
+}
+
+func validateSetTagsRequest(req *pbs.SetWorkerTagsRequest) error {
+	badFields := map[string]string{}
+	if !handlers.ValidId(handlers.Id(req.GetId()), server.WorkerPrefix) {
+		badFields[globals.IdField] = "Incorrectly formatted identifier."
+	}
+	if req.GetVersion() == 0 {
+		badFields[globals.VersionField] = "Required field."
+	}
+	for k, lv := range req.GetApiTags() {
+		if err := validateStringForDb(k); err != "" {
+			badFields[globals.ApiTagsField] = "Tag keys " + err
+			break
+		}
+		if lv.GetValues() == nil {
+			badFields[globals.ApiTagsField] = "Tag values must be non-empty."
+			break
+		}
+		for _, v := range lv.GetValues() {
+			if _, ok := v.GetKind().(*structpb.Value_StringValue); !ok {
+				badFields[globals.ApiTagsField] = "Tag values must be strings."
+				break
+			}
+			if err := validateStringForDb(v.GetStringValue()); err != "" {
+				badFields[globals.ApiTagsField] = "Tag values " + err
+				break
+			}
+		}
+	}
+	if len(badFields) > 0 {
+		return handlers.InvalidArgumentErrorf("Errors in provided fields.", badFields)
+	}
+	return nil
+}
+
+func validateRemoveTagsRequest(req *pbs.RemoveWorkerTagsRequest) error {
+	badFields := map[string]string{}
+	if !handlers.ValidId(handlers.Id(req.GetId()), server.WorkerPrefix) {
+		badFields[globals.IdField] = "Incorrectly formatted identifier."
+	}
+	if req.GetVersion() == 0 {
+		badFields[globals.VersionField] = "Required field."
+	}
+	if req.GetApiTags() == nil {
+		badFields[globals.ApiTagsField] = "Must be non-empty."
+	}
+	for k, lv := range req.GetApiTags() {
+		if err := validateStringForDb(k); err != "" {
+			badFields[globals.ApiTagsField] = "Tag keys " + err
+			break
+		}
+		if lv.GetValues() == nil {
+			badFields[globals.ApiTagsField] = "Tag values must be non-empty."
+			break
+		}
+		for _, v := range lv.GetValues() {
+			if _, ok := v.GetKind().(*structpb.Value_StringValue); !ok {
+				badFields[globals.ApiTagsField] = "Tag values must be strings."
+				break
+			}
+			if err := validateStringForDb(v.GetStringValue()); err != "" {
+				badFields[globals.ApiTagsField] = "Tag values " + err
+				break
+			}
+		}
+	}
+	if len(badFields) > 0 {
+		return handlers.InvalidArgumentErrorf("Errors in provided fields.", badFields)
+	}
+	return nil
+}

internal/daemon/controller/handlers/workers/worker_service_test.go

@@ -37,7 +37,7 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
-var testAuthorizedActions = []string{"no-op", "read", "update", "delete"}
+var testAuthorizedActions = []string{"no-op", "read", "update", "delete", "add-worker-tags", "set-worker-tags", "remove-worker-tags"}
 
 func structListValue(t *testing.T, ss ...string) *structpb.ListValue {
 	t.Helper()
@@ -50,6 +50,26 @@ func structListValue(t *testing.T, ss ...string) *structpb.ListValue {
 	return lv
 }
 
+func equalTags(t *testing.T, expected map[string]*structpb.ListValue, actual map[string]*structpb.ListValue) bool {
+	t.Helper()
+	if len(expected) != len(actual) {
+		return false
+	}
+	for eK, eValList := range expected {
+		found := false
+		for k, valList := range actual {
+			if eK == k && eValList.String() == valList.String() {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false
+		}
+	}
+	return true
+}
+
 func TestGet(t *testing.T) {
 	conn, _ := db.TestSetup(t, "postgres")
 	wrap := db.TestWrapper(t)
@@ -1295,3 +1315,479 @@ func TestCreateWorkerLed(t *testing.T) {
 		})
 	}
 }
+
+func TestService_AddWorkerTags(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	iamRepoFn := func() (*iam.Repository, error) {
+		return iamRepo, nil
+	}
+	_, proj := iam.TestScopes(t, iamRepo)
+	rw := db.New(conn)
+	testKms := kms.TestKms(t, conn, wrapper)
+	repoFn := func() (*server.Repository, error) {
+		return server.NewRepository(rw, rw, testKms)
+	}
+	s, err := NewService(context.Background(), repoFn, iamRepoFn)
+	require.NoError(err)
+	worker := server.TestKmsWorker(t, conn, wrapper)
+
+	tests := []struct {
+		name            string
+		req             *pbs.AddWorkerTagsRequest
+		wantTags        map[string]*structpb.ListValue
+		wantErrContains string
+	}{
+		{
+			name: "bad-id",
+			req: &pbs.AddWorkerTagsRequest{
+				Id:      "bad_id",
+				Version: worker.Version,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantErrContains: "Incorrectly formatted identifier.",
+		},
+		{
+			name: "nil-version",
+			req: &pbs.AddWorkerTagsRequest{
+				Id:      worker.PublicId,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantErrContains: "Required field.",
+		},
+		{
+			name: "nil-tags",
+			req: &pbs.AddWorkerTagsRequest{
+				Id:      worker.PublicId,
+				Version: worker.Version,
+			},
+			wantErrContains: "Must be non-empty.",
+		},
+		{
+			name: "valid-tags",
+			req: &pbs.AddWorkerTagsRequest{
+				Id:      worker.PublicId,
+				Version: worker.Version,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+		},
+		{
+			name: "many-valid-tags",
+			req: func() *pbs.AddWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.AddWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key":  {Values: []*structpb.Value{structpb.NewStringValue("value"), structpb.NewStringValue("value2")}},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantTags: map[string]*structpb.ListValue{
+				"key":  {Values: []*structpb.Value{structpb.NewStringValue("value"), structpb.NewStringValue("value2")}},
+				"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+				"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+				"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+			},
+		},
+		{
+			name: "invalid-tag-value",
+			req: func() *pbs.AddWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.AddWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewBoolValue(false)}}},
+				}
+			}(),
+			wantErrContains: "Tag values must be strings.",
+		},
+		{
+			name: "mixed-invalid-tags",
+			req: func() *pbs.AddWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.AddWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"":     {Values: nil},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{nil}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantErrContains: "Tag keys must be non-empty.",
+		},
+		{
+			name: "duplicate-tags",
+			req: func() *pbs.AddWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.AddWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2"), structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantErrContains: "Unable to add worker tags in repo",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := s.AddWorkerTags(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), tc.req)
+			if len(tc.wantErrContains) > 0 {
+				assert.Nil(got)
+				assert.Contains(err.Error(), tc.wantErrContains)
+			}
+			if tc.wantTags != nil {
+				assert.Nil(err)
+				assert.True(equalTags(t, tc.wantTags, got.GetItem().GetApiTags()), "want tags: %q got: %q", tc.wantTags, got.GetItem().GetApiTags())
+				assert.Equal(tc.req.Version+1, got.GetItem().GetVersion())
+			}
+		})
+	}
+}
+
+func TestService_SetWorkerTags(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	iamRepoFn := func() (*iam.Repository, error) {
+		return iamRepo, nil
+	}
+	_, proj := iam.TestScopes(t, iamRepo)
+	rw := db.New(conn)
+	testKms := kms.TestKms(t, conn, wrapper)
+	repoFn := func() (*server.Repository, error) {
+		return server.NewRepository(rw, rw, testKms)
+	}
+	s, err := NewService(context.Background(), repoFn, iamRepoFn)
+	require.NoError(err)
+	worker := server.TestKmsWorker(t, conn, wrapper)
+
+	tests := []struct {
+		name            string
+		req             *pbs.SetWorkerTagsRequest
+		wantTags        map[string]*structpb.ListValue
+		wantErrContains string
+	}{
+		{
+			name: "bad-id",
+			req: &pbs.SetWorkerTagsRequest{
+				Id:      "bad_id",
+				Version: worker.Version,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantErrContains: "Incorrectly formatted identifier.",
+		},
+		{
+			name: "nil-version",
+			req: &pbs.SetWorkerTagsRequest{
+				Id:      worker.PublicId,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantErrContains: "Required field.",
+		},
+		{
+			name: "nil-tags",
+			req: &pbs.SetWorkerTagsRequest{
+				Id:      worker.PublicId,
+				Version: worker.Version,
+			},
+			wantTags: map[string]*structpb.ListValue{},
+		},
+		{
+			name: "valid-tags",
+			req: func() *pbs.SetWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.SetWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+				}
+			}(),
+			wantTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+		},
+		{
+			name: "many-valid-tags",
+			req: func() *pbs.SetWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.SetWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key":  {Values: []*structpb.Value{structpb.NewStringValue("value"), structpb.NewStringValue("value2")}},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantTags: map[string]*structpb.ListValue{
+				"key":  {Values: []*structpb.Value{structpb.NewStringValue("value"), structpb.NewStringValue("value2")}},
+				"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+				"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+				"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+			},
+		},
+		{
+			name: "invalid-tag-value",
+			req: func() *pbs.SetWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.SetWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewBoolValue(false)}}},
+				}
+			}(),
+			wantErrContains: "Tag values must be strings.",
+		},
+		{
+			name: "mixed-invalid-tags",
+			req: func() *pbs.SetWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.SetWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key":  {Values: nil},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantErrContains: "Tag values must be non-empty.",
+		},
+		{
+			name: "duplicate-tags",
+			req: func() *pbs.SetWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.SetWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2"), structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantErrContains: "Unable to set worker tags in repo",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := s.SetWorkerTags(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), tc.req)
+			if len(tc.wantErrContains) > 0 {
+				assert.Nil(got)
+				assert.Contains(err.Error(), tc.wantErrContains)
+			}
+			if tc.wantTags != nil {
+				assert.Nil(err)
+				assert.True(equalTags(t, tc.wantTags, got.GetItem().GetApiTags()))
+				assert.Equal(tc.req.Version+1, got.GetItem().GetVersion())
+			}
+		})
+	}
+}
+
+func TestService_RemoveWorkerTags(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	iamRepoFn := func() (*iam.Repository, error) {
+		return iamRepo, nil
+	}
+	_, proj := iam.TestScopes(t, iamRepo)
+	rw := db.New(conn)
+	testKms := kms.TestKms(t, conn, wrapper)
+	repoFn := func() (*server.Repository, error) {
+		return server.NewRepository(rw, rw, testKms)
+	}
+	s, err := NewService(context.Background(), repoFn, iamRepoFn)
+	require.NoError(err)
+	worker := server.TestKmsWorker(t, conn, wrapper)
+
+	tests := []struct {
+		name            string
+		req             *pbs.RemoveWorkerTagsRequest
+		wantDeletedTags map[string]*structpb.ListValue
+		wantErrContains string
+	}{
+		{
+			name: "bad-id",
+			req: &pbs.RemoveWorkerTagsRequest{
+				Id:      "bad_id",
+				Version: worker.Version,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantErrContains: "Incorrectly formatted identifier.",
+		},
+		{
+			name: "nil-version",
+			req: &pbs.RemoveWorkerTagsRequest{
+				Id:      worker.PublicId,
+				ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+			},
+			wantErrContains: "Required field.",
+		},
+		{
+			name: "nil-tags",
+			req: &pbs.RemoveWorkerTagsRequest{
+				Id:      worker.PublicId,
+				Version: worker.Version,
+			},
+			wantErrContains: "Must be non-empty.",
+		},
+		{
+			name: "remove-valid-tag",
+			req: func() *pbs.RemoveWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				s.addTagsInRepo(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), worker.PublicId,
+					worker.Version, map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}})
+				return &pbs.RemoveWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version + 1,
+					ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+				}
+			}(),
+			wantDeletedTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewStringValue("value")}}},
+		},
+		{
+			name: "remove-many-valid-tags",
+			req: func() *pbs.RemoveWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				s.addTagsInRepo(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), worker.PublicId,
+					worker.Version, map[string]*structpb.ListValue{
+						"key":  {Values: []*structpb.Value{structpb.NewStringValue("value"), structpb.NewStringValue("value1")}},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+						"key5": {Values: []*structpb.Value{structpb.NewStringValue("value5")}},
+						"key6": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+					})
+				return &pbs.RemoveWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version + 1,
+					ApiTags: map[string]*structpb.ListValue{
+						"key":  {Values: []*structpb.Value{structpb.NewStringValue("value"), structpb.NewStringValue("value1")}},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+						"key5": {Values: []*structpb.Value{structpb.NewStringValue("value5")}},
+						"key6": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+					},
+				}
+			}(),
+			wantDeletedTags: map[string]*structpb.ListValue{
+				"key":  {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+				"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+				"key3": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+				"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+				"key5": {Values: []*structpb.Value{structpb.NewStringValue("value5")}},
+				"key6": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+			},
+		},
+		{
+			name: "invalid-tag-value",
+			req: func() *pbs.RemoveWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.RemoveWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{"key": {Values: []*structpb.Value{structpb.NewBoolValue(false)}}},
+				}
+			}(),
+			wantErrContains: "Tag values must be strings.",
+		},
+		{
+			name: "mixed-invalid-tags",
+			req: func() *pbs.RemoveWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.RemoveWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key":  {Values: nil},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+					},
+				}
+			}(),
+			wantErrContains: "Tag values must be non-empty.",
+		},
+		{
+			name: "remove-nonexistent-tags",
+			req: func() *pbs.RemoveWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				s.addTagsInRepo(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), worker.PublicId,
+					worker.Version, map[string]*structpb.ListValue{
+						"key":  {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+						"key5": {Values: []*structpb.Value{structpb.NewStringValue("value5")}},
+						"key6": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+					})
+				return &pbs.RemoveWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version + 1,
+					ApiTags: map[string]*structpb.ListValue{
+						"no-key": {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+						"key2":   {Values: []*structpb.Value{structpb.NewStringValue("value2")}},
+						"key":    {Values: []*structpb.Value{structpb.NewStringValue("value")}},
+					},
+				}
+			}(),
+			wantErrContains: "Unable to remove worker tags in repo:",
+		},
+		{
+			name: "duplicate-tags",
+			req: func() *pbs.RemoveWorkerTagsRequest {
+				worker := server.TestKmsWorker(t, conn, wrapper)
+				return &pbs.RemoveWorkerTagsRequest{
+					Id:      worker.PublicId,
+					Version: worker.Version,
+					ApiTags: map[string]*structpb.ListValue{
+						"key2": {Values: []*structpb.Value{structpb.NewStringValue("value2"), structpb.NewStringValue("value2")}},
+						"key3": {Values: []*structpb.Value{structpb.NewStringValue("value3")}},
+						"key4": {Values: []*structpb.Value{structpb.NewStringValue("value4")}},
+					},
+				}
+			}(),
+			wantErrContains: "Unable to remove worker tags in repo",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := s.RemoveWorkerTags(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), tc.req)
+			if len(tc.wantErrContains) > 0 {
+				assert.Nil(got)
+				assert.Contains(err.Error(), tc.wantErrContains)
+			}
+			if tc.wantDeletedTags != nil {
+				assert.Nil(err)
+				assert.False(equalTags(t, tc.wantDeletedTags, got.GetItem().GetApiTags()))
+				assert.Equal(tc.req.Version+1, got.GetItem().GetVersion())
+			}
+		})
+	}
+}