add a SyncChangeSource for BGP peer watcher, fire it from status_file_watcher

Author: aaaaaaaalex

confd/pkg/backends/calico/client.go

@@ -192,7 +192,7 @@ func NewCalicoClient(confdConfig *config.Config) (*client, error) {
 	c.localBGPPeerWatcher.Start()
 
 	// Create a conditional that we use to wake up all of the watcher threads when there
-	// may some actionable updates.
+	// may be some actionable updates.
 	c.watcherCond = sync.NewCond(&c.cacheLock)
 
 	// Increment the waitForSync wait group.  This blocks the GetValues call until the
@@ -292,9 +292,11 @@ func NewCalicoClient(confdConfig *config.Config) (*client, error) {
 	return c, nil
 }
 
+// Strings for keying in-sync messages by their upstream source.
 var (
 	SourceSyncer         string = "SourceSyncer"
 	SourceRouteGenerator string = "SourceRouteGenerator"
+	SourceLocalBGPPeerWatcher string = "LocalBGPPeerWatcher"
 )
 
 // client implements the StoreClient interface for confd, and also implements the
@@ -425,7 +427,7 @@ func (c *client) ExcludeServiceAdvertisement() bool {
 	return false
 }
 
-// OnInSync handles multiplexing in-sync messages from multiple data sources
+// OnSyncChange handles multiplexing in-sync messages from multiple data sources
 // into a single representation of readiness.
 func (c *client) OnSyncChange(source string, ready bool) {
 	c.cacheLock.Lock()
@@ -439,13 +441,13 @@ func (c *client) OnSyncChange(source string, ready bool) {
 	log.Infof("Source %v readiness changed, ready=%v", source, ready)
 
 	// Check if we are fully in sync, before applying this change.
-	oldFullSync := c.sourceReady[SourceSyncer] && c.sourceReady[SourceRouteGenerator]
+	oldFullSync := c.sourceReady[SourceSyncer] && c.sourceReady[SourceRouteGenerator] && c.sourceReady[SourceLocalBGPPeerWatcher]
 
 	// Apply the change.
 	c.sourceReady[source] = ready
 
 	// Check if we are fully in sync now.
-	newFullSync := c.sourceReady[SourceSyncer] && c.sourceReady[SourceRouteGenerator]
+	newFullSync := c.sourceReady[SourceSyncer] && c.sourceReady[SourceRouteGenerator] && c.sourceReady[SourceLocalBGPPeerWatcher]
 
 	if newFullSync == oldFullSync {
 		log.Debugf("No change to full sync status (%v)", newFullSync)

confd/pkg/backends/calico/local_bgp_peer_watcher.go

@@ -53,6 +53,7 @@ func NewLocalBGPPeerWatcher(client *client, prefix string, pollIntervalSeconds i
 		OnFileCreation: w.OnFileCreation,
 		OnFileUpdate:   w.OnFileUpdate,
 		OnFileDeletion: w.OnFileDeletion,
+		OnInSync:       w.OnInSync,
 	})
 
 	return w, nil
@@ -99,6 +100,12 @@ func (w *localBGPPeerWatcher) OnFileDeletion(fileName string) {
 	w.client.recheckPeerConfig()
 }
 
+func (w *localBGPPeerWatcher) OnInSync(inSync bool) {
+	log.WithField("newValue", inSync).Debug("Received new inSync msg from upstream")
+
+	w.client.OnSyncChange(SourceLocalBGPPeerWatcher, inSync)
+}
+
 func (w *localBGPPeerWatcher) updateEpStatus(fileName string, epStatus *epstatus.WorkloadEndpointStatus) {
 	w.mutex.Lock()
 	defer w.mutex.Unlock()

libcalico-go/lib/epstatusfile/status_file_watcher.go

@@ -32,6 +32,7 @@ type Callbacks struct {
 	OnFileCreation func(fileName string)
 	OnFileUpdate   func(fileName string)
 	OnFileDeletion func(fileName string)
+	OnInSync       func(inSync bool)
 }
 
 type FileWatcher struct {
@@ -101,11 +102,6 @@ func (w *FileWatcher) newFsnotifyWatcher() error {
 }
 
 func (w *FileWatcher) runFsnotifyWatcher(watcher *fsnotify.Watcher) error {
-	w.fsnotifyActive = true
-
-	// Get current state of the directory and emit initial events.
-	w.scanDirectory()
-
 	// Listen for events and loop until error occurs.
 	for {
 		select {
@@ -138,6 +134,7 @@ func (w *FileWatcher) runFsnotifyWatcher(watcher *fsnotify.Watcher) error {
 					}
 				}
 			}
+			w.callbacks.OnInSync(true)
 
 		case err, ok := <-watcher.Errors:
 			if !ok {
@@ -168,6 +165,9 @@ func (w *FileWatcher) runWatcher() {
 		}
 
 		if w.fsWatcher != nil {
+			// Get current state of the directory and emit initial events.
+			w.fsnotifyActive = true
+			w.scanDirectory()
 			// Run fsnotify watcher loop if possible.
 			err := w.runFsnotifyWatcher(w.fsWatcher)
 			w.fsnotifyActive = false
@@ -239,6 +239,7 @@ func (w *FileWatcher) scanDirectory() {
 
 	// Update lastState for next iteration.
 	w.lastState = currentState
+	w.callbacks.OnInSync(true)
 }
 
 // Stop stops the watcher.