resource: add MutateAndValidate endpoint (#20311)

agent/grpc-external/services/resource/mutate_and_validate.go

@@ -0,0 +1,139 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resource
+
+import (
+	"context"
+	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func (s *Server) MutateAndValidate(ctx context.Context, req *pbresource.MutateAndValidateRequest) (*pbresource.MutateAndValidateResponse, error) {
+	tenancyMarkedForDeletion, err := s.mutateAndValidate(ctx, req.Resource)
+	if err != nil {
+		return nil, err
+	}
+
+	if tenancyMarkedForDeletion {
+		return nil, status.Errorf(
+			codes.InvalidArgument,
+			"tenancy marked for deletion: %s/%s",
+			req.Resource.Id.Tenancy.Partition,
+			req.Resource.Id.Tenancy.Namespace,
+		)
+	}
+	return &pbresource.MutateAndValidateResponse{Resource: req.Resource}, nil
+}
+
+// private DRY impl that is used by both the Write and MutateAndValidate RPCs.
+func (s *Server) mutateAndValidate(ctx context.Context, res *pbresource.Resource) (tenancyMarkedForDeletion bool, err error) {
+	reg, err := s.ensureResourceValid(res)
+	if err != nil {
+		return false, err
+	}
+
+	v1EntMeta := v2TenancyToV1EntMeta(res.Id.Tenancy)
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), v1EntMeta)
+	if err != nil {
+		return false, err
+	}
+	v1EntMetaToV2Tenancy(reg, v1EntMeta, res.Id.Tenancy)
+
+	// Check the user sent the correct type of data.
+	if res.Data != nil && !res.Data.MessageIs(reg.Proto) {
+		got := strings.TrimPrefix(res.Data.TypeUrl, "type.googleapis.com/")
+
+		return false, status.Errorf(
+			codes.InvalidArgument,
+			"resource.data is of wrong type (expected=%q, got=%q)",
+			reg.Proto.ProtoReflect().Descriptor().FullName(),
+			got,
+		)
+	}
+
+	if err = reg.Mutate(res); err != nil {
+		return false, status.Errorf(codes.Internal, "failed mutate hook: %v", err.Error())
+	}
+
+	if err = reg.Validate(res); err != nil {
+		return false, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	// ACL check comes before tenancy existence checks to not leak tenancy "existence".
+	err = reg.ACLs.Write(authz, authzContext, res)
+	switch {
+	case acl.IsErrPermissionDenied(err):
+		return false, status.Error(codes.PermissionDenied, err.Error())
+	case err != nil:
+		return false, status.Errorf(codes.Internal, "failed write acl: %v", err)
+	}
+
+	// Check tenancy exists for the V2 resource
+	if err = tenancyExists(reg, s.TenancyBridge, res.Id.Tenancy, codes.InvalidArgument); err != nil {
+		return false, err
+	}
+
+	// This is used later in the "create" and "update" paths to block non-delete related writes
+	// when a tenancy unit has been marked for deletion.
+	tenancyMarkedForDeletion, err = isTenancyMarkedForDeletion(reg, s.TenancyBridge, res.Id.Tenancy)
+	if err != nil {
+		return false, status.Errorf(codes.Internal, "failed tenancy marked for deletion check: %v", err)
+	}
+	if tenancyMarkedForDeletion {
+		return true, nil
+	}
+	return false, nil
+}
+
+func (s *Server) ensureResourceValid(res *pbresource.Resource) (*resource.Registration, error) {
+	var field string
+	switch {
+	case res == nil:
+		field = "resource"
+	case res.Id == nil:
+		field = "resource.id"
+	}
+
+	if field != "" {
+		return nil, status.Errorf(codes.InvalidArgument, "%s is required", field)
+	}
+
+	if err := validateId(res.Id, "resource.id"); err != nil {
+		return nil, err
+	}
+
+	if res.Owner != nil {
+		if err := validateId(res.Owner, "resource.owner"); err != nil {
+			return nil, err
+		}
+	}
+
+	// Check type exists.
+	reg, err := s.resolveType(res.Id.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = checkV2Tenancy(s.UseV2Tenancy, res.Id.Type); err != nil {
+		return nil, err
+	}
+
+	// Check scope
+	if reg.Scope == resource.ScopePartition && res.Id.Tenancy.Namespace != "" {
+		return nil, status.Errorf(
+			codes.InvalidArgument,
+			"partition scoped resource %s cannot have a namespace. got: %s",
+			resource.ToGVK(res.Id.Type),
+			res.Id.Tenancy.Namespace,
+		)
+	}
+
+	return reg, nil
+}

agent/grpc-external/services/resource/mutate_and_validate_test.go

@@ -0,0 +1,212 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resource_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	svc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
+	svctest "github.com/hashicorp/consul/agent/grpc-external/services/resource/testing"
+	"github.com/hashicorp/consul/internal/resource/demo"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	pbdemov2 "github.com/hashicorp/consul/proto/private/pbdemo/v2"
+	"github.com/hashicorp/consul/proto/private/prototest"
+)
+
+func TestMutateAndValidate_InputValidation(t *testing.T) {
+	run := func(t *testing.T, client pbresource.ResourceServiceClient, tc resourceValidTestCase) {
+		artist, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		recordLabel, err := demo.GenerateV1RecordLabel("looney-tunes")
+		require.NoError(t, err)
+
+		req := &pbresource.MutateAndValidateRequest{Resource: tc.modFn(artist, recordLabel)}
+		_, err = client.MutateAndValidate(testContext(t), req)
+		require.Error(t, err)
+		require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+		require.ErrorContains(t, err, tc.errContains)
+	}
+
+	for _, v2tenancy := range []bool{false, true} {
+		t.Run(fmt.Sprintf("v2tenancy %v", v2tenancy), func(t *testing.T) {
+			client := svctest.NewResourceServiceBuilder().
+				WithRegisterFns(demo.RegisterTypes).
+				WithV2Tenancy(v2tenancy).
+				Run(t)
+
+			for desc, tc := range resourceValidTestCases(t) {
+				t.Run(desc, func(t *testing.T) {
+					run(t, client, tc)
+				})
+			}
+		})
+	}
+}
+
+func TestMutateAndValidate_OwnerValidation(t *testing.T) {
+	run := func(t *testing.T, client pbresource.ResourceServiceClient, tc ownerValidTestCase) {
+		artist, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		album, err := demo.GenerateV2Album(artist.Id)
+		require.NoError(t, err)
+
+		tc.modFn(album)
+
+		_, err = client.MutateAndValidate(testContext(t), &pbresource.MutateAndValidateRequest{Resource: album})
+		require.Error(t, err)
+		require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+		require.ErrorContains(t, err, tc.errorContains)
+	}
+
+	for _, v2tenancy := range []bool{false, true} {
+		t.Run(fmt.Sprintf("v2tenancy %v", v2tenancy), func(t *testing.T) {
+			client := svctest.NewResourceServiceBuilder().
+				WithRegisterFns(demo.RegisterTypes).
+				WithV2Tenancy(v2tenancy).
+				Run(t)
+
+			for desc, tc := range ownerValidationTestCases(t) {
+				t.Run(desc, func(t *testing.T) {
+					run(t, client, tc)
+				})
+			}
+		})
+	}
+}
+
+func TestMutateAndValidate_TypeNotFound(t *testing.T) {
+	run := func(t *testing.T, client pbresource.ResourceServiceClient) {
+		res, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		_, err = client.MutateAndValidate(testContext(t), &pbresource.MutateAndValidateRequest{Resource: res})
+		require.Error(t, err)
+		require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+		require.Contains(t, err.Error(), "resource type demo.v2.Artist not registered")
+	}
+
+	for _, v2tenancy := range []bool{false, true} {
+		t.Run(fmt.Sprintf("v2tenancy %v", v2tenancy), func(t *testing.T) {
+			client := svctest.NewResourceServiceBuilder().WithV2Tenancy(v2tenancy).Run(t)
+			run(t, client)
+		})
+	}
+}
+
+func TestMutateAndValidate_Success(t *testing.T) {
+	run := func(t *testing.T, client pbresource.ResourceServiceClient, tc mavOrWriteSuccessTestCase) {
+		recordLabel, err := demo.GenerateV1RecordLabel("looney-tunes")
+		require.NoError(t, err)
+
+		artist, err := demo.GenerateV2Artist()
+		require.NoError(t, err)
+
+		rsp, err := client.MutateAndValidate(testContext(t), &pbresource.MutateAndValidateRequest{Resource: tc.modFn(artist, recordLabel)})
+		require.NoError(t, err)
+		prototest.AssertDeepEqual(t, tc.expectedTenancy, rsp.Resource.Id.Tenancy)
+	}
+
+	for _, v2tenancy := range []bool{false, true} {
+		t.Run(fmt.Sprintf("v2tenancy %v", v2tenancy), func(t *testing.T) {
+			client := svctest.NewResourceServiceBuilder().
+				WithRegisterFns(demo.RegisterTypes).
+				WithV2Tenancy(v2tenancy).
+				Run(t)
+
+			for desc, tc := range mavOrWriteSuccessTestCases(t) {
+				t.Run(desc, func(t *testing.T) {
+					run(t, client, tc)
+				})
+			}
+		})
+	}
+}
+
+func TestMutateAndValidate_Mutate(t *testing.T) {
+	for _, v2tenancy := range []bool{false, true} {
+		t.Run(fmt.Sprintf("v2tenancy %v", v2tenancy), func(t *testing.T) {
+			client := svctest.NewResourceServiceBuilder().
+				WithRegisterFns(demo.RegisterTypes).
+				WithV2Tenancy(v2tenancy).
+				Run(t)
+
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+
+			artistData := &pbdemov2.Artist{}
+			artist.Data.UnmarshalTo(artistData)
+			require.NoError(t, err)
+
+			// mutate hook sets genre to disco when unspecified
+			artistData.Genre = pbdemov2.Genre_GENRE_UNSPECIFIED
+			artist.Data.MarshalFrom(artistData)
+			require.NoError(t, err)
+
+			rsp, err := client.MutateAndValidate(testContext(t), &pbresource.MutateAndValidateRequest{Resource: artist})
+			require.NoError(t, err)
+
+			// verify mutate hook set genre to disco
+			require.NoError(t, rsp.Resource.Data.UnmarshalTo(artistData))
+			require.Equal(t, pbdemov2.Genre_GENRE_DISCO, artistData.Genre)
+		})
+	}
+}
+
+func TestMutateAndValidate_Tenancy_NotFound(t *testing.T) {
+	for desc, tc := range mavOrWriteTenancyNotFoundTestCases(t) {
+		t.Run(desc, func(t *testing.T) {
+			client := svctest.NewResourceServiceBuilder().
+				WithV2Tenancy(true).
+				WithRegisterFns(demo.RegisterTypes).
+				Run(t)
+
+			recordLabel, err := demo.GenerateV1RecordLabel("looney-tunes")
+			require.NoError(t, err)
+
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+
+			_, err = client.MutateAndValidate(testContext(t), &pbresource.MutateAndValidateRequest{Resource: tc.modFn(artist, recordLabel)})
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+			require.Contains(t, err.Error(), tc.errContains)
+		})
+	}
+}
+
+func TestMutateAndValidate_TenancyMarkedForDeletion_Fails(t *testing.T) {
+	for desc, tc := range mavOrWriteTenancyMarkedForDeletionTestCases(t) {
+		t.Run(desc, func(t *testing.T) {
+			server := testServer(t)
+			client := testClient(t, server)
+			demo.RegisterTypes(server.Registry)
+
+			recordLabel, err := demo.GenerateV1RecordLabel("looney-tunes")
+			require.NoError(t, err)
+			recordLabel.Id.Tenancy.Partition = "ap1"
+
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+			artist.Id.Tenancy.Partition = "ap1"
+			artist.Id.Tenancy.Namespace = "ns1"
+
+			mockTenancyBridge := &svc.MockTenancyBridge{}
+			mockTenancyBridge.On("PartitionExists", "ap1").Return(true, nil)
+			mockTenancyBridge.On("NamespaceExists", "ap1", "ns1").Return(true, nil)
+			server.TenancyBridge = mockTenancyBridge
+
+			_, err = client.MutateAndValidate(testContext(t), &pbresource.MutateAndValidateRequest{Resource: tc.modFn(artist, recordLabel, mockTenancyBridge)})
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+			require.Contains(t, err.Error(), tc.errContains)
+		})
+	}
+}

agent/grpc-external/services/resource/testing/builder.go

@@ -193,7 +193,7 @@ func (b *Builder) Run(t testutil.TestingTB) pbresource.ResourceServiceClient {
 	switch config.TenancyBridge.(type) {
 	case *tenancy.V2TenancyBridge:
 		config.TenancyBridge.(*tenancy.V2TenancyBridge).WithClient(client)
-		// Default partition namespace can finally be created
+		// Default partition and namespace can finally be created
 		require.NoError(t, initTenancy(ctx, backend))
 
 		for _, tenancy := range b.tenancies {

agent/grpc-external/services/resource/testing/testing_ce.go

@@ -29,7 +29,8 @@ func FillAuthorizerContext(authzContext *acl.AuthorizerContext) {
 	// nothing to to in CE.
 }
 
-// initTenancy create the base tenancy objects (default/default)
+// initTenancy creates the builtin v2 namespace resource only. The builtin
+// v2 partition is not created because we're in CE.
 func initTenancy(ctx context.Context, b *inmem.Backend) error {
 	nsData, err := anypb.New(&pbtenancy.Namespace{Description: "default namespace in default partition"})
 	if err != nil {

agent/grpc-external/services/resource/write.go

@@ -6,7 +6,6 @@ package resource
 import (
 	"context"
 	"errors"
-	"strings"
 
 	"github.com/oklog/ulid/v2"
 	"golang.org/x/exp/maps"
@@ -14,7 +13,6 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
-	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/storage"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -37,59 +35,11 @@ import (
 var errUseWriteStatus = status.Error(codes.InvalidArgument, "resource.status can only be set using the WriteStatus endpoint")
 
 func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbresource.WriteResponse, error) {
-	reg, err := s.ensureWriteRequestValid(req)
+	tenancyMarkedForDeletion, err := s.mutateAndValidate(ctx, req.Resource)
 	if err != nil {
 		return nil, err
 	}
 
-	v1EntMeta := v2TenancyToV1EntMeta(req.Resource.Id.Tenancy)
-	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), v1EntMeta)
-	if err != nil {
-		return nil, err
-	}
-	v1EntMetaToV2Tenancy(reg, v1EntMeta, req.Resource.Id.Tenancy)
-
-	// Check the user sent the correct type of data.
-	if req.Resource.Data != nil && !req.Resource.Data.MessageIs(reg.Proto) {
-		got := strings.TrimPrefix(req.Resource.Data.TypeUrl, "type.googleapis.com/")
-
-		return nil, status.Errorf(
-			codes.InvalidArgument,
-			"resource.data is of wrong type (expected=%q, got=%q)",
-			reg.Proto.ProtoReflect().Descriptor().FullName(),
-			got,
-		)
-	}
-
-	if err = reg.Mutate(req.Resource); err != nil {
-		return nil, status.Errorf(codes.Internal, "failed mutate hook: %v", err.Error())
-	}
-
-	if err = reg.Validate(req.Resource); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
-	// ACL check comes before tenancy existence checks to not leak tenancy "existence".
-	err = reg.ACLs.Write(authz, authzContext, req.Resource)
-	switch {
-	case acl.IsErrPermissionDenied(err):
-		return nil, status.Error(codes.PermissionDenied, err.Error())
-	case err != nil:
-		return nil, status.Errorf(codes.Internal, "failed write acl: %v", err)
-	}
-
-	// Check tenancy exists for the V2 resource
-	if err = tenancyExists(reg, s.TenancyBridge, req.Resource.Id.Tenancy, codes.InvalidArgument); err != nil {
-		return nil, err
-	}
-
-	// This is used later in the "create" and "update" paths to block non-delete related writes
-	// when a tenancy unit has been marked for deletion.
-	tenancyMarkedForDeletion, err := isTenancyMarkedForDeletion(reg, s.TenancyBridge, req.Resource.Id.Tenancy)
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "failed tenancy marked for deletion check: %v", err)
-	}
-
 	// At the storage backend layer, all writes are CAS operations.
 	//
 	// This makes it possible to *safely* do things like keeping the Uid stable
@@ -251,52 +201,6 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 	return &pbresource.WriteResponse{Resource: result}, nil
 }
 
-func (s *Server) ensureWriteRequestValid(req *pbresource.WriteRequest) (*resource.Registration, error) {
-	var field string
-	switch {
-	case req.Resource == nil:
-		field = "resource"
-	case req.Resource.Id == nil:
-		field = "resource.id"
-	}
-
-	if field != "" {
-		return nil, status.Errorf(codes.InvalidArgument, "%s is required", field)
-	}
-
-	if err := validateId(req.Resource.Id, "resource.id"); err != nil {
-		return nil, err
-	}
-
-	if req.Resource.Owner != nil {
-		if err := validateId(req.Resource.Owner, "resource.owner"); err != nil {
-			return nil, err
-		}
-	}
-
-	// Check type exists.
-	reg, err := s.resolveType(req.Resource.Id.Type)
-	if err != nil {
-		return nil, err
-	}
-
-	if err = checkV2Tenancy(s.UseV2Tenancy, req.Resource.Id.Type); err != nil {
-		return nil, err
-	}
-
-	// Check scope
-	if reg.Scope == resource.ScopePartition && req.Resource.Id.Tenancy.Namespace != "" {
-		return nil, status.Errorf(
-			codes.InvalidArgument,
-			"partition scoped resource %s cannot have a namespace. got: %s",
-			resource.ToGVK(req.Resource.Id.Type),
-			req.Resource.Id.Tenancy.Namespace,
-		)
-	}
-
-	return reg, nil
-}
-
 func ensureMetadataSameExceptFor(input *pbresource.Resource, existing *pbresource.Resource, ignoreKey string) error {
 	// Work on copies since we're mutating them
 	inputCopy := maps.Clone(input.Metadata)

agent/grpc-external/services/resource/write_mav_common_test.go

@@ -0,0 +1,328 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resource_test
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	svc "github.com/hashicorp/consul/agent/grpc-external/services/resource"
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	pbdemov2 "github.com/hashicorp/consul/proto/private/pbdemo/v2"
+)
+
+// Common test structs and test cases shared by the Write and MutateAndValidate RPCs
+// only. These are not intended to be used by other tests.
+
+type resourceValidTestCase struct {
+	modFn       func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
+	errContains string
+}
+
+func resourceValidTestCases(t *testing.T) map[string]resourceValidTestCase {
+	return map[string]resourceValidTestCase{
+		"no resource": {
+			modFn: func(_, _ *pbresource.Resource) *pbresource.Resource {
+				return nil
+			},
+			errContains: "resource is required",
+		},
+		"no id": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id = nil
+				return artist
+			},
+			errContains: "resource.id is required",
+		},
+		"no type": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Type = nil
+				return artist
+			},
+			errContains: "resource.id.type is required",
+		},
+		"no name": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Name = ""
+				return artist
+			},
+			errContains: "resource.id.name invalid",
+		},
+		"name is mixed case": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Name = "MixedCaseNotAllowed"
+				return artist
+			},
+			errContains: "resource.id.name invalid",
+		},
+		"name too long": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Name = strings.Repeat("a", resource.MaxNameLength+1)
+				return artist
+			},
+			errContains: "resource.id.name invalid",
+		},
+		"wrong data type": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				var err error
+				artist.Data, err = anypb.New(&pbdemov2.Album{})
+				require.NoError(t, err)
+				return artist
+			},
+			errContains: "resource.data is of wrong type",
+		},
+		"partition is mixed case": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = "Default"
+				return artist
+			},
+			errContains: "resource.id.tenancy.partition invalid",
+		},
+		"partition too long": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = strings.Repeat("p", resource.MaxNameLength+1)
+				return artist
+			},
+			errContains: "resource.id.tenancy.partition invalid",
+		},
+		"namespace is mixed case": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Namespace = "Default"
+				return artist
+			},
+			errContains: "resource.id.tenancy.namespace invalid",
+		},
+		"namespace too long": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Namespace = strings.Repeat("n", resource.MaxNameLength+1)
+				return artist
+			},
+			errContains: "resource.id.tenancy.namespace invalid",
+		},
+		"fail validation hook": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				buffer := &pbdemov2.Artist{}
+				require.NoError(t, artist.Data.UnmarshalTo(buffer))
+				buffer.Name = "" // name cannot be empty
+				require.NoError(t, artist.Data.MarshalFrom(buffer))
+				return artist
+			},
+			errContains: "artist.name required",
+		},
+		"partition scope with non-empty namespace": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.Namespace = "bogus"
+				return recordLabel
+			},
+			errContains: "cannot have a namespace",
+		},
+	}
+}
+
+type ownerValidTestCase struct {
+	modFn         func(res *pbresource.Resource)
+	errorContains string
+}
+
+func ownerValidationTestCases(t *testing.T) map[string]ownerValidTestCase {
+	return map[string]ownerValidTestCase{
+		"no owner type": {
+			modFn:         func(res *pbresource.Resource) { res.Owner.Type = nil },
+			errorContains: "resource.owner.type is required",
+		},
+		"no owner name": {
+			modFn:         func(res *pbresource.Resource) { res.Owner.Name = "" },
+			errorContains: "resource.owner.name invalid",
+		},
+		"mixed case owner name": {
+			modFn:         func(res *pbresource.Resource) { res.Owner.Name = strings.ToUpper(res.Owner.Name) },
+			errorContains: "resource.owner.name invalid",
+		},
+		"owner name too long": {
+			modFn: func(res *pbresource.Resource) {
+				res.Owner.Name = strings.Repeat("a", resource.MaxNameLength+1)
+			},
+			errorContains: "resource.owner.name invalid",
+		},
+		"owner partition is mixed case": {
+			modFn: func(res *pbresource.Resource) {
+				res.Owner.Tenancy.Partition = "Default"
+			},
+			errorContains: "resource.owner.tenancy.partition invalid",
+		},
+		"owner partition too long": {
+			modFn: func(res *pbresource.Resource) {
+				res.Owner.Tenancy.Partition = strings.Repeat("p", resource.MaxNameLength+1)
+			},
+			errorContains: "resource.owner.tenancy.partition invalid",
+		},
+		"owner namespace is mixed case": {
+			modFn: func(res *pbresource.Resource) {
+				res.Owner.Tenancy.Namespace = "Default"
+			},
+			errorContains: "resource.owner.tenancy.namespace invalid",
+		},
+		"owner namespace too long": {
+			modFn: func(res *pbresource.Resource) {
+				res.Owner.Tenancy.Namespace = strings.Repeat("n", resource.MaxNameLength+1)
+			},
+			errorContains: "resource.owner.tenancy.namespace invalid",
+		},
+	}
+}
+
+// Test case struct shared by MutateAndValidate and Write success test cases
+type mavOrWriteSuccessTestCase struct {
+	modFn           func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
+	expectedTenancy *pbresource.Tenancy
+}
+
+// Test case struct shared by MutateAndValidate and Write success test cases
+func mavOrWriteSuccessTestCases(t *testing.T) map[string]mavOrWriteSuccessTestCase {
+	return map[string]mavOrWriteSuccessTestCase{
+		"namespaced resource provides nonempty partition and namespace": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens partition when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens namespace when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Namespace = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens partition and namespace when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = ""
+				artist.Id.Tenancy.Namespace = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource inherits tokens partition and namespace when tenancy nil": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy = nil
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"namespaced resource defaults peername to local when empty": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.PeerName = ""
+				return artist
+			},
+			expectedTenancy: resource.DefaultNamespacedTenancy(),
+		},
+		"partitioned resource provides nonempty partition": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+		"partitioned resource inherits tokens partition when empty": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.Partition = ""
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+		"partitioned resource inherits tokens partition when tenancy nil": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy = nil
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+		"partitioned resource defaults peername to local when empty": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.PeerName = ""
+				return recordLabel
+			},
+			expectedTenancy: resource.DefaultPartitionedTenancy(),
+		},
+	}
+}
+
+// Test case struct shared by MutateAndValidate and Write test cases where tenancy is not found
+type mavOrWriteTenancyNotFoundTestCase map[string]struct {
+	modFn       func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
+	errCode     codes.Code
+	errContains string
+}
+
+// Test case struct shared by MutateAndValidate and Write test cases where tenancy is not found
+func mavOrWriteTenancyNotFoundTestCases(t *testing.T) mavOrWriteTenancyNotFoundTestCase {
+	return mavOrWriteTenancyNotFoundTestCase{
+		"namespaced resource provides nonexistant partition": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Partition = "boguspartition"
+				return artist
+			},
+			errCode:     codes.InvalidArgument,
+			errContains: "partition not found",
+		},
+		"namespaced resource provides nonexistant namespace": {
+			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
+				artist.Id.Tenancy.Namespace = "bogusnamespace"
+				return artist
+			},
+			errCode:     codes.InvalidArgument,
+			errContains: "namespace not found",
+		},
+		"partitioned resource provides nonexistant partition": {
+			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
+				recordLabel.Id.Tenancy.Partition = "boguspartition"
+				return recordLabel
+			},
+			errCode:     codes.InvalidArgument,
+			errContains: "partition not found",
+		},
+	}
+}
+
+type mavOrWriteTenancyMarkedForDeletionTestCase struct {
+	modFn       func(artist, recordLabel *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource
+	errContains string
+}
+
+func mavOrWriteTenancyMarkedForDeletionTestCases(t *testing.T) map[string]mavOrWriteTenancyMarkedForDeletionTestCase {
+	return map[string]mavOrWriteTenancyMarkedForDeletionTestCase{
+		"namespaced resources partition marked for deletion": {
+			modFn: func(artist, _ *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource {
+				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "ap1").Return(true, nil)
+				return artist
+			},
+			errContains: "tenancy marked for deletion",
+		},
+		"namespaced resources namespace marked for deletion": {
+			modFn: func(artist, _ *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource {
+				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "ap1").Return(false, nil)
+				mockTenancyBridge.On("IsNamespaceMarkedForDeletion", "ap1", "ns1").Return(true, nil)
+				return artist
+			},
+			errContains: "tenancy marked for deletion",
+		},
+		"partitioned resources partition marked for deletion": {
+			modFn: func(_, recordLabel *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource {
+				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "ap1").Return(true, nil)
+				return recordLabel
+			},
+			errContains: "tenancy marked for deletion",
+		},
+	}
+}

agent/grpc-external/services/resource/write_test.go

@@ -5,7 +5,6 @@ package resource_test
 
 import (
 	"context"
-	"strings"
 	"testing"
 	"time"
 
@@ -36,109 +35,7 @@ func TestWrite_InputValidation(t *testing.T) {
 		WithRegisterFns(demo.RegisterTypes).
 		Run(t)
 
-	type testCase struct {
+	for desc, tc := range resourceValidTestCases(t) {
-		modFn       func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
-		errContains string
-	}
-
-	testCases := map[string]testCase{
-		"no resource": {
-			modFn: func(_, _ *pbresource.Resource) *pbresource.Resource {
-				return nil
-			},
-			errContains: "resource is required",
-		},
-		"no id": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id = nil
-				return artist
-			},
-			errContains: "resource.id is required",
-		},
-		"no type": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Type = nil
-				return artist
-			},
-			errContains: "resource.id.type is required",
-		},
-		"no name": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Name = ""
-				return artist
-			},
-			errContains: "resource.id.name invalid",
-		},
-		"name is mixed case": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Name = "MixedCaseNotAllowed"
-				return artist
-			},
-			errContains: "resource.id.name invalid",
-		},
-		"name too long": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Name = strings.Repeat("a", resource.MaxNameLength+1)
-				return artist
-			},
-			errContains: "resource.id.name invalid",
-		},
-		"wrong data type": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				var err error
-				artist.Data, err = anypb.New(&pbdemov2.Album{})
-				require.NoError(t, err)
-				return artist
-			},
-			errContains: "resource.data is of wrong type",
-		},
-		"partition is mixed case": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Partition = "Default"
-				return artist
-			},
-			errContains: "resource.id.tenancy.partition invalid",
-		},
-		"partition too long": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Partition = strings.Repeat("p", resource.MaxNameLength+1)
-				return artist
-			},
-			errContains: "resource.id.tenancy.partition invalid",
-		},
-		"namespace is mixed case": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Namespace = "Default"
-				return artist
-			},
-			errContains: "resource.id.tenancy.namespace invalid",
-		},
-		"namespace too long": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Namespace = strings.Repeat("n", resource.MaxNameLength+1)
-				return artist
-			},
-			errContains: "resource.id.tenancy.namespace invalid",
-		},
-		"fail validation hook": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				buffer := &pbdemov2.Artist{}
-				require.NoError(t, artist.Data.UnmarshalTo(buffer))
-				buffer.Name = "" // name cannot be empty
-				require.NoError(t, artist.Data.MarshalFrom(buffer))
-				return artist
-			},
-			errContains: "artist.name required",
-		},
-		"partition scope with non-empty namespace": {
-			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
-				recordLabel.Id.Tenancy.Namespace = "bogus"
-				return recordLabel
-			},
-			errContains: "cannot have a namespace",
-		},
-	}
-	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
 			artist, err := demo.GenerateV2Artist()
 			require.NoError(t, err)
