controlplane: connect enable/disable peer to heartbeat

Signed-off-by: Kfir Toledo <[email protected]>

pkg/controlplane/client.go

@@ -39,12 +39,13 @@ const (
 // client for accessing a remote peer.
 type client struct {
 	// jsonapi clients for connecting to the remote peer (one per each gateway)
-	clients    []*jsonapi.Client
-	lastSeen   time.Time
-	active     bool
-	stopSignal chan struct{}
-	lock       sync.RWMutex
-	logger     *logrus.Entry
+	clients            []*jsonapi.Client
+	lastSeen           time.Time
+	active             bool
+	stopSignal         chan struct{}
+	lock               sync.RWMutex
+	logger             *logrus.Entry
+	peerStatusCallback func(bool) // Callback function for notifying changes in peer
 }
 
 // remoteServerAuthorizationResponse represents an authorization response received from a remote controlplane server.
@@ -114,11 +115,17 @@ func (c *client) IsActive() bool {
 // setActive the peer status (active or not).
 func (c *client) setActive(active bool) {
 	c.lock.Lock()
-	defer c.lock.Unlock()
+	activePrevState := c.active
 	c.active = active
 	if active || c.lastSeen.IsZero() {
 		c.lastSeen = time.Now()
 	}
+	c.lock.Unlock()
+
+	// Update other components like the policy engine with the peer status.
+	if active != activePrevState && c.peerStatusCallback != nil {
+		c.peerStatusCallback(active)
+	}
 }
 
 // GetHeartbeat get a heartbeat from other peers.
@@ -183,6 +190,11 @@ func (c *client) heartbeatMonitor() {
 	}
 }
 
+// SetPeerStatusCallback set the peerStatusCallback.
+func (c *client) SetPeerStatusCallback(callback func(bool)) {
+	c.peerStatusCallback = callback
+}
+
 // newClient returns a new Peer API client.
 func newClient(peer *store.Peer, tlsConfig *tls.Config) *client {
 	clients := make([]*jsonapi.Client, len(peer.Gateways))

pkg/controlplane/instance.go

@@ -87,6 +87,15 @@ func (cp *Instance) CreatePeer(peer *cpstore.Peer) error {
 
 	cp.policyDecider.AddPeer(&api.Peer{Name: peer.Name, Spec: peer.PeerSpec})
 
+	client.SetPeerStatusCallback(func(isActive bool) {
+		if isActive {
+			cp.policyDecider.EnablePeer(peer.Name)
+			return
+		}
+
+		cp.policyDecider.DisablePeer(peer.Name)
+	})
+
 	return nil
 }