connect: validate upstreams and prevent duplicates (#6224)

* connect: validate upstreams and prevent duplicates

* Actually run Upstream.Validate() instead of ignoring it as dead code.

* Prevent two upstreams from declaring the same bind address and port.
  It wouldn't work anyway.

* Prevent two upstreams from being declared that use the same
  type+name+namespace+datacenter. Due to how the Upstream.Identity()
  function worked this ended up mostly being enforced in xDS at use-time,
  but it should be enforced more clearly at register-time.

agent/structs/connect_proxy_config.go

@@ -196,9 +196,11 @@ type Upstream struct {
 
 // Validate sanity checks the struct is valid
 func (u *Upstream) Validate() error {
-	if u.DestinationType != UpstreamDestTypeService &&
-		u.DestinationType != UpstreamDestTypePreparedQuery {
-		return fmt.Errorf("unknown upstream destination type")
+	switch u.DestinationType {
+	case UpstreamDestTypePreparedQuery:
+	case UpstreamDestTypeService, "":
+	default:
+		return fmt.Errorf("unknown upstream destination type: %q", u.DestinationType)
 	}
 
 	if u.DestinationName == "" {
@@ -228,6 +230,38 @@ func (u *Upstream) ToAPI() api.Upstream {
 	}
 }
 
+// ToKey returns a value-type representation that uniquely identifies the
+// upstream in a canonical way. Set and unset values are deliberately handled
+// differently.
+//
+// These fields should be user-specificed explicit values and not inferred
+// values.
+func (u *Upstream) ToKey() UpstreamKey {
+	return UpstreamKey{
+		DestinationType:      u.DestinationType,
+		DestinationNamespace: u.DestinationNamespace,
+		DestinationName:      u.DestinationName,
+		Datacenter:           u.Datacenter,
+	}
+}
+
+type UpstreamKey struct {
+	DestinationType      string
+	DestinationName      string
+	DestinationNamespace string
+	Datacenter           string
+}
+
+func (k UpstreamKey) String() string {
+	return fmt.Sprintf(
+		"[type=%q, name=%q, namespace=%q, datacenter=%q]",
+		k.DestinationType,
+		k.DestinationName,
+		k.DestinationNamespace,
+		k.Datacenter,
+	)
+}
+
 // Identifier returns a string representation that uniquely identifies the
 // upstream in a canonical but human readable way.
 func (u *Upstream) Identifier() string {

agent/structs/structs.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/rand"
+	"net"
 	"reflect"
 	"regexp"
 	"sort"
@@ -948,6 +949,39 @@ func (s *NodeService) Validate() error {
 			result = multierror.Append(result, fmt.Errorf(
 				"A Proxy cannot also be Connect Native, only typical services"))
 		}
+
+		// ensure we don't have multiple upstreams for the same service
+		var (
+			upstreamKeys = make(map[UpstreamKey]struct{})
+			bindAddrs    = make(map[string]struct{})
+		)
+		for _, u := range s.Proxy.Upstreams {
+			if err := u.Validate(); err != nil {
+				result = multierror.Append(result, err)
+				continue
+			}
+
+			uk := u.ToKey()
+			if _, ok := upstreamKeys[uk]; ok {
+				result = multierror.Append(result, fmt.Errorf(
+					"upstreams cannot contain duplicates of %s", uk))
+				continue
+			}
+			upstreamKeys[uk] = struct{}{}
+
+			addr := u.LocalBindAddress
+			if addr == "" {
+				addr = "127.0.0.1"
+			}
+			addr = net.JoinHostPort(addr, fmt.Sprintf("%d", u.LocalBindPort))
+
+			if _, ok := bindAddrs[addr]; ok {
+				result = multierror.Append(result, fmt.Errorf(
+					"upstreams cannot contain duplicates by local bind address and port; %q is specified twice", addr))
+				continue
+			}
+			bindAddrs[addr] = struct{}{}
+		}
 	}
 
 	// MeshGateway validation

agent/structs/structs_test.go

@@ -467,6 +467,206 @@ func TestStructs_NodeService_ValidateConnectProxy(t *testing.T) {
 			func(x *NodeService) { x.Connect.Native = true },
 			"cannot also be",
 		},
+
+		{
+			"connect-proxy: upstream missing type (defaulted)",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{{
+					DestinationName: "foo",
+					LocalBindPort:   5000,
+				}}
+			},
+			"",
+		},
+		{
+			"connect-proxy: upstream invalid type",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{{
+					DestinationType: "garbage",
+					DestinationName: "foo",
+					LocalBindPort:   5000,
+				}}
+			},
+			"unknown upstream destination type",
+		},
+		{
+			"connect-proxy: upstream empty name",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{{
+					DestinationType: UpstreamDestTypeService,
+					LocalBindPort:   5000,
+				}}
+			},
+			"upstream destination name cannot be empty",
+		},
+		{
+			"connect-proxy: upstream empty bind port",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{{
+					DestinationType: UpstreamDestTypeService,
+					DestinationName: "foo",
+					LocalBindPort:   0,
+				}}
+			},
+			"upstream local bind port cannot be zero",
+		},
+		{
+			"connect-proxy: Upstreams almost-but-not-quite-duplicated in various ways",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{ // baseline
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						LocalBindPort:   5000,
+					},
+					{ // different bind address
+						DestinationType:  UpstreamDestTypeService,
+						DestinationName:  "bar",
+						LocalBindAddress: "127.0.0.2",
+						LocalBindPort:    5000,
+					},
+					{ // different datacenter
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						Datacenter:      "dc2",
+						LocalBindPort:   5001,
+					},
+					{ // explicit default namespace
+						DestinationType:      UpstreamDestTypeService,
+						DestinationName:      "foo",
+						DestinationNamespace: "default",
+						LocalBindPort:        5003,
+					},
+					{ // different namespace
+						DestinationType:      UpstreamDestTypeService,
+						DestinationName:      "foo",
+						DestinationNamespace: "alternate",
+						LocalBindPort:        5002,
+					},
+					{ // different type
+						DestinationType: UpstreamDestTypePreparedQuery,
+						DestinationName: "foo",
+						LocalBindPort:   5004,
+					},
+				}
+			},
+			"",
+		},
+		{
+			"connect-proxy: Upstreams duplicated by port",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						LocalBindPort:   5000,
+					},
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						LocalBindPort:   5000,
+					},
+				}
+			},
+			"upstreams cannot contain duplicates",
+		},
+		{
+			"connect-proxy: Upstreams duplicated by ip and port",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{
+						DestinationType:  UpstreamDestTypeService,
+						DestinationName:  "foo",
+						LocalBindAddress: "127.0.0.2",
+						LocalBindPort:    5000,
+					},
+					{
+						DestinationType:  UpstreamDestTypeService,
+						DestinationName:  "bar",
+						LocalBindAddress: "127.0.0.2",
+						LocalBindPort:    5000,
+					},
+				}
+			},
+			"upstreams cannot contain duplicates",
+		},
+		{
+			"connect-proxy: Upstreams duplicated by ip and port with ip defaulted in one",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						LocalBindPort:   5000,
+					},
+					{
+						DestinationType:  UpstreamDestTypeService,
+						DestinationName:  "foo",
+						LocalBindAddress: "127.0.0.1",
+						LocalBindPort:    5000,
+					},
+				}
+			},
+			"upstreams cannot contain duplicates",
+		},
+		{
+			"connect-proxy: Upstreams duplicated by name",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						LocalBindPort:   5000,
+					},
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						LocalBindPort:   5001,
+					},
+				}
+			},
+			"upstreams cannot contain duplicates",
+		},
+		{
+			"connect-proxy: Upstreams duplicated by name and datacenter",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						Datacenter:      "dc2",
+						LocalBindPort:   5000,
+					},
+					{
+						DestinationType: UpstreamDestTypeService,
+						DestinationName: "foo",
+						Datacenter:      "dc2",
+						LocalBindPort:   5001,
+					},
+				}
+			},
+			"upstreams cannot contain duplicates",
+		},
+		{
+			"connect-proxy: Upstreams duplicated by name and namespace",
+			func(x *NodeService) {
+				x.Proxy.Upstreams = Upstreams{
+					{
+						DestinationType:      UpstreamDestTypeService,
+						DestinationName:      "foo",
+						DestinationNamespace: "alternate",
+						LocalBindPort:        5000,
+					},
+					{
+						DestinationType:      UpstreamDestTypeService,
+						DestinationName:      "foo",
+						DestinationNamespace: "alternate",
+						LocalBindPort:        5001,
+					},
+				}
+			},
+			"upstreams cannot contain duplicates",
+		},
 	}
 
 	for _, tc := range cases {