Allow ingress gateways to send empty clusters, routes, and listeners (#7795)

This is useful when updating an config entry with no services, and the
expected behavior is that envoy closes all listeners and clusters.

We also allow empty routes because ingress gateways name route
configurations based on the port of the listener, so it is important we
remove any stale routes. Then, if a new listener with an old port is
added, we will not have to deal with stale routes hanging around routing
to the wrong place.

Endpoints are associated with clusters, and thus by deleting the
clusters we don't have to care about sending empty endpoint responses.

agent/xds/server.go

@@ -219,19 +219,28 @@ func (s *Server) process(stream ADSStream, reqCh <-chan *envoy.DiscoveryRequest)
 			resources: s.clustersFromSnapshot,
 			stream:    stream,
 			allowEmptyFn: func(cfgSnap *proxycfg.ConfigSnapshot) bool {
-				// Mesh and Terminating gateways are allowed to inform CDS of no clusters.
-				return cfgSnap.Kind == structs.ServiceKindMeshGateway || cfgSnap.Kind == structs.ServiceKindTerminatingGateway
+				// Mesh, Ingress, and Terminating gateways are allowed to inform CDS of
+				// no clusters.
+				return cfgSnap.Kind == structs.ServiceKindMeshGateway ||
+					cfgSnap.Kind == structs.ServiceKindTerminatingGateway ||
+					cfgSnap.Kind == structs.ServiceKindIngressGateway
 			},
 		},
 		RouteType: {
 			typeURL:   RouteType,
 			resources: routesFromSnapshot,
 			stream:    stream,
+			allowEmptyFn: func(cfgSnap *proxycfg.ConfigSnapshot) bool {
+				return cfgSnap.Kind == structs.ServiceKindIngressGateway
+			},
 		},
 		ListenerType: {
 			typeURL:   ListenerType,
 			resources: s.listenersFromSnapshot,
 			stream:    stream,
+			allowEmptyFn: func(cfgSnap *proxycfg.ConfigSnapshot) bool {
+				return cfgSnap.Kind == structs.ServiceKindIngressGateway
+			},
 		},
 	}
 

agent/xds/server_test.go

@@ -853,3 +853,70 @@ func TestServer_Check(t *testing.T) {
 		})
 	}
 }
+
+func TestServer_StreamAggregatedResources_IngressEmptyResponse(t *testing.T) {
+	logger := testutil.Logger(t)
+	mgr := newTestManager(t)
+	aclResolve := func(id string) (acl.Authorizer, error) {
+		// Allow all
+		return acl.RootAuthorizer("manage"), nil
+	}
+	envoy := NewTestEnvoy(t, "ingress-gateway", "")
+	defer envoy.Close()
+
+	s := Server{
+		Logger:       logger,
+		CfgMgr:       mgr,
+		Authz:        mgr,
+		ResolveToken: aclResolve,
+	}
+	s.Initialize()
+
+	sid := structs.NewServiceID("ingress-gateway", nil)
+
+	go func() {
+		err := s.StreamAggregatedResources(envoy.stream)
+		require.NoError(t, err)
+	}()
+
+	// Register the proxy to create state needed to Watch() on
+	mgr.RegisterProxy(t, sid)
+
+	// Send initial cluster discover
+	envoy.SendReq(t, ClusterType, 0, 0)
+
+	// Check no response sent yet
+	assertChanBlocked(t, envoy.stream.sendCh)
+
+	// Deliver a new snapshot with no services
+	snap := proxycfg.TestConfigSnapshotIngressGatewayNoServices(t)
+	mgr.DeliverConfig(t, sid, snap)
+
+	emptyClusterJSON := `{
+		"versionInfo": "` + hexString(1) + `",
+		"resources": [],
+		"typeUrl": "type.googleapis.com/envoy.api.v2.Cluster",
+		"nonce": "` + hexString(1) + `"
+		}`
+	emptyListenerJSON := `{
+		"versionInfo": "` + hexString(1) + `",
+		"resources": [],
+		"typeUrl": "type.googleapis.com/envoy.api.v2.Listener",
+		"nonce": "` + hexString(2) + `"
+		}`
+	emptyRouteJSON := `{
+		"versionInfo": "` + hexString(1) + `",
+		"resources": [],
+		"typeUrl": "type.googleapis.com/envoy.api.v2.RouteConfiguration",
+		"nonce": "` + hexString(3) + `"
+		}`
+
+	assertResponseSent(t, envoy.stream.sendCh, emptyClusterJSON)
+
+	// Send initial listener discover
+	envoy.SendReq(t, ListenerType, 0, 0)
+	assertResponseSent(t, envoy.stream.sendCh, emptyListenerJSON)
+
+	envoy.SendReq(t, RouteType, 0, 0)
+	assertResponseSent(t, envoy.stream.sendCh, emptyRouteJSON)
+}