[release/1.14.x] peering: peering partition failover fixes (#16693)

add local source partition for peered upstreams

.changelog/16693.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+peering: Fixes a bug where the importing partition was not added to peered failover targets, which causes issues when the importing partition is a non-default partition.
+```

agent/proxycfg/upstreams.go

@@ -462,6 +462,9 @@ func (s *handlerUpstreams) watchUpstreamTarget(ctx context.Context, snap *Config
 
 	if opts.peer != "" {
 		uid = NewUpstreamIDFromTargetID(opts.chainID)
+		// chainID has the partition stripped. However, when a target is in a cluster peer, the partition should be set
+		// to the local partition (i.e chain.Partition), since the peered target is imported into the local partition.
+		uid.OverridePartition(opts.entMeta.PartitionOrDefault())
 		correlationID = upstreamPeerWatchIDPrefix + uid.String()
 	}
 

agent/xds/clusters.go

@@ -950,7 +950,7 @@ func (s *ResourceGenerator) makeUpstreamClusterForPeerService(
 	// entire cluster.
 	outlierDetection.MaxEjectionPercent = &wrappers.UInt32Value{Value: 100}
 
-	s.Logger.Trace("generating cluster for", "cluster", clusterName)
+	s.Logger.Trace("generating cluster for", "cluster", clusterName, "uid", uid)
 	if c == nil {
 		c = &envoy_cluster_v3.Cluster{
 			Name:           clusterName,
@@ -1018,10 +1018,13 @@ func (s *ResourceGenerator) makeUpstreamClusterForPeerService(
 		makeTLSParametersFromProxyTLSConfig(cfgSnap.MeshConfigTLSOutgoing()),
 	)
 	err = injectSANMatcher(commonTLSContext, peerMeta.SpiffeID...)
+	s.Logger.Trace("injecting SAN matcher rules for cluster %q with SPIFFE IDs: %+v", clusterName, peerMeta.SpiffeID)
+
 	if err != nil {
 		return nil, fmt.Errorf("failed to inject SAN matcher rules for cluster %q: %v", clusterName, err)
 	}
 
+	s.Logger.Trace("injecting TLS context for cluster %q with SNI: %+v", clusterName, peerMeta.PrimarySNI())
 	tlsContext := &envoy_tls_v3.UpstreamTlsContext{
 		CommonTlsContext: commonTLSContext,
 		Sni:              peerMeta.PrimarySNI(),
@@ -1279,6 +1282,10 @@ func (s *ResourceGenerator) makeUpstreamClustersForDiscoveryChain(
 
 			targetUID := proxycfg.NewUpstreamIDFromTargetID(targetData.targetID)
 			if targetUID.Peer != "" {
+				// targetID already has a stripped partition, so targetUID will not have a partition either. However,
+				// when a failover target is in a cluster peer, the partition should be set to the local partition (i.e
+				// chain.Partition), since that's where the data is imported to.
+				targetUID.OverridePartition(chain.Partition)
 				peerMeta, found := upstreamsSnapshot.UpstreamPeerMeta(targetUID)
 				if !found {
 					s.Logger.Warn("failed to fetch upstream peering metadata for cluster", "target", targetUID)