From c5c290c5032a95798375842a16fd95b066c5bb66 Mon Sep 17 00:00:00 2001
From: freddygv <freddy@hashicorp.com>
Date: Mon, 13 Dec 2021 15:30:49 -0700
Subject: [PATCH] Validate chains are associated with upstreams

Previously we could get into a state where discovery chain entries were
not cleaned up after the associated watch was cancelled. These changes
add handling for that case where stray chain references are encountered.
---
 agent/xds/clusters.go  | 10 +++++++++-
 agent/xds/endpoints.go | 10 +++++++++-
 agent/xds/listeners.go |  7 +++++++
 agent/xds/routes.go    | 12 +++++++++---
 4 files changed, 34 insertions(+), 5 deletions(-)

diff --git a/agent/xds/clusters.go b/agent/xds/clusters.go
index 441b587355..fe978c2d98 100644
--- a/agent/xds/clusters.go
+++ b/agent/xds/clusters.go
@@ -78,13 +78,21 @@ func (s *ResourceGenerator) clustersFromSnapshotConnectProxy(cfgSnap *proxycfg.C
 	}
 
 	for id, chain := range cfgSnap.ConnectProxy.DiscoveryChain {
+		upstreamCfg := cfgSnap.ConnectProxy.UpstreamConfig[id]
+
+		explicit := upstreamCfg.HasLocalPortOrSocket()
+		if _, implicit := cfgSnap.ConnectProxy.IntentionUpstreams[id]; !implicit && !explicit {
+			// Discovery chain is not associated with a known explicit or implicit upstream so it is skipped.
+			continue
+		}
+
 		chainEndpoints, ok := cfgSnap.ConnectProxy.WatchedUpstreamEndpoints[id]
 		if !ok {
 			// this should not happen
 			return nil, fmt.Errorf("no endpoint map for upstream %q", id)
 		}
 
-		upstreamClusters, err := s.makeUpstreamClustersForDiscoveryChain(id, cfgSnap.ConnectProxy.UpstreamConfig[id], chain, chainEndpoints, cfgSnap)
+		upstreamClusters, err := s.makeUpstreamClustersForDiscoveryChain(id, upstreamCfg, chain, chainEndpoints, cfgSnap)
 		if err != nil {
 			return nil, err
 		}
diff --git a/agent/xds/endpoints.go b/agent/xds/endpoints.go
index e8521670c0..b93b6da760 100644
--- a/agent/xds/endpoints.go
+++ b/agent/xds/endpoints.go
@@ -48,11 +48,19 @@ func (s *ResourceGenerator) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg.
 		len(cfgSnap.ConnectProxy.PreparedQueryEndpoints)+len(cfgSnap.ConnectProxy.WatchedUpstreamEndpoints))
 
 	for id, chain := range cfgSnap.ConnectProxy.DiscoveryChain {
+		upstreamCfg := cfgSnap.ConnectProxy.UpstreamConfig[id]
+
+		explicit := upstreamCfg.HasLocalPortOrSocket()
+		if _, implicit := cfgSnap.ConnectProxy.IntentionUpstreams[id]; !implicit && !explicit {
+			// Discovery chain is not associated with a known explicit or implicit upstream so it is skipped.
+			continue
+		}
+
 		es := s.endpointsFromDiscoveryChain(
 			id,
 			chain,
 			cfgSnap.Locality,
-			cfgSnap.ConnectProxy.UpstreamConfig[id],
+			upstreamCfg,
 			cfgSnap.ConnectProxy.WatchedUpstreamEndpoints[id],
 			cfgSnap.ConnectProxy.WatchedGatewayEndpoints[id],
 		)
diff --git a/agent/xds/listeners.go b/agent/xds/listeners.go
index 4cb85aea04..d4de6feb56 100644
--- a/agent/xds/listeners.go
+++ b/agent/xds/listeners.go
@@ -95,6 +95,13 @@ func (s *ResourceGenerator) listenersFromSnapshotConnectProxy(cfgSnap *proxycfg.
 
 	for id, chain := range cfgSnap.ConnectProxy.DiscoveryChain {
 		upstreamCfg := cfgSnap.ConnectProxy.UpstreamConfig[id]
+
+		explicit := upstreamCfg.HasLocalPortOrSocket()
+		if _, implicit := cfgSnap.ConnectProxy.IntentionUpstreams[id]; !implicit && !explicit {
+			// Discovery chain is not associated with a known explicit or implicit upstream so it is skipped.
+			continue
+		}
+
 		cfg := s.getAndModifyUpstreamConfigForListener(id, upstreamCfg, chain)
 
 		// If escape hatch is present, create a listener from it and move on to the next
diff --git a/agent/xds/routes.go b/agent/xds/routes.go
index fd1a4ab9d4..7cb714daab 100644
--- a/agent/xds/routes.go
+++ b/agent/xds/routes.go
@@ -28,7 +28,7 @@ func (s *ResourceGenerator) routesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot)
 
 	switch cfgSnap.Kind {
 	case structs.ServiceKindConnectProxy:
-		return s.routesForConnectProxy(cfgSnap.ConnectProxy.DiscoveryChain)
+		return s.routesForConnectProxy(cfgSnap)
 	case structs.ServiceKindIngressGateway:
 		return s.routesForIngressGateway(
 			cfgSnap.IngressGateway.Listeners,
@@ -46,13 +46,19 @@ func (s *ResourceGenerator) routesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot)
 
 // routesFromSnapshotConnectProxy returns the xDS API representation of the
 // "routes" in the snapshot.
-func (s *ResourceGenerator) routesForConnectProxy(chains map[string]*structs.CompiledDiscoveryChain) ([]proto.Message, error) {
+func (s *ResourceGenerator) routesForConnectProxy(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
 	var resources []proto.Message
-	for id, chain := range chains {
+	for id, chain := range cfgSnap.ConnectProxy.DiscoveryChain {
 		if chain.IsDefault() {
 			continue
 		}
 
+		explicit := cfgSnap.ConnectProxy.UpstreamConfig[id].HasLocalPortOrSocket()
+		if _, implicit := cfgSnap.ConnectProxy.IntentionUpstreams[id]; !implicit && !explicit {
+			// Discovery chain is not associated with a known explicit or implicit upstream so it is skipped.
+			continue
+		}
+
 		virtualHost, err := makeUpstreamRouteForDiscoveryChain(id, chain, []string{"*"})
 		if err != nil {
 			return nil, err