Merge pull request openziti/fabric#611 from openziti/add-cluster-events

Add cluster events

controller/controller.go

@@ -30,7 +30,6 @@ import (
 	"github.com/openziti/fabric/controller/db"
 	"github.com/pkg/errors"
 
-	hcraft "github.com/hashicorp/raft"
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/channel/v2"
 	"github.com/openziti/channel/v2/protobufs"
@@ -117,6 +116,10 @@ func (c *Controller) GetCloseNotify() <-chan struct{} {
 	return c.shutdownC
 }
 
+func (c *Controller) GetRaftConfig() *raft.Config {
+	return c.config.Raft
+}
+
 func (c *Controller) RenderJsonConfig() (string, error) {
 	jsonMap, err := config.ToJsonCompatibleMap(c.config.src)
 	if err != nil {
@@ -143,7 +146,7 @@ func NewController(cfg *Config, versionProvider versions.VersionProvider) (*Cont
 	}
 
 	if cfg.Raft != nil {
-		c.raftController = raft.NewController(cfg.Id, versionProvider, cfg.Raft, metricRegistry, c, c.routerDispatchCallback)
+		c.raftController = raft.NewController(c, c)
 		if err := c.raftController.Init(); err != nil {
 			log.WithError(err).Panic("error starting raft")
 		}
@@ -360,7 +363,7 @@ func (c *Controller) registerXts() {
 
 func (c *Controller) registerComponents() error {
 	c.ctrlConnectHandler = handler_ctrl.NewConnectHandler(c.config.Id, c.network)
-
+	c.eventDispatcher.AddClusterEventHandler(event.ClusterEventHandlerF(c.routerDispatchCallback))
 	return nil
 }
 
@@ -405,25 +408,24 @@ func (c *Controller) GetEventDispatcher() event.Dispatcher {
 	return c.eventDispatcher
 }
 
-func (c *Controller) routerDispatchCallback(cfg *hcraft.Configuration) error {
-	if c.raftController != nil {
-		data := make([]string, 0)
-		for _, srvr := range cfg.Servers {
-			data = append(data, string(srvr.Address))
+func (c *Controller) routerDispatchCallback(evt *event.ClusterEvent) {
+	if evt.EventType == event.ClusterMembersChanged {
+		var endpoints []string
+		for _, peer := range evt.Peers {
+			endpoints = append(endpoints, peer.Addr)
 		}
 		updMsg := &ctrl_pb.UpdateCtrlAddresses{
-			Addresses: data,
+			Addresses: endpoints,
 			IsLeader:  c.raftController.IsLeader(),
-			Index:     c.raftController.GetRaft().LastIndex(),
+			Index:     evt.Index,
 		}
 
 		for _, r := range c.network.AllConnectedRouters() {
 			if err := protobufs.MarshalTyped(updMsg).Send(r.Control); err != nil {
-				return err
+				pfxlog.Logger().WithError(err).WithField("routerId", r.Id).Error("unable to update controller endpoints on router")
 			}
 		}
 	}
-	return nil
 }
 
 func (c *Controller) TryInitializeRaftFromBoltDb() error {

controller/raft/fsm.go

@@ -23,6 +23,7 @@ import (
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/fabric/controller/command"
 	"github.com/openziti/fabric/controller/db"
+	"github.com/openziti/fabric/event"
 	"github.com/openziti/storage/boltz"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -31,23 +32,22 @@ import (
 	"path"
 )
 
-func NewFsm(dataDir string, decoders command.Decoders, indexTracker IndexTracker, routerDispatchCallback RouterDispatchCallback) *BoltDbFsm {
+func NewFsm(dataDir string, decoders command.Decoders, indexTracker IndexTracker, eventDispatcher event.Dispatcher) *BoltDbFsm {
 	return &BoltDbFsm{
-		decoders:               decoders,
-		dbPath:                 path.Join(dataDir, "ctrl-ha.db"),
-		indexTracker:           indexTracker,
-		routerDispatchCallback: routerDispatchCallback,
+		decoders:        decoders,
+		dbPath:          path.Join(dataDir, "ctrl-ha.db"),
+		indexTracker:    indexTracker,
+		eventDispatcher: eventDispatcher,
 	}
 }
 
 type BoltDbFsm struct {
-	db           boltz.Db
-	dbPath       string
-	decoders     command.Decoders
-	indexTracker IndexTracker
-
-	currentState           *raft.Configuration
-	routerDispatchCallback RouterDispatchCallback
+	db              boltz.Db
+	dbPath          string
+	decoders        command.Decoders
+	indexTracker    IndexTracker
+	eventDispatcher event.Dispatcher
+	currentState    *raft.Configuration
 }
 
 func (self *BoltDbFsm) Init() error {
@@ -88,11 +88,16 @@ func (self *BoltDbFsm) GetCurrentState(raft *raft.Raft) (uint64, *raft.Configura
 }
 
 func (self *BoltDbFsm) StoreConfiguration(index uint64, configuration raft.Configuration) {
-	log := pfxlog.Logger()
 	self.currentState = &configuration
-	if err := self.routerDispatchCallback(&configuration); err != nil {
-		log.Errorf("Unable to dispatch router callback: %v", err)
+	evt := event.NewClusterEvent(event.ClusterMembersChanged)
+	evt.Index = index
+	for _, srv := range configuration.Servers {
+		evt.Peers = append(evt.Peers, &event.ClusterPeer{
+			Id:   string(srv.ID),
+			Addr: string(srv.Address),
+		})
 	}
+	self.eventDispatcher.AcceptClusterEvent(evt)
 }
 
 func (self *BoltDbFsm) Apply(log *raft.Log) interface{} {

controller/raft/member.go

@@ -58,7 +58,7 @@ func (self *Controller) ListMembers() ([]*Member, error) {
 			Leader: srv.Address == leaderAddr,
 			Version: func() string {
 				if srv.Address == leaderAddr {
-					return self.version.Version()
+					return self.env.GetVersionProvider().Version()
 				}
 				if peer, exists := peers[string(srv.Address)]; exists {
 					return peer.Version.Version

controller/raft/mesh/mesh.go

@@ -18,6 +18,7 @@ package mesh
 
 import (
 	"crypto/x509"
+	"github.com/openziti/fabric/event"
 	"github.com/openziti/foundation/v2/concurrenz"
 	"github.com/openziti/foundation/v2/versions"
 	"net"
@@ -54,7 +55,7 @@ type Peer struct {
 }
 
 func (self *Peer) HandleClose(channel.Channel) {
-	self.mesh.RemovePeer(self)
+	self.mesh.PeerDisconnected(self)
 }
 
 func (self *Peer) ContentType() int32 {
@@ -113,6 +114,12 @@ const (
 	ClusterReadOnly  ClusterState = 1
 )
 
+type Env interface {
+	GetId() *identity.TokenId
+	GetVersionProvider() versions.VersionProvider
+	GetEventDispatcher() event.Dispatcher
+}
+
 // Mesh provides the networking layer to raft
 type Mesh interface {
 	raft.StreamLayer
@@ -132,24 +139,25 @@ type Mesh interface {
 	Init(bindHandler channel.BindHandler)
 }
 
-func New(id *identity.TokenId, version versions.VersionProvider, raftAddr raft.ServerAddress) Mesh {
-	versionEncoded, err := version.EncoderDecoder().Encode(version.AsVersionInfo())
+func New(env Env, raftAddr raft.ServerAddress) Mesh {
+	versionEncoded, err := env.GetVersionProvider().EncoderDecoder().Encode(env.GetVersionProvider().AsVersionInfo())
 	if err != nil {
 		panic(err)
 	}
 
 	return &impl{
-		id:       id,
+		id:       env.GetId(),
 		raftAddr: raftAddr,
 		netAddr: &meshAddr{
 			network: "mesh",
 			addr:    string(raftAddr),
 		},
-		Peers:          map[string]*Peer{},
-		closeNotify:    make(chan struct{}),
-		raftAccepts:    make(chan net.Conn),
-		version:        version,
-		versionEncoded: versionEncoded,
+		Peers:           map[string]*Peer{},
+		closeNotify:     make(chan struct{}),
+		raftAccepts:     make(chan net.Conn),
+		version:         env.GetVersionProvider(),
+		versionEncoded:  versionEncoded,
+		eventDispatcher: env.GetEventDispatcher(),
 	}
 }
 
@@ -167,6 +175,7 @@ type impl struct {
 	versionEncoded       []byte
 	readonly             atomic.Bool
 	clusterStateHandlers concurrenz.CopyOnWriteSlice[func(state ClusterState)]
+	eventDispatcher      event.Dispatcher
 }
 
 func (self *impl) RegisterClusterStateHandler(f func(state ClusterState)) {
@@ -284,7 +293,7 @@ func (self *impl) GetOrConnectPeer(address string, timeout time.Duration) (*Peer
 		return nil, errors.Wrapf(err, "unable to dial %v", address)
 	}
 
-	self.AddPeer(peer)
+	self.PeerConnected(peer)
 	return peer, nil
 }
 
@@ -350,12 +359,21 @@ func (self *impl) extractPeerId(peerAddr string, certs []*x509.Certificate) (str
 	return "", errors.New("no controller SPIFFE ID found in peer certificates")
 }
 
-func (self *impl) AddPeer(peer *Peer) {
+func (self *impl) PeerConnected(peer *Peer) {
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	self.Peers[peer.Address] = peer
 	self.updateClusterState()
 	logrus.Infof("added peer at %v", peer.Address)
+
+	evt := event.NewClusterEvent(event.ClusterPeerConnected)
+	evt.Peers = append(evt.Peers, &event.ClusterPeer{
+		Id:      string(peer.Id),
+		Addr:    peer.Address,
+		Version: peer.Version.Version,
+	})
+
+	self.eventDispatcher.AcceptClusterEvent(evt)
 }
 
 func (self *impl) GetPeer(addr raft.ServerAddress) *Peer {
@@ -364,11 +382,20 @@ func (self *impl) GetPeer(addr raft.ServerAddress) *Peer {
 	return self.Peers[string(addr)]
 }
 
-func (self *impl) RemovePeer(peer *Peer) {
+func (self *impl) PeerDisconnected(peer *Peer) {
 	self.lock.RLock()
 	defer self.lock.RUnlock()
 	delete(self.Peers, peer.Address)
 	self.updateClusterState()
+
+	evt := event.NewClusterEvent(event.ClusterPeerDisconnected)
+	evt.Peers = append(evt.Peers, &event.ClusterPeer{
+		Id:      string(peer.Id),
+		Addr:    peer.Address,
+		Version: peer.Version.Version,
+	})
+
+	self.eventDispatcher.AcceptClusterEvent(evt)
 }
 
 func (self *impl) updateClusterState() {
@@ -455,7 +482,7 @@ func (self *impl) AcceptUnderlay(underlay channel.Underlay) error {
 		return err
 	}
 
-	self.AddPeer(peer)
+	self.PeerConnected(peer)
 	logrus.Infof("connected peer %v at %v", peer.Id, peer.Address)
 
 	return nil

controller/raft/mesh/mesh_test.go

@@ -1,6 +1,7 @@
 package mesh
 
 import (
+	"github.com/openziti/fabric/event"
 	"github.com/openziti/foundation/v2/versions"
 	"runtime"
 	"testing"
@@ -51,25 +52,27 @@ func Test_checkState_ReadonlySetToFalseWhenPreviouslyTrueAndAllVersionsNowMatch(
 
 func Test_AddPeer_PassesReadonlyWhenVersionsMatch(t *testing.T) {
 	m := &impl{
-		Peers:   map[string]*Peer{},
-		version: NewVersionProviderTest(),
+		Peers:           map[string]*Peer{},
+		version:         NewVersionProviderTest(),
+		eventDispatcher: event.DispatcherMock{},
 	}
 
 	p := &Peer{Version: testVersion("1")}
 
-	m.AddPeer(p)
+	m.PeerConnected(p)
 	assert.Equal(t, false, m.readonly.Load(), "Expected readonly to be false, got ", m.readonly.Load())
 }
 
 func Test_AddPeer_TurnsReadonlyWhenVersionsDoNotMatch(t *testing.T) {
 	m := &impl{
-		Peers:   map[string]*Peer{},
-		version: NewVersionProviderTest(),
+		Peers:           map[string]*Peer{},
+		version:         NewVersionProviderTest(),
+		eventDispatcher: event.DispatcherMock{},
 	}
 
 	p := &Peer{Version: testVersion("dne")}
 
-	m.AddPeer(p)
+	m.PeerConnected(p)
 	assert.Equal(t, true, m.readonly.Load(), "Expected readonly to be true, got ", m.readonly.Load())
 }
 
@@ -79,11 +82,12 @@ func Test_RemovePeer_StaysReadonlyWhenDeletingPeerAndStillHasMismatchedVersions(
 			"1": {Version: testVersion("dne"), Address: "1"},
 			"2": {Version: testVersion("dne"), Address: "2"},
 		},
-		version: NewVersionProviderTest(),
+		version:         NewVersionProviderTest(),
+		eventDispatcher: event.DispatcherMock{},
 	}
 	m.readonly.Store(true)
 
-	m.RemovePeer(m.Peers["1"])
+	m.PeerDisconnected(m.Peers["1"])
 	assert.Equal(t, true, m.readonly.Load(), "Expected readonly to be true, got ", m.readonly.Load())
 }
 
@@ -93,11 +97,12 @@ func Test_RemovePeer_RemovesReadonlyWhenDeletingPeerWithNoOtherMismatches(t *tes
 			"1": {Version: testVersion("dne"), Address: "1"},
 			"2": {Version: testVersion("1"), Address: "2"},
 		},
-		version: NewVersionProviderTest(),
+		version:         NewVersionProviderTest(),
+		eventDispatcher: event.DispatcherMock{},
 	}
 	m.readonly.Store(true)
 
-	m.RemovePeer(m.Peers["1"])
+	m.PeerDisconnected(m.Peers["1"])
 	assert.Equal(t, false, m.readonly.Load(), "Expected readonly to be false, got ", m.readonly.Load())
 }
 
@@ -106,11 +111,12 @@ func Test_RemovePeer_RemovesReadonlyWhenDeletingLastPeer(t *testing.T) {
 		Peers: map[string]*Peer{
 			"1": {Version: testVersion("dne"), Address: "1"},
 		},
-		version: NewVersionProviderTest(),
+		version:         NewVersionProviderTest(),
+		eventDispatcher: event.DispatcherMock{},
 	}
 	m.readonly.Store(true)
 
-	m.RemovePeer(m.Peers["1"])
+	m.PeerDisconnected(m.Peers["1"])
 	assert.Equal(t, false, m.readonly.Load(), "Expected readonly to be false, got ", m.readonly.Load())
 }
 

controller/raft/raft.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"github.com/hashicorp/go-hclog"
 	"github.com/openziti/fabric/controller/peermsg"
+	"github.com/openziti/fabric/event"
 	"github.com/openziti/fabric/pb/cmd_pb"
 	"github.com/openziti/foundation/v2/concurrenz"
 	"github.com/openziti/foundation/v2/versions"
@@ -183,23 +184,28 @@ func newClusterState(isLeader, isReadWrite bool) ClusterState {
 	return ClusterState(val)
 }
 
-func NewController(id *identity.TokenId, version versions.VersionProvider, config *Config, metricsRegistry metrics.Registry, migrationMgr MigrationManager, routerDispatchCallback RouterDispatchCallback) *Controller {
+type Env interface {
+	GetId() *identity.TokenId
+	GetVersionProvider() versions.VersionProvider
+	GetRaftConfig() *Config
+	GetMetricsRegistry() metrics.Registry
+	GetEventDispatcher() event.Dispatcher
+}
+
+func NewController(env Env, migrationMgr MigrationManager) *Controller {
 	result := &Controller{
-		Id:                     id,
-		Config:                 config,
-		metricsRegistry:        metricsRegistry,
-		indexTracker:           NewIndexTracker(),
-		version:                version,
-		migrationMgr:           migrationMgr,
-		routerDispatchCallback: routerDispatchCallback,
-		clusterEvents:          make(chan raft.Observation, 16),
+		env:           env,
+		Config:        env.GetRaftConfig(),
+		indexTracker:  NewIndexTracker(),
+		migrationMgr:  migrationMgr,
+		clusterEvents: make(chan raft.Observation, 16),
 	}
 	return result
 }
 
 // Controller manages RAFT related state and operations
 type Controller struct {
-	Id                         *identity.TokenId
+	env                        Env
 	Config                     *Config
 	Mesh                       mesh.Mesh
 	Raft                       *raft.Raft
@@ -207,12 +213,9 @@ type Controller struct {
 	bootstrapped               atomic.Bool
 	clusterLock                sync.Mutex
 	servers                    []raft.Server
-	metricsRegistry            metrics.Registry
 	closeNotify                <-chan struct{}
 	indexTracker               IndexTracker
-	version                    versions.VersionProvider
 	migrationMgr               MigrationManager
-	routerDispatchCallback     RouterDispatchCallback
 	clusterStateChangeHandlers concurrenz.CopyOnWriteSlice[func(event ClusterEvent, state ClusterState)]
 	isLeader                   atomic.Bool
 	clusterEvents              chan raft.Observation
@@ -280,7 +283,7 @@ func (self *Controller) GetCloseNotify() <-chan struct{} {
 }
 
 func (self *Controller) GetMetricsRegistry() metrics.Registry {
-	return self.metricsRegistry
+	return self.env.GetMetricsRegistry()
 }
 
 // Dispatch dispatches the given command to the current leader. If the current node is the leader, the command
@@ -483,7 +486,7 @@ func (self *Controller) Init() error {
 
 	localAddr := raft.ServerAddress(raftConfig.AdvertiseAddress.String())
 	conf := raft.DefaultConfig()
-	conf.LocalID = raft.ServerID(self.Id.Token)
+	conf.LocalID = raft.ServerID(self.env.GetId().Token)
 	conf.NoSnapshotRestoreOnStart = false
 	raftConfig.Configure(conf)
 
@@ -502,7 +505,7 @@ func (self *Controller) Init() error {
 		return err
 	}
 
-	self.Mesh = mesh.New(self.Id, self.version, localAddr)
+	self.Mesh = mesh.New(self.env, localAddr)
 	self.Mesh.RegisterClusterStateHandler(func(state mesh.ClusterState) {
 		obs := raft.Observation{
 			Raft: self.Raft,
@@ -511,7 +514,7 @@ func (self *Controller) Init() error {
 		self.clusterEvents <- obs
 	})
 
-	self.Fsm = NewFsm(raftConfig.DataDir, command.GetDefaultDecoders(), self.indexTracker, self.routerDispatchCallback)
+	self.Fsm = NewFsm(raftConfig.DataDir, command.GetDefaultDecoders(), self.indexTracker, self.env.GetEventDispatcher())
 
 	if err = self.Fsm.Init(); err != nil {
 		return errors.Wrap(err, "failed to init FSM")
@@ -545,6 +548,7 @@ func (self *Controller) Init() error {
 	}
 
 	self.Raft = r
+	self.addEventsHandlers()
 	self.ObserveLeaderChanges()
 
 	return nil
@@ -603,7 +607,7 @@ func (self *Controller) Bootstrap() error {
 		logrus.Infof("waiting for cluster size: %v", self.Config.MinClusterSize)
 		req := &cmd_pb.AddPeerRequest{
 			Addr:    string(self.Mesh.GetAdvertiseAddr()),
-			Id:      self.Id.Token,
+			Id:      self.env.GetId().Token,
 			IsVoter: true,
 		}
 		if err := self.Join(req); err != nil {
@@ -718,6 +722,23 @@ func (self *Controller) RenderJsonConfig() (string, error) {
 	return string(b), err
 }
 
+func (self *Controller) addEventsHandlers() {
+	self.RegisterClusterEventHandler(func(evt ClusterEvent, state ClusterState) {
+		switch evt {
+		case ClusterEventLeadershipGained:
+			self.env.GetEventDispatcher().AcceptClusterEvent(event.NewClusterEvent(event.ClusterLeadershipGained))
+		case ClusterEventLeadershipLost:
+			self.env.GetEventDispatcher().AcceptClusterEvent(event.NewClusterEvent(event.ClusterLeadershipLost))
+		case ClusterEventReadOnly:
+			self.env.GetEventDispatcher().AcceptClusterEvent(event.NewClusterEvent(event.ClusterStateReadOnly))
+		case ClusterEventReadWrite:
+			self.env.GetEventDispatcher().AcceptClusterEvent(event.NewClusterEvent(event.ClusterStateReadWrite))
+		default:
+			pfxlog.Logger().Errorf("unhandled cluster event type: %v", evt)
+		}
+	})
+}
+
 type MigrationManager interface {
 	TryInitializeRaftFromBoltDb() error
 	InitializeRaftFromBoltDb(srcDb string) error

event/cluster.go

@@ -0,0 +1,76 @@
+/*
+	Copyright NetFoundry Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package event
+
+import (
+	"fmt"
+	"time"
+)
+
+type ClusterEventType string
+
+const (
+	ClusterEventsNs = "fabric.cluster"
+
+	ClusterPeerConnected    ClusterEventType = "peer.connected"
+	ClusterPeerDisconnected ClusterEventType = "peer.disconnected"
+	ClusterMembersChanged   ClusterEventType = "members.changed"
+	ClusterLeadershipGained ClusterEventType = "leadership.gained"
+	ClusterLeadershipLost   ClusterEventType = "leadership.lost"
+	ClusterStateReadOnly    ClusterEventType = "state.ro"
+	ClusterStateReadWrite   ClusterEventType = "state.rw"
+)
+
+type ClusterPeer struct {
+	Id      string `json:"id,omitempty"`
+	Addr    string `json:"addr,omitempty"`
+	Version string `json:"version,omitempty"`
+}
+
+func (self *ClusterPeer) String() string {
+	return fmt.Sprintf("id=%v addr=%v version=%v", self.Id, self.Addr, self.Version)
+}
+
+type ClusterEvent struct {
+	Namespace string           `json:"namespace"`
+	EventType ClusterEventType `json:"event_type"`
+	Timestamp time.Time        `json:"timestamp"`
+	Index     uint64           `json:"index,omitempty"`
+	Peers     []*ClusterPeer   `json:"peers,omitempty"`
+}
+
+func (event *ClusterEvent) String() string {
+	return fmt.Sprintf("%v.%v time=%v peers=%v", event.Namespace, event.EventType, event.Timestamp, event.Peers)
+}
+
+type ClusterEventHandler interface {
+	AcceptClusterEvent(event *ClusterEvent)
+}
+
+type ClusterEventHandlerF func(event *ClusterEvent)
+
+func (f ClusterEventHandlerF) AcceptClusterEvent(event *ClusterEvent) {
+	f(event)
+}
+
+func NewClusterEvent(eventType ClusterEventType) *ClusterEvent {
+	return &ClusterEvent{
+		Namespace: ClusterEventsNs,
+		EventType: eventType,
+		Timestamp: time.Now(),
+	}
+}

event/dispatcher.go

@@ -106,6 +106,9 @@ type Dispatcher interface {
 	AddUsageEventHandler(handler UsageEventHandler)
 	RemoveUsageEventHandler(handler UsageEventHandler)
 
+	AddClusterEventHandler(handler ClusterEventHandler)
+	RemoveClusterEventHandler(handler ClusterEventHandler)
+
 	CircuitEventHandler
 	LinkEventHandler
 	MetricsEventHandler
@@ -114,6 +117,7 @@ type Dispatcher interface {
 	ServiceEventHandler
 	TerminatorEventHandler
 	UsageEventHandler
+	ClusterEventHandler
 }
 
 type Event interface {

event/dispatcher_mock.go

@@ -99,3 +99,9 @@ func (d DispatcherMock) AcceptServiceEvent(*ServiceEvent) {}
 func (d DispatcherMock) AcceptTerminatorEvent(*TerminatorEvent) {}
 
 func (d DispatcherMock) AcceptUsageEvent(*UsageEvent) {}
+
+func (d DispatcherMock) AddClusterEventHandler(ClusterEventHandler) {}
+
+func (d DispatcherMock) RemoveClusterEventHandler(ClusterEventHandler) {}
+
+func (d DispatcherMock) AcceptClusterEvent(*ClusterEvent) {}

events/dispatcher.go

@@ -53,6 +53,7 @@ func NewDispatcher(closeNotify <-chan struct{}) *Dispatcher {
 	result.RegisterEventTypeFunctions(event.ServiceEventsNs, result.registerServiceEventHandler, result.unregisterServiceEventHandler)
 	result.RegisterEventTypeFunctions(event.TerminatorEventsNs, result.registerTerminatorEventHandler, result.unregisterTerminatorEventHandler)
 	result.RegisterEventTypeFunctions(event.UsageEventsNs, result.registerUsageEventHandler, result.unregisterUsageEventHandler)
+	result.RegisterEventTypeFunctions(event.ClusterEventsNs, result.registerClusterEventHandler, result.unregisterClusterEventHandler)
 
 	result.RegisterFormatterFactory("json", event.FormatterFactoryF(func(sink event.FormattedEventSink) io.Closer {
 		return NewJsonFormatter(16, sink)
@@ -78,6 +79,7 @@ type Dispatcher struct {
 	terminatorEventHandlers concurrenz.CopyOnWriteSlice[event.TerminatorEventHandler]
 	usageEventHandlers      concurrenz.CopyOnWriteSlice[event.UsageEventHandler]
 	usageEventV3Handlers    concurrenz.CopyOnWriteSlice[event.UsageEventV3Handler]
+	clusterEventHandlers    concurrenz.CopyOnWriteSlice[event.ClusterEventHandler]
 
 	metricsMappers concurrenz.CopyOnWriteSlice[event.MetricsMapper]
 

events/dispatcher_cluster.go

@@ -0,0 +1,57 @@
+/*
+	Copyright NetFoundry Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package events
+
+import (
+	"github.com/openziti/fabric/event"
+	"github.com/pkg/errors"
+	"reflect"
+)
+
+func (self *Dispatcher) AddClusterEventHandler(handler event.ClusterEventHandler) {
+	self.clusterEventHandlers.Append(handler)
+}
+
+func (self *Dispatcher) RemoveClusterEventHandler(handler event.ClusterEventHandler) {
+	self.clusterEventHandlers.Delete(handler)
+}
+
+func (self *Dispatcher) AcceptClusterEvent(event *event.ClusterEvent) {
+	go func() {
+		for _, handler := range self.clusterEventHandlers.Value() {
+			handler.AcceptClusterEvent(event)
+		}
+	}()
+}
+
+func (self *Dispatcher) registerClusterEventHandler(val interface{}, _ map[string]interface{}) error {
+	handler, ok := val.(event.ClusterEventHandler)
+
+	if !ok {
+		return errors.Errorf("type %v doesn't implement github.com/openziti/fabric/event/ClusterEventHandler interface.", reflect.TypeOf(val))
+	}
+
+	self.clusterEventHandlers.Append(handler)
+
+	return nil
+}
+
+func (self *Dispatcher) unregisterClusterEventHandler(val interface{}) {
+	if handler, ok := val.(event.ClusterEventHandler); ok {
+		self.RemoveClusterEventHandler(handler)
+	}
+}

events/formatter.go

@@ -175,6 +175,16 @@ func (event *JsonUsageEventV3) Format() ([]byte, error) {
 	return MarshalJson(event)
 }
 
+type JsonClusterEvent event.ClusterEvent
+
+func (event *JsonClusterEvent) GetEventType() string {
+	return "cluster"
+}
+
+func (event *JsonClusterEvent) Format() ([]byte, error) {
+	return MarshalJson(event)
+}
+
 func NewJsonFormatter(queueDepth int, sink event.FormattedEventSink) *JsonFormatter {
 	result := &JsonFormatter{
 		BaseFormatter: BaseFormatter{
@@ -223,6 +233,10 @@ func (formatter *JsonFormatter) AcceptUsageEventV3(evt *event.UsageEventV3) {
 	formatter.AcceptLoggingEvent((*JsonUsageEventV3)(evt))
 }
 
+func (formatter *JsonFormatter) AcceptClusterEvent(evt *event.ClusterEvent) {
+	formatter.AcceptLoggingEvent((*JsonClusterEvent)(evt))
+}
+
 var histogramBuckets = map[string]string{"p50": "0.50", "p75": "0.75", "p95": "0.95", "p99": "0.99", "p999": "0.999", "p9999": "0.9999"}
 
 type PrometheusMetricsEvent event.MetricsEvent

router/env/ctrl.go

@@ -131,7 +131,7 @@ func (self *networkControllers) Add(ch channel.Channel) (NetworkController, erro
 		ch: ch,
 	}
 	if existing := self.ctrls.Get(ctrl.Channel().Id()); existing != nil {
-		return nil, fmt.Errorf("Duplicate channel with id %v", ctrl.Channel().Id())
+		return nil, fmt.Errorf("duplicate channel with id %v", ctrl.Channel().Id())
 	}
 	self.ctrls.Put(ctrl.Channel().Id(), ctrl)
 	return ctrl, nil