catalog: normalize/default/validate tenancy components of FailoverPolicy internal References (#18825)

FailoverPolicy resources contain inner Reference fields. We want to ensure 
that components of those reference Tenancy fields left unspecified are defaulted 
using the tenancy of the enclosing FailoverPolicy resource.

As the underlying helper being used to do the normalization calls the function 
modified in #18822, it also means that the PeerName field will be set to "local" for 
now automatically to avoid "local" != "" issues downstream.

internal/catalog/exports.go

@@ -126,3 +126,17 @@ type FailoverPolicyMapper interface {
 func NewFailoverPolicyMapper() FailoverPolicyMapper {
 	return failovermapper.New()
 }
+
+// ValidateLocalServiceRefNoSection ensures the following:
+//
+// - ref is non-nil
+// - type is ServiceType
+// - section is empty
+// - tenancy is set and partition/namespace are both non-empty
+// - peer_name must be "local"
+//
+// Each possible validation error is wrapped in the wrapErr function before
+// being collected in a multierror.Error.
+func ValidateLocalServiceRefNoSection(ref *pbresource.Reference, wrapErr func(error) error) error {
+	return types.ValidateLocalServiceRefNoSection(ref, wrapErr)
+}

internal/catalog/internal/types/failover_policy.go

@@ -4,7 +4,6 @@
 package types
 
 import (
-	"errors"
 	"fmt"
 
 	"github.com/hashicorp/go-multierror"
@@ -53,10 +52,21 @@ func MutateFailoverPolicy(res *pbresource.Resource) error {
 		failover.Config = nil
 		changed = true
 	}
+
+	if failover.Config != nil {
+		if mutateFailoverConfig(res.Id.Tenancy, failover.Config) {
+			changed = true
+		}
+	}
+
 	for port, pc := range failover.PortConfigs {
 		if pc.IsEmpty() {
 			delete(failover.PortConfigs, port)
 			changed = true
+		} else {
+			if mutateFailoverConfig(res.Id.Tenancy, pc) {
+				changed = true
+			}
 		}
 	}
 	if len(failover.PortConfigs) == 0 {
@@ -64,8 +74,6 @@ func MutateFailoverPolicy(res *pbresource.Resource) error {
 		changed = true
 	}
 
-	// TODO(rb): normalize dest ref tenancies
-
 	if !changed {
 		return nil
 	}
@@ -73,6 +81,42 @@ func MutateFailoverPolicy(res *pbresource.Resource) error {
 	return res.Data.MarshalFrom(&failover)
 }
 
+func mutateFailoverConfig(policyTenancy *pbresource.Tenancy, config *pbcatalog.FailoverConfig) (changed bool) {
+	if policyTenancy != nil && !isLocalPeer(policyTenancy.PeerName) {
+		// TODO(peering/v2): remove this bypass when we know what to do with
+		// non-local peer references.
+		return false
+	}
+
+	for _, dest := range config.Destinations {
+		if dest.Ref == nil {
+			continue
+		}
+		if dest.Ref.Tenancy != nil && !isLocalPeer(dest.Ref.Tenancy.PeerName) {
+			// TODO(peering/v2): remove this bypass when we know what to do with
+			// non-local peer references.
+			continue
+		}
+
+		orig := proto.Clone(dest.Ref).(*pbresource.Reference)
+		resource.DefaultReferenceTenancy(
+			dest.Ref,
+			policyTenancy,
+			resource.DefaultNamespacedTenancy(), // Services are all namespace scoped.
+		)
+
+		if !proto.Equal(orig, dest.Ref) {
+			changed = true
+		}
+	}
+
+	return changed
+}
+
+func isLocalPeer(p string) bool {
+	return p == "local" || p == ""
+}
+
 func ValidateFailoverPolicy(res *pbresource.Resource) error {
 	var failover pbcatalog.FailoverPolicy
 
@@ -90,15 +134,25 @@ func ValidateFailoverPolicy(res *pbresource.Resource) error {
 	}
 
 	if failover.Config != nil {
-		for _, err := range validateFailoverConfig(failover.Config, false) {
+		wrapConfigErr := func(err error) error {
-			merr = multierror.Append(merr, resource.ErrInvalidField{
+			return resource.ErrInvalidField{
 				Name:    "config",
 				Wrapped: err,
-			})
+			}
+		}
+		if cfgErr := validateFailoverConfig(failover.Config, false, wrapConfigErr); cfgErr != nil {
+			merr = multierror.Append(merr, cfgErr)
 		}
 	}
 
 	for portName, pc := range failover.PortConfigs {
+		wrapConfigErr := func(err error) error {
+			return resource.ErrInvalidMapValue{
+				Map:     "port_configs",
+				Key:     portName,
+				Wrapped: err,
+			}
+		}
 		if portNameErr := validatePortName(portName); portNameErr != nil {
 			merr = multierror.Append(merr, resource.ErrInvalidMapKey{
 				Map:     "port_configs",
@@ -107,12 +161,8 @@ func ValidateFailoverPolicy(res *pbresource.Resource) error {
 			})
 		}
 
-		for _, err := range validateFailoverConfig(pc, true) {
+		if cfgErr := validateFailoverConfig(pc, true, wrapConfigErr); cfgErr != nil {
-			merr = multierror.Append(merr, resource.ErrInvalidMapValue{
+			merr = multierror.Append(merr, cfgErr)
-				Map:     "port_configs",
-				Key:     portName,
-				Wrapped: err,
-			})
 		}
 
 		// TODO: should sameness group be a ref once that's a resource?
@@ -121,23 +171,29 @@ func ValidateFailoverPolicy(res *pbresource.Resource) error {
 	return merr
 }
 
-func validateFailoverConfig(config *pbcatalog.FailoverConfig, ported bool) []error {
+func validateFailoverConfig(config *pbcatalog.FailoverConfig, ported bool, wrapErr func(error) error) error {
-	var errs []error
+	var merr error
 
 	if (len(config.Destinations) > 0) == (config.SamenessGroup != "") {
-		errs = append(errs, resource.ErrInvalidField{
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
 			Name: "destinations",
-			Wrapped: fmt.Errorf("exactly one of destinations or sameness_group should be set"),
+			// Wrapped: fmt.Errorf("exactly one of destinations or sameness_group should be set"),
-		})
+			Wrapped: fmt.Errorf("exactly one of destinations or sameness_group should be set: %v || %v",
+				(len(config.Destinations) > 0), (config.SamenessGroup != ""),
+			),
+		}))
 	}
 	for i, dest := range config.Destinations {
-		for _, err := range validateFailoverPolicyDestination(dest, ported) {
+		wrapDestErr := func(err error) error {
-			errs = append(errs, resource.ErrInvalidListElement{
+			return wrapErr(resource.ErrInvalidListElement{
 				Name:    "destinations",
 				Index:   i,
 				Wrapped: err,
 			})
 		}
+		if destErr := validateFailoverPolicyDestination(dest, ported, wrapDestErr); destErr != nil {
+			merr = multierror.Append(merr, destErr)
+		}
 	}
 
 	switch config.Mode {
@@ -146,72 +202,62 @@ func validateFailoverConfig(config *pbcatalog.FailoverConfig, ported bool) []err
 	case pbcatalog.FailoverMode_FAILOVER_MODE_SEQUENTIAL:
 	case pbcatalog.FailoverMode_FAILOVER_MODE_ORDER_BY_LOCALITY:
 	default:
-		errs = append(errs, resource.ErrInvalidField{
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
 			Name:    "mode",
 			Wrapped: fmt.Errorf("not a supported enum value: %v", config.Mode),
-		})
+		}))
 	}
 
 	// TODO: validate sameness group requirements
 
-	return errs
+	return merr
 }
 
-func validateFailoverPolicyDestination(dest *pbcatalog.FailoverDestination, ported bool) []error {
+func validateFailoverPolicyDestination(dest *pbcatalog.FailoverDestination, ported bool, wrapErr func(error) error) error {
-	var errs []error
+	var merr error
-	if dest.Ref == nil {
+
-		errs = append(errs, resource.ErrInvalidField{
+	wrapRefErr := func(err error) error {
+		return wrapErr(resource.ErrInvalidField{
 			Name:    "ref",
-			Wrapped: resource.ErrMissing,
+			Wrapped: err,
-		})
-	} else if !resource.EqualType(dest.Ref.Type, ServiceType) {
-		errs = append(errs, resource.ErrInvalidField{
-			Name: "ref",
-			Wrapped: resource.ErrInvalidReferenceType{
-				AllowedType: ServiceType,
-			},
-		})
-	} else if dest.Ref.Section != "" {
-		errs = append(errs, resource.ErrInvalidField{
-			Name: "ref",
-			Wrapped: resource.ErrInvalidField{
-				Name:    "section",
-				Wrapped: errors.New("section not supported for failover policy dest refs"),
-			},
 		})
 	}
 
+	if refErr := ValidateLocalServiceRefNoSection(dest.Ref, wrapRefErr); refErr != nil {
+		merr = multierror.Append(merr, refErr)
+	}
+
 	// NOTE: Destinations here cannot define ports. Port equality is
 	// assumed and will be reconciled.
 	if dest.Port != "" {
 		if ported {
 			if portNameErr := validatePortName(dest.Port); portNameErr != nil {
-				errs = append(errs, resource.ErrInvalidField{
+				merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
 					Name:    "port",
 					Wrapped: portNameErr,
-				})
+				}))
 			}
 		} else {
-			errs = append(errs, resource.ErrInvalidField{
+			merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
 				Name:    "port",
 				Wrapped: fmt.Errorf("ports cannot be specified explicitly for the general failover section since it relies upon port alignment"),
-			})
+			}))
 		}
 	}
 
 	hasPeer := false
 	if dest.Ref != nil {
-		hasPeer = dest.Ref.Tenancy.PeerName != "local"
+		hasPeer = dest.Ref.Tenancy.PeerName != "" && dest.Ref.Tenancy.PeerName != "local"
 	}
 
 	if hasPeer && dest.Datacenter != "" {
-		errs = append(errs, resource.ErrInvalidField{
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
 			Name:    "datacenter",
 			Wrapped: fmt.Errorf("ref.tenancy.peer_name and datacenter are mutually exclusive fields"),
-		})
+		}))
 	}
 
-	return errs
+	return merr
 }
 
 // SimplifyFailoverPolicy fully populates the PortConfigs map and clears the

internal/catalog/internal/types/failover_policy_test.go

@@ -4,6 +4,7 @@
 package types
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -19,6 +20,7 @@ import (
 
 func TestMutateFailoverPolicy(t *testing.T) {
 	type testcase struct {
+		policyTenancy *pbresource.Tenancy
 		failover      *pbcatalog.FailoverPolicy
 		expect        *pbcatalog.FailoverPolicy
 		expectErr     string
@@ -26,6 +28,7 @@ func TestMutateFailoverPolicy(t *testing.T) {
 
 	run := func(t *testing.T, tc testcase) {
 		res := resourcetest.Resource(FailoverPolicyType, "api").
+			WithTenancy(tc.policyTenancy).
 			WithData(t, tc.failover).
 			Build()
 
@@ -134,6 +137,49 @@ func TestMutateFailoverPolicy(t *testing.T) {
 				},
 			},
 		},
+		"dest ref tenancy defaulting": {
+			policyTenancy: newTestTenancy("foo.bar"),
+			failover: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Mode:    pbcatalog.FailoverMode_FAILOVER_MODE_SEQUENTIAL,
+					Regions: []string{"foo", "bar"},
+					Destinations: []*pbcatalog.FailoverDestination{
+						{Ref: newRefWithTenancy(ServiceType, "", "api")},
+						{Ref: newRefWithTenancy(ServiceType, ".zim", "api")},
+						{Ref: newRefWithTenancy(ServiceType, "gir.zim", "api")},
+					},
+				},
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRefWithTenancy(ServiceType, "", "api")},
+							{Ref: newRefWithTenancy(ServiceType, ".luthor", "api")},
+							{Ref: newRefWithTenancy(ServiceType, "lex.luthor", "api")},
+						},
+					},
+				},
+			},
+			expect: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Mode:    pbcatalog.FailoverMode_FAILOVER_MODE_SEQUENTIAL,
+					Regions: []string{"foo", "bar"},
+					Destinations: []*pbcatalog.FailoverDestination{
+						{Ref: newRefWithTenancy(ServiceType, "foo.bar", "api")},
+						{Ref: newRefWithTenancy(ServiceType, "foo.zim", "api")},
+						{Ref: newRefWithTenancy(ServiceType, "gir.zim", "api")},
+					},
+				},
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRefWithTenancy(ServiceType, "foo.bar", "api")},
+							{Ref: newRefWithTenancy(ServiceType, "foo.luthor", "api")},
+							{Ref: newRefWithTenancy(ServiceType, "lex.luthor", "api")},
+						},
+					},
+				},
+			},
+		},
 	}
 
 	for name, tc := range cases {
@@ -224,7 +270,7 @@ func TestValidateFailoverPolicy(t *testing.T) {
 					{Ref: newRef(WorkloadType, "api-backup")},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "destinations": invalid "ref" field: reference must have type catalog.v1alpha1.Service`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "ref" field: invalid "type" field: reference must have type catalog.v1alpha1.Service`,
 		},
 		"dest: ref with section": {
 			config: &pbcatalog.FailoverConfig{
@@ -232,23 +278,25 @@ func TestValidateFailoverPolicy(t *testing.T) {
 					{Ref: resourcetest.Resource(ServiceType, "api").WithTenancy(resource.DefaultNamespacedTenancy()).Reference("blah")},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "destinations": invalid "ref" field: invalid "section" field: section not supported for failover policy dest refs`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "ref" field: invalid "section" field: section cannot be set here`,
-		},
-		"dest: ref peer and datacenter": {
-			config: &pbcatalog.FailoverConfig{
-				Destinations: []*pbcatalog.FailoverDestination{
-					{Ref: newRefWithPeer(ServiceType, "api", "peer1"), Datacenter: "dc2"},
-				},
-			},
-			expectErr: `invalid element at index 0 of list "destinations": invalid "datacenter" field: ref.tenancy.peer_name and datacenter are mutually exclusive fields`,
-		},
-		"dest: ref peer without datacenter": {
-			config: &pbcatalog.FailoverConfig{
-				Destinations: []*pbcatalog.FailoverDestination{
-					{Ref: newRefWithPeer(ServiceType, "api", "peer1")},
-				},
-			},
 		},
+		// TODO(v2/peering): re-enable when peering can exist
+		// "dest: ref peer and datacenter": {
+		// 	config: &pbcatalog.FailoverConfig{
+		// 		Destinations: []*pbcatalog.FailoverDestination{
+		// 			{Ref: newRefWithPeer(ServiceType, "api", "peer1"), Datacenter: "dc2"},
+		// 		},
+		// 	},
+		// 	expectErr: `invalid element at index 0 of list "destinations": invalid "datacenter" field: ref.tenancy.peer_name and datacenter are mutually exclusive fields`,
+		// },
+		// TODO(v2/peering): re-enable when peering can exist
+		// "dest: ref peer without datacenter": {
+		// 	config: &pbcatalog.FailoverConfig{
+		// 		Destinations: []*pbcatalog.FailoverDestination{
+		// 			{Ref: newRefWithPeer(ServiceType, "api", "peer1")},
+		// 		},
+		// 	},
+		// },
 		"dest: ref datacenter without peer": {
 			config: &pbcatalog.FailoverConfig{
 				Destinations: []*pbcatalog.FailoverDestination{
@@ -595,8 +643,33 @@ func newRef(typ *pbresource.Type, name string) *pbresource.Reference {
 		Reference("")
 }
 
+func newRefWithTenancy(typ *pbresource.Type, tenancyStr, name string) *pbresource.Reference {
+	return resourcetest.Resource(typ, name).
+		WithTenancy(newTestTenancy(tenancyStr)).
+		Reference("")
+}
+
 func newRefWithPeer(typ *pbresource.Type, name string, peer string) *pbresource.Reference {
 	ref := newRef(typ, name)
 	ref.Tenancy.PeerName = peer
 	return ref
 }
+
+func newTestTenancy(s string) *pbresource.Tenancy {
+	parts := strings.Split(s, ".")
+	switch len(parts) {
+	case 0:
+		return resource.DefaultClusteredTenancy()
+	case 1:
+		v := resource.DefaultPartitionedTenancy()
+		v.Partition = parts[0]
+		return v
+	case 2:
+		v := resource.DefaultNamespacedTenancy()
+		v.Partition = parts[0]
+		v.Namespace = parts[1]
+		return v
+	default:
+		return &pbresource.Tenancy{Partition: "BAD", Namespace: "BAD", PeerName: "BAD"}
+	}
+}

internal/catalog/internal/types/validators.go

@@ -4,6 +4,7 @@
 package types
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"regexp"
@@ -237,3 +238,81 @@ func validateHealth(health pbcatalog.Health) error {
 		return resource.NewConstError(fmt.Sprintf("not a supported enum value: %v", health))
 	}
 }
+
+// ValidateLocalServiceRefNoSection ensures the following:
+//
+// - ref is non-nil
+// - type is ServiceType
+// - section is empty
+// - tenancy is set and partition/namespace are both non-empty
+// - peer_name must be "local"
+//
+// Each possible validation error is wrapped in the wrapErr function before
+// being collected in a multierror.Error.
+func ValidateLocalServiceRefNoSection(ref *pbresource.Reference, wrapErr func(error) error) error {
+	if ref == nil {
+		return wrapErr(resource.ErrMissing)
+	}
+
+	if !resource.EqualType(ref.Type, ServiceType) {
+		return wrapErr(resource.ErrInvalidField{
+			Name: "type",
+			Wrapped: resource.ErrInvalidReferenceType{
+				AllowedType: ServiceType,
+			},
+		})
+	}
+
+	var merr error
+	if ref.Section != "" {
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+			Name:    "section",
+			Wrapped: errors.New("section cannot be set here"),
+		}))
+	}
+
+	if ref.Tenancy == nil {
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+			Name:    "tenancy",
+			Wrapped: resource.ErrMissing,
+		}))
+	} else {
+		// NOTE: these are Service specific, since that's a Namespace-scoped type.
+		if ref.Tenancy.Partition == "" {
+			merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+				Name: "tenancy",
+				Wrapped: resource.ErrInvalidField{
+					Name:    "partition",
+					Wrapped: resource.ErrEmpty,
+				},
+			}))
+		}
+		if ref.Tenancy.Namespace == "" {
+			merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+				Name: "tenancy",
+				Wrapped: resource.ErrInvalidField{
+					Name:    "namespace",
+					Wrapped: resource.ErrEmpty,
+				},
+			}))
+		}
+		if ref.Tenancy.PeerName != "local" {
+			merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+				Name: "tenancy",
+				Wrapped: resource.ErrInvalidField{
+					Name:    "peer_name",
+					Wrapped: errors.New(`must be set to "local"`),
+				},
+			}))
+		}
+	}
+
+	if ref.Name == "" {
+		merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
+			Name:    "name",
+			Wrapped: resource.ErrMissing,
+		}))
+	}
+
+	return merr
+}