From bfc519f293f946ea737e5fe090e524bd964f20a7 Mon Sep 17 00:00:00 2001
From: "R.B. Boyer" <4903+rboyer@users.noreply.github.com>
Date: Tue, 8 Aug 2023 14:42:43 -0500
Subject: [PATCH] catalog: add FailoverPolicy mutation and validation hooks
 (#18390)

Add most of the validation and mutation hooks for the FailoverPolicy resource.
---
 internal/catalog/exports.go                   |   7 +
 .../catalog/internal/types/failover_policy.go | 224 ++++++-
 .../internal/types/failover_policy_test.go    | 590 ++++++++++++++++++
 .../v1alpha1/failover_policy_extras.go        |  65 ++
 .../v1alpha1/failover_policy_extras_test.go   | 171 +++++
 5 files changed, 1055 insertions(+), 2 deletions(-)
 create mode 100644 internal/catalog/internal/types/failover_policy_test.go
 create mode 100644 proto-public/pbcatalog/v1alpha1/failover_policy_extras.go
 create mode 100644 proto-public/pbcatalog/v1alpha1/failover_policy_extras_test.go

diff --git a/internal/catalog/exports.go b/internal/catalog/exports.go
index 06af51d37a..463b11e16e 100644
--- a/internal/catalog/exports.go
+++ b/internal/catalog/exports.go
@@ -13,6 +13,7 @@ import (
 	"github.com/hashicorp/consul/internal/catalog/internal/types"
 	"github.com/hashicorp/consul/internal/controller"
 	"github.com/hashicorp/consul/internal/resource"
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v1alpha1"
 )
 
 var (
@@ -94,3 +95,9 @@ func DefaultControllerDependencies() ControllerDependencies {
 func RegisterControllers(mgr *controller.Manager, deps ControllerDependencies) {
 	controllers.Register(mgr, deps)
 }
+
+// SimplifyFailoverPolicy fully populates the PortConfigs map and clears the
+// Configs map using the provided Service.
+func SimplifyFailoverPolicy(svc *pbcatalog.Service, failover *pbcatalog.FailoverPolicy) *pbcatalog.FailoverPolicy {
+	return types.SimplifyFailoverPolicy(svc, failover)
+}
diff --git a/internal/catalog/internal/types/failover_policy.go b/internal/catalog/internal/types/failover_policy.go
index 61930bc1c5..49d38674bb 100644
--- a/internal/catalog/internal/types/failover_policy.go
+++ b/internal/catalog/internal/types/failover_policy.go
@@ -4,6 +4,12 @@
 package types
 
 import (
+	"errors"
+	"fmt"
+
+	"github.com/hashicorp/go-multierror"
+	"google.golang.org/protobuf/proto"
+
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v1alpha1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -27,7 +33,221 @@ func RegisterFailoverPolicy(r resource.Registry) {
 	r.Register(resource.Registration{
 		Type:     FailoverPolicyV1Alpha1Type,
 		Proto:    &pbcatalog.FailoverPolicy{},
-		Validate: nil,
-		Mutate:   nil,
+		Mutate:   MutateFailoverPolicy,
+		Validate: ValidateFailoverPolicy,
 	})
 }
+
+func MutateFailoverPolicy(res *pbresource.Resource) error {
+	var failover pbcatalog.FailoverPolicy
+
+	if err := res.Data.UnmarshalTo(&failover); err != nil {
+		return resource.NewErrDataParse(&failover, err)
+	}
+
+	changed := false
+
+	// Handle eliding empty configs.
+	if failover.Config != nil && failover.Config.IsEmpty() {
+		failover.Config = nil
+		changed = true
+	}
+	for port, pc := range failover.PortConfigs {
+		if pc.IsEmpty() {
+			delete(failover.PortConfigs, port)
+			changed = true
+		}
+	}
+	if len(failover.PortConfigs) == 0 {
+		failover.PortConfigs = nil
+		changed = true
+	}
+
+	// TODO(rb): normalize dest ref tenancies
+
+	if !changed {
+		return nil
+	}
+
+	return res.Data.MarshalFrom(&failover)
+}
+
+func ValidateFailoverPolicy(res *pbresource.Resource) error {
+	var failover pbcatalog.FailoverPolicy
+
+	if err := res.Data.UnmarshalTo(&failover); err != nil {
+		return resource.NewErrDataParse(&failover, err)
+	}
+
+	var merr error
+
+	if failover.Config == nil && len(failover.PortConfigs) == 0 {
+		merr = multierror.Append(merr, resource.ErrInvalidField{
+			Name:    "config",
+			Wrapped: fmt.Errorf("at least one of config or port_configs must be set"),
+		})
+	}
+
+	if failover.Config != nil {
+		for _, err := range validateFailoverConfig(failover.Config, false) {
+			merr = multierror.Append(merr, resource.ErrInvalidField{
+				Name:    "config",
+				Wrapped: err,
+			})
+		}
+	}
+
+	for portName, pc := range failover.PortConfigs {
+		if portNameErr := validatePortName(portName); portNameErr != nil {
+			merr = multierror.Append(merr, resource.ErrInvalidMapKey{
+				Map:     "port_configs",
+				Key:     portName,
+				Wrapped: portNameErr,
+			})
+		}
+
+		for _, err := range validateFailoverConfig(pc, true) {
+			merr = multierror.Append(merr, resource.ErrInvalidMapValue{
+				Map:     "port_configs",
+				Key:     portName,
+				Wrapped: err,
+			})
+		}
+
+		// TODO: should sameness group be a ref once that's a resource?
+	}
+
+	return merr
+}
+
+func validateFailoverConfig(config *pbcatalog.FailoverConfig, ported bool) []error {
+	var errs []error
+
+	if (len(config.Destinations) > 0) == (config.SamenessGroup != "") {
+		errs = append(errs, resource.ErrInvalidField{
+			Name:    "destinations",
+			Wrapped: fmt.Errorf("exactly one of destinations or sameness_group should be set"),
+		})
+	}
+	for i, dest := range config.Destinations {
+		for _, err := range validateFailoverPolicyDestination(dest, ported) {
+			errs = append(errs, resource.ErrInvalidListElement{
+				Name:    "destinations",
+				Index:   i,
+				Wrapped: err,
+			})
+		}
+	}
+
+	// TODO: validate sameness group requirements
+
+	return errs
+}
+
+func validateFailoverPolicyDestination(dest *pbcatalog.FailoverDestination, ported bool) []error {
+	var errs []error
+	if dest.Ref == nil {
+		errs = append(errs, resource.ErrInvalidField{
+			Name:    "ref",
+			Wrapped: resource.ErrMissing,
+		})
+	} 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"),
+			},
+		})
+	}
+
+	// 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{
+					Name:    "port",
+					Wrapped: portNameErr,
+				})
+			}
+		} else {
+			errs = append(errs, 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"
+	}
+
+	if hasPeer && dest.Datacenter != "" {
+		errs = append(errs, resource.ErrInvalidField{
+			Name:    "datacenter",
+			Wrapped: fmt.Errorf("ref.tenancy.peer_name and datacenter are mutually exclusive fields"),
+		})
+	}
+
+	return errs
+}
+
+// SimplifyFailoverPolicy fully populates the PortConfigs map and clears the
+// Configs map using the provided Service.
+func SimplifyFailoverPolicy(svc *pbcatalog.Service, failover *pbcatalog.FailoverPolicy) *pbcatalog.FailoverPolicy {
+	if failover == nil {
+		panic("failover is required")
+	}
+	if svc == nil {
+		panic("service is required")
+	}
+
+	// Copy so we can edit it.
+	dup := proto.Clone(failover)
+	failover = dup.(*pbcatalog.FailoverPolicy)
+
+	if failover.PortConfigs == nil {
+		failover.PortConfigs = make(map[string]*pbcatalog.FailoverConfig)
+	}
+
+	for _, port := range svc.Ports {
+		if port.Protocol == pbcatalog.Protocol_PROTOCOL_MESH {
+			continue // skip
+		}
+
+		if pc, ok := failover.PortConfigs[port.TargetPort]; ok {
+			for i, dest := range pc.Destinations {
+				// Assume port alignment.
+				if dest.Port == "" {
+					dest.Port = port.TargetPort
+					pc.Destinations[i] = dest
+				}
+			}
+			continue
+		}
+
+		if failover.Config != nil {
+			// Duplicate because each port will get this uniquely.
+			pc2 := proto.Clone(failover.Config).(*pbcatalog.FailoverConfig)
+			for _, dest := range pc2.Destinations {
+				dest.Port = port.TargetPort
+			}
+			failover.PortConfigs[port.TargetPort] = pc2
+		}
+	}
+
+	if failover.Config != nil {
+		failover.Config = nil
+	}
+
+	return failover
+}
diff --git a/internal/catalog/internal/types/failover_policy_test.go b/internal/catalog/internal/types/failover_policy_test.go
new file mode 100644
index 0000000000..3d41cb25fc
--- /dev/null
+++ b/internal/catalog/internal/types/failover_policy_test.go
@@ -0,0 +1,590 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package types
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/resource/resourcetest"
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v1alpha1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	"github.com/hashicorp/consul/proto/private/prototest"
+	"github.com/hashicorp/consul/sdk/testutil"
+)
+
+func TestMutateFailoverPolicy(t *testing.T) {
+	type testcase struct {
+		failover  *pbcatalog.FailoverPolicy
+		expect    *pbcatalog.FailoverPolicy
+		expectErr string
+	}
+
+	run := func(t *testing.T, tc testcase) {
+		res := resourcetest.Resource(FailoverPolicyType, "api").
+			WithData(t, tc.failover).
+			Build()
+
+		err := MutateFailoverPolicy(res)
+
+		got := resourcetest.MustDecode[pbcatalog.FailoverPolicy, *pbcatalog.FailoverPolicy](t, res)
+
+		if tc.expectErr == "" {
+			require.NoError(t, err)
+			prototest.AssertDeepEqual(t, tc.expect, got.Data)
+		} else {
+			testutil.RequireErrorContains(t, err, tc.expectErr)
+		}
+	}
+
+	cases := map[string]testcase{
+		"empty-1": {
+			failover: &pbcatalog.FailoverPolicy{},
+			expect:   &pbcatalog.FailoverPolicy{},
+		},
+		"empty-config-1": {
+			failover: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{},
+			},
+			expect: &pbcatalog.FailoverPolicy{},
+		},
+		"empty-config-2": {
+			failover: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Destinations: make([]*pbcatalog.FailoverDestination, 0),
+				},
+			},
+			expect: &pbcatalog.FailoverPolicy{},
+		},
+		"empty-map-1": {
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: make(map[string]*pbcatalog.FailoverConfig),
+			},
+			expect: &pbcatalog.FailoverPolicy{},
+		},
+		"empty-map-config-1": {
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {},
+				},
+			},
+			expect: &pbcatalog.FailoverPolicy{},
+		},
+		"empty-map-config-2": {
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: make([]*pbcatalog.FailoverDestination, 0),
+					},
+				},
+			},
+			expect: &pbcatalog.FailoverPolicy{},
+		},
+		"normal": {
+			failover: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Mode:    pbcatalog.FailoverMode_FAILOVER_MODE_SEQUENTIAL,
+					Regions: []string{"foo", "bar"},
+					Destinations: []*pbcatalog.FailoverDestination{
+						{Ref: newRef(ServiceType, "a")},
+						{Ref: newRef(ServiceType, "b")},
+					},
+				},
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "foo")},
+							{Ref: newRef(ServiceType, "bar")},
+						},
+					},
+					"admin": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "y")},
+							{Ref: newRef(ServiceType, "z")},
+						},
+					},
+				},
+			},
+			expect: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Mode:    pbcatalog.FailoverMode_FAILOVER_MODE_SEQUENTIAL,
+					Regions: []string{"foo", "bar"},
+					Destinations: []*pbcatalog.FailoverDestination{
+						{Ref: newRef(ServiceType, "a")},
+						{Ref: newRef(ServiceType, "b")},
+					},
+				},
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "foo")},
+							{Ref: newRef(ServiceType, "bar")},
+						},
+					},
+					"admin": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "y")},
+							{Ref: newRef(ServiceType, "z")},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func TestValidateFailoverPolicy(t *testing.T) {
+	type configTestcase struct {
+		config    *pbcatalog.FailoverConfig
+		expectErr string
+	}
+
+	type testcase struct {
+		failover  *pbcatalog.FailoverPolicy
+		expectErr string
+	}
+
+	run := func(t *testing.T, tc testcase) {
+		res := resourcetest.Resource(FailoverPolicyType, "api").
+			WithData(t, tc.failover).
+			Build()
+
+		require.NoError(t, MutateFailoverPolicy(res))
+
+		// Verify that mutate didn't actually change the object.
+		got := resourcetest.MustDecode[pbcatalog.FailoverPolicy, *pbcatalog.FailoverPolicy](t, res)
+		prototest.AssertDeepEqual(t, tc.failover, got.Data)
+
+		err := ValidateFailoverPolicy(res)
+
+		// Verify that validate didn't actually change the object.
+		got = resourcetest.MustDecode[pbcatalog.FailoverPolicy, *pbcatalog.FailoverPolicy](t, res)
+		prototest.AssertDeepEqual(t, tc.failover, got.Data)
+
+		if tc.expectErr == "" {
+			require.NoError(t, err)
+		} else {
+			testutil.RequireErrorContains(t, err, tc.expectErr)
+		}
+	}
+
+	configCases := map[string]configTestcase{
+		"dest with sameness": {
+			config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{Ref: newRef(ServiceType, "api-backup")},
+				},
+				SamenessGroup: "blah",
+			},
+			expectErr: `invalid "destinations" field: exactly one of destinations or sameness_group should be set`,
+		},
+		"dest without sameness": {
+			config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{Ref: newRef(ServiceType, "api-backup")},
+				},
+			},
+		},
+		"sameness without dest": {
+			config: &pbcatalog.FailoverConfig{
+				SamenessGroup: "blah",
+			},
+		},
+		"dest: no ref": {
+			config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "destinations": invalid "ref" field: missing required field`,
+		},
+		"dest: non-service ref": {
+			config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{Ref: newRef(WorkloadType, "api-backup")},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "destinations": invalid "ref" field: reference must have type catalog.v1alpha1.Service`,
+		},
+		"dest: ref with section": {
+			config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{Ref: resourcetest.Resource(ServiceType, "api").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`,
+		},
+		"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")},
+				},
+			},
+		},
+		"dest: ref datacenter without peer": {
+			config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{Ref: newRef(ServiceType, "api"), Datacenter: "dc2"},
+				},
+			},
+		},
+	}
+
+	cases := map[string]testcase{
+		// emptiness
+		"empty": {
+			failover:  &pbcatalog.FailoverPolicy{},
+			expectErr: `invalid "config" field: at least one of config or port_configs must be set`,
+		},
+		"non-empty: one port config but no plain config": {
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "api-backup")},
+						},
+					},
+				},
+			},
+		},
+		"non-empty: some plain config but no port configs": {
+			failover: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Destinations: []*pbcatalog.FailoverDestination{
+						{Ref: newRef(ServiceType, "api-backup")},
+					},
+				},
+			},
+		},
+		// plain config
+		"plain config: bad dest: any port name": {
+			failover: &pbcatalog.FailoverPolicy{
+				Config: &pbcatalog.FailoverConfig{
+					Destinations: []*pbcatalog.FailoverDestination{
+						{Ref: newRef(ServiceType, "api-backup"), Port: "web"},
+					},
+				},
+			},
+			expectErr: `invalid "config" field: invalid element at index 0 of list "destinations": invalid "port" field: ports cannot be specified explicitly for the general failover section since it relies upon port alignment`,
+		},
+		// ported config
+		"ported config: bad dest: invalid port name": {
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "api-backup"), Port: "$bad$"},
+						},
+					},
+				},
+			},
+			expectErr: `invalid value of key "http" within port_configs: invalid element at index 0 of list "destinations": invalid "port" field: value must match regex: ^[a-z0-9]([a-z0-9\-_]*[a-z0-9])?$`,
+		},
+		"ported config: bad ported in map": {
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"$bad$": {
+						Destinations: []*pbcatalog.FailoverDestination{
+							{Ref: newRef(ServiceType, "api-backup"), Port: "http"},
+						},
+					},
+				},
+			},
+			expectErr: `map port_configs contains an invalid key - "$bad$": value must match regex: ^[a-z0-9]([a-z0-9\-_]*[a-z0-9])?$`,
+		},
+	}
+
+	maybeWrap := func(wrapPrefix, base string) string {
+		if base != "" {
+			return wrapPrefix + base
+		}
+		return ""
+	}
+
+	for name, tc := range configCases {
+		cases["plain config: "+name] = testcase{
+			failover: &pbcatalog.FailoverPolicy{
+				Config: proto.Clone(tc.config).(*pbcatalog.FailoverConfig),
+			},
+			expectErr: maybeWrap(`invalid "config" field: `, tc.expectErr),
+		}
+
+		cases["ported config: "+name] = testcase{
+			failover: &pbcatalog.FailoverPolicy{
+				PortConfigs: map[string]*pbcatalog.FailoverConfig{
+					"http": proto.Clone(tc.config).(*pbcatalog.FailoverConfig),
+				},
+			},
+			expectErr: maybeWrap(`invalid value of key "http" within port_configs: `, tc.expectErr),
+		}
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func TestSimplifyFailoverPolicy(t *testing.T) {
+	registry := resource.NewRegistry()
+	Register(registry)
+
+	type testcase struct {
+		svc      *pbresource.Resource
+		failover *pbresource.Resource
+		expect   *pbresource.Resource
+	}
+	run := func(t *testing.T, tc testcase) {
+		// Ensure we only use valid inputs.
+		resourcetest.ValidateAndNormalize(t, registry, tc.svc)
+		resourcetest.ValidateAndNormalize(t, registry, tc.failover)
+		resourcetest.ValidateAndNormalize(t, registry, tc.expect)
+
+		svc := resourcetest.MustDecode[pbcatalog.Service, *pbcatalog.Service](t, tc.svc)
+		failover := resourcetest.MustDecode[pbcatalog.FailoverPolicy, *pbcatalog.FailoverPolicy](t, tc.failover)
+		expect := resourcetest.MustDecode[pbcatalog.FailoverPolicy, *pbcatalog.FailoverPolicy](t, tc.expect)
+
+		inputFailoverCopy := proto.Clone(failover.Data).(*pbcatalog.FailoverPolicy)
+
+		got := SimplifyFailoverPolicy(svc.Data, failover.Data)
+		prototest.AssertDeepEqual(t, expect.Data, got)
+
+		// verify input was not altered
+		prototest.AssertDeepEqual(t, inputFailoverCopy, failover.Data)
+	}
+
+	newPort := func(name string, virtualPort uint32, protocol pbcatalog.Protocol) *pbcatalog.ServicePort {
+		return &pbcatalog.ServicePort{
+			VirtualPort: virtualPort,
+			TargetPort:  name,
+			Protocol:    protocol,
+		}
+	}
+
+	cases := map[string]testcase{
+		"implicit with mesh port skipping": {
+			svc: resourcetest.Resource(ServiceType, "api").
+				WithData(t, &pbcatalog.Service{
+					Ports: []*pbcatalog.ServicePort{
+						newPort("mesh", 21001, pbcatalog.Protocol_PROTOCOL_MESH),
+						newPort("http", 8080, pbcatalog.Protocol_PROTOCOL_HTTP),
+					},
+				}).
+				Build(),
+			failover: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					Config: &pbcatalog.FailoverConfig{
+						Destinations: []*pbcatalog.FailoverDestination{
+							{
+								Ref: newRef(ServiceType, "api-backup"),
+							},
+						},
+					},
+				}).
+				Build(),
+			expect: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					PortConfigs: map[string]*pbcatalog.FailoverConfig{
+						"http": {
+							Destinations: []*pbcatalog.FailoverDestination{
+								{
+									Ref:  newRef(ServiceType, "api-backup"),
+									Port: "http", // port defaulted
+								},
+							},
+						},
+					},
+				}).
+				Build(),
+		},
+		"explicit with port aligned defaulting": {
+			svc: resourcetest.Resource(ServiceType, "api").
+				WithData(t, &pbcatalog.Service{
+					Ports: []*pbcatalog.ServicePort{
+						newPort("mesh", 9999, pbcatalog.Protocol_PROTOCOL_MESH),
+						newPort("http", 8080, pbcatalog.Protocol_PROTOCOL_HTTP),
+						newPort("rest", 8282, pbcatalog.Protocol_PROTOCOL_HTTP2),
+					},
+				}).
+				Build(),
+			failover: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					PortConfigs: map[string]*pbcatalog.FailoverConfig{
+						"http": {
+							Destinations: []*pbcatalog.FailoverDestination{
+								{
+									Ref:  newRef(ServiceType, "api-backup"),
+									Port: "www",
+								},
+								{
+									Ref: newRef(ServiceType, "api-double-backup"),
+								},
+							},
+						},
+					},
+				}).
+				Build(),
+			expect: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					PortConfigs: map[string]*pbcatalog.FailoverConfig{
+						"http": {
+							Destinations: []*pbcatalog.FailoverDestination{
+								{
+									Ref:  newRef(ServiceType, "api-backup"),
+									Port: "www",
+								},
+								{
+									Ref:  newRef(ServiceType, "api-double-backup"),
+									Port: "http", // port defaulted
+								},
+							},
+						},
+					},
+				}).
+				Build(),
+		},
+		"implicit port explosion": {
+			svc: resourcetest.Resource(ServiceType, "api").
+				WithData(t, &pbcatalog.Service{
+					Ports: []*pbcatalog.ServicePort{
+						newPort("http", 8080, pbcatalog.Protocol_PROTOCOL_HTTP),
+						newPort("rest", 8282, pbcatalog.Protocol_PROTOCOL_HTTP2),
+					},
+				}).
+				Build(),
+			failover: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					Config: &pbcatalog.FailoverConfig{
+						Destinations: []*pbcatalog.FailoverDestination{
+							{
+								Ref: newRef(ServiceType, "api-backup"),
+							},
+							{
+								Ref: newRef(ServiceType, "api-double-backup"),
+							},
+						},
+					},
+				}).
+				Build(),
+			expect: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					PortConfigs: map[string]*pbcatalog.FailoverConfig{
+						"http": {
+							Destinations: []*pbcatalog.FailoverDestination{
+								{
+									Ref:  newRef(ServiceType, "api-backup"),
+									Port: "http",
+								},
+								{
+									Ref:  newRef(ServiceType, "api-double-backup"),
+									Port: "http",
+								},
+							},
+						},
+						"rest": {
+							Destinations: []*pbcatalog.FailoverDestination{
+								{
+									Ref:  newRef(ServiceType, "api-backup"),
+									Port: "rest",
+								},
+								{
+									Ref:  newRef(ServiceType, "api-double-backup"),
+									Port: "rest",
+								},
+							},
+						},
+					},
+				}).
+				Build(),
+		},
+		"mixed port explosion with skip": {
+			svc: resourcetest.Resource(ServiceType, "api").
+				WithData(t, &pbcatalog.Service{
+					Ports: []*pbcatalog.ServicePort{
+						newPort("http", 8080, pbcatalog.Protocol_PROTOCOL_HTTP),
+						newPort("rest", 8282, pbcatalog.Protocol_PROTOCOL_HTTP2),
+					},
+				}).
+				Build(),
+			failover: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					Config: &pbcatalog.FailoverConfig{
+						Destinations: []*pbcatalog.FailoverDestination{
+							{
+								Ref: newRef(ServiceType, "api-backup"),
+							},
+							{
+								Ref: newRef(ServiceType, "api-double-backup"),
+							},
+						},
+					},
+					PortConfigs: map[string]*pbcatalog.FailoverConfig{
+						"rest": {
+							Mode:          pbcatalog.FailoverMode_FAILOVER_MODE_ORDER_BY_LOCALITY,
+							Regions:       []string{"us", "eu"},
+							SamenessGroup: "sameweb",
+						},
+					},
+				}).
+				Build(),
+			expect: resourcetest.Resource(FailoverPolicyType, "api").
+				WithData(t, &pbcatalog.FailoverPolicy{
+					PortConfigs: map[string]*pbcatalog.FailoverConfig{
+						"http": {
+							Destinations: []*pbcatalog.FailoverDestination{
+								{
+									Ref:  newRef(ServiceType, "api-backup"),
+									Port: "http",
+								},
+								{
+									Ref:  newRef(ServiceType, "api-double-backup"),
+									Port: "http",
+								},
+							},
+						},
+						"rest": {
+							Mode:          pbcatalog.FailoverMode_FAILOVER_MODE_ORDER_BY_LOCALITY,
+							Regions:       []string{"us", "eu"},
+							SamenessGroup: "sameweb",
+						},
+					},
+				}).
+				Build(),
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func newRef(typ *pbresource.Type, name string) *pbresource.Reference {
+	return resourcetest.Resource(typ, name).Reference("")
+}
+
+func newRefWithPeer(typ *pbresource.Type, name string, peer string) *pbresource.Reference {
+	ref := newRef(typ, name)
+	ref.Tenancy.PeerName = peer
+	return ref
+}
diff --git a/proto-public/pbcatalog/v1alpha1/failover_policy_extras.go b/proto-public/pbcatalog/v1alpha1/failover_policy_extras.go
new file mode 100644
index 0000000000..8c1f2d104c
--- /dev/null
+++ b/proto-public/pbcatalog/v1alpha1/failover_policy_extras.go
@@ -0,0 +1,65 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package catalogv1alpha1
+
+import pbresource "github.com/hashicorp/consul/proto-public/pbresource"
+
+// GetUnderlyingDestinations will collect FailoverDestinations from all
+// internal fields and bundle them up in one slice.
+//
+// NOTE: no deduplication occurs.
+func (x *FailoverPolicy) GetUnderlyingDestinations() []*FailoverDestination {
+	if x == nil {
+		return nil
+	}
+
+	estimate := 0
+	if x.Config != nil {
+		estimate += len(x.Config.Destinations)
+	}
+	for _, pc := range x.PortConfigs {
+		estimate += len(pc.Destinations)
+	}
+
+	out := make([]*FailoverDestination, 0, estimate)
+	if x.Config != nil {
+		out = append(out, x.Config.Destinations...)
+	}
+	for _, pc := range x.PortConfigs {
+		out = append(out, pc.Destinations...)
+	}
+	return out
+}
+
+// GetUnderlyingDestinationRefs is like GetUnderlyingDestinations except it
+// returns a slice of References.
+//
+// NOTE: no deduplication occurs.
+func (x *FailoverPolicy) GetUnderlyingDestinationRefs() []*pbresource.Reference {
+	if x == nil {
+		return nil
+	}
+
+	dests := x.GetUnderlyingDestinations()
+
+	out := make([]*pbresource.Reference, 0, len(dests))
+	for _, dest := range dests {
+		if dest.Ref != nil {
+			out = append(out, dest.Ref)
+		}
+	}
+
+	return out
+}
+
+// IsEmpty returns true if a config has no definition.
+func (x *FailoverConfig) IsEmpty() bool {
+	if x == nil {
+		return true
+	}
+	return len(x.Destinations) == 0 &&
+		x.Mode == 0 &&
+		len(x.Regions) == 0 &&
+		x.SamenessGroup == ""
+}
diff --git a/proto-public/pbcatalog/v1alpha1/failover_policy_extras_test.go b/proto-public/pbcatalog/v1alpha1/failover_policy_extras_test.go
new file mode 100644
index 0000000000..f9fe3e879d
--- /dev/null
+++ b/proto-public/pbcatalog/v1alpha1/failover_policy_extras_test.go
@@ -0,0 +1,171 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package catalogv1alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+
+	pbresource "github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func TestFailoverPolicy_IsEmpty(t *testing.T) {
+	t.Run("nil", func(t *testing.T) {
+		var fc *FailoverConfig
+		require.True(t, fc.IsEmpty())
+	})
+	t.Run("empty", func(t *testing.T) {
+		fc := &FailoverConfig{}
+		require.True(t, fc.IsEmpty())
+	})
+	t.Run("dest", func(t *testing.T) {
+		fc := &FailoverConfig{
+			Destinations: []*FailoverDestination{
+				newFailoverDestination("foo"),
+			},
+		}
+		require.False(t, fc.IsEmpty())
+	})
+	t.Run("regions", func(t *testing.T) {
+		fc := &FailoverConfig{
+			Regions: []string{"us-east"},
+		}
+		require.False(t, fc.IsEmpty())
+	})
+	t.Run("regions", func(t *testing.T) {
+		fc := &FailoverConfig{
+			SamenessGroup: "blah",
+		}
+		require.False(t, fc.IsEmpty())
+	})
+}
+
+func TestFailoverPolicy_GetUnderlyingDestinations_AndRefs(t *testing.T) {
+	type testcase struct {
+		failover    *FailoverPolicy
+		expectDests []*FailoverDestination
+		expectRefs  []*pbresource.Reference
+	}
+
+	run := func(t *testing.T, tc testcase) {
+		assertSliceEquals(t, tc.expectDests, tc.failover.GetUnderlyingDestinations())
+		assertSliceEquals(t, tc.expectRefs, tc.failover.GetUnderlyingDestinationRefs())
+	}
+
+	cases := map[string]testcase{
+		"nil": {},
+		"kitchen sink dests": {
+			failover: &FailoverPolicy{
+				Config: &FailoverConfig{
+					Destinations: []*FailoverDestination{
+						newFailoverDestination("foo"),
+						newFailoverDestination("bar"),
+					},
+				},
+				PortConfigs: map[string]*FailoverConfig{
+					"admin": {
+						Destinations: []*FailoverDestination{
+							newFailoverDestination("admin"),
+						},
+					},
+					"web": {
+						Destinations: []*FailoverDestination{
+							newFailoverDestination("foo"), // duplicated
+							newFailoverDestination("www"),
+						},
+					},
+				},
+			},
+			expectDests: []*FailoverDestination{
+				newFailoverDestination("foo"),
+				newFailoverDestination("bar"),
+				newFailoverDestination("admin"),
+				newFailoverDestination("foo"), // duplicated
+				newFailoverDestination("www"),
+			},
+			expectRefs: []*pbresource.Reference{
+				newFailoverRef("foo"),
+				newFailoverRef("bar"),
+				newFailoverRef("admin"),
+				newFailoverRef("foo"), // duplicated
+				newFailoverRef("www"),
+			},
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func assertSliceEquals[V proto.Message](t *testing.T, expect, got []V) {
+	t.Helper()
+
+	require.Len(t, got, len(expect))
+
+	// O(N*M) scan
+	var expectedMissing []string
+	for _, expectVal := range expect {
+		found := false
+		for j, gotVal := range got {
+			if proto.Equal(expectVal, gotVal) {
+				found = true
+				got = append(got[:j], got[j+1:]...) // remove found item
+				break
+			}
+		}
+
+		if !found {
+			expectedMissing = append(expectedMissing, protoToString(t, expectVal))
+		}
+	}
+
+	if len(expectedMissing) > 0 || len(got) > 0 {
+		var gotMissing []string
+		for _, gotVal := range got {
+			gotMissing = append(gotMissing, protoToString(t, gotVal))
+		}
+
+		t.Fatalf("assertion failed: unmatched values\n\texpected: %s\n\tactual: %s",
+			expectedMissing,
+			gotMissing,
+		)
+	}
+}
+
+func protoToString[V proto.Message](t *testing.T, pb V) string {
+	m := protojson.MarshalOptions{
+		Indent: "  ",
+	}
+	gotJSON, err := m.Marshal(pb)
+	require.NoError(t, err)
+	return string(gotJSON)
+}
+
+func newFailoverRef(name string) *pbresource.Reference {
+	return &pbresource.Reference{
+		Type: &pbresource.Type{
+			Group:        "fake",
+			GroupVersion: "v1alpha1",
+			Kind:         "fake",
+		},
+		Tenancy: &pbresource.Tenancy{
+			Partition: "default",
+			Namespace: "default",
+			PeerName:  "local",
+		},
+		Name: name,
+	}
+}
+
+func newFailoverDestination(name string) *FailoverDestination {
+	return &FailoverDestination{
+		Ref: newFailoverRef(name),
+	}
+}