Implement UpdateWorker service method (#2133)

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

@@ -14,6 +14,7 @@ import (
 	"github.com/hashicorp/boundary/internal/perms"
 	"github.com/hashicorp/boundary/internal/requests"
 	"github.com/hashicorp/boundary/internal/servers"
+	"github.com/hashicorp/boundary/internal/servers/store"
 	"github.com/hashicorp/boundary/internal/types/action"
 	"github.com/hashicorp/boundary/internal/types/resource"
 	"github.com/hashicorp/boundary/internal/types/scope"
@@ -24,6 +25,8 @@ import (
 )
 
 var (
+	maskManager handlers.MaskManager
+
 	// IdActions contains the set of actions that can be performed on
 	// individual resources
 	IdActions = action.ActionSet{
@@ -41,6 +44,13 @@ var (
 	}
 )
 
+func init() {
+	var err error
+	if maskManager, err = handlers.NewMaskManager(handlers.MaskDestination{&store.Worker{}}, handlers.MaskSource{&pb.Worker{}}); err != nil {
+		panic(err)
+	}
+}
+
 // Service handles request as described by the pbs.WorkerServiceServer interface.
 type Service struct {
 	pbs.UnimplementedWorkerServiceServer
@@ -192,6 +202,44 @@ func (s Service) DeleteWorker(ctx context.Context, req *pbs.DeleteWorkerRequest)
 	return nil, nil
 }
 
+// UpdateWorker implements the interface pbs.WorkerServiceServer.
+func (s Service) UpdateWorker(ctx context.Context, req *pbs.UpdateWorkerRequest) (*pbs.UpdateWorkerResponse, error) {
+	const op = "workers.(Service).UpdateWorker"
+
+	if err := validateUpdateRequest(req); err != nil {
+		return nil, err
+	}
+	authResults := s.authResult(ctx, req.GetId(), action.Update)
+	if authResults.Error != nil {
+		return nil, authResults.Error
+	}
+	w, err := s.updateInRepo(ctx, authResults.Scope.GetId(), req.GetId(), req.GetUpdateMask().GetPaths(), req.GetItem())
+	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.UpdateWorkerResponse{Item: item}, nil
+}
+
 func (s Service) listFromRepo(ctx context.Context, scopeIds []string) ([]*servers.Worker, error) {
 	repo, err := s.repoFn()
 	if err != nil {
@@ -238,6 +286,38 @@ func (s Service) deleteFromRepo(ctx context.Context, id string) (bool, error) {
 	return rows > 0, nil
 }
 
+func (s Service) updateInRepo(ctx context.Context, scopeId, id string, mask []string, item *pb.Worker) (*servers.Worker, error) {
+	const op = "workers.(Service).updateInRepo"
+	var opts []servers.Option
+	if desc := item.GetDescription(); desc != nil {
+		opts = append(opts, servers.WithDescription(desc.GetValue()))
+	}
+	if name := item.GetName(); name != nil {
+		opts = append(opts, servers.WithName(name.GetValue()))
+	}
+	if addr := item.GetAddress(); addr != nil {
+		opts = append(opts, servers.WithAddress(addr.GetValue()))
+	}
+	w := servers.NewWorker(scopeId, opts...)
+	w.PublicId = id
+	dbMask := maskManager.Translate(mask)
+	if len(dbMask) == 0 {
+		return nil, handlers.InvalidArgumentErrorf("No valid fields included in the update mask.", map[string]string{"update_mask": "No valid fields provided in the update mask."})
+	}
+	repo, err := s.repoFn()
+	if err != nil {
+		return nil, err
+	}
+	out, rowsUpdated, err := repo.UpdateWorker(ctx, w, item.GetVersion(), dbMask)
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to update worker"))
+	}
+	if rowsUpdated == 0 {
+		return nil, handlers.NotFoundErrorf("Worker %q doesn't exist or incorrect version provided.", id)
+	}
+	return out, nil
+}
+
 func (s Service) authResult(ctx context.Context, id string, a action.Type) auth.VerifyResults {
 	res := auth.VerifyResults{}
 	repo, err := s.repoFn()
@@ -393,3 +473,22 @@ func validateListRequest(req *pbs.ListWorkersRequest) error {
 func validateDeleteRequest(req *pbs.DeleteWorkerRequest) error {
 	return handlers.ValidateDeleteRequest(handlers.NoopValidatorFn, req, servers.WorkerPrefix)
 }
+
+func validateUpdateRequest(req *pbs.UpdateWorkerRequest) error {
+	return handlers.ValidateUpdateRequest(req, req.GetItem(), func() map[string]string {
+		badFields := map[string]string{}
+		if req.GetItem().GetWorkerConfig() != nil {
+			badFields[globals.ConfigurationField] = "This is a read only field."
+		}
+		if req.GetItem().GetCanonicalAddress() != "" {
+			badFields[globals.CanonicalAddressField] = "This is a read only field."
+		}
+		if req.GetItem().GetCanonicalTags() != nil {
+			badFields[globals.CanonicalAddressField] = "This is a read only field."
+		}
+		if req.GetItem().GetTags() != nil {
+			badFields[globals.TagsField] = "These values cannot be updated in this type of request."
+		}
+		return badFields
+	}, servers.WorkerPrefix)
+}

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

@@ -20,8 +20,12 @@ import (
 	pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/workers"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/genproto/protobuf/field_mask"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/structpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
@@ -284,3 +288,462 @@ func TestDelete(t *testing.T) {
 		})
 	}
 }
+
+func TestUpdate(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	ctx := context.Background()
+	rw := db.New(conn)
+
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	iamRepoFn := func() (*iam.Repository, error) {
+		return iamRepo, nil
+	}
+	repo, err := servers.NewRepository(rw, rw, kms)
+	require.NoError(t, err)
+	repoFn := func() (*servers.Repository, error) {
+		return repo, nil
+	}
+
+	wkr := servers.TestWorker(t, conn, wrapper,
+		servers.WithName("default"),
+		servers.WithDescription("default"),
+		servers.WithAddress("default"))
+
+	version := wkr.GetVersion()
+
+	resetWorker := func() {
+		version++
+		_, _, err = repo.UpdateWorker(context.Background(), wkr, version, []string{"Name", "Description", "Address"})
+		require.NoError(t, err, "Failed to reset worker.")
+		version++
+	}
+
+	wCreated := wkr.GetCreateTime().GetTimestamp().AsTime()
+	toMerge := &pbs.UpdateWorkerRequest{
+		Id: wkr.GetPublicId(),
+	}
+	workerService, err := NewService(ctx, repoFn, iamRepoFn)
+	require.NoError(t, err)
+	expectedScope := &scopes.ScopeInfo{Id: scope.Global.String(), Type: scope.Global.String(), Name: scope.Global.String(), Description: "Global Scope"}
+	expectedConfig := &pb.WorkerConfig{
+		Address: wkr.GetWorkerReportedAddress(),
+		Name:    wkr.GetWorkerReportedName(),
+	}
+
+	cases := []struct {
+		name string
+		req  *pbs.UpdateWorkerRequest
+		res  *pbs.UpdateWorkerResponse
+		err  error
+	}{
+		{
+			name: "Update an Existing Worker",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"name", "description", "address"},
+				},
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("name"),
+					Description: wrapperspb.String("desc"),
+					Address:     wrapperspb.String("address"),
+				},
+			},
+			res: &pbs.UpdateWorkerResponse{
+				Item: &pb.Worker{
+					Id:                wkr.GetPublicId(),
+					ScopeId:           wkr.GetScopeId(),
+					Scope:             expectedScope,
+					Name:              wrapperspb.String("name"),
+					Description:       wrapperspb.String("desc"),
+					Address:           wrapperspb.String("address"),
+					CanonicalAddress:  "address",
+					CreatedTime:       wkr.GetCreateTime().GetTimestamp(),
+					LastStatusTime:    wkr.GetLastStatusTime().GetTimestamp(),
+					WorkerConfig:      expectedConfig,
+					AuthorizedActions: testAuthorizedActions,
+				},
+			},
+		},
+		{
+			name: "Multiple Paths in single string",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"name,description,type"},
+				},
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("name"),
+					Description: wrapperspb.String("desc"),
+				},
+			},
+			res: &pbs.UpdateWorkerResponse{
+				Item: &pb.Worker{
+					Id:                wkr.GetPublicId(),
+					ScopeId:           wkr.GetScopeId(),
+					Scope:             expectedScope,
+					Name:              wrapperspb.String("name"),
+					Description:       wrapperspb.String("desc"),
+					Address:           wrapperspb.String("default"),
+					CanonicalAddress:  "default",
+					CreatedTime:       wkr.GetCreateTime().GetTimestamp(),
+					LastStatusTime:    wkr.GetLastStatusTime().GetTimestamp(),
+					WorkerConfig:      expectedConfig,
+					AuthorizedActions: testAuthorizedActions,
+				},
+			},
+		},
+		{
+			name: "No Update Mask",
+			req: &pbs.UpdateWorkerRequest{
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("updated name"),
+					Description: wrapperspb.String("updated desc"),
+				},
+			},
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Empty Path",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{Paths: []string{}},
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("updated name"),
+					Description: wrapperspb.String("updated desc"),
+				},
+			},
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Update port to 0",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{Paths: []string{"default_port"}},
+				Item:       &pb.Worker{},
+			},
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Only non-existant paths in Mask",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{Paths: []string{"nonexistant_field"}},
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("updated name"),
+					Description: wrapperspb.String("updated desc"),
+				},
+			},
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Unset Name",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"name"},
+				},
+				Item: &pb.Worker{
+					Description: wrapperspb.String("ignored"),
+				},
+			},
+			res: &pbs.UpdateWorkerResponse{
+				Item: &pb.Worker{
+					Id:                wkr.GetPublicId(),
+					ScopeId:           wkr.GetScopeId(),
+					Scope:             expectedScope,
+					Description:       wrapperspb.String("default"),
+					Address:           wrapperspb.String("default"),
+					CanonicalAddress:  "default",
+					CreatedTime:       wkr.GetCreateTime().GetTimestamp(),
+					LastStatusTime:    wkr.GetLastStatusTime().GetTimestamp(),
+					WorkerConfig:      expectedConfig,
+					AuthorizedActions: testAuthorizedActions,
+				},
+			},
+		},
+		{
+			name: "Unset Description",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"description"},
+				},
+				Item: &pb.Worker{
+					Name: wrapperspb.String("ignored"),
+				},
+			},
+			res: &pbs.UpdateWorkerResponse{
+				Item: &pb.Worker{
+					Id:                wkr.GetPublicId(),
+					ScopeId:           wkr.GetScopeId(),
+					Scope:             expectedScope,
+					Name:              wrapperspb.String("default"),
+					Address:           wrapperspb.String("default"),
+					CanonicalAddress:  "default",
+					CreatedTime:       wkr.GetCreateTime().GetTimestamp(),
+					LastStatusTime:    wkr.GetLastStatusTime().GetTimestamp(),
+					WorkerConfig:      expectedConfig,
+					AuthorizedActions: testAuthorizedActions,
+				},
+			},
+		},
+		{
+			name: "Update Only Name",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"name"},
+				},
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("updated"),
+					Description: wrapperspb.String("ignored"),
+				},
+			},
+			res: &pbs.UpdateWorkerResponse{
+				Item: &pb.Worker{
+					Id:                wkr.GetPublicId(),
+					ScopeId:           wkr.GetScopeId(),
+					Scope:             expectedScope,
+					Name:              wrapperspb.String("updated"),
+					Description:       wrapperspb.String("default"),
+					Address:           wrapperspb.String("default"),
+					CanonicalAddress:  "default",
+					CreatedTime:       wkr.GetCreateTime().GetTimestamp(),
+					LastStatusTime:    wkr.GetLastStatusTime().GetTimestamp(),
+					WorkerConfig:      expectedConfig,
+					AuthorizedActions: testAuthorizedActions,
+				},
+			},
+		},
+		{
+			name: "Update Only Description",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"description"},
+				},
+				Item: &pb.Worker{
+					Name:        wrapperspb.String("ignored"),
+					Description: wrapperspb.String("notignored"),
+				},
+			},
+			res: &pbs.UpdateWorkerResponse{
+				Item: &pb.Worker{
+					Id:                wkr.GetPublicId(),
+					ScopeId:           wkr.GetScopeId(),
+					Scope:             expectedScope,
+					Name:              wrapperspb.String("default"),
+					Description:       wrapperspb.String("notignored"),
+					Address:           wrapperspb.String("default"),
+					CanonicalAddress:  "default",
+					CreatedTime:       wkr.GetCreateTime().GetTimestamp(),
+					LastStatusTime:    wkr.GetLastStatusTime().GetTimestamp(),
+					WorkerConfig:      expectedConfig,
+					AuthorizedActions: testAuthorizedActions,
+				},
+			},
+		},
+		{
+			name: "Update a Non Existing Worker",
+			req: &pbs.UpdateWorkerRequest{
+				Id: servers.WorkerPrefix + "_DoesntExis",
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"description"},
+				},
+				Item: &pb.Worker{
+					Description: wrapperspb.String("desc"),
+				},
+			},
+			err: handlers.ApiErrorWithCode(codes.NotFound),
+		},
+		{
+			name: "Cant change Worker Defined Name",
+			req: &pbs.UpdateWorkerRequest{
+				Id: wkr.GetPublicId(),
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"worker_config.name"},
+				},
+				Item: &pb.Worker{
+					WorkerConfig: &pb.WorkerConfig{
+						Name: "name",
+					},
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant change Worker Defined Address",
+			req: &pbs.UpdateWorkerRequest{
+				Id: wkr.GetPublicId(),
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"worker_config.address"},
+				},
+				Item: &pb.Worker{
+					WorkerConfig: &pb.WorkerConfig{
+						Address: "address",
+					},
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant change Id",
+			req: &pbs.UpdateWorkerRequest{
+				Id: wkr.GetPublicId(),
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"id"},
+				},
+				Item: &pb.Worker{
+					Id: "w_somethinge",
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant specify Created Time",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"created_time"},
+				},
+				Item: &pb.Worker{
+					CreatedTime: timestamppb.Now(),
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant specify Updated Time",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"updated_time"},
+				},
+				Item: &pb.Worker{
+					UpdatedTime: timestamppb.Now(),
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant specify tags",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"tags"},
+				},
+				Item: &pb.Worker{
+					Tags: map[string]*structpb.ListValue{
+						"foo": func() *structpb.ListValue {
+							l, err := structpb.NewList([]interface{}{"bar"})
+							require.NoError(t, err)
+							return l
+						}(),
+					},
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant specify canonical tags",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"canonical_tags"},
+				},
+				Item: &pb.Worker{
+					CanonicalTags: map[string]*structpb.ListValue{
+						"foo": func() *structpb.ListValue {
+							l, err := structpb.NewList([]interface{}{"bar"})
+							require.NoError(t, err)
+							return l
+						}(),
+					},
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+		{
+			name: "Cant specify canonical address",
+			req: &pbs.UpdateWorkerRequest{
+				UpdateMask: &field_mask.FieldMask{
+					Paths: []string{"canonical_address"},
+				},
+				Item: &pb.Worker{
+					CanonicalAddress: "should_fail",
+				},
+			},
+			res: nil,
+			err: handlers.ApiErrorWithCode(codes.InvalidArgument),
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			tc.req.Item.Version = version
+
+			req := proto.Clone(toMerge).(*pbs.UpdateWorkerRequest)
+			proto.Merge(req, tc.req)
+
+			got, gErr := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, scope.Global.String()), req)
+			if tc.err != nil {
+				require.Error(t, gErr)
+				assert.True(t, errors.Is(gErr, tc.err), "UpdateWorker(%+v) got error %v, wanted %v", req, gErr, tc.err)
+			}
+
+			if tc.err == nil {
+				defer resetWorker()
+			}
+
+			if got != nil {
+				assert.NotNilf(t, tc.res, "Expected UpdateWorker response to be nil, but was %v", got)
+				gotUpdateTime := got.GetItem().GetUpdatedTime().AsTime()
+				// Verify updated set after it was created
+				assert.True(t, gotUpdateTime.After(wCreated), "Updated resource should have been updated after its creation. Was updated %v, which is after %v", gotUpdateTime, wCreated)
+
+				// Clear all values which are hard to compare against.
+				got.Item.UpdatedTime, tc.res.Item.UpdatedTime = nil, nil
+			}
+			if tc.res != nil {
+				tc.res.Item.Version = version + 1
+			}
+			assert.Empty(t, cmp.Diff(got, tc.res, protocmp.Transform()), "UpdateWorker(%q) got response %q, wanted %q", req, got, tc.res)
+		})
+	}
+}
+
+func TestUpdate_BadVersion(t *testing.T) {
+	t.Parallel()
+	conn, _ := db.TestSetup(t, "postgres")
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+
+	iamRepo := iam.TestRepo(t, conn, wrapper)
+	iamRepoFn := func() (*iam.Repository, error) {
+		return iamRepo, nil
+	}
+
+	_, proj := iam.TestScopes(t, iamRepo)
+
+	rw := db.New(conn)
+	repo, err := servers.NewRepository(rw, rw, kms)
+	require.NoError(t, err, "Couldn't create new worker repo.")
+	repoFn := func() (*servers.Repository, error) {
+		return repo, nil
+	}
+	ctx := context.Background()
+
+	workerService, err := NewService(ctx, repoFn, iamRepoFn)
+	require.NoError(t, err, "Failed to create a new host set service.")
+
+	wkr := servers.TestWorker(t, conn, wrapper)
+
+	upTar, err := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, proj.GetPublicId()), &pbs.UpdateWorkerRequest{
+		Id: wkr.GetPublicId(),
+		Item: &pb.Worker{
+			Description: wrapperspb.String("updated"),
+			Version:     72,
+		},
+		UpdateMask: &field_mask.FieldMask{Paths: []string{"description"}},
+	})
+	assert.Nil(t, upTar)
+	assert.Error(t, err)
+	assert.True(t, errors.Is(err, handlers.NotFoundError()), "Got %v, wanted not found error.", err)
+}