diff --git a/pkg/controlplane/client.go b/pkg/controlplane/client.go index ab32cc15..b5f8b9d3 100644 --- a/pkg/controlplane/client.go +++ b/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)) diff --git a/pkg/controlplane/instance.go b/pkg/controlplane/instance.go index 140830c2..4424af23 100644 --- a/pkg/controlplane/instance.go +++ b/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 }