@@ -160,58 +57,15 @@ func TestWrite_OwnerValidation(t *testing.T) {
 		WithRegisterFns(demo.RegisterTypes).
 		Run(t)
 
-	type testCase struct {
+	testCases := ownerValidationTestCases(t)
-		modReqFn      func(req *pbresource.WriteRequest)
+
-		errorContains string
+	// This is not part of ownerValidationTestCases because it is a special case
-	}
+	// that only gets caught deeper into the write path.
-	testCases := map[string]testCase{
+	testCases["no owner tenancy"] = ownerValidTestCase{
-		"no owner type": {
+		modFn:         func(res *pbresource.Resource) { res.Owner.Tenancy = nil },
-			modReqFn:      func(req *pbresource.WriteRequest) { req.Resource.Owner.Type = nil },
+		errorContains: "resource.owner does not exist",
-			errorContains: "resource.owner.type is required",
-		},
-		"no owner tenancy": {
-			modReqFn:      func(req *pbresource.WriteRequest) { req.Resource.Owner.Tenancy = nil },
-			errorContains: "resource.owner does not exist",
-		},
-		"no owner name": {
-			modReqFn:      func(req *pbresource.WriteRequest) { req.Resource.Owner.Name = "" },
-			errorContains: "resource.owner.name invalid",
-		},
-		"mixed case owner name": {
-			modReqFn:      func(req *pbresource.WriteRequest) { req.Resource.Owner.Name = strings.ToUpper(req.Resource.Owner.Name) },
-			errorContains: "resource.owner.name invalid",
-		},
-		"owner name too long": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Name = strings.Repeat("a", resource.MaxNameLength+1)
-			},
-			errorContains: "resource.owner.name invalid",
-		},
-		"owner partition is mixed case": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy.Partition = "Default"
-			},
-			errorContains: "resource.owner.tenancy.partition invalid",
-		},
-		"owner partition too long": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy.Partition = strings.Repeat("p", resource.MaxNameLength+1)
-			},
-			errorContains: "resource.owner.tenancy.partition invalid",
-		},
-		"owner namespace is mixed case": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy.Namespace = "Default"
-			},
-			errorContains: "resource.owner.tenancy.namespace invalid",
-		},
-		"owner namespace too long": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy.Namespace = strings.Repeat("n", resource.MaxNameLength+1)
-			},
-			errorContains: "resource.owner.tenancy.namespace invalid",
-		},
 	}
