add local source partition for peered upstreams

.changelog/16673.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

@@ -470,6 +470,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

@@ -1957,11 +1957,17 @@ func (s *ResourceGenerator) mapDiscoChainTargets(cfgSnap *proxycfg.ConfigSnapsho
 		}
 
 		if targetUID.Peer != "" {
+			// targetID has the partition stripped, 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 the peered failover target is imported into the local partition.
+			targetUID.OverridePartition(chain.Partition)
+
 			tbs, _ := upstreamsSnapshot.UpstreamPeerTrustBundles.Get(targetUID.Peer)
 			rootPEMs = tbs.ConcatenatedRootPEMs()
 
 			peerMeta, found := upstreamsSnapshot.UpstreamPeerMeta(targetUID)
 			if !found {
+				s.Logger.Warn("failed to fetch upstream peering metadata", "target", targetUID)
 				continue
 			}
 			sni = peerMeta.PrimarySNI()