Backport of Fix multiple issues related to proxycfg health queries. into release/1.14.x (#17267)

* backport of commit 464f79a5d3

* backport of commit ff4ba957b6

* backport of commit ab3e904470

---------

Co-authored-by: Derek Menteer <derek.menteer@hashicorp.com>

.changelog/17241.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+connect: Fix multiple inefficient behaviors when querying service health.
+```

agent/consul/health_endpoint.go

@@ -211,28 +211,10 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 		f = h.serviceNodesDefault
 	}
 
-	authzContext := acl.AuthorizerContext{
-		Peer: args.PeerName,
-	}
-	authz, err := h.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, &authzContext)
-	if err != nil {
-		return err
-	}
-
 	if err := h.srv.validateEnterpriseRequest(&args.EnterpriseMeta, false); err != nil {
 		return err
 	}
 
-	// If we're doing a connect or ingress query, we need read access to the service
-	// we're trying to find proxies for, so check that.
-	if args.Connect || args.Ingress {
-		// TODO(acl-error-enhancements) Look for ways to percolate this information up to give any feedback to the user.
-		if authz.ServiceRead(args.ServiceName, &authzContext) != acl.Allow {
-			// Just return nil, which will return an empty response (tested)
-			return nil
-		}
-	}
-
 	filter, err := bexpr.CreateFilter(args.Filter, nil, reply.Nodes)
 	if err != nil {
 		return err
@@ -254,6 +236,25 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 				return err
 			}
 
+			authzContext := acl.AuthorizerContext{
+				Peer: args.PeerName,
+			}
+			authz, err := h.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, &authzContext)
+			if err != nil {
+				return err
+			}
+
+			// If we're doing a connect or ingress query, we need read access to the service
+			// we're trying to find proxies for, so check that.
+			if args.Connect || args.Ingress {
+				// TODO(acl-error-enhancements) Look for ways to percolate this information up to give any feedback to the user.
+				if authz.ServiceRead(args.ServiceName, &authzContext) != acl.Allow {
+					// Return the index here so that the agent cache does not infinitely loop.
+					reply.Index = index
+					return nil
+				}
+			}
+
 			resolvedNodes := nodes
 			if args.MergeCentralConfig {
 				for _, node := range resolvedNodes {

agent/consul/health_endpoint_test.go

@@ -1124,6 +1124,7 @@ node "foo" {
 	var resp structs.IndexedCheckServiceNodes
 	assert.Nil(t, msgpackrpc.CallWithCodec(codec, "Health.ServiceNodes", &req, &resp))
 	assert.Len(t, resp.Nodes, 0)
+	assert.Greater(t, resp.Index, uint64(0))
 
 	// List w/ token. This should work since we're requesting "foo", but should
 	// also only contain the proxies with names that adhere to our ACL.

agent/proxycfg/state_test.go

@@ -781,7 +781,7 @@ func TestState_WatchesAndUpdates(t *testing.T) {
 				fmt.Sprintf("upstream-target:api-failover-remote.default.default.dc2:%s-failover-remote?dc=dc2", apiUID.String()): genVerifyServiceSpecificRequest("api-failover-remote", "", "dc2", true),
 				fmt.Sprintf("upstream-target:api-failover-local.default.default.dc2:%s-failover-local?dc=dc2", apiUID.String()):   genVerifyServiceSpecificRequest("api-failover-local", "", "dc2", true),
 				fmt.Sprintf("upstream-target:api-failover-direct.default.default.dc2:%s-failover-direct?dc=dc2", apiUID.String()): genVerifyServiceSpecificRequest("api-failover-direct", "", "dc2", true),
-				upstreamPeerWatchIDPrefix + fmt.Sprintf("%s-failover-to-peer?peer=cluster-01", apiUID.String()):                   genVerifyServiceSpecificPeeredRequest("api-failover-to-peer", "", "", "cluster-01", true),
+				upstreamPeerWatchIDPrefix + fmt.Sprintf("%s-failover-to-peer?peer=cluster-01", apiUID.String()):                   genVerifyServiceSpecificPeeredRequest("api-failover-to-peer", "", "dc1", "cluster-01", true),
 				fmt.Sprintf("mesh-gateway:dc2:%s-failover-remote?dc=dc2", apiUID.String()):                                        genVerifyGatewayWatch("dc2"),
 				fmt.Sprintf("mesh-gateway:dc1:%s-failover-local?dc=dc2", apiUID.String()):                                         genVerifyGatewayWatch("dc1"),
 			},

agent/proxycfg/upstreams.go

@@ -305,8 +305,19 @@ func (s *handlerUpstreams) resetWatchesFromChain(
 			watchedChainEndpoints = true
 		}
 
-		opts := targetWatchOpts{upstreamID: uid}
-		opts.fromChainTarget(target)
+		opts := targetWatchOpts{
+			upstreamID: uid,
+			chainID:    target.ID,
+			service:    target.Service,
+			filter:     target.Subset.Filter,
+			datacenter: target.Datacenter,
+			peer:       target.Peer,
+			entMeta:    target.GetEnterpriseMetadata(),
+		}
+		// Peering targets do not set the datacenter field, so we should default it here.
+		if opts.datacenter == "" {
+			opts.datacenter = s.source.Datacenter
+		}
 
 		err := s.watchUpstreamTarget(ctx, snap, opts)
 		if err != nil {
@@ -424,15 +435,6 @@ type targetWatchOpts struct {
 	entMeta    *acl.EnterpriseMeta
 }
 
-func (o *targetWatchOpts) fromChainTarget(t *structs.DiscoveryTarget) {
-	o.chainID = t.ID
-	o.service = t.Service
-	o.filter = t.Subset.Filter
-	o.datacenter = t.Datacenter
-	o.peer = t.Peer
-	o.entMeta = t.GetEnterpriseMetadata()
-}
-
 func (s *handlerUpstreams) watchUpstreamTarget(ctx context.Context, snap *ConfigSnapshotUpstreams, opts targetWatchOpts) error {
 	s.logger.Trace("initializing watch of target",
 		"upstream", opts.upstreamID,