+
 	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
 			artist, err := demo.GenerateV2Artist()
@@ -220,10 +74,9 @@ func TestWrite_OwnerValidation(t *testing.T) {
 			album, err := demo.GenerateV2Album(artist.Id)
 			require.NoError(t, err)
 
-			albumReq := &pbresource.WriteRequest{Resource: album}
+			tc.modFn(album)
-			tc.modReqFn(albumReq)
 
-			_, err = client.Write(testContext(t), albumReq)
+			_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
 			require.Error(t, err)
 			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
 			require.ErrorContains(t, err, tc.errorContains)
@@ -311,84 +164,7 @@ func TestWrite_Mutate(t *testing.T) {
 }
 
 func TestWrite_Create_Success(t *testing.T) {
-	testCases := map[string]struct {
+	for desc, tc := range mavOrWriteSuccessTestCases(t) {
-		modFn           func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
-		expectedTenancy *pbresource.Tenancy
-	}{
-		"namespaced resource provides nonempty partition and namespace": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				return artist
-			},
-			expectedTenancy: resource.DefaultNamespacedTenancy(),
-		},
-		"namespaced resource inherits tokens partition when empty": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Partition = ""
-				return artist
-			},
-			expectedTenancy: resource.DefaultNamespacedTenancy(),
-		},
-		"namespaced resource inherits tokens namespace when empty": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Namespace = ""
-				return artist
-			},
-			expectedTenancy: resource.DefaultNamespacedTenancy(),
-		},
-		"namespaced resource inherits tokens partition and namespace when empty": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Partition = ""
-				artist.Id.Tenancy.Namespace = ""
-				return artist
-			},
-			expectedTenancy: resource.DefaultNamespacedTenancy(),
-		},
-		"namespaced resource inherits tokens partition and namespace when tenancy nil": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy = nil
-				return artist
-			},
-			expectedTenancy: resource.DefaultNamespacedTenancy(),
-		},
-		// TODO(spatel): NET-5475 - Remove as part of peer_name moving to PeerTenancy
-		"namespaced resource defaults peername to local when empty": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.PeerName = ""
-				return artist
-			},
-			expectedTenancy: resource.DefaultNamespacedTenancy(),
-		},
-		"partitioned resource provides nonempty partition": {
-			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
-				return recordLabel
-			},
-			expectedTenancy: resource.DefaultPartitionedTenancy(),
-		},
-		"partitioned resource inherits tokens partition when empty": {
-			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
-				recordLabel.Id.Tenancy.Partition = ""
-				return recordLabel
-			},
-			expectedTenancy: resource.DefaultPartitionedTenancy(),
-		},
-		"partitioned resource inherits tokens partition when tenancy nil": {
-			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
-				recordLabel.Id.Tenancy = nil
-				return recordLabel
-			},
-			expectedTenancy: resource.DefaultPartitionedTenancy(),
-		},
-		// TODO(spatel): NET-5475 - Remove as part of peer_name moving to PeerTenancy
-		"partitioned resource defaults peername to local when empty": {
-			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
-				recordLabel.Id.Tenancy.PeerName = ""
-				return recordLabel
-			},
-			expectedTenancy: resource.DefaultPartitionedTenancy(),
-		},
-		// TODO(spatel): Add cluster scope tests when we have an actual cluster scoped resource (e.g. partition)
-	}
-	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
 			client := svctest.NewResourceServiceBuilder().
 				WithRegisterFns(demo.RegisterTypes).
