resource: add helper to normalize inner Reference tenancy during mutate (#18765)

When one resource contains an inner field that is of type *pbresource.Reference we want the
Tenancy to be reasonably defaulted by the following rules:

1. The final values will be limited by the scope of the referenced type.
2. Values will be inferred from the parent's tenancy, and if that is insufficient then using
   the default tenancy for the type's scope.
3. Namespace will only be used from a parent if the reference and the parent share a
   partition, otherwise the default namespace will be used.

Until we tackle peering, this hard codes an assumption of peer name being local. The
logic for defaulting may need adjustment when that is addressed.

internal/resource/tenancy.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"strings"
 
+	"google.golang.org/protobuf/proto"
+
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
@@ -78,3 +80,96 @@ func DefaultNamespacedTenancy() *pbresource.Tenancy {
 		PeerName: "local",
 	}
 }
+
+// DefaultReferenceTenancy will default/normalize the Tenancy of the provided
+// Reference in the context of some parent resource containing that Reference.
+// The default tenancy for the Reference's type is also provided in cases where
+// "default" is needed selectively or the parent is more precise than the
+// child.
+func DefaultReferenceTenancy(ref *pbresource.Reference, parentTenancy, scopeTenancy *pbresource.Tenancy) {
+	if ref == nil {
+		return
+	}
+	if ref.Tenancy == nil {
+		ref.Tenancy = &pbresource.Tenancy{}
+	}
+
+	if parentTenancy != nil {
+		dup := proto.Clone(parentTenancy).(*pbresource.Tenancy)
+		parentTenancy = dup
+	}
+
+	defaultTenancy(ref.Tenancy, parentTenancy, scopeTenancy)
+}
+
+func defaultTenancy(itemTenancy, parentTenancy, scopeTenancy *pbresource.Tenancy) {
+	if itemTenancy == nil {
+		panic("item tenancy is required")
+	}
+	if scopeTenancy == nil {
+		panic("scope tenancy is required")
+	}
+
+	if itemTenancy.PeerName == "" {
+		itemTenancy.PeerName = "local"
+	}
+	Normalize(itemTenancy)
+
+	if parentTenancy != nil {
+		// Recursively normalize this tenancy as well.
+		defaultTenancy(parentTenancy, nil, scopeTenancy)
+	}
+
+	// use scope defaults for parent
+	if parentTenancy == nil {
+		parentTenancy = scopeTenancy
+	}
+	Normalize(parentTenancy)
+
+	if !equalOrEmpty(itemTenancy.PeerName, "local") {
+		panic("peering is not supported yet for resource tenancies")
+	}
+	if !equalOrEmpty(parentTenancy.PeerName, "local") {
+		panic("peering is not supported yet for parent tenancies")
+	}
+	if !equalOrEmpty(scopeTenancy.PeerName, "local") {
+		panic("peering is not supported yet for scopes")
+	}
+
+	// Only retain the parts of the parent that apply to this resource.
+	if scopeTenancy.Partition == "" {
+		parentTenancy.Partition = ""
+		itemTenancy.Partition = ""
+	}
+	if scopeTenancy.Namespace == "" {
+		parentTenancy.Namespace = ""
+		itemTenancy.Namespace = ""
+	}
+
+	if parentTenancy.Partition == "" {
+		// (cluster scoped)
+	} else {
+		if itemTenancy.Partition == "" {
+			itemTenancy.Partition = parentTenancy.Partition
+		}
+		if parentTenancy.Namespace == "" {
+			// (partition scoped)
+		} else {
+			// (namespace scoped)
+
+			if itemTenancy.Namespace == "" {
+				if itemTenancy.Partition == parentTenancy.Partition {
+					// safe to copy the namespace
+					itemTenancy.Namespace = parentTenancy.Namespace
+				} else {
+					// cross-peer, the namespace must come from the scope default
+					itemTenancy.Namespace = scopeTenancy.Namespace
+				}
+			}
+		}
+	}
+}
+
+func equalOrEmpty(a, b string) bool {
+	return (a == b) || (a == "") || (b == "")
+}

