Envoy Ingress£ºContour»ù±¾Ô­ÀíºÍÔ´Âë·ÖÎö

# yaml °²×°
kubectl apply -f https://projectcontour.io/quickstart/contour.yaml
# ÑéÖ¤
kubectl get pod -n projectcontour

# Ìí¼Ó²Ö¿â
helm repo add bitnami https://charts.bitnami.com/bitnami  
helm repo update
helm install contour --namespace projectcontour bitnami/contour

$ kubectl get deploy,svc,ds,job -n projectcontour
NAME                      READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/contour   2/2     2            2           12d

NAME              TYPE           CLUSTER-IP       EXTERNAL-IP 
service/contour   ClusterIP      172.20.126.255   <none>    8001/TCP                     12d
service/envoy     LoadBalancer   172.20.12.164    ...

NAME                   DESIRED   CURRENT   READY   UP-TO-DATE   AVAILABLE   NODE SELECTOR   AGE
daemonset.apps/envoy   10        10        10      10           10          <none>          12d

$ kubectl get crd|grep contour
contourconfigurations.projectcontour.io                      2021-11-04T13:30:06Z
contourdeployments.projectcontour.io                         2021-11-04T13:30:06Z
extensionservices.projectcontour.io                          2021-11-04T13:30:06Z
httpproxies.projectcontour.io                                2021-11-04T13:30:06Z
tlscertificatedelegations.projectcontour.io                  2021-11-04T13:30:06Z

apiVersion: projectcontour.io/v1
kind: HTTPProxy
metadata:
  name: name-example-foo
  namespace: default
spec:
  virtualhost:
    fqdn: foo1.bar.com
  routes:
    - conditions:
      - prefix: /
      services:
        - name: s1
          port: 80

# ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: basic
spec:
  rules:
  - host: foo-basic.bar.com
    http:
      paths:
      - backend:
          service:
            name: s1
            port:
              number: 80

# httpproxy.yaml
apiVersion: projectcontour.io/v1
kind: HTTPProxy
metadata:
  name: basic
spec:
  virtualhost:
    fqdn: foo-basic.bar.com
  routes:
    - conditions:
      - prefix: /
      services:
        - name: s1
          port: 80

// contour/cmd/contour/serve.go
func doServe(log logrus.FieldLogger, ctx *serveContext) error {
  // ÕâÀï¼àÌý contour µÄ crd ×ÊÔ´, ±ÈÈ磺httpproxy
	for _, r := range k8s.DefaultResources() {
		inf, err := clients.InformerForResource(r)
		......
		inf.AddEventHandler(&dynamicHandler)
	}
  // ¼àÌý ingress ×ÊÔ´
	for _, r := range k8s.IngressV1Resources() {
		if err := informOnResource(clients, r, &dynamicHandler); err != nil {
			log.WithError(err).WithField("resource", r).Fatal("failed to create informer")
		}
	}
  ......
  // ¼àÌý secret ×ÊÔ´
	for _, r := range k8s.SecretsResources() {
    ......
		if err := informOnResource(clients, r, handler); err != nil {
			log.WithError(err).WithField("resource", r).Fatal("failed to create informer")
		}
	}

	// ¼àÌý endpoint ×ÊÔ´
	for _, r := range k8s.EndpointsResources() {
		if err := informOnResource(clients, r, &k8s.DynamicClientHandler{
			Next: &contour.EventRecorder{
				Next:    endpointHandler,
				Counter: contourMetrics.EventHandlerOperations,
			},
			Converter: converter,
			Logger:    log.WithField("context", "endpointstranslator"),
		}); err != nil {
			log.WithError(err).WithField("resource", r).Fatal("failed to create informer")
		}
	}
  ......
}

// k8s.io/client-go/tools/cache/controller.go
type ResourceEventHandler interface {
	OnAdd(obj interface{})
	OnUpdate(oldObj, newObj interface{})
	OnDelete(obj interface{})
}

func (e *EventHandler) OnAdd(obj interface{}) {
	e.update <- opAdd{obj: obj}
}

func (e *EventHandler) OnUpdate(oldObj, newObj interface{}) {
	e.update <- opUpdate{oldObj: oldObj, newObj: newObj}
}

func (e *EventHandler) OnDelete(obj interface{}) {
	e.update <- opDelete{obj: obj}
}

// contour/cmd/contour/server.go
func doServe(log logrus.FieldLogger, ctx *serveContext) error {
	......
  // Æô¶¯Ð­³Ì
  g.Add(eventHandler.Start())
  ......
}

// counter/internal/contour/handler.go
func (e *EventHandler) Start() func(<-chan struct{}) error {
	e.update = make(chan interface{})
	return e.run
}

func (e *EventHandler) run(stop <-chan struct{}) error {
	......
  // ËÀÑ­»·£¬Ò»Ö±¼àÌý channel ÖеÄʼþ
	for {
		select {
		case op := <-e.update:
			if e.onUpdate(op) {
				......
			}
      ......
 }
  
// ´¦Àíʼþ
// ½«×ÊÔ´±£´æµ½ e.Builder.Source
func (e *EventHandler) onUpdate(op interface{}) bool {
	switch op := op.(type) {
	case opAdd:
		return e.Builder.Source.Insert(op.obj)
	case opUpdate:
		......
		remove := e.Builder.Source.Remove(op.oldObj)
		insert := e.Builder.Source.Insert(op.newObj)
		return remove || insert
	case opDelete:
		return e.Builder.Source.Remove(op.obj)
	case bool:
		return op
	default:
		return false
	}
}

// contour/internal/xdscache/v3/endpointstranslator.go
func (e *EndpointsTranslator) OnAdd(obj interface{}) {
	switch obj := obj.(type) {
	case *v1.Endpoints:
    // ¸üб¾µØ»º´æ
		e.cache.UpdateEndpoint(obj)
    // ºÏ²¢×ÊÔ´
		e.Merge(e.cache.Recalculate())
		e.Notify()
		if e.Observer != nil {
      // ˢб£´æ¿ìÕÕ
			e.Observer.Refresh()
		}
	default:
		e.Errorf("OnAdd unexpected type %T: %#v", obj, obj)
	}
}

// contour/internal/dag/cache.go
type KubernetesCache struct {
	......
	ingresses                 map[types.NamespacedName]*networking_v1.Ingress
	ingressclass              *networking_v1.IngressClass
	httpproxies               map[types.NamespacedName]*contour_api_v1.HTTPProxy
	secrets                   map[types.NamespacedName]*v1.Secret
	tlscertificatedelegations map[types.NamespacedName]*contour_api_v1.TLSCertificateDelegation
	services                  map[types.NamespacedName]*v1.Service
	namespaces                map[string]*v1.Namespace
	gatewayclass              *gatewayapi_v1alpha1.GatewayClass
	gateway                   *gatewayapi_v1alpha1.Gateway
	httproutes                map[types.NamespacedName]*gatewayapi_v1alpha1.HTTPRoute
	tlsroutes                 map[types.NamespacedName]*gatewayapi_v1alpha1.TLSRoute
	tcproutes                 map[types.NamespacedName]*gatewayapi_v1alpha1.TCPRoute
	udproutes                 map[types.NamespacedName]*gatewayapi_v1alpha1.UDPRoute
	backendpolicies           map[types.NamespacedName]*gatewayapi_v1alpha1.BackendPolicy
	extensions                map[types.NamespacedName]*contour_api_v1alpha1.ExtensionService
  ......
}

// mapÖÐkeyµÄÊý¾ÝÀàÐÍ£ºÓÉ namespace ºÍ name ¹²Í¬¾ö¶¨
type NamespacedName struct {
	Namespace string
	Name      string
}

// contour/internal/dag/cache.go
func (kc *KubernetesCache) Insert(obj interface{}) bool {
  // ³õʼ»¯ map£¬È·±£Ö´ÐÐÒ»´Î
	kc.initialize.Do(kc.init)
  // ¸ù¾Ý²»Í¬µÄ×ÊÔ´ÀàÐÍ£¬·ÅÈ벻ͬµÄ map
	switch obj := obj.(type) {
	case *v1.Secret:
		......
    // ±£´æ secret ×ÊÔ´
		kc.secrets[k8s.NamespacedNameOf(obj)] = obj
		return kc.secretTriggersRebuild(obj)
	case *v1.Service:
		kc.services[k8s.NamespacedNameOf(obj)] = obj
		return kc.serviceTriggersRebuild(obj)
	case *v1.Namespace:
		kc.namespaces[obj.Name] = obj
		return true
	......
	}

	return false
}

// contour/internal/contour/handler.go
func (e *EventHandler) rebuildDAG() {
  // ½« KubernetesCache ¹¹½¨³É DAG
	latestDAG := e.Builder.Build()
  // DAG ±ä¸üʱ£¬Í¨ÖªËùÓÐµÄ Observer£¬
  // ÿÖÖ Observer ÊÇÒ»¸ö XDS ¶ÔÏó£¬È¡³ö DAG ×ÊԴת»»Îª XDS ÅäÖÃ
	e.Observer.OnChange(latestDAG)
  // ¸üРsnapshot ÐÅÏ¢
	for _, upd := range latestDAG.StatusCache.GetStatusUpdates() {
		e.StatusUpdater.Send(upd)
	}

}

// contour/internal/dag/dag.go
type DAG struct {
	// StatusCache holds a cache of status updates to send.
	StatusCache status.Cache

	// roots are the root vertices of this DAG.
	roots []Vertex
}

// Vertex is a node in the DAG that can be visited.
type Vertex interface {
	Visit(func(Vertex))
}

// contour/internal/dag/builder.go
func (b *Builder) Build() *DAG {
	// Ïȹ¹ÔìÒ»¸ö¿ÕµÄ DAG
  dag := DAG{
		StatusCache: status.NewCache(b.Source.ConfiguredGateway),
	}
 
  // ±éÀúËùÓÐµÄ processor£¬Ã¿¸ö processor °Ñ²»Í¬µÄ k8s ×ÊÔ´±ä³É DAG µÄÒ»²¿·ÖÐÅÏ¢
  // ËùÓÐµÄ processor ¹²Í¬¹¹³ÉÍêÕûµÄ DAG
	for _, p := range b.Processors {
		p.Run(&dag, &b.Source)
	}
	return &dag
}

func getDAGBuilder(ctx *serveContext, clients *k8s.Clients, clientCert, fallbackCert *types.NamespacedName, 	log logrus.FieldLogger) dag.Builder {

  dagProcessors := []dag.Processor{
      &dag.IngressProcessor{
        ...
      },
      &dag.ExtensionServiceProcessor{
        ...
      },
      &dag.HTTPProxyProcessor{
       ...
      },
    }
    if ctx.Config.GatewayConfig != nil && clients.ResourcesExist(k8s.GatewayAPIResources()...) {
      dagProcessors = append(dagProcessors, &dag.GatewayAPIProcessor{
        FieldLogger: log.WithField("context", "GatewayAPIProcessor"),
      })
    }
    ...
    dagProcessors = append(dagProcessors, &dag.ListenerProcessor{})
  ...
}

func (p *HTTPProxyProcessor) Run(dag *DAG, source *KubernetesCache) {
	...
  // УÑéºÏ·¨ÐÔ
	for _, proxy := range p.validHTTPProxies() {
    // ¼ÆËã·ÓɹæÔò
    // Èë²ÎΪ crd ×ÊÔ´ HttpProxy
    // ÄÚ²¿¾­¹ý¸÷ÖÖ¼ÆË㣬¹¹Ôì·ÓɹØϵ£¬×îÖÕ±£´æÔÚDAGÖÐ
		p.computeHTTPProxy(proxy)
	}

	for meta := range p.orphaned {
		proxy, ok := p.source.httpproxies[meta]
		if ok {
			pa, commit := p.dag.StatusCache.ProxyAccessor(proxy)
			pa.ConditionFor(status.ValidCondition).AddError(contour_api_v1.ConditionTypeOrphanedError,
				"Orphaned",
				"this HTTPProxy is not part of a delegation chain from a root HTTPProxy")
			commit()
		}
	}
}

// contour/internal/contour/handler.go
func (e *EventHandler) rebuildDAG() {
  // ½« KubernetesCache ¹¹½¨³É DAG
	latestDAG := e.Builder.Build()
  // DAG ±ä¸üʱ£¬Í¨ÖªËùÓÐµÄ Observer£¬
  // ÿÖÖ Observer ÊÇÒ»¸ö XDS ¶ÔÏó£¬È¡³ö DAG ×ÊԴת»»Îª XDS ÅäÖÃ
	e.Observer.OnChange(latestDAG)
  // ¸üРsnapshot ÐÅÏ¢
	for _, upd := range latestDAG.StatusCache.GetStatusUpdates() {
		e.StatusUpdater.Send(upd)
	}
}

// contour/cmd/contour/serve.go
func doServe(log logrus.FieldLogger, ctx *serveContext) error {
	 ...
  // ResoureCache Áбí
  resources := []xdscache.ResourceCache{
		xdscache_v3.NewListenerCache(listenerConfig, ctx.statsAddr, ctx.statsPort),
		&xdscache_v3.SecretCache{},
		&xdscache_v3.RouteCache{},
		&xdscache_v3.ClusterCache{},
		endpointHandler,
	}
  ...
	eventHandler := &contour.EventHandler{
		...
    // ×¢²áÒ»¸ö×éºÏµÄ Observer
		Observer:        dag.ComposeObservers(append(xdscache.ObserversOf(resources), snapshotHandler)...),
		...
	}
}

// ×éºÏ¶à¸öObserver
func ComposeObservers(observers ...Observer) Observer {
	return ObserverFunc(func(d *DAG) {
		for _, o := range observers {
			o.OnChange(d)
		}
	})
}

// contour/internal/xdscache/resources.go
type ResourceCache interface {
	dag.Observer
	xds.Resource
}

// contour/internal/dag/dag.go
type Observer interface {
  // µ± DAG±ä¸üʱ£¬´¥·¢ÏàÓ¦ÒµÎñÂß¼­
	OnChange(*DAG)
}

// contour/internal/xds/resource.go
type Resource interface {
  // ·µ»ØÈ«²¿×ÊÔ´ÄÚÈÝ
	// Contents returns the contents of this resource.
	Contents() []proto.Message
  
  // ͨ¹ý¸ø¶¨µÄÃû³Æ£¬·µ»Ø¶ÔÓ¦µÄ×ÊÔ´ÄÚÈÝ
	// Query returns an entry for each resource name supplied.
	Query(names []string) []proto.Message

  // ×¢²á×ÊÔ´
	// Register registers ch to receive a value when Notify is called.
	Register(chan int, int, ...string)

  // ×ÊÔ´ÀàÐÍ
	// TypeURL returns the typeURL of messages returned from Values.
	TypeURL() string
}

// contour/internal/xdscache/v3/cluster.go
// ClusterCache manages the contents of the gRPC CDS cache.
type ClusterCache struct {
	mu     sync.Mutex
	values map[string]*envoy_cluster_v3.Cluster
	contour.Cond
}

// contour/internal/xdscache/v3/listener.go
// ListenerCache manages the contents of the gRPC LDS cache.
type ListenerCache struct {
	mu           sync.Mutex
	values       map[string]*envoy_listener_v3.Listener
	staticValues map[string]*envoy_listener_v3.Listener

	Config ListenerConfig
	contour.Cond
}

// contour/internal/xdscache/v3/route.go
// RouteCache manages the contents of the gRPC RDS cache.
type RouteCache struct {
	mu     sync.Mutex
	values map[string]*envoy_route_v3.RouteConfiguration
	contour.Cond
}

// contour/internal/xdscache/v3/secret.go
// SecretCache manages the contents of the gRPC SDS cache.
type SecretCache struct {
	mu     sync.Mutex
	values map[string]*envoy_tls_v3.Secret
	contour.Cond
}

// contour/internal/xdscache/v3/endpointstranslator.go
// A EndpointsTranslator translates Kubernetes Endpoints objects into Envoy
// ClusterLoadAssignment resources.
type EndpointsTranslator struct {
	// Observer notifies when the endpoints cache has been updated.
	Observer contour.Observer

	contour.Cond
	logrus.FieldLogger

	cache EndpointsCache

	mu      sync.Mutex // Protects entries.
	entries map[string]*envoy_endpoint_v3.ClusterLoadAssignment
}

// contour/internal/xdscache/v3/route.go
func (c *RouteCache) OnChange(root *dag.DAG) {
  // ¶ÁÈ¡ DAG ÖеÄÐÅÏ¢£¬Éú³É envoy µÄ RDS ×ÊÔ´ envoy_route_v3.RouteConfiguration
	routes := visitRoutes(root)
  // ½«ÅäÖÃÎļþ·Åµ½ value ×Ö¶ÎÖУ¬¹© Query¡¢Context µÈ·½·¨µ÷ÓÃʱ²é¿´
	c.Update(routes)
}

// contour/internal/xdscache/v3/route.go
func visitRoutes(root dag.Vertex) map[string]*envoy_route_v3.RouteConfiguration {
	// Collect the route configurations for all the routes we can
	// find. For HTTP hosts, the routes will all be collected on the
	// well-known ENVOY_HTTP_LISTENER, but for HTTPS hosts, we will
	// generate a per-vhost collection. This lets us keep different
	// SNI names disjoint when we later configure the listener.
  
  // http ºÍ https ÐèÒª×ö²»Í¬µÄ´¦Àí
	rv := routeVisitor{
		routes: map[string]*envoy_route_v3.RouteConfiguration{
			ENVOY_HTTP_LISTENER: envoy_v3.RouteConfiguration(ENVOY_HTTP_LISTENER),
		},
	}

	rv.visit(root)

	for _, v := range rv.routes {
		sort.Stable(sorter.For(v.VirtualHosts))
	}

	return rv.routes
}

// contour/internal/xdscache/v3/route.go
// ´Ó¸ù½Úµã¿ªÊ¼²éÕÒËùÓÐµÄ Route£¬²éÕÒ˳Ðò£º Listener -> VirtualHost(SecureVirtualHost) -> Route
func (v *routeVisitor) visit(vertex dag.Vertex) {
	switch l := vertex.(type) {
   // ´¦Àí DAG ÖÐµÄ Listener
	case *dag.Listener:
		l.Visit(func(vertex dag.Vertex) {
			switch vh := vertex.(type) {
      // ´¦Àí VirtualHost
			case *dag.VirtualHost:
				v.onVirtualHost(vh)
      // ´¦Àí SecureVirtualHost
			case *dag.SecureVirtualHost:
				v.onSecureVirtualHost(vh)
			default:
				// recurse
				vertex.Visit(v.visit)
			}
		})
	default:
		// recurse
		vertex.Visit(v.visit)
	}
}

// ²éÕÒ Listener.VirtualHost.Route
func (v *routeVisitor) onVirtualHost(vh *dag.VirtualHost) {
	var routes []*dag.Route

	vh.Visit(func(v dag.Vertex) {
		route, ok := v.(*dag.Route)
		if !ok {
			return
		}
    // ½«ËùÓÐµÄ route ÊÕ¼¯µ½ÁбíÖÐ
		routes = append(routes, route)
	})

	if len(routes) == 0 {
		return
	}

  // ¶¨ÒåÁË°Ñ dag.Route ¶ÔÏóת»»Îª envoy_route_v3.Route ¶ÔÏóµÄ·½·¨
  // ½«Á½ÖÖÊý¾Ý¸ñʽ½øÐÐת»»
	toEnvoyRoute := func(route *dag.Route) *envoy_route_v3.Route {
		...
		rt := &envoy_route_v3.Route{
			Match:  envoy_v3.RouteMatch(route),
			Action: envoy_v3.RouteRoute(route),
		}
		...
		return rt
	}

  // ¶Ô route ÅÅÐò
	sortRoutes(routes)
  // ÕâÀïÉú³ÉµÄÅäÖ㬴洢µ½ routes Õâ¸ömapÖУ¬key ÊÇ ingress_http
	v.routes[ENVOY_HTTP_LISTENER].VirtualHosts = append(v.routes[ENVOY_HTTP_LISTENER].VirtualHosts, toEnvoyVirtualHost(vh, routes, toEnvoyRoute))
}

func doServe(log logrus.FieldLogger, ctx *serveContext) error {
	...
  // ÓÃÓÚÉú³É¿ìÕÕ£¬Õⲿ·Ö±È½Ï¼òµ¥£¬²»×ö½éÉÜ
  snapshotHandler := xdscache.NewSnapshotHandler(resources, log.WithField("context", "snapshotHandler"))

  ...
  // ×¢²á·þÎñµ½ GRPC
  // ÏÈͨ¹ý NewContourServer Éú³ÉÒ»¸ö GRPC ·þÎñ£¬ÔÙ×¢²áµ½ grpcServer
  switch ctx.Config.Server.XDSServerType {
    // ¸ù¾ÝÆô¶¯²ÎÊýµÄ²»Í¬ÅäÖã¬ÊµÀý»¯²»Í¬µÄ¶ÔÏó
    
    // serverType = "envoy"
		case config.EnvoyServerType:
			v3cache := contour_xds_v3.NewSnapshotCache(false, log)
			snapshotHandler.AddSnapshotter(v3cache)
      // µ÷Óà NewServer ¹¹Ôì¶ÔÏó
      // Õâ¸ö·½·¨ÊÇ go-control-plane ÄÚÖõķ½·¨
			contour_xds_v3.RegisterServer(envoy_server_v3.NewServer(taskCtx, v3cache, contour_xds_v3.NewRequestLoggingCallbacks(log)), grpcServer)
    // serverType = "contour"£¬Õâ¸öÊÇĬÈÏÅäÖÃ
		case config.ContourServerType:
      // µ÷Óà NewContourServer ¹¹Ôì¶ÔÏó
			contour_xds_v3.RegisterServer(contour_xds_v3.NewContourServer(log, xdscache.ResourcesOf(resources)...), grpcServer)
  ...
}

// contour/internal/xds/v3/server.go

// ÒÔÏÂ×¢²á·½·¨¾ùµ÷ÓÃµÄ go-control-plane sdk
// °üÀ¨×¢²á ADS¡¢SDS¡¢CDS¡¢EDS¡¢LDS¡¢RDS
func RegisterServer(srv Server, g *grpc.Server) {
	// register services
	envoy_service_discovery_v3.RegisterAggregatedDiscoveryServiceServer(g, srv)
	envoy_service_secret_v3.RegisterSecretDiscoveryServiceServer(g, srv)
	envoy_service_cluster_v3.RegisterClusterDiscoveryServiceServer(g, srv)
	envoy_service_endpoint_v3.RegisterEndpointDiscoveryServiceServer(g, srv)
	envoy_service_listener_v3.RegisterListenerDiscoveryServiceServer(g, srv)
	envoy_service_route_v3.RegisterRouteDiscoveryServiceServer(g, srv)
}

// ÕæÕýʵÀý»¯µÄ Server ¶ÔÏóÊÇ `contourServer`
// ½«Ç°Ãæ×¼±¸ºÃµÄ ResourceCache ÁÐ±í´«½øÀ´
func NewContourServer(log logrus.FieldLogger, resources ...xds.Resource) Server {
	c := contourServer{
		FieldLogger: log,
		resources:   map[string]xds.Resource{},
	}
  // °´ÕÕ TypeURL ×÷Ϊ key£¬±£´æ×ÊÔ´
	for i, r := range resources {
		c.resources[r.TypeURL()] = resources[i]
	}

	return &c
}

// contour/internal/xds/v3/contour.go
type contourServer struct {
   // Since we only implement the streaming state of the world
   // protocol, embed the default null implementations to handle
   // the unimplemented gRPC endpoints.
   envoy_service_discovery_v3.UnimplementedAggregatedDiscoveryServiceServer
   envoy_service_secret_v3.UnimplementedSecretDiscoveryServiceServer
   envoy_service_route_v3.UnimplementedRouteDiscoveryServiceServer
   envoy_service_endpoint_v3.UnimplementedEndpointDiscoveryServiceServer
   envoy_service_cluster_v3.UnimplementedClusterDiscoveryServiceServer
   envoy_service_listener_v3.UnimplementedListenerDiscoveryServiceServer

   logrus.FieldLogger
   // ËùÓÐÉú³ÉµÄ Envoy XDS ÐÅÏ¢¶¼´«½øÀ´±£´æµ½Õâ¸ö×Ö¶Î
   resources   map[string]xds.Resource
   connections xds.Counter
}

func (s *contourServer) StreamClusters(srv envoy_service_cluster_v3.ClusterDiscoveryService_StreamClustersServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamEndpoints(srv envoy_service_endpoint_v3.EndpointDiscoveryService_StreamEndpointsServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamListeners(srv envoy_service_listener_v3.ListenerDiscoveryService_StreamListenersServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamRoutes(srv envoy_service_route_v3.RouteDiscoveryService_StreamRoutesServer) error {
	return s.stream(srv)
}

func (s *contourServer) StreamSecrets(srv envoy_service_secret_v3.SecretDiscoveryService_StreamSecretsServer) error {
	return s.stream(srv)
}

func (s *contourServer) stream(st grpcStream) error {
	...
  // ³ÖÐø¼àÌýÁ¬½Ó£¬²¢´¦ÀíÇëÇó
	// now stick in this loop until the client disconnects.
	for {
		// »ñÈ¡ envoy ·¢ËÍÀ´µÄÇëÇó
		req, err := st.Recv()
		...
    // ¸ù¾ÝÇëÇóµÄ TypeURL£¬ÕÒµ½Æ¥ÅäµÄ ResourceCache
		r, ok := s.resources[req.GetTypeUrl()]
		...
    // ×¢²á¼àÌýÆ÷
		r.Register(ch, last, req.ResourceNames...)
		select {
		case last = <-ch:
			
			var resources []proto.Message
			switch len(req.ResourceNames) {
      // Èç¹ûûÓÐÖ¸¶¨×ÊÔ´Ãû³Æ£¬·µ»ØËùÓÐ×ÊÔ´
			case 0:
				// µ÷ÓÃÇ°Ãæ½éÉܹýµÄ Contents ·½·¨£¬´Ó ResourceCache µÄ value ÖÐÈ¡µ½ envoy ÅäÖÃ
				resources = r.Contents()
      // Èç¹ûÖ¸¶¨×ÊÔ´Ãû³Æ£¬²éÕÒÌض¨×ÊÔ´
			default:
				// resource hints supplied, return exactly those
				resources = r.Query(req.ResourceNames)
			}
      // ½« ResourceCache ת»»³É PB Êý¾Ý
			any := make([]*any.Any, 0, len(resources))
			for _, r := range resources {
				a, err := anypb.New(proto.MessageV2(r))
				if err != nil {
					return done(log, err)
				}
				any = append(any, a)
			}
      // ¹¹Ôì grpc response ¶ÔÏó
			resp := &envoy_service_discovery_v3.DiscoveryResponse{
				VersionInfo: strconv.Itoa(last),
				Resources:   any,
				TypeUrl:     req.GetTypeUrl(),
				Nonce:       strconv.Itoa(last),
			}

      // ½« response ͨ¹ý grpc ·¢Ë͸ø¿Í»§¶Ë envoy
			if err := st.Send(resp); err != nil {
				return done(log, err)
			}

		case <-ctx.Done():
			return done(log, ctx.Err())
		}
	}
}

initContainers:
- args:
  - bootstrap
  - /config/envoy.json
  - --xds-address=contour
  - --xds-port=8001
  - --xds-resource-version=v3
  - --resources-dir=/config/resources
  - --envoy-cafile=/certs/ca.crt
  - --envoy-cert-file=/certs/tls.crt
  - --envoy-key-file=/certs/tls.key

kubectl -n projectcontour exec -it envoy-65qfq -c envoy -- cat /config/envoy.json

{
    "static_resources": {
        "clusters": [
            {
                "name": "contour",
                "alt_stat_name": "projectcontour_contour_8001",
                "type": "STRICT_DNS",
                "connect_timeout": "5s",
                "load_assignment": {
                    "cluster_name": "contour",
                    "endpoints": [
                        {
                            "lb_endpoints": [
                                {
                                    "endpoint": {
                                        "address": {
                                            "socket_address": {
                                                "address": "contour",
                                                "port_value": 8001
                                            }
                                        }
                                    }
                                }
                            ]
                        }
                    ]
                },
                "circuit_breakers": {
                    "thresholds": [
                        {
                            "priority": "HIGH",
                            "max_connections": 100000,
                            "max_pending_requests": 100000,
                            "max_requests": 60000000,
                            "max_retries": 50
                        },
                        {
                            "max_connections": 100000,
                            "max_pending_requests": 100000,
                            "max_requests": 60000000,
                            "max_retries": 50
                        }
                    ]
                },
                "typed_extension_protocol_options": {
                    "envoy.extensions.upstreams.http.v3.HttpProtocolOptions": {
                        "@type": "type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions",
                        "explicit_http_config": {
                            "http2_protocol_options": {}
                        }
                    }
                },
                "transport_socket": {
                    "name": "envoy.transport_sockets.tls",
                    "typed_config": {
                        "@type": "type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.UpstreamTlsContext",
                        "common_tls_context": {
                            "tls_params": {
                                "tls_maximum_protocol_version": "TLSv1_3"
                            },
                            "tls_certificate_sds_secret_configs": [
                                {
                                    "name": "contour_xds_tls_certificate",
                                    "sds_config": {
                                        "path": "/config/resources/sds/xds-tls-certificate.json",
                                        "resource_api_version": "V3"
                                    }
                                }
                            ],
                            "validation_context_sds_secret_config": {
                                "name": "contour_xds_tls_validation_context",
                                "sds_config": {
                                    "path": "/config/resources/sds/xds-validation-context.json",
                                    "resource_api_version": "V3"
                                }
                            }
                        }
                    }
                },
                "upstream_connection_options": {
                    "tcp_keepalive": {
                        "keepalive_probes": 3,
                        "keepalive_time": 30,
                        "keepalive_interval": 5
                    }
                }
            },
            {
                "name": "envoy-admin",
                "alt_stat_name": "projectcontour_envoy-admin_9001",
                "type": "STATIC",
                "connect_timeout": "0.250s",
                "load_assignment": {
                    "cluster_name": "envoy-admin",
                    "endpoints": [
                        {
                            "lb_endpoints": [
                                {
                                    "endpoint": {
                                        "address": {
                                            "pipe": {
                                                "path": "/admin/admin.sock",
                                                "mode": 420
                                            }
                                        }
                                    }
                                }
                            ]
                        }
                    ]
                }
            }
        ]
    },
    "dynamic_resources": {
        "lds_config": {
            "api_config_source": {
                "api_type": "GRPC",
                "transport_api_version": "V3",
                "grpc_services": [
                    {
                        "envoy_grpc": {
                            "cluster_name": "contour"
                        }
                    }
                ]
            },
            "resource_api_version": "V3"
        },
        "cds_config": {
            "api_config_source": {
                "api_type": "GRPC",
                "transport_api_version": "V3",
                "grpc_services": [
                    {
                        "envoy_grpc": {
                            "cluster_name": "contour"
                        }
                    }
                ]
            },
            "resource_api_version": "V3"
        }
    },
    "admin": {
        "access_log": [
            {
                "name": "envoy.access_loggers.file",
                "typed_config": {
                    "@type": "type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog",
                    "path": "/dev/null"
                }
            }
        ],
        "address": {
            "pipe": {
                "path": "/admin/admin.sock",
                "mode": 420
            }
        }
    }
}