@@ -411,37 +187,7 @@ func TestWrite_Create_Success(t *testing.T) {
 }
 
 func TestWrite_Create_Tenancy_NotFound(t *testing.T) {
-	testCases := map[string]struct {
+	for desc, tc := range mavOrWriteTenancyNotFoundTestCases(t) {
-		modFn       func(artist, recordLabel *pbresource.Resource) *pbresource.Resource
-		errCode     codes.Code
-		errContains string
-	}{
-		"namespaced resource provides nonexistant partition": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Partition = "boguspartition"
-				return artist
-			},
-			errCode:     codes.InvalidArgument,
-			errContains: "partition not found",
-		},
-		"namespaced resource provides nonexistant namespace": {
-			modFn: func(artist, _ *pbresource.Resource) *pbresource.Resource {
-				artist.Id.Tenancy.Namespace = "bogusnamespace"
-				return artist
-			},
-			errCode:     codes.InvalidArgument,
-			errContains: "namespace not found",
-		},
-		"partitioned resource provides nonexistant partition": {
-			modFn: func(_, recordLabel *pbresource.Resource) *pbresource.Resource {
-				recordLabel.Id.Tenancy.Partition = "boguspartition"
-				return recordLabel
-			},
-			errCode:     codes.InvalidArgument,
-			errContains: "partition not found",
-		},
-	}
-	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
 			client := svctest.NewResourceServiceBuilder().
 				WithV2Tenancy(true).
@@ -481,35 +227,7 @@ func TestWrite_Create_With_DeletionTimestamp_Fails(t *testing.T) {
 }
 
 func TestWrite_Create_With_TenancyMarkedForDeletion_Fails(t *testing.T) {
-	// Verify resource write fails when its partition or namespace is marked for deletion.
+	for desc, tc := range mavOrWriteTenancyMarkedForDeletionTestCases(t) {
-	testCases := map[string]struct {
-		modFn       func(artist, recordLabel *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource
-		errContains string
-	}{
-		"namespaced resources partition marked for deletion": {
-			modFn: func(artist, _ *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource {
-				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "ap1").Return(true, nil)
-				return artist
-			},
-			errContains: "tenancy marked for deletion",
-		},
-		"namespaced resources namespace marked for deletion": {
-			modFn: func(artist, _ *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource {
-				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "ap1").Return(false, nil)
-				mockTenancyBridge.On("IsNamespaceMarkedForDeletion", "ap1", "ns1").Return(true, nil)
-				return artist
-			},
-			errContains: "tenancy marked for deletion",
-		},
-		"partitioned resources partition marked for deletion": {
-			modFn: func(_, recordLabel *pbresource.Resource, mockTenancyBridge *svc.MockTenancyBridge) *pbresource.Resource {
-				mockTenancyBridge.On("IsPartitionMarkedForDeletion", "ap1").Return(true, nil)
-				return recordLabel
-			},
-			errContains: "tenancy marked for deletion",
-		},
-	}
-	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
 			server := testServer(t)
 			client := testClient(t, server)
@@ -1199,7 +917,7 @@ func TestWrite_NonCASWritePreservesDeletionTimestamp(t *testing.T) {
 			rsp, err := client.Write(context.Background(), &pbresource.WriteRequest{Resource: userRes})
 			require.NoError(t, err)
 
-			// Verify write result preserved metadata based on testcase.expecteMetadata
+			// Verify write result preserved metadata based on testcase.expectedMetadata
 			for k := range tc.expectedMeta {
 				require.Equal(t, tc.expectedMeta[k], rsp.Resource.Metadata[k])
 			}

agent/grpc-middleware/rate_limit_mappings.gen.go

@@ -30,6 +30,7 @@ var rpcRateLimitSpecs = map[string]rate.OperationSpec{
 	"/hashicorp.consul.resource.ResourceService/Delete":                                     {Type: rate.OperationTypeWrite, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/List":                                       {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/ListByOwner":                                {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
+	"/hashicorp.consul.resource.ResourceService/MutateAndValidate":                          {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/Read":                                       {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/WatchList":                                  {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/Write":                                      {Type: rate.OperationTypeWrite, Category: rate.OperationCategoryResource},

grpcmocks/proto-public/pbresource/mock_ResourceServiceClient.go

@@ -235,6 +235,76 @@ func (_c *ResourceServiceClient_ListByOwner_Call) RunAndReturn(run func(context.
 	return _c
 }
 
+// MutateAndValidate provides a mock function with given fields: ctx, in, opts
+func (_m *ResourceServiceClient) MutateAndValidate(ctx context.Context, in *pbresource.MutateAndValidateRequest, opts ...grpc.CallOption) (*pbresource.MutateAndValidateResponse, error) {
+	_va := make([]interface{}, len(opts))
+	for _i := range opts {
+		_va[_i] = opts[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, in)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	var r0 *pbresource.MutateAndValidateResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *pbresource.MutateAndValidateRequest, ...grpc.CallOption) (*pbresource.MutateAndValidateResponse, error)); ok {
+		return rf(ctx, in, opts...)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *pbresource.MutateAndValidateRequest, ...grpc.CallOption) *pbresource.MutateAndValidateResponse); ok {
+		r0 = rf(ctx, in, opts...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*pbresource.MutateAndValidateResponse)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *pbresource.MutateAndValidateRequest, ...grpc.CallOption) error); ok {
+		r1 = rf(ctx, in, opts...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ResourceServiceClient_MutateAndValidate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MutateAndValidate'
+type ResourceServiceClient_MutateAndValidate_Call struct {
+	*mock.Call
+}
+
+// MutateAndValidate is a helper method to define mock.On call
+//   - ctx context.Context
+//   - in *pbresource.MutateAndValidateRequest
+//   - opts ...grpc.CallOption
+func (_e *ResourceServiceClient_Expecter) MutateAndValidate(ctx interface{}, in interface{}, opts ...interface{}) *ResourceServiceClient_MutateAndValidate_Call {
+	return &ResourceServiceClient_MutateAndValidate_Call{Call: _e.mock.On("MutateAndValidate",
+		append([]interface{}{ctx, in}, opts...)...)}
+}
+
+func (_c *ResourceServiceClient_MutateAndValidate_Call) Run(run func(ctx context.Context, in *pbresource.MutateAndValidateRequest, opts ...grpc.CallOption)) *ResourceServiceClient_MutateAndValidate_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		variadicArgs := make([]grpc.CallOption, len(args)-2)
+		for i, a := range args[2:] {
+			if a != nil {
+				variadicArgs[i] = a.(grpc.CallOption)
+			}
+		}
+		run(args[0].(context.Context), args[1].(*pbresource.MutateAndValidateRequest), variadicArgs...)
+	})
+	return _c
+}
+
+func (_c *ResourceServiceClient_MutateAndValidate_Call) Return(_a0 *pbresource.MutateAndValidateResponse, _a1 error) *ResourceServiceClient_MutateAndValidate_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *ResourceServiceClient_MutateAndValidate_Call) RunAndReturn(run func(context.Context, *pbresource.MutateAndValidateRequest, ...grpc.CallOption) (*pbresource.MutateAndValidateResponse, error)) *ResourceServiceClient_MutateAndValidate_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Read provides a mock function with given fields: ctx, in, opts
 func (_m *ResourceServiceClient) Read(ctx context.Context, in *pbresource.ReadRequest, opts ...grpc.CallOption) (*pbresource.ReadResponse, error) {
 	_va := make([]interface{}, len(opts))

grpcmocks/proto-public/pbresource/mock_ResourceServiceServer.go

@@ -187,6 +187,61 @@ func (_c *ResourceServiceServer_ListByOwner_Call) RunAndReturn(run func(context.
 	return _c
 }
 
+// MutateAndValidate provides a mock function with given fields: _a0, _a1
+func (_m *ResourceServiceServer) MutateAndValidate(_a0 context.Context, _a1 *pbresource.MutateAndValidateRequest) (*pbresource.MutateAndValidateResponse, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *pbresource.MutateAndValidateResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *pbresource.MutateAndValidateRequest) (*pbresource.MutateAndValidateResponse, error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *pbresource.MutateAndValidateRequest) *pbresource.MutateAndValidateResponse); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*pbresource.MutateAndValidateResponse)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *pbresource.MutateAndValidateRequest) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ResourceServiceServer_MutateAndValidate_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MutateAndValidate'
+type ResourceServiceServer_MutateAndValidate_Call struct {
+	*mock.Call
+}
+
+// MutateAndValidate is a helper method to define mock.On call
+//   - _a0 context.Context
+//   - _a1 *pbresource.MutateAndValidateRequest
+func (_e *ResourceServiceServer_Expecter) MutateAndValidate(_a0 interface{}, _a1 interface{}) *ResourceServiceServer_MutateAndValidate_Call {
+	return &ResourceServiceServer_MutateAndValidate_Call{Call: _e.mock.On("MutateAndValidate", _a0, _a1)}
+}
+
+func (_c *ResourceServiceServer_MutateAndValidate_Call) Run(run func(_a0 context.Context, _a1 *pbresource.MutateAndValidateRequest)) *ResourceServiceServer_MutateAndValidate_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*pbresource.MutateAndValidateRequest))
+	})
+	return _c
+}
+
+func (_c *ResourceServiceServer_MutateAndValidate_Call) Return(_a0 *pbresource.MutateAndValidateResponse, _a1 error) *ResourceServiceServer_MutateAndValidate_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *ResourceServiceServer_MutateAndValidate_Call) RunAndReturn(run func(context.Context, *pbresource.MutateAndValidateRequest) (*pbresource.MutateAndValidateResponse, error)) *ResourceServiceServer_MutateAndValidate_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Read provides a mock function with given fields: _a0, _a1
 func (_m *ResourceServiceServer) Read(_a0 context.Context, _a1 *pbresource.ReadRequest) (*pbresource.ReadResponse, error) {
 	ret := _m.Called(_a0, _a1)

proto-public/pbresource/resource.pb.binary.go

@@ -226,3 +226,23 @@ func (msg *WatchEvent) MarshalBinary() ([]byte, error) {
 func (msg *WatchEvent) UnmarshalBinary(b []byte) error {
 	return proto.Unmarshal(b, msg)
 }
+
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *MutateAndValidateRequest) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *MutateAndValidateRequest) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *MutateAndValidateResponse) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *MutateAndValidateResponse) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}

proto-public/pbresource/resource.pb.go

@@ -1555,6 +1555,102 @@ func (x *WatchEvent) GetResource() *Resource {
 	return nil
 }
 
+// MutateAndValidateRequest contains the parameters to the MutateAndValidate endpoint.
+type MutateAndValidateRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Resource *Resource `protobuf:"bytes,1,opt,name=resource,proto3" json:"resource,omitempty"`
+}
+
+func (x *MutateAndValidateRequest) Reset() {
+	*x = MutateAndValidateRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pbresource_resource_proto_msgTypes[22]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *MutateAndValidateRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*MutateAndValidateRequest) ProtoMessage() {}
+
+func (x *MutateAndValidateRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_pbresource_resource_proto_msgTypes[22]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use MutateAndValidateRequest.ProtoReflect.Descriptor instead.
+func (*MutateAndValidateRequest) Descriptor() ([]byte, []int) {
+	return file_pbresource_resource_proto_rawDescGZIP(), []int{22}
+}
+
+func (x *MutateAndValidateRequest) GetResource() *Resource {
+	if x != nil {
+		return x.Resource
+	}
+	return nil
+}
+
+// MutateAndValidateResponse contains the results of calling the MutateAndValidate endpoint.
+type MutateAndValidateResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Resource *Resource `protobuf:"bytes,1,opt,name=resource,proto3" json:"resource,omitempty"`
+}
+
+func (x *MutateAndValidateResponse) Reset() {
+	*x = MutateAndValidateResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pbresource_resource_proto_msgTypes[23]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *MutateAndValidateResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*MutateAndValidateResponse) ProtoMessage() {}
+
+func (x *MutateAndValidateResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_pbresource_resource_proto_msgTypes[23]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use MutateAndValidateResponse.ProtoReflect.Descriptor instead.
+func (*MutateAndValidateResponse) Descriptor() ([]byte, []int) {
+	return file_pbresource_resource_proto_rawDescGZIP(), []int{23}
+}
+
+func (x *MutateAndValidateResponse) GetResource() *Resource {
+	if x != nil {
+		return x.Resource
+	}
+	return nil
+}
+
 var File_pbresource_resource_proto protoreflect.FileDescriptor
 
 var file_pbresource_resource_proto_rawDesc = []byte{
@@ -1759,71 +1855,91 @@ var file_pbresource_resource_proto_rawDesc = []byte{
 	0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x14,
 	0x0a, 0x10, 0x4f, 0x50, 0x45, 0x52, 0x41, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x50, 0x53, 0x45,
 	0x52, 0x54, 0x10, 0x01, 0x12, 0x14, 0x0a, 0x10, 0x4f, 0x50, 0x45, 0x52, 0x41, 0x54, 0x49, 0x4f,
-	0x4e, 0x5f, 0x44, 0x45, 0x4c, 0x45, 0x54, 0x45, 0x10, 0x02, 0x32, 0x83, 0x06, 0x0a, 0x0f, 0x52,
+	0x4e, 0x5f, 0x44, 0x45, 0x4c, 0x45, 0x54, 0x45, 0x10, 0x02, 0x22, 0x5b, 0x0a, 0x18, 0x4d, 0x75,
-	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x61,
+	0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52,
-	0x0a, 0x04, 0x52, 0x65, 0x61, 0x64, 0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69,
+	0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0x5c, 0x0a, 0x19, 0x4d, 0x75, 0x74, 0x61, 0x74,
+	0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3f, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
 	0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
-	0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27,
+	0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x08, 0x72, 0x65, 0x73,
-	0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x32, 0x8e, 0x07, 0x0a, 0x0f, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72,
-	0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52,
+	0x63, 0x65, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x61, 0x0a, 0x04, 0x52, 0x65, 0x61,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10,
+	0x64, 0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
-	0x0b, 0x12, 0x64, 0x0a, 0x05, 0x57, 0x72, 0x69, 0x74, 0x65, 0x12, 0x27, 0x2e, 0x68, 0x61, 0x73,
+	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65,
-	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65,
+	0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x68, 0x61, 0x73, 0x68,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75,
+	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73,
-	0x65, 0x73, 0x74, 0x1a, 0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x61, 0x64, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
-	0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e,
+	0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x64, 0x0a, 0x05,
-	0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2,
+	0x57, 0x72, 0x69, 0x74, 0x65, 0x12, 0x27, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72,
-	0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x76, 0x0a, 0x0b, 0x57, 0x72, 0x69, 0x74, 0x65,
-	0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x2d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
-	0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
-	0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72,
 	0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73,
+	0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x28,
-	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12,
-	0x61, 0x0a, 0x04, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63,
-	0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x27, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
-	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02,
-	0x10, 0x0b, 0x12, 0x76, 0x0a, 0x0b, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65,
-	0x72, 0x12, 0x2d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
-	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69,
-	0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x1a, 0x2e, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e,
-	0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73,
-	0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x67, 0x0a, 0x06, 0x44, 0x65,
-	0x6c, 0x65, 0x74, 0x65, 0x12, 0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
-	0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x29,
 	0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75,
-	0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74,
+	0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65,
-	0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03,
-	0x03, 0x10, 0x0b, 0x12, 0x6b, 0x0a, 0x09, 0x57, 0x61, 0x74, 0x63, 0x68, 0x4c, 0x69, 0x73, 0x74,
+	0x10, 0x0b, 0x12, 0x76, 0x0a, 0x0b, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75,
-	0x12, 0x2b, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e,
+	0x73, 0x12, 0x2d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
-	0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74,
+	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72,
-	0x63, 0x68, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x25, 0x2e,
+	0x69, 0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c,
+	0x1a, 0x2e, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e,
-	0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45,
+	0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x72, 0x69,
-	0x76, 0x65, 0x6e, 0x74, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x30, 0x01,
+	0x74, 0x65, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x42, 0xe9, 0x01, 0x0a, 0x1d, 0x63, 0x6f, 0x6d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x61, 0x0a, 0x04, 0x4c, 0x69,
+	0x73, 0x74, 0x12, 0x26, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63,
+	0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c,
+	0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x68, 0x61, 0x73,
+	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x76, 0x0a,
+	0x0b, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x12, 0x2d, 0x2e, 0x68,
+	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e,
+	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f,
+	0x77, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x68, 0x61,
+	0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x42, 0x79, 0x4f, 0x77,
+	0x6e, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04,
+	0x04, 0x08, 0x02, 0x10, 0x0b, 0x12, 0x67, 0x0a, 0x06, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x12,
+	0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73,
+	0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x44, 0x65, 0x6c, 0x65,
+	0x74, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x29, 0x2e, 0x68, 0x61, 0x73, 0x68,
+	0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x03, 0x10, 0x0b, 0x12, 0x6b,
+	0x0a, 0x09, 0x57, 0x61, 0x74, 0x63, 0x68, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x2b, 0x2e, 0x68, 0x61,
+	0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x4c, 0x69, 0x73,
+	0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x25, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69,
+	0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x22,
+	0x08, 0xe2, 0x86, 0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x30, 0x01, 0x12, 0x88, 0x01, 0x0a, 0x11,
+	0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74,
+	0x65, 0x12, 0x33, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
+	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x2e, 0x4d, 0x75,
+	0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x34, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
 	0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72,
-	0x63, 0x65, 0x42, 0x0d, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50, 0x72, 0x6f, 0x74,
+	0x63, 0x65, 0x2e, 0x4d, 0x75, 0x74, 0x61, 0x74, 0x65, 0x41, 0x6e, 0x64, 0x56, 0x61, 0x6c, 0x69,
-	0x6f, 0x50, 0x01, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
+	0x64, 0x61, 0x74, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x08, 0xe2, 0x86,
-	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c,
+	0x04, 0x04, 0x08, 0x02, 0x10, 0x0b, 0x42, 0xe9, 0x01, 0x0a, 0x1d, 0x63, 0x6f, 0x6d, 0x2e, 0x68,
-	0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x2f, 0x70, 0x62,
+	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xa2, 0x02, 0x03, 0x48, 0x43, 0x52, 0xaa, 0x02,
+	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x42, 0x0d, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72,
-	0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75,
+	0x63, 0x65, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75,
-	0x6c, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xca, 0x02, 0x19, 0x48, 0x61, 0x73,
+	0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f,
-	0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65,
+	0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2d, 0x70, 0x75, 0x62,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xe2, 0x02, 0x25, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x6c, 0x69, 0x63, 0x2f, 0x70, 0x62, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xa2, 0x02,
-	0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72,
+	0x03, 0x48, 0x43, 0x52, 0xaa, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
-	0x63, 0x65, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02,
+	0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x1b, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x3a, 0x3a, 0x43, 0x6f, 0x6e, 0x73,
+	0xca, 0x02, 0x19, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e,
-	0x75, 0x6c, 0x3a, 0x3a, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x62, 0x06, 0x70, 0x72,
+	0x73, 0x75, 0x6c, 0x5c, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0xe2, 0x02, 0x25, 0x48,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c,
+	0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61,
+	0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x1b, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
+	0x3a, 0x3a, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x3a, 0x3a, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1839,47 +1955,49 @@ func file_pbresource_resource_proto_rawDescGZIP() []byte {
 }
 
 var file_pbresource_resource_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
-var file_pbresource_resource_proto_msgTypes = make([]protoimpl.MessageInfo, 24)
+var file_pbresource_resource_proto_msgTypes = make([]protoimpl.MessageInfo, 26)
 var file_pbresource_resource_proto_goTypes = []interface{}{
-	(Condition_State)(0),          // 0: hashicorp.consul.resource.Condition.State
+	(Condition_State)(0),              // 0: hashicorp.consul.resource.Condition.State
-	(WatchEvent_Operation)(0),     // 1: hashicorp.consul.resource.WatchEvent.Operation
+	(WatchEvent_Operation)(0),         // 1: hashicorp.consul.resource.WatchEvent.Operation
-	(*Type)(nil),                  // 2: hashicorp.consul.resource.Type
+	(*Type)(nil),                      // 2: hashicorp.consul.resource.Type
-	(*Tenancy)(nil),               // 3: hashicorp.consul.resource.Tenancy
+	(*Tenancy)(nil),                   // 3: hashicorp.consul.resource.Tenancy
-	(*ID)(nil),                    // 4: hashicorp.consul.resource.ID
+	(*ID)(nil),                        // 4: hashicorp.consul.resource.ID
-	(*Resource)(nil),              // 5: hashicorp.consul.resource.Resource
+	(*Resource)(nil),                  // 5: hashicorp.consul.resource.Resource
-	(*Status)(nil),                // 6: hashicorp.consul.resource.Status
+	(*Status)(nil),                    // 6: hashicorp.consul.resource.Status
-	(*Condition)(nil),             // 7: hashicorp.consul.resource.Condition
+	(*Condition)(nil),                 // 7: hashicorp.consul.resource.Condition
-	(*Reference)(nil),             // 8: hashicorp.consul.resource.Reference
+	(*Reference)(nil),                 // 8: hashicorp.consul.resource.Reference
-	(*Tombstone)(nil),             // 9: hashicorp.consul.resource.Tombstone
+	(*Tombstone)(nil),                 // 9: hashicorp.consul.resource.Tombstone
-	(*ReadRequest)(nil),           // 10: hashicorp.consul.resource.ReadRequest
+	(*ReadRequest)(nil),               // 10: hashicorp.consul.resource.ReadRequest
-	(*ReadResponse)(nil),          // 11: hashicorp.consul.resource.ReadResponse
+	(*ReadResponse)(nil),              // 11: hashicorp.consul.resource.ReadResponse
-	(*ListRequest)(nil),           // 12: hashicorp.consul.resource.ListRequest
+	(*ListRequest)(nil),               // 12: hashicorp.consul.resource.ListRequest
-	(*ListResponse)(nil),          // 13: hashicorp.consul.resource.ListResponse
+	(*ListResponse)(nil),              // 13: hashicorp.consul.resource.ListResponse
-	(*ListByOwnerRequest)(nil),    // 14: hashicorp.consul.resource.ListByOwnerRequest
+	(*ListByOwnerRequest)(nil),        // 14: hashicorp.consul.resource.ListByOwnerRequest
-	(*ListByOwnerResponse)(nil),   // 15: hashicorp.consul.resource.ListByOwnerResponse
+	(*ListByOwnerResponse)(nil),       // 15: hashicorp.consul.resource.ListByOwnerResponse
-	(*WriteRequest)(nil),          // 16: hashicorp.consul.resource.WriteRequest
+	(*WriteRequest)(nil),              // 16: hashicorp.consul.resource.WriteRequest
-	(*WriteResponse)(nil),         // 17: hashicorp.consul.resource.WriteResponse
+	(*WriteResponse)(nil),             // 17: hashicorp.consul.resource.WriteResponse
-	(*WriteStatusRequest)(nil),    // 18: hashicorp.consul.resource.WriteStatusRequest
+	(*WriteStatusRequest)(nil),        // 18: hashicorp.consul.resource.WriteStatusRequest
-	(*WriteStatusResponse)(nil),   // 19: hashicorp.consul.resource.WriteStatusResponse
+	(*WriteStatusResponse)(nil),       // 19: hashicorp.consul.resource.WriteStatusResponse
-	(*DeleteRequest)(nil),         // 20: hashicorp.consul.resource.DeleteRequest
+	(*DeleteRequest)(nil),             // 20: hashicorp.consul.resource.DeleteRequest
-	(*DeleteResponse)(nil),        // 21: hashicorp.consul.resource.DeleteResponse
+	(*DeleteResponse)(nil),            // 21: hashicorp.consul.resource.DeleteResponse
-	(*WatchListRequest)(nil),      // 22: hashicorp.consul.resource.WatchListRequest
+	(*WatchListRequest)(nil),          // 22: hashicorp.consul.resource.WatchListRequest
-	(*WatchEvent)(nil),            // 23: hashicorp.consul.resource.WatchEvent
+	(*WatchEvent)(nil),                // 23: hashicorp.consul.resource.WatchEvent
-	nil,                           // 24: hashicorp.consul.resource.Resource.MetadataEntry
+	(*MutateAndValidateRequest)(nil),  // 24: hashicorp.consul.resource.MutateAndValidateRequest
-	nil,                           // 25: hashicorp.consul.resource.Resource.StatusEntry
+	(*MutateAndValidateResponse)(nil), // 25: hashicorp.consul.resource.MutateAndValidateResponse
-	(*anypb.Any)(nil),             // 26: google.protobuf.Any
+	nil,                               // 26: hashicorp.consul.resource.Resource.MetadataEntry
-	(*timestamppb.Timestamp)(nil), // 27: google.protobuf.Timestamp
+	nil,                               // 27: hashicorp.consul.resource.Resource.StatusEntry
+	(*anypb.Any)(nil),                 // 28: google.protobuf.Any
+	(*timestamppb.Timestamp)(nil),     // 29: google.protobuf.Timestamp
 }
 var file_pbresource_resource_proto_depIdxs = []int32{
 	2,  // 0: hashicorp.consul.resource.ID.type:type_name -> hashicorp.consul.resource.Type
 	3,  // 1: hashicorp.consul.resource.ID.tenancy:type_name -> hashicorp.consul.resource.Tenancy
 	4,  // 2: hashicorp.consul.resource.Resource.id:type_name -> hashicorp.consul.resource.ID
 	4,  // 3: hashicorp.consul.resource.Resource.owner:type_name -> hashicorp.consul.resource.ID
-	24, // 4: hashicorp.consul.resource.Resource.metadata:type_name -> hashicorp.consul.resource.Resource.MetadataEntry
+	26, // 4: hashicorp.consul.resource.Resource.metadata:type_name -> hashicorp.consul.resource.Resource.MetadataEntry
-	25, // 5: hashicorp.consul.resource.Resource.status:type_name -> hashicorp.consul.resource.Resource.StatusEntry
+	27, // 5: hashicorp.consul.resource.Resource.status:type_name -> hashicorp.consul.resource.Resource.StatusEntry
-	26, // 6: hashicorp.consul.resource.Resource.data:type_name -> google.protobuf.Any
+	28, // 6: hashicorp.consul.resource.Resource.data:type_name -> google.protobuf.Any
 	7,  // 7: hashicorp.consul.resource.Status.conditions:type_name -> hashicorp.consul.resource.Condition
-	27, // 8: hashicorp.consul.resource.Status.updated_at:type_name -> google.protobuf.Timestamp
+	29, // 8: hashicorp.consul.resource.Status.updated_at:type_name -> google.protobuf.Timestamp
 	0,  // 9: hashicorp.consul.resource.Condition.state:type_name -> hashicorp.consul.resource.Condition.State
 	8,  // 10: hashicorp.consul.resource.Condition.resource:type_name -> hashicorp.consul.resource.Reference
 	2,  // 11: hashicorp.consul.resource.Reference.type:type_name -> hashicorp.consul.resource.Type
@@ -1902,26 +2020,30 @@ var file_pbresource_resource_proto_depIdxs = []int32{
 	3,  // 28: hashicorp.consul.resource.WatchListRequest.tenancy:type_name -> hashicorp.consul.resource.Tenancy
 	1,  // 29: hashicorp.consul.resource.WatchEvent.operation:type_name -> hashicorp.consul.resource.WatchEvent.Operation
 	5,  // 30: hashicorp.consul.resource.WatchEvent.resource:type_name -> hashicorp.consul.resource.Resource
-	6,  // 31: hashicorp.consul.resource.Resource.StatusEntry.value:type_name -> hashicorp.consul.resource.Status
+	5,  // 31: hashicorp.consul.resource.MutateAndValidateRequest.resource:type_name -> hashicorp.consul.resource.Resource
-	10, // 32: hashicorp.consul.resource.ResourceService.Read:input_type -> hashicorp.consul.resource.ReadRequest
+	5,  // 32: hashicorp.consul.resource.MutateAndValidateResponse.resource:type_name -> hashicorp.consul.resource.Resource
-	16, // 33: hashicorp.consul.resource.ResourceService.Write:input_type -> hashicorp.consul.resource.WriteRequest
+	6,  // 33: hashicorp.consul.resource.Resource.StatusEntry.value:type_name -> hashicorp.consul.resource.Status
-	18, // 34: hashicorp.consul.resource.ResourceService.WriteStatus:input_type -> hashicorp.consul.resource.WriteStatusRequest
+	10, // 34: hashicorp.consul.resource.ResourceService.Read:input_type -> hashicorp.consul.resource.ReadRequest
-	12, // 35: hashicorp.consul.resource.ResourceService.List:input_type -> hashicorp.consul.resource.ListRequest
+	16, // 35: hashicorp.consul.resource.ResourceService.Write:input_type -> hashicorp.consul.resource.WriteRequest
-	14, // 36: hashicorp.consul.resource.ResourceService.ListByOwner:input_type -> hashicorp.consul.resource.ListByOwnerRequest
+	18, // 36: hashicorp.consul.resource.ResourceService.WriteStatus:input_type -> hashicorp.consul.resource.WriteStatusRequest
-	20, // 37: hashicorp.consul.resource.ResourceService.Delete:input_type -> hashicorp.consul.resource.DeleteRequest
+	12, // 37: hashicorp.consul.resource.ResourceService.List:input_type -> hashicorp.consul.resource.ListRequest
-	22, // 38: hashicorp.consul.resource.ResourceService.WatchList:input_type -> hashicorp.consul.resource.WatchListRequest
+	14, // 38: hashicorp.consul.resource.ResourceService.ListByOwner:input_type -> hashicorp.consul.resource.ListByOwnerRequest
-	11, // 39: hashicorp.consul.resource.ResourceService.Read:output_type -> hashicorp.consul.resource.ReadResponse
+	20, // 39: hashicorp.consul.resource.ResourceService.Delete:input_type -> hashicorp.consul.resource.DeleteRequest
-	17, // 40: hashicorp.consul.resource.ResourceService.Write:output_type -> hashicorp.consul.resource.WriteResponse
+	22, // 40: hashicorp.consul.resource.ResourceService.WatchList:input_type -> hashicorp.consul.resource.WatchListRequest
-	19, // 41: hashicorp.consul.resource.ResourceService.WriteStatus:output_type -> hashicorp.consul.resource.WriteStatusResponse
+	24, // 41: hashicorp.consul.resource.ResourceService.MutateAndValidate:input_type -> hashicorp.consul.resource.MutateAndValidateRequest
-	13, // 42: hashicorp.consul.resource.ResourceService.List:output_type -> hashicorp.consul.resource.ListResponse
+	11, // 42: hashicorp.consul.resource.ResourceService.Read:output_type -> hashicorp.consul.resource.ReadResponse
-	15, // 43: hashicorp.consul.resource.ResourceService.ListByOwner:output_type -> hashicorp.consul.resource.ListByOwnerResponse
+	17, // 43: hashicorp.consul.resource.ResourceService.Write:output_type -> hashicorp.consul.resource.WriteResponse
-	21, // 44: hashicorp.consul.resource.ResourceService.Delete:output_type -> hashicorp.consul.resource.DeleteResponse
+	19, // 44: hashicorp.consul.resource.ResourceService.WriteStatus:output_type -> hashicorp.consul.resource.WriteStatusResponse
-	23, // 45: hashicorp.consul.resource.ResourceService.WatchList:output_type -> hashicorp.consul.resource.WatchEvent
+	13, // 45: hashicorp.consul.resource.ResourceService.List:output_type -> hashicorp.consul.resource.ListResponse
-	39, // [39:46] is the sub-list for method output_type
+	15, // 46: hashicorp.consul.resource.ResourceService.ListByOwner:output_type -> hashicorp.consul.resource.ListByOwnerResponse
-	32, // [32:39] is the sub-list for method input_type
+	21, // 47: hashicorp.consul.resource.ResourceService.Delete:output_type -> hashicorp.consul.resource.DeleteResponse
-	32, // [32:32] is the sub-list for extension type_name
+	23, // 48: hashicorp.consul.resource.ResourceService.WatchList:output_type -> hashicorp.consul.resource.WatchEvent
-	32, // [32:32] is the sub-list for extension extendee
+	25, // 49: hashicorp.consul.resource.ResourceService.MutateAndValidate:output_type -> hashicorp.consul.resource.MutateAndValidateResponse
-	0,  // [0:32] is the sub-list for field type_name
+	42, // [42:50] is the sub-list for method output_type
+	34, // [34:42] is the sub-list for method input_type
+	34, // [34:34] is the sub-list for extension type_name
+	34, // [34:34] is the sub-list for extension extendee
+	0,  // [0:34] is the sub-list for field type_name
 }
 
 func init() { file_pbresource_resource_proto_init() }
@@ -2194,6 +2316,30 @@ func file_pbresource_resource_proto_init() {
 				return nil
 			}
 		}
+		file_pbresource_resource_proto_msgTypes[22].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*MutateAndValidateRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pbresource_resource_proto_msgTypes[23].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*MutateAndValidateResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -2201,7 +2347,7 @@ func file_pbresource_resource_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_pbresource_resource_proto_rawDesc,
 			NumEnums:      2,
-			NumMessages:   24,
+			NumMessages:   26,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

proto-public/pbresource/resource.proto

@@ -351,6 +351,19 @@ service ResourceService {
       operation_category: OPERATION_CATEGORY_RESOURCE
     };
   }
+
+  // MutateAndValidate a resource.
+  //
+  // Applies a resource type's registered mutation and validation hooks to
+  // a resource. This is useful for filling in defaults and validating inputs before
+  // doing a Write. It's not a pre-requisite since the Write endpoint will also apply
+  // the hooks.
+  rpc MutateAndValidate(MutateAndValidateRequest) returns (MutateAndValidateResponse) {
+    option (hashicorp.consul.internal.ratelimit.spec) = {
+      operation_type: OPERATION_TYPE_READ,
+      operation_category: OPERATION_CATEGORY_RESOURCE
+    };
+  }
 }
 
 // ReadRequest contains the parameters to the Read endpoint.
@@ -488,3 +501,13 @@ message WatchEvent {
   // Resource the event relates to.
   Resource resource = 2;
 }
+
+// MutateAndValidateRequest contains the parameters to the MutateAndValidate endpoint.
+message MutateAndValidateRequest {
+  Resource resource = 1;
+}
+
+// MutateAndValidateResponse contains the results of calling the MutateAndValidate endpoint.
+message MutateAndValidateResponse {
+  Resource resource = 1;
+}

proto-public/pbresource/resource_cloning_grpc.pb.go

@@ -112,6 +112,17 @@ func (c CloningResourceServiceClient) Delete(ctx context.Context, in *DeleteRequ
 	return proto.Clone(out).(*DeleteResponse), nil
 }
 
+func (c CloningResourceServiceClient) MutateAndValidate(ctx context.Context, in *MutateAndValidateRequest, opts ...grpc.CallOption) (*MutateAndValidateResponse, error) {
+	in = proto.Clone(in).(*MutateAndValidateRequest)
+
+	out, err := c.ResourceServiceClient.MutateAndValidate(ctx, in)
+	if err != nil {
+		return nil, err
+	}
+
+	return proto.Clone(out).(*MutateAndValidateResponse), nil
+}
+
 func (c CloningResourceServiceClient) WatchList(ctx context.Context, in *WatchListRequest, opts ...grpc.CallOption) (ResourceService_WatchListClient, error) {
 	in = proto.Clone(in).(*WatchListRequest)
 

proto-public/pbresource/resource_deepcopy.gen.go

@@ -466,3 +466,45 @@ func (in *WatchEvent) DeepCopy() *WatchEvent {
 func (in *WatchEvent) DeepCopyInterface() interface{} {
 	return in.DeepCopy()
 }
+
+// DeepCopyInto supports using MutateAndValidateRequest within kubernetes types, where deepcopy-gen is used.
+func (in *MutateAndValidateRequest) DeepCopyInto(out *MutateAndValidateRequest) {
+	proto.Reset(out)
+	proto.Merge(out, proto.Clone(in))
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MutateAndValidateRequest. Required by controller-gen.
+func (in *MutateAndValidateRequest) DeepCopy() *MutateAndValidateRequest {
+	if in == nil {
+		return nil
+	}
+	out := new(MutateAndValidateRequest)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInterface is an autogenerated deepcopy function, copying the receiver, creating a new MutateAndValidateRequest. Required by controller-gen.
+func (in *MutateAndValidateRequest) DeepCopyInterface() interface{} {
+	return in.DeepCopy()
+}
+
+// DeepCopyInto supports using MutateAndValidateResponse within kubernetes types, where deepcopy-gen is used.
+func (in *MutateAndValidateResponse) DeepCopyInto(out *MutateAndValidateResponse) {
+	proto.Reset(out)
+	proto.Merge(out, proto.Clone(in))
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MutateAndValidateResponse. Required by controller-gen.
+func (in *MutateAndValidateResponse) DeepCopy() *MutateAndValidateResponse {
+	if in == nil {
+		return nil
+	}
+	out := new(MutateAndValidateResponse)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInterface is an autogenerated deepcopy function, copying the receiver, creating a new MutateAndValidateResponse. Required by controller-gen.
+func (in *MutateAndValidateResponse) DeepCopyInterface() interface{} {
+	return in.DeepCopy()
+}

proto-public/pbresource/resource_grpc.pb.go

@@ -107,6 +107,13 @@ type ResourceServiceClient interface {
 	//
 	// buf:lint:ignore RPC_RESPONSE_STANDARD_NAME
 	WatchList(ctx context.Context, in *WatchListRequest, opts ...grpc.CallOption) (ResourceService_WatchListClient, error)
+	// MutateAndValidate a resource.
+	//
+	// Applies a resource type's registered mutation and validation hooks to
+	// a resource. This is useful for filling in defaults and validating inputs before
+	// doing a Write. It's not a pre-requisite since the Write endpoint will also apply
+	// the hooks.
+	MutateAndValidate(ctx context.Context, in *MutateAndValidateRequest, opts ...grpc.CallOption) (*MutateAndValidateResponse, error)
 }
 
 type resourceServiceClient struct {
@@ -203,6 +210,15 @@ func (x *resourceServiceWatchListClient) Recv() (*WatchEvent, error) {
 	return m, nil
 }
 
+func (c *resourceServiceClient) MutateAndValidate(ctx context.Context, in *MutateAndValidateRequest, opts ...grpc.CallOption) (*MutateAndValidateResponse, error) {
+	out := new(MutateAndValidateResponse)
+	err := c.cc.Invoke(ctx, "/hashicorp.consul.resource.ResourceService/MutateAndValidate", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // ResourceServiceServer is the server API for ResourceService service.
 // All implementations should embed UnimplementedResourceServiceServer
 // for forward compatibility
@@ -292,6 +308,13 @@ type ResourceServiceServer interface {
 	//
 	// buf:lint:ignore RPC_RESPONSE_STANDARD_NAME
 	WatchList(*WatchListRequest, ResourceService_WatchListServer) error
+	// MutateAndValidate a resource.
+	//
+	// Applies a resource type's registered mutation and validation hooks to
+	// a resource. This is useful for filling in defaults and validating inputs before
+	// doing a Write. It's not a pre-requisite since the Write endpoint will also apply
+	// the hooks.
+	MutateAndValidate(context.Context, *MutateAndValidateRequest) (*MutateAndValidateResponse, error)
 }
 
 // UnimplementedResourceServiceServer should be embedded to have forward compatible implementations.
@@ -319,6 +342,9 @@ func (UnimplementedResourceServiceServer) Delete(context.Context, *DeleteRequest
 func (UnimplementedResourceServiceServer) WatchList(*WatchListRequest, ResourceService_WatchListServer) error {
 	return status.Errorf(codes.Unimplemented, "method WatchList not implemented")
 }
+func (UnimplementedResourceServiceServer) MutateAndValidate(context.Context, *MutateAndValidateRequest) (*MutateAndValidateResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method MutateAndValidate not implemented")
+}
 
 // UnsafeResourceServiceServer may be embedded to opt out of forward compatibility for this service.
 // Use of this interface is not recommended, as added methods to ResourceServiceServer will
@@ -460,6 +486,24 @@ func (x *resourceServiceWatchListServer) Send(m *WatchEvent) error {
 	return x.ServerStream.SendMsg(m)
 }
 
+func _ResourceService_MutateAndValidate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(MutateAndValidateRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ResourceServiceServer).MutateAndValidate(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/hashicorp.consul.resource.ResourceService/MutateAndValidate",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ResourceServiceServer).MutateAndValidate(ctx, req.(*MutateAndValidateRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // ResourceService_ServiceDesc is the grpc.ServiceDesc for ResourceService service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -491,6 +535,10 @@ var ResourceService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "Delete",
 			Handler:    _ResourceService_Delete_Handler,
 		},
+		{
+			MethodName: "MutateAndValidate",
+			Handler:    _ResourceService_MutateAndValidate_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{

proto-public/pbresource/resource_json.gen.go

@@ -247,6 +247,28 @@ func (this *WatchEvent) UnmarshalJSON(b []byte) error {
 	return ResourceUnmarshaler.Unmarshal(b, this)
 }
 
+// MarshalJSON is a custom marshaler for MutateAndValidateRequest
+func (this *MutateAndValidateRequest) MarshalJSON() ([]byte, error) {
+	str, err := ResourceMarshaler.Marshal(this)
+	return []byte(str), err
+}
+
+// UnmarshalJSON is a custom unmarshaler for MutateAndValidateRequest
+func (this *MutateAndValidateRequest) UnmarshalJSON(b []byte) error {
+	return ResourceUnmarshaler.Unmarshal(b, this)
+}
+
+// MarshalJSON is a custom marshaler for MutateAndValidateResponse
+func (this *MutateAndValidateResponse) MarshalJSON() ([]byte, error) {
+	str, err := ResourceMarshaler.Marshal(this)
+	return []byte(str), err
+}
+
+// UnmarshalJSON is a custom unmarshaler for MutateAndValidateResponse
+func (this *MutateAndValidateResponse) UnmarshalJSON(b []byte) error {
+	return ResourceUnmarshaler.Unmarshal(b, this)
+}
+
 var (
 	ResourceMarshaler   = &protojson.MarshalOptions{}
 	ResourceUnmarshaler = &protojson.UnmarshalOptions{DiscardUnknown: false}