internal/resource/tenancy_test.go

@@ -0,0 +1,237 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resource
+
+import (
+	"strings"
+	"testing"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	"github.com/hashicorp/consul/proto/private/prototest"
+)
+
+func TestDefaultReferenceTenancy(t *testing.T) {
+	// Just do a few small tests here and let the more complicated cases be covered by
+	// TestDefaultTenancy below.
+
+	t.Run("partition inference", func(t *testing.T) {
+		ref := &pbresource.Reference{
+			Type: &pbresource.Type{
+				Group:        "fake",
+				GroupVersion: "v1fake",
+				Kind:         "artificial",
+			},
+			Name: "blah",
+			Tenancy: &pbresource.Tenancy{
+				Namespace: "zim",
+			},
+		}
+
+		expect := &pbresource.Reference{
+			Type: &pbresource.Type{
+				Group:        "fake",
+				GroupVersion: "v1fake",
+				Kind:         "artificial",
+			},
+			Name:    "blah",
+			Tenancy: newTestTenancy("gir.zim"),
+		}
+
+		parent := newTestTenancy("gir.gaz")
+
+		DefaultReferenceTenancy(ref, parent, DefaultNamespacedTenancy())
+		prototest.AssertDeepEqual(t, expect, ref)
+	})
+
+	t.Run("full default", func(t *testing.T) {
+		ref := &pbresource.Reference{
+			Type: &pbresource.Type{
+				Group:        "fake",
+				GroupVersion: "v1fake",
+				Kind:         "artificial",
+			},
+			Name: "blah",
+		}
+
+		expect := &pbresource.Reference{
+			Type: &pbresource.Type{
+				Group:        "fake",
+				GroupVersion: "v1fake",
+				Kind:         "artificial",
+			},
+			Name:    "blah",
+			Tenancy: newTestTenancy("gir.gaz"),
+		}
+
+		parent := newTestTenancy("gir.gaz")
+
+		DefaultReferenceTenancy(ref, parent, DefaultNamespacedTenancy())
+		prototest.AssertDeepEqual(t, expect, ref)
+	})
+}
+
+func TestDefaultTenancy(t *testing.T) {
+	type testcase struct {
+		ref    *pbresource.Tenancy
+		parent *pbresource.Tenancy
+		scope  *pbresource.Tenancy
+		expect *pbresource.Tenancy
+	}
+
+	run := func(t *testing.T, tc testcase) {
+		got := proto.Clone(tc.ref).(*pbresource.Tenancy)
+
+		defaultTenancy(got, tc.parent, tc.scope)
+		prototest.AssertDeepEqual(t, tc.expect, got)
+	}
+
+	cases := map[string]testcase{
+		// Completely empty values get backfilled from the scope.
+		"clustered/empty/no-parent": {
+			ref:    newTestTenancy(""),
+			parent: nil,
+			scope:  DefaultClusteredTenancy(),
+			expect: DefaultClusteredTenancy(),
+		},
+		"partitioned/empty/no-parent": {
+			ref:    newTestTenancy(""),
+			parent: nil,
+			scope:  DefaultPartitionedTenancy(),
+			expect: DefaultPartitionedTenancy(),
+		},
+		"namespaced/empty/no-parent": {
+			ref:    newTestTenancy(""),
+			parent: nil,
+			scope:  DefaultNamespacedTenancy(),
+			expect: DefaultNamespacedTenancy(),
+		},
+		// Completely provided values are limited by the scope.
+		"clustered/full/no-parent": {
+			ref:    newTestTenancy("foo.bar"),
+			parent: nil,
+			scope:  DefaultClusteredTenancy(),
+			expect: DefaultClusteredTenancy(),
+		},
+		"partitioned/full/no-parent": {
+			ref:    newTestTenancy("foo.bar"),
+			parent: nil,
+			scope:  DefaultPartitionedTenancy(),
+			expect: newTestTenancy("foo"),
+		},
+		"namespaced/full/no-parent": {
+			ref:    newTestTenancy("foo.bar"),
+			parent: nil,
+			scope:  DefaultNamespacedTenancy(),
+			expect: newTestTenancy("foo.bar"),
+		},
+		// Completely provided parent values are limited by the scope before
+		// being blindly used for to fill in for the empty provided value.
+		"clustered/empty/full-parent": {
+			ref:    newTestTenancy(""),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultClusteredTenancy(),
+			expect: DefaultClusteredTenancy(),
+		},
+		"partitioned/empty/full-parent": {
+			ref:    newTestTenancy(""),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultPartitionedTenancy(),
+			expect: newTestTenancy("foo"),
+		},
+		"namespaced/empty/full-parent": {
+			ref:    newTestTenancy(""),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultNamespacedTenancy(),
+			expect: newTestTenancy("foo.bar"),
+		},
+		// (1) Partially filled values are only partially populated by parents.
+		"clustered/part-only/full-parent": {
+			ref:    newTestTenancy("zim"),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultClusteredTenancy(),
+			expect: DefaultClusteredTenancy(),
+		},
+		"partitioned/part-only/full-parent": {
+			ref:    newTestTenancy("zim"),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultPartitionedTenancy(),
+			expect: newTestTenancy("zim"),
+		},
+		"namespaced/part-only/full-parent": {
+			ref:    newTestTenancy("zim"),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultNamespacedTenancy(),
+			// partitions don't match so the namespace comes from the scope
+			expect: newTestTenancy("zim.default"),
+		},
+		// (2) Partially filled values are only partially populated by parents.
+		"clustered/ns-only/full-parent": {
+			// Leading dot implies no partition
+			ref:    newTestTenancy(".gir"),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultClusteredTenancy(),
+			expect: DefaultClusteredTenancy(),
+		},
+		"partitioned/ns-only/full-parent": {
+			// Leading dot implies no partition
+			ref:    newTestTenancy(".gir"),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultPartitionedTenancy(),
+			expect: newTestTenancy("foo"),
+		},
+		"namespaced/ns-only/full-parent": {
+			// Leading dot implies no partition
+			ref:    newTestTenancy(".gir"),
+			parent: newTestTenancy("foo.bar"),
+			scope:  DefaultNamespacedTenancy(),
+			expect: newTestTenancy("foo.gir"),
+		},
+		// Fully specified ignores parent.
+		"clustered/full/full-parent": {
+			ref:    newTestTenancy("foo.bar"),
+			parent: newTestTenancy("zim.gir"),
+			scope:  DefaultClusteredTenancy(),
+			expect: DefaultClusteredTenancy(),
+		},
+		"partitioned/full/full-parent": {
+			ref:    newTestTenancy("foo.bar"),
+			parent: newTestTenancy("zim.gir"),
+			scope:  DefaultPartitionedTenancy(),
+			expect: newTestTenancy("foo"),
+		},
+		"namespaced/full/full-parent": {
+			ref:    newTestTenancy("foo.bar"),
+			parent: newTestTenancy("zim.gir"),
+			scope:  DefaultNamespacedTenancy(),
+			expect: newTestTenancy("foo.bar"),
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func newTestTenancy(s string) *pbresource.Tenancy {
+	parts := strings.Split(s, ".")
+	switch len(parts) {
+	case 0:
+		return DefaultClusteredTenancy()
+	case 1:
+		v := DefaultPartitionedTenancy()
+		v.Partition = parts[0]
+		return v
+	case 2:
+		v := DefaultNamespacedTenancy()
+		v.Partition = parts[0]
+		v.Namespace = parts[1]
+		return v
+	default:
+		return &pbresource.Tenancy{Partition: "BAD", Namespace: "BAD", PeerName: "BAD"}
+	}
+}