Merge pull request openziti/fabric#113 from openziti/events

Convert metrics, traces and session streams to new events framework. …

controller/handler_ctrl/bind.go

@@ -34,6 +34,7 @@ func newBindHandler(router *network.Router, network *network.Network, xctrls []x
 }
 
 func (bindHandler *bindHandler) BindChannel(ch channel2.Channel) error {
+	traceDispatchWrapper := trace.NewDispatchWrapper(bindHandler.network.GetEventDispatcher().Dispatch)
 	ch.SetLogicalName(bindHandler.router.Id)
 	ch.AddReceiveHandler(newSessionRequestHandler(bindHandler.router, bindHandler.network))
 	ch.AddReceiveHandler(newCreateTerminatorHandler(bindHandler.network, bindHandler.router))
@@ -42,9 +43,9 @@ func (bindHandler *bindHandler) BindChannel(ch channel2.Channel) error {
 	ch.AddReceiveHandler(newLinkHandler(bindHandler.router, bindHandler.network))
 	ch.AddReceiveHandler(newFaultHandler(bindHandler.router, bindHandler.network))
 	ch.AddReceiveHandler(newMetricsHandler(bindHandler.network))
-	ch.AddReceiveHandler(bindHandler.network.GetTraceEventController())
+	ch.AddReceiveHandler(newTraceHandler(traceDispatchWrapper))
 	ch.AddReceiveHandler(newInspectHandler(bindHandler.network))
-	ch.AddPeekHandler(trace.NewChannelPeekHandler(bindHandler.network.GetAppId(), ch, bindHandler.network.GetTraceController(), bindHandler.network.GetTraceEventController()))
+	ch.AddPeekHandler(trace.NewChannelPeekHandler(bindHandler.network.GetAppId(), ch, bindHandler.network.GetTraceController(), traceDispatchWrapper))
 
 	xctrlDone := make(chan struct{})
 	for _, x := range bindHandler.xctrls {

controller/handler_ctrl/metrics.go

@@ -26,11 +26,11 @@ import (
 )
 
 type metricsHandler struct {
-	metrics.EventController
+	metrics.Handler
 }
 
 func newMetricsHandler(network *network.Network) *metricsHandler {
-	return &metricsHandler{network.GetMetricsEventController()}
+	return &metricsHandler{metrics.NewDispatchWrapper(network.GetEventDispatcher().Dispatch)}
 }
 
 func (h *metricsHandler) ContentType() int32 {

controller/handler_ctrl/trace.go

@@ -0,0 +1,31 @@
+package handler_ctrl
+
+import (
+	"github.com/golang/protobuf/proto"
+	"github.com/michaelquigley/pfxlog"
+	"github.com/openziti/fabric/pb/ctrl_pb"
+	"github.com/openziti/fabric/trace"
+	"github.com/openziti/foundation/channel2"
+	trace_pb "github.com/openziti/foundation/trace/pb"
+)
+
+type traceHandler struct {
+	dispatcher trace.EventHandler
+}
+
+func newTraceHandler(dispatcher trace.EventHandler) *traceHandler {
+	return &traceHandler{dispatcher: dispatcher}
+}
+
+func (*traceHandler) ContentType() int32 {
+	return int32(ctrl_pb.ContentType_TraceEventType)
+}
+
+func (handler *traceHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
+	event := &trace_pb.ChannelMessage{}
+	if err := proto.Unmarshal(msg.Body, event); err == nil {
+		handler.dispatcher.Accept(event)
+	} else {
+		pfxlog.Logger().Errorf("unexpected error decoding trace message (%s)", err)
+	}
+}

controller/handler_mgmt/bind.go

@@ -67,7 +67,9 @@ func (bindHandler *BindHandler) BindChannel(ch channel2.Channel) error {
 	ch.AddCloseHandler(streamTracesHandler)
 
 	ch.AddReceiveHandler(newTogglePipeTracesHandler(network))
-	ch.AddPeekHandler(trace.NewChannelPeekHandler(network.GetAppId(), ch, network.GetTraceController(), network.GetTraceEventController()))
+
+	traceDispatchWrapper := trace.NewDispatchWrapper(network.GetEventDispatcher().Dispatch)
+	ch.AddPeekHandler(trace.NewChannelPeekHandler(network.GetAppId(), ch, network.GetTraceController(), traceDispatchWrapper))
 
 	ch.AddReceiveHandler(newSnapshotDbHandler(network))
 

controller/handler_mgmt/stream_metrics.go

@@ -24,6 +24,7 @@ import (
 	"github.com/openziti/fabric/controller/network"
 	"github.com/openziti/fabric/pb/mgmt_pb"
 	"github.com/openziti/foundation/channel2"
+	"github.com/openziti/foundation/events"
 	"github.com/openziti/foundation/metrics"
 	"github.com/openziti/foundation/metrics/metrics_pb"
 	"regexp"
@@ -56,18 +57,17 @@ func (handler *streamMetricsHandler) HandleReceive(msg *channel2.Message, ch cha
 	}
 
 	metricsStreamHandler := &MetricsStreamHandler{
-		ch:              ch,
-		filters:         filters,
-		eventController: handler.network.GetMetricsEventController(),
+		ch:      ch,
+		filters: filters,
 	}
 
 	handler.streamHandlers = append(handler.streamHandlers, metricsStreamHandler)
-	handler.network.GetMetricsEventController().AddHandler(metricsStreamHandler)
+	events.AddMetricsEventHandler(metricsStreamHandler)
 }
 
-func (handler *streamMetricsHandler) HandleClose(ch channel2.Channel) {
+func (handler *streamMetricsHandler) HandleClose(channel2.Channel) {
 	for _, streamHandler := range handler.streamHandlers {
-		handler.network.GetMetricsEventController().RemoveHandler(streamHandler)
+		events.RemoveMetricsEventHandler(streamHandler)
 	}
 }
 
@@ -103,9 +103,8 @@ type metricsFilter struct {
 }
 
 type MetricsStreamHandler struct {
-	ch              channel2.Channel
-	filters         []*metricsFilter
-	eventController metrics.EventController
+	ch      channel2.Channel
+	filters []*metricsFilter
 }
 
 func (handler *MetricsStreamHandler) AcceptMetrics(msg *metrics_pb.MetricsMessage) {
@@ -232,5 +231,5 @@ func (handler *MetricsStreamHandler) close() {
 	if err := handler.ch.Close(); err != nil {
 		pfxlog.Logger().WithError(err).Errorf("failure while closing handler")
 	}
-	handler.eventController.RemoveHandler(handler)
+	events.RemoveMetricsEventHandler(handler)
 }

controller/handler_mgmt/stream_sessions.go

@@ -21,6 +21,7 @@ import (
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/fabric/controller/handler_common"
 	"github.com/openziti/fabric/controller/network"
+	"github.com/openziti/fabric/events"
 	"github.com/openziti/fabric/pb/mgmt_pb"
 	"github.com/openziti/foundation/channel2"
 	"github.com/openziti/foundation/identity/identity"
@@ -28,7 +29,7 @@ import (
 
 type streamSessionsHandler struct {
 	network        *network.Network
-	streamHandlers []network.SessionLifeCycleListener
+	streamHandlers []network.SessionEventHandler
 }
 
 func newStreamSessionsHandler(network *network.Network) *streamSessionsHandler {
@@ -46,24 +47,19 @@ func (handler *streamSessionsHandler) HandleReceive(msg *channel2.Message, ch ch
 		return
 	}
 
-	sessionEventRouter := handler.network.GetSessionLifeCycleController()
-	sessionsStreamHandler := &SessionsStreamHandler{
-		ch:     ch,
-		router: sessionEventRouter,
-	}
+	sessionsStreamHandler := &SessionsStreamHandler{ch: ch}
 	handler.streamHandlers = append(handler.streamHandlers, sessionsStreamHandler)
-	sessionEventRouter.AddListener(sessionsStreamHandler)
+	events.AddSessionEventHandler(sessionsStreamHandler)
 }
 
 func (handler *streamSessionsHandler) HandleClose(ch channel2.Channel) {
 	for _, listener := range handler.streamHandlers {
-		handler.network.GetSessionLifeCycleController().RemoveListener(listener)
+		events.RemoveSessionEventHandler(listener)
 	}
 }
 
 type SessionsStreamHandler struct {
-	ch     channel2.Channel
-	router network.SessionLifeCycleController
+	ch channel2.Channel
 }
 
 func (handler *SessionsStreamHandler) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *network.Circuit) {
@@ -112,5 +108,5 @@ func (handler *SessionsStreamHandler) close() {
 	if err := handler.ch.Close(); err != nil {
 		pfxlog.Logger().WithError(err).Error("unexpected error closing mgmt channel")
 	}
-	handler.router.RemoveListener(handler)
+	events.RemoveSessionEventHandler(handler)
 }

controller/handler_mgmt/stream_traces.go

@@ -21,6 +21,7 @@ import (
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/fabric/controller/handler_common"
 	"github.com/openziti/fabric/controller/network"
+	"github.com/openziti/fabric/events"
 	"github.com/openziti/fabric/pb/mgmt_pb"
 	"github.com/openziti/fabric/trace"
 	"github.com/openziti/foundation/channel2"
@@ -47,18 +48,16 @@ func (handler *streamTracesHandler) HandleReceive(msg *channel2.Message, ch chan
 		return
 	}
 
-	eventRouter := handler.network.GetTraceEventController()
-
 	filter := createFilter(request)
-	eventHandler := &traceEventsHandler{ch, eventRouter, filter}
+	eventHandler := &traceEventsHandler{ch, filter}
 
 	handler.streamHandlers = append(handler.streamHandlers, eventHandler)
-	eventRouter.AddHandler(eventHandler)
+	events.AddTraceEventHandler(eventHandler)
 }
 
-func (handler *streamTracesHandler) HandleClose(ch channel2.Channel) {
+func (handler *streamTracesHandler) HandleClose(channel2.Channel) {
 	for _, streamHandler := range handler.streamHandlers {
-		handler.network.GetTraceEventController().RemoveHandler(streamHandler)
+		events.RemoveTraceEventHandler(streamHandler)
 	}
 }
 
@@ -74,7 +73,6 @@ func createFilter(request *mgmt_pb.StreamTracesRequest) trace.Filter {
 
 type traceEventsHandler struct {
 	ch     channel2.Channel
-	router trace.EventController
 	filter trace.Filter
 }
 
@@ -99,5 +97,5 @@ func (handler *traceEventsHandler) close() {
 	if err := handler.ch.Close(); err != nil {
 		pfxlog.Logger().WithError(err).Error("unexpected error while closing mgmt channel")
 	}
-	handler.router.RemoveHandler(handler)
+	events.RemoveTraceEventHandler(handler)
 }

controller/network/network.go

@@ -27,6 +27,8 @@ import (
 	"github.com/openziti/fabric/pb/ctrl_pb"
 	"github.com/openziti/fabric/trace"
 	"github.com/openziti/foundation/channel2"
+	"github.com/openziti/foundation/event"
+	"github.com/openziti/foundation/events"
 	"github.com/openziti/foundation/identity/identity"
 	"github.com/openziti/foundation/metrics"
 	"github.com/openziti/foundation/metrics/metrics_pb"
@@ -44,24 +46,22 @@ import (
 
 type Network struct {
 	*Controllers
-	nodeId                     *identity.TokenId
-	options                    *Options
-	routerChanged              chan *Router
-	linkController             *linkController
-	linkChanged                chan *Link
-	sessionController          *sessionController
-	sequence                   *sequence.Sequence
-	metricsEventController     metrics.EventController
-	sessionLifeCycleController SessionLifeCycleController
-	traceEventController       trace.EventController
-	traceController            trace.Controller
-	routerPresenceHandlers     []RouterPresenceHandler
-	capabilities               []string
-	shutdownChan               chan struct{}
-	isShutdown                 concurrenz.AtomicBoolean
-	lock                       sync.Mutex
-	strategyRegistry           xt.Registry
-	lastSnapshot               time.Time
+	nodeId                 *identity.TokenId
+	options                *Options
+	routerChanged          chan *Router
+	linkController         *linkController
+	linkChanged            chan *Link
+	sessionController      *sessionController
+	sequence               *sequence.Sequence
+	eventDispatcher        event.Dispatcher
+	traceController        trace.Controller
+	routerPresenceHandlers []RouterPresenceHandler
+	capabilities           []string
+	shutdownChan           chan struct{}
+	isShutdown             concurrenz.AtomicBoolean
+	lock                   sync.Mutex
+	strategyRegistry       xt.Registry
+	lastSnapshot           time.Time
 }
 
 func NewNetwork(nodeId *identity.TokenId, options *Options, database boltz.Db, metricsCfg *metrics.Config) (*Network, error) {
@@ -72,23 +72,22 @@ func NewNetwork(nodeId *identity.TokenId, options *Options, database boltz.Db, m
 	controllers := NewControllers(database, stores)
 
 	network := &Network{
-		Controllers:                controllers,
-		nodeId:                     nodeId,
-		options:                    options,
-		routerChanged:              make(chan *Router),
-		linkController:             newLinkController(),
-		linkChanged:                make(chan *Link),
-		sessionController:          newSessionController(),
-		sequence:                   sequence.NewSequence(),
-		metricsEventController:     metrics.NewEventController(metricsCfg),
-		sessionLifeCycleController: NewSessionLifeCyleController(),
-		traceEventController:       trace.NewEventController(),
-		traceController:            trace.NewController(),
-		shutdownChan:               make(chan struct{}),
-		strategyRegistry:           xt.GlobalRegistry(),
-		lastSnapshot:               time.Now().Add(-time.Hour),
+		Controllers:       controllers,
+		nodeId:            nodeId,
+		options:           options,
+		routerChanged:     make(chan *Router),
+		linkController:    newLinkController(),
+		linkChanged:       make(chan *Link),
+		sessionController: newSessionController(),
+		sequence:          sequence.NewSequence(),
+		eventDispatcher:   event.NewDispatcher(),
+		traceController:   trace.NewController(),
+		shutdownChan:      make(chan struct{}),
+		strategyRegistry:  xt.GlobalRegistry(),
+		lastSnapshot:      time.Now().Add(-time.Hour),
 	}
-	network.metricsEventController.AddHandler(network)
+	metrics.Init(metricsCfg)
+	events.AddMetricsEventHandler(network)
 	network.AddCapability("ziti.fabric")
 	network.showOptions()
 	return network, nil
@@ -146,16 +145,8 @@ func (network *Network) GetAllSessions() []*session {
 	return network.sessionController.all()
 }
 
-func (network *Network) GetMetricsEventController() metrics.EventController {
-	return network.metricsEventController
-}
-
-func (network *Network) GetSessionLifeCycleController() SessionLifeCycleController {
-	return network.sessionLifeCycleController
-}
-
-func (network *Network) GetTraceEventController() trace.EventController {
-	return network.traceEventController
+func (network *Network) GetEventDispatcher() event.Dispatcher {
+	return network.eventDispatcher
 }
 
 func (network *Network) GetTraceController() trace.Controller {
@@ -332,7 +323,7 @@ func (network *Network) CreateSession(srcR *Router, clientId *identity.TokenId,
 			PeerData:   peerData,
 		}
 		network.sessionController.add(ss)
-		network.sessionLifeCycleController.SessionCreated(ss.Id, ss.ClientId, ss.Service.Id, ss.Circuit)
+		network.SessionCreated(ss.Id, ss.ClientId, ss.Service.Id, ss.Circuit)
 
 		log.Debugf("created session [s/%s] ==> %s", sessionId.Token, ss.Circuit)
 		return ss, nil
@@ -399,6 +390,14 @@ func (network *Network) selectPath(srcR *Router, svc *Service) (xt.Strategy, xt.
 
 	terminator, err := strategy.Select(weightedTerminators)
 
+	if err != nil {
+		return nil, nil, nil, errors2.Errorf("strategy %v errored selecting terminator for service %v: %v", svc.TerminatorStrategy, svc.Id, err)
+	}
+
+	if terminator == nil {
+		return nil, nil, nil, errors2.Errorf("strategy %v did not select terminator for service %v", svc.TerminatorStrategy, svc.Id)
+	}
+
 	if logrus.IsLevelEnabled(logrus.DebugLevel) {
 		buf := strings.Builder{}
 		buf.WriteString("[")
@@ -413,14 +412,6 @@ func (network *Network) selectPath(srcR *Router, svc *Service) (xt.Strategy, xt.
 		pfxlog.Logger().Infof("selected %v for path from %v", terminator.GetId(), buf.String())
 	}
 
-	if err != nil {
-		return nil, nil, nil, errors2.Errorf("strategy %v errored selecting terminator for service %v: %v", svc.TerminatorStrategy, svc.Id, err)
-	}
-
-	if terminator == nil {
-		return nil, nil, nil, errors2.Errorf("strategy %v did not select terminator for service %v", svc.TerminatorStrategy, svc.Id)
-	}
-
 	return strategy, terminator, paths[terminator.GetRouterId()].path, nil
 }
 
@@ -435,7 +426,7 @@ func (network *Network) RemoveSession(sessionId *identity.TokenId, now bool) err
 			}
 		}
 		network.sessionController.remove(ss)
-		network.sessionLifeCycleController.SessionDeleted(sessionId)
+		network.SessionDeleted(sessionId)
 
 		if strategy, err := network.strategyRegistry.GetStrategy(ss.Service.TerminatorStrategy); strategy != nil {
 			strategy.NotifyEvent(xt.NewSessionEnded(ss.Terminator))
@@ -562,6 +553,7 @@ func (network *Network) Run() {
 func (network *Network) Shutdown() {
 	if network.isShutdown.CompareAndSwap(false, true) {
 		close(network.shutdownChan)
+		events.RemoveMetricsEventHandler(network)
 	}
 }
 
@@ -618,7 +610,7 @@ func (network *Network) rerouteSession(s *session) error {
 
 		log.Infof("rerouted session [s/%s]", s.Id.Token)
 
-		network.sessionLifeCycleController.CircuitUpdated(s.Id, s.Circuit)
+		network.CircuitUpdated(s.Id, s.Circuit)
 
 		return nil
 	} else {
@@ -645,7 +637,7 @@ func (network *Network) smartReroute(s *session, cq *Circuit) error {
 
 	log.Debugf("rerouted session [s/%s]", s.Id.Token)
 
-	network.sessionLifeCycleController.CircuitUpdated(s.Id, s.Circuit)
+	network.CircuitUpdated(s.Id, s.Circuit)
 
 	return nil
 }
@@ -655,7 +647,7 @@ func (network *Network) AcceptMetrics(metrics *metrics_pb.MetricsMessage) {
 
 	router, err := network.Routers.Read(metrics.SourceId)
 	if err != nil {
-		log.Warnf("could not find router [r/%s] while processing metrics", metrics.SourceId)
+		log.Debugf("could not find router [r/%s] while processing metrics", metrics.SourceId)
 		return
 	}
 
@@ -667,7 +659,7 @@ func (network *Network) AcceptMetrics(metrics *metrics_pb.MetricsMessage) {
 			} else if link.Dst.Id == router.Id {
 				link.DstLatency = int64(latency.Mean)
 			} else {
-				log.Warnf("link not for router (wtf?)")
+				log.Warnf("link not for router")
 			}
 		}
 	}

controller/network/session_lifecycle.go

@@ -18,116 +18,44 @@ package network
 
 import (
 	"github.com/openziti/foundation/identity/identity"
+	"github.com/openziti/foundation/util/cowslice"
 )
 
-type SessionLifeCycleListener interface {
+var SessionEventHandlerRegistry = cowslice.NewCowSlice(make([]SessionEventHandler, 0))
+
+func getSessionEventHandlers() []SessionEventHandler {
+	return SessionEventHandlerRegistry.Value().([]SessionEventHandler)
+}
+
+type SessionEventHandler interface {
 	SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *Circuit)
 	SessionDeleted(sessionId *identity.TokenId)
 	CircuitUpdated(sessionId *identity.TokenId, circuit *Circuit)
 }
 
-// SessionLifeCycleController allows for sending session changes to multiple Listeners
-type SessionLifeCycleController interface {
-	SessionLifeCycleListener
-
-	// AddListener adds the given listener to the controller
-	AddListener(listener SessionLifeCycleListener)
-
-	// RemoveListener removes the given listener from the controller
-	RemoveListener(listener SessionLifeCycleListener)
-
-	// Shutdown stops the controller
-	Shutdown()
-}
-
-// NewSessionLifeCycleController creates a new SessionLifeCycleController instance
-func NewSessionLifeCyleController() SessionLifeCycleController {
-	controller := &sessionLifeCycleControllerImpl{
-		changeChan: make(chan sessionLifeCycleControllerChange, 1),
-		eventChan:  make(chan sessionLifeCycleEvent, 25),
-		listeners:  make(map[SessionLifeCycleListener]SessionLifeCycleListener),
-	}
-
-	go controller.run()
-	return controller
-}
-
-type sessionLifeCycleControllerImpl struct {
-	changeChan chan sessionLifeCycleControllerChange
-	eventChan  chan sessionLifeCycleEvent
-	listeners  map[SessionLifeCycleListener]SessionLifeCycleListener
-}
-
-func (controller *sessionLifeCycleControllerImpl) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *Circuit) {
+func (network *Network) SessionCreated(sessionId *identity.TokenId, clientId *identity.TokenId, serviceId string, circuit *Circuit) {
 	event := &sessionCreatedEvent{
 		sessionId: sessionId,
 		clientId:  clientId,
 		serviceId: serviceId,
 		circuit:   circuit,
 	}
-	controller.eventChan <- event
+	network.eventDispatcher.Dispatch(event)
 }
 
-func (controller *sessionLifeCycleControllerImpl) SessionDeleted(sessionId *identity.TokenId) {
+func (network *Network) SessionDeleted(sessionId *identity.TokenId) {
 	event := &sessionDeletedEvent{
 		sessionId: sessionId,
 	}
-	controller.eventChan <- event
+	network.eventDispatcher.Dispatch(event)
 }
 
-func (controller *sessionLifeCycleControllerImpl) CircuitUpdated(sessionId *identity.TokenId, circuit *Circuit) {
+func (network *Network) CircuitUpdated(sessionId *identity.TokenId, circuit *Circuit) {
 	event := &sessionCircuitEvent{
 		sessionId: sessionId,
 		circuit:   circuit,
 	}
-	controller.eventChan <- event
-}
-
-func (controller *sessionLifeCycleControllerImpl) AddListener(listener SessionLifeCycleListener) {
-	controller.changeChan <- sessionLifeCycleControllerChange{listener, sessionLifeCycleControllerAddListener}
-}
-
-func (controller *sessionLifeCycleControllerImpl) RemoveListener(listener SessionLifeCycleListener) {
-	controller.changeChan <- sessionLifeCycleControllerChange{listener, sessionLifeCycleControllerRemoveListener}
-}
-
-func (controller *sessionLifeCycleControllerImpl) Shutdown() {
-	controller.changeChan <- sessionLifeCycleControllerChange{nil, sessionLifeCycleControllerShutdown}
-}
-
-func (controller *sessionLifeCycleControllerImpl) run() {
-	running := true
-	for running {
-		select {
-		case change := <-controller.changeChan:
-			if sessionLifeCycleControllerAddListener == change.changeType {
-				controller.listeners[change.listener] = change.listener
-			} else if sessionLifeCycleControllerRemoveListener == change.changeType {
-				delete(controller.listeners, change.listener)
-			} else if sessionLifeCycleControllerShutdown == change.changeType {
-				running = false
-			}
-		case event := <-controller.eventChan:
-			event.handle(controller)
-		}
-	}
-}
-
-type sessionLifecycleControllerChangeType uint8
-
-const (
-	sessionLifeCycleControllerAddListener = iota
-	sessionLifeCycleControllerRemoveListener
-	sessionLifeCycleControllerShutdown
-)
-
-type sessionLifeCycleControllerChange struct {
-	listener   SessionLifeCycleListener
-	changeType sessionLifecycleControllerChangeType
-}
-
-type sessionLifeCycleEvent interface {
-	handle(controller *sessionLifeCycleControllerImpl)
+	network.eventDispatcher.Dispatch(event)
 }
 
 type sessionCreatedEvent struct {
@@ -137,8 +65,9 @@ type sessionCreatedEvent struct {
 	circuit   *Circuit
 }
 
-func (event *sessionCreatedEvent) handle(controller *sessionLifeCycleControllerImpl) {
-	for _, handler := range controller.listeners {
+func (event *sessionCreatedEvent) Handle() {
+	handlers := getSessionEventHandlers()
+	for _, handler := range handlers {
 		go handler.SessionCreated(event.sessionId, event.clientId, event.serviceId, event.circuit)
 	}
 }
@@ -147,8 +76,9 @@ type sessionDeletedEvent struct {
 	sessionId *identity.TokenId
 }
 
-func (event *sessionDeletedEvent) handle(controller *sessionLifeCycleControllerImpl) {
-	for _, handler := range controller.listeners {
+func (event *sessionDeletedEvent) Handle() {
+	handlers := getSessionEventHandlers()
+	for _, handler := range handlers {
 		go handler.SessionDeleted(event.sessionId)
 	}
 }
@@ -158,8 +88,9 @@ type sessionCircuitEvent struct {
 	circuit   *Circuit
 }
 
-func (event *sessionCircuitEvent) handle(controller *sessionLifeCycleControllerImpl) {
-	for _, handler := range controller.listeners {
+func (event *sessionCircuitEvent) Handle() {
+	handlers := getSessionEventHandlers()
+	for _, handler := range handlers {
 		go handler.CircuitUpdated(event.sessionId, event.circuit)
 	}
 }

events/registrations.go

@@ -0,0 +1,23 @@
+package events
+
+import (
+	"github.com/openziti/fabric/controller/network"
+	"github.com/openziti/fabric/trace"
+	"github.com/openziti/foundation/util/cowslice"
+)
+
+func AddSessionEventHandler(handler network.SessionEventHandler) {
+	cowslice.Append(network.SessionEventHandlerRegistry, handler)
+}
+
+func RemoveSessionEventHandler(handler network.SessionEventHandler) {
+	cowslice.Delete(network.SessionEventHandlerRegistry, handler)
+}
+
+func AddTraceEventHandler(handler trace.EventHandler) {
+	cowslice.Append(trace.EventHandlerRegistry, handler)
+}
+
+func RemoveTraceEventHandler(handler trace.EventHandler) {
+	cowslice.Delete(trace.EventHandlerRegistry, handler)
+}

router/handler_ctrl/validate_terminators.go

@@ -73,7 +73,7 @@ func (handler *validateTerminatorsHandler) validateTerminators(req *ctrl_pb.Vali
 		// TODO: We could distringuish between gone and invalid configuration and allow disabling terminators
 		//       rather than deleting them
 		if dialer == nil || !dialer.IsTerminatorValid(terminator.Id, terminator.Address) {
-			log.Infof("removing invalid terminator %v", terminator.Id)
+			log.Infof("removing invalid terminator %v with binding: %v. had dialer? %v", terminator.Id, terminator.Binding, dialer != nil)
 			if err := xgress.RemoveTerminator(handler.ctrl, terminator.Id); err != nil {
 				log.Errorf("failed to remove invalid terminator %v (%v)", terminator.Id, err)
 			}

router/router.go

@@ -34,6 +34,8 @@ import (
 	"github.com/openziti/fabric/router/xlink"
 	"github.com/openziti/fabric/router/xlink_transport"
 	"github.com/openziti/foundation/channel2"
+	"github.com/openziti/foundation/event"
+	"github.com/openziti/foundation/events"
 	"github.com/openziti/foundation/metrics"
 	"github.com/openziti/foundation/profiler"
 	"github.com/openziti/foundation/util/concurrenz"
@@ -59,6 +61,8 @@ type Router struct {
 	metricsRegistry metrics.Registry
 	shutdownC       chan struct{}
 	isShutdown      concurrenz.AtomicBoolean
+	eventDispatcher event.Dispatcher
+	metricsReporter metrics.Handler
 }
 
 func (self *Router) Channel() channel2.Channel {
@@ -66,13 +70,15 @@ func (self *Router) Channel() channel2.Channel {
 }
 
 func Create(config *Config) *Router {
-	metricsRegistry := metrics.NewRegistry(config.Id.Token, make(map[string]string), time.Second*15, nil)
+	eventDispatcher := event.NewDispatcher()
+	metricsRegistry := metrics.NewRegistry(config.Id.Token, make(map[string]string), time.Second*15, metrics.NewDispatchWrapper(eventDispatcher.Dispatch))
 
 	return &Router{
 		config:          config,
 		forwarder:       forwarder.NewForwarder(metricsRegistry, config.Forwarder),
 		metricsRegistry: metricsRegistry,
 		shutdownC:       make(chan struct{}),
+		eventDispatcher: eventDispatcher,
 	}
 }
 
@@ -110,6 +116,11 @@ func (self *Router) Start() error {
 func (self *Router) Shutdown() error {
 	var errors []error
 	if self.isShutdown.CompareAndSwap(false, true) {
+		self.eventDispatcher.Stop()
+		if self.metricsReporter != nil {
+			events.RemoveMetricsEventHandler(self.metricsReporter)
+		}
+
 		if err := self.ctrl.Close(); err != nil {
 			errors = append(errors, err)
 		}
@@ -278,7 +289,8 @@ func (self *Router) startControlPlane() error {
 		}
 	}
 
-	self.metricsRegistry.EventController().AddHandler(metrics.NewChannelReporter(self.ctrl))
+	self.metricsReporter = metrics.NewChannelReporter(self.ctrl)
+	events.AddMetricsEventHandler(self.metricsReporter)
 
 	return nil
 }

trace/channel_peekhandler.go

@@ -64,7 +64,6 @@ func (handler *ChannelPeekHandler) ToggleTracing(sourceType SourceType, matcher
 }
 
 func NewChannelPeekHandler(appId *identity.TokenId, ch channel2.Channel, controller Controller, eventSink EventHandler) *ChannelPeekHandler {
-
 	handler := &ChannelPeekHandler{
 		appId:      appId,
 		ch:         ch,

trace/event_controller.go

@@ -1,125 +0,0 @@
-/*
-	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 trace
-
-import (
-	"github.com/golang/protobuf/proto"
-	"github.com/michaelquigley/pfxlog"
-	"github.com/openziti/fabric/pb/ctrl_pb"
-	"github.com/openziti/foundation/channel2"
-	"github.com/openziti/foundation/trace/pb"
-)
-
-// EventHandler is for types wishing to receive trace messages
-type EventHandler interface {
-	Accept(event *trace_pb.ChannelMessage)
-}
-
-// EventController allows for sending session changes to multiple Listeners
-type EventController interface {
-	EventHandler
-	channel2.ReceiveHandler
-
-	// AddHandler adds the given handler to the router
-	AddHandler(handler EventHandler)
-
-	// RemoveHandler removes the given handler from the router
-	RemoveHandler(handler EventHandler)
-
-	// Shutdown stops the router
-	Shutdown()
-}
-
-// NewEventController creates a new EventController instance
-func NewEventController() EventController {
-	router := &eventControllerImpl{
-		changeChan: make(chan routerEvent, 1),
-		eventChan:  make(chan *trace_pb.ChannelMessage, 25),
-		handlers:   make(map[EventHandler]EventHandler),
-	}
-
-	go router.run()
-	return router
-}
-
-type eventControllerImpl struct {
-	changeChan chan routerEvent
-	eventChan  chan *trace_pb.ChannelMessage
-	handlers   map[EventHandler]EventHandler
-}
-
-func (controller *eventControllerImpl) Accept(event *trace_pb.ChannelMessage) {
-	controller.eventChan <- event
-}
-
-func (controller *eventControllerImpl) AddHandler(handler EventHandler) {
-	controller.changeChan <- routerEvent{handler, addHandlerChange}
-}
-
-func (controller *eventControllerImpl) RemoveHandler(handler EventHandler) {
-	controller.changeChan <- routerEvent{handler, removeHandlerChange}
-}
-
-func (controller *eventControllerImpl) Shutdown() {
-	controller.changeChan <- routerEvent{nil, shutdownEventControllerChange}
-}
-
-func (controller *eventControllerImpl) run() {
-	running := true
-	for running {
-		select {
-		case change := <-controller.changeChan:
-			if addHandlerChange == change.changeType {
-				controller.handlers[change.handler] = change.handler
-			} else if removeHandlerChange == change.changeType {
-				delete(controller.handlers, change.handler)
-			} else if shutdownEventControllerChange == change.changeType {
-				running = false
-			}
-		case event := <-controller.eventChan:
-			for _, handler := range controller.handlers {
-				go handler.Accept(event)
-			}
-		}
-	}
-}
-
-type eventControllerChangeType uint8
-
-const (
-	addHandlerChange = iota
-	removeHandlerChange
-	shutdownEventControllerChange
-)
-
-type routerEvent struct {
-	handler    EventHandler
-	changeType eventControllerChangeType
-}
-
-func (*eventControllerImpl) ContentType() int32 {
-	return int32(ctrl_pb.ContentType_TraceEventType)
-}
-
-func (controller *eventControllerImpl) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
-	event := &trace_pb.ChannelMessage{}
-	if err := proto.Unmarshal(msg.Body, event); err == nil {
-		controller.Accept(event)
-	} else {
-		pfxlog.Logger().Errorf("unexpected error decoding trace message (%s)", err)
-	}
-}

trace/event_handler.go

@@ -0,0 +1,60 @@
+/*
+	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 trace
+
+import (
+	"github.com/openziti/foundation/event"
+	"github.com/openziti/foundation/trace/pb"
+	"github.com/openziti/foundation/util/cowslice"
+)
+
+var EventHandlerRegistry = cowslice.NewCowSlice(make([]EventHandler, 0))
+
+func getTraceEventHandlers() []EventHandler {
+	return EventHandlerRegistry.Value().([]EventHandler)
+}
+
+// EventHandler is for types wishing to receive trace messages
+type EventHandler interface {
+	Accept(event *trace_pb.ChannelMessage)
+}
+
+type eventWrapper struct {
+	wrapped *trace_pb.ChannelMessage
+}
+
+func (event *eventWrapper) Handle() {
+	handlers := getTraceEventHandlers()
+	for _, handler := range handlers {
+		handler.Accept(event.wrapped)
+	}
+}
+
+func NewDispatchWrapper(delegate func(event event.Event)) EventHandler {
+	return &DispatchWrapper{
+		delegate: delegate,
+	}
+}
+
+type DispatchWrapper struct {
+	delegate func(event event.Event)
+}
+
+func (dispatchWrapper *DispatchWrapper) Accept(event *trace_pb.ChannelMessage) {
+	eventWrapper := &eventWrapper{wrapped: event}
+	dispatchWrapper.delegate(eventWrapper)
+}

util/stackdump-tool/stackdump_tool.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"regexp"
+)
+
+/*
+This tool can take stackdumps generated from ziti-fabric inspect stackdump and separate them into one dump per file
+So if the file contains 4 stackdumps, 2 for controller and 2 for a router with id 001, it will seprate them into
+controller.0.dump
+controller.1.dump
+001.0.dump
+001.1.dump
+*/
+
+func main() {
+	inputFileName := os.Args[1]
+	inputFile, err := os.Open(inputFileName)
+	if err != nil {
+		panic(err)
+	}
+	scanner := bufio.NewScanner(inputFile)
+
+	var outputFile *os.File
+
+	resultsRegEx, err := regexp.Compile("Results:.*")
+	if err != nil {
+		panic(err)
+	}
+
+	stackdumpIdRegEx, err := regexp.Compile(`(^.*)\.stackdump$`)
+	if err != nil {
+		panic(err)
+	}
+
+	counts := map[string]int{}
+
+	for scanner.Scan() {
+		if scanner.Err() != nil {
+			panic(scanner.Err())
+		}
+		line := scanner.Text()
+		if resultsRegEx.MatchString(line) {
+			fmt.Printf("Line matches: %v\n", line)
+		} else if stackdumpIdRegEx.MatchString(line) {
+			stackDumpId := stackdumpIdRegEx.FindStringSubmatch(line)[1]
+			count := counts[stackDumpId]
+			counts[stackDumpId] = count + 1
+			if outputFile != nil {
+				if err = outputFile.Close(); err != nil {
+					panic(err)
+				}
+			}
+			outputFileName := fmt.Sprintf("%v.%v.dump", stackDumpId, count)
+			fmt.Printf("New stackdump found: %v dumping to %v\n", stackDumpId, outputFileName)
+			outputFile, err = os.Create(outputFileName)
+			if err != nil {
+				panic(err)
+			}
+		} else if outputFile != nil {
+			_, err := outputFile.WriteString(line + "\n")
+			if err != nil {
+				panic(err)
+			}
+		}
+	}
+}