Complete ADS snapshot cache implementation (openservicemesh#5017)

Signed-off-by: Sean Teeling <[email protected]>

Author: steeling

pkg/envoy/ads/cache_callbacks.go

@@ -7,6 +7,7 @@ import (
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 
 	"github.com/openservicemesh/osm/pkg/envoy"
+	"github.com/openservicemesh/osm/pkg/messaging"
 	"github.com/openservicemesh/osm/pkg/metricsstore"
 	"github.com/openservicemesh/osm/pkg/utils"
 )
@@ -44,18 +45,59 @@ func (s *Server) OnStreamOpen(ctx context.Context, streamID int64, typ string) e
 	}
 
 	s.proxyRegistry.RegisterProxy(proxy)
+	go func() {
+		// Register for proxy config updates broadcasted by the message broker
+		proxyUpdatePubSub := s.msgBroker.GetProxyUpdatePubSub()
+		proxyUpdateChan := proxyUpdatePubSub.Sub(messaging.ProxyUpdateTopic, messaging.GetPubSubTopicForProxyUUID(proxy.UUID.String()))
+		defer s.msgBroker.Unsub(proxyUpdatePubSub, proxyUpdateChan)
+
+		certRotations, unsubRotations := s.certManager.SubscribeRotations(proxy.Identity.String())
+		defer unsubRotations()
+
+		for {
+			select {
+			case <-proxyUpdateChan:
+				log.Debug().Str("proxy", proxy.String()).Msg("Broadcast update received")
+				s.update(proxy)
+			case <-certRotations:
+				log.Debug().Str("proxy", proxy.String()).Msg("Certificate has been updated for proxy")
+				s.update(proxy)
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
 	return nil
 }
 
+func (s *Server) update(proxy *envoy.Proxy) {
+	ch := s.workqueues.AddJob(&proxyResponseJob{
+		proxy:     proxy,
+		xdsServer: s,
+		typeURIs:  envoy.XDSResponseOrder,
+		done:      make(chan struct{}),
+	})
+	<-ch
+	close(ch)
+}
+
 // OnStreamClosed is called on stream closed
 func (s *Server) OnStreamClosed(streamID int64) {
 	log.Debug().Msgf("OnStreamClosed id: %d", streamID)
 	s.proxyRegistry.UnregisterProxy(streamID)
+
+	metricsstore.DefaultMetricsStore.ProxyConnectCount.Dec()
 }
 
-// OnStreamRequest is called when a request happens on an open string
-func (s *Server) OnStreamRequest(a int64, req *discovery.DiscoveryRequest) error {
+// OnStreamRequest is called when a request happens on an open connection
+func (s *Server) OnStreamRequest(streamID int64, req *discovery.DiscoveryRequest) error {
 	log.Debug().Msgf("OnStreamRequest node: %s, type: %s, v: %s, nonce: %s, resNames: %s", req.Node.Id, req.TypeUrl, req.VersionInfo, req.ResponseNonce, req.ResourceNames)
+
+	proxy := s.proxyRegistry.GetConnectedProxy(streamID)
+	if proxy != nil {
+		metricsstore.DefaultMetricsStore.ProxyXDSRequestCount.WithLabelValues(proxy.UUID.String(), proxy.Identity.String(), req.TypeUrl).Inc()
+	}
+
 	return nil
 }
 

pkg/envoy/ads/cache_stream.go

@@ -8,46 +8,8 @@ import (
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
 
 	"github.com/openservicemesh/osm/pkg/envoy"
-	"github.com/openservicemesh/osm/pkg/messaging"
 )
 
-// Routine which fulfills listening to proxy broadcasts
-func (s *Server) watchForUpdates(ctx context.Context) {
-	// Register for proxy config updates broadcasted by the message broker
-	proxyUpdatePubSub := s.msgBroker.GetProxyUpdatePubSub()
-	proxyUpdateChan := proxyUpdatePubSub.Sub(messaging.ProxyUpdateTopic)
-	defer s.msgBroker.Unsub(proxyUpdatePubSub, proxyUpdateChan)
-
-	for {
-		select {
-		case <-proxyUpdateChan:
-			s.allPodUpdater()
-			// TODO(2683): listen to specific pod updates and cert rotations
-		case <-ctx.Done():
-			return
-		}
-	}
-}
-
-func (s *Server) allPodUpdater() {
-	for _, proxy := range s.proxyRegistry.ListConnectedProxies() {
-		s.update(proxy)
-	}
-}
-
-func (s *Server) update(proxy *envoy.Proxy) {
-	// Queue update for this proxy/pod
-	job := proxyResponseJob{
-		proxy:     proxy,
-		adsStream: nil, // Since it goes in the cache, stream is not needed
-		request:   nil, // No request is used, as we fill all verticals
-		xdsServer: s,
-		typeURIs:  envoy.XDSResponseOrder,
-		done:      make(chan struct{}),
-	}
-	s.workqueues.AddJob(&job)
-}
-
 // RecordFullSnapshot stores a group of resources as a new Snapshot with a new version in the cache.
 // It also runs a consistency check on the snapshot (will warn if there are missing resources referenced in
 // the snapshot)

pkg/envoy/ads/jobs.go

@@ -22,7 +22,7 @@ type proxyResponseJob struct {
 }
 
 // GetDoneCh returns the channel, which when closed, indicates the job has been finished.
-func (proxyJob *proxyResponseJob) GetDoneCh() <-chan struct{} {
+func (proxyJob *proxyResponseJob) GetDoneCh() chan struct{} {
 	return proxyJob.done
 }
 

pkg/envoy/ads/server.go

@@ -87,12 +87,6 @@ func (s *Server) Start(ctx context.Context, cancel context.CancelFunc, port int,
 		return fmt.Errorf("error starting ADS server: %w", err)
 	}
 
-	if s.cacheEnabled {
-		// Start broadcast listener thread when cache is enabled and we are ready to start handling
-		// proxy broadcast updates
-		go s.watchForUpdates(ctx)
-	}
-
 	s.ready = true
 
 	return nil

pkg/workerpool/workerpool.go

@@ -29,7 +29,7 @@ type Job interface {
 	Run()
 
 	// GetDoneCh returns the channel, which when closed, indicates that the job was finished.
-	GetDoneCh() <-chan struct{}
+	GetDoneCh() chan struct{}
 }
 
 // NewWorkerPool creates a new work group.
@@ -60,7 +60,7 @@ func NewWorkerPool(nWorkers int) *WorkerPool {
 
 // AddJob posts the job on a worker queue
 // Uses Hash underneath to choose worker to post the job to
-func (wp *WorkerPool) AddJob(job Job) <-chan struct{} {
+func (wp *WorkerPool) AddJob(job Job) chan struct{} {
 	wp.jobs <- job
 	return job.GetDoneCh()
 }

pkg/workerpool/workerpool_test.go

@@ -25,7 +25,7 @@ type testJob struct {
 	hash    uint64
 }
 
-func (tj *testJob) GetDoneCh() <-chan struct{} {
+func (tj *testJob) GetDoneCh() chan struct{} {
 	return tj.jobDone
 }