diff --git a/controller/controller.go b/controller/controller.go index a1fa73fcf..a3d1f99cf 100644 --- a/controller/controller.go +++ b/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 { diff --git a/controller/raft/fsm.go b/controller/raft/fsm.go index 4930a5aec..5502b25d8 100644 --- a/controller/raft/fsm.go +++ b/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{} { diff --git a/controller/raft/member.go b/controller/raft/member.go index 610de157f..b07f8ea71 100644 --- a/controller/raft/member.go +++ b/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 diff --git a/controller/raft/mesh/mesh.go b/controller/raft/mesh/mesh.go index efc2d6f85..2f309c714 100644 --- a/controller/raft/mesh/mesh.go +++ b/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 diff --git a/controller/raft/mesh/mesh_test.go b/controller/raft/mesh/mesh_test.go index 050037ccf..3b10602eb 100644 --- a/controller/raft/mesh/mesh_test.go +++ b/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()) } diff --git a/controller/raft/raft.go b/controller/raft/raft.go index 109dc3059..2a07ca863 100644 --- a/controller/raft/raft.go +++ b/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 diff --git a/event/cluster.go b/event/cluster.go new file mode 100644 index 000000000..ed0908f73 --- /dev/null +++ b/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(), + } +} diff --git a/event/dispatcher.go b/event/dispatcher.go index ff53cbf51..3f61834b2 100644 --- a/event/dispatcher.go +++ b/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 { diff --git a/event/dispatcher_mock.go b/event/dispatcher_mock.go index d92173683..d05c9002d 100644 --- a/event/dispatcher_mock.go +++ b/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) {} diff --git a/events/dispatcher.go b/events/dispatcher.go index 7432445be..3ae0dc7cb 100644 --- a/events/dispatcher.go +++ b/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] diff --git a/events/dispatcher_cluster.go b/events/dispatcher_cluster.go new file mode 100644 index 000000000..49e22be5e --- /dev/null +++ b/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) + } +} diff --git a/events/formatter.go b/events/formatter.go index fad6e6a50..1ef608f37 100644 --- a/events/formatter.go +++ b/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 diff --git a/router/env/ctrl.go b/router/env/ctrl.go index e4224594d..93369b4bd 100644 --- a/router/env/ctrl.go +++ b/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