diff --git a/controller/handler_ctrl/bind.go b/controller/handler_ctrl/bind.go index 932df19a0..4bb381574 100644 --- a/controller/handler_ctrl/bind.go +++ b/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 { diff --git a/controller/handler_ctrl/metrics.go b/controller/handler_ctrl/metrics.go index 71b6f2bee..f229f78ea 100644 --- a/controller/handler_ctrl/metrics.go +++ b/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 { diff --git a/controller/handler_ctrl/trace.go b/controller/handler_ctrl/trace.go new file mode 100644 index 000000000..a3ccf7068 --- /dev/null +++ b/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) + } +} diff --git a/controller/handler_mgmt/bind.go b/controller/handler_mgmt/bind.go index 3d271d087..d7ae2731f 100644 --- a/controller/handler_mgmt/bind.go +++ b/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)) diff --git a/controller/handler_mgmt/stream_metrics.go b/controller/handler_mgmt/stream_metrics.go index 631ddea31..ae665df6e 100644 --- a/controller/handler_mgmt/stream_metrics.go +++ b/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) } diff --git a/controller/handler_mgmt/stream_sessions.go b/controller/handler_mgmt/stream_sessions.go index 1945e5096..648d30dc9 100644 --- a/controller/handler_mgmt/stream_sessions.go +++ b/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) } diff --git a/controller/handler_mgmt/stream_traces.go b/controller/handler_mgmt/stream_traces.go index 3cc2e5aba..b2399a09c 100644 --- a/controller/handler_mgmt/stream_traces.go +++ b/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) } diff --git a/controller/network/network.go b/controller/network/network.go index 623aac230..23a9af15f 100644 --- a/controller/network/network.go +++ b/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") } } } diff --git a/controller/network/session_lifecycle.go b/controller/network/session_lifecycle.go index e54a77f2b..38a12f245 100644 --- a/controller/network/session_lifecycle.go +++ b/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) } } diff --git a/events/registrations.go b/events/registrations.go new file mode 100644 index 000000000..e4a7a5d89 --- /dev/null +++ b/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) +} diff --git a/router/handler_ctrl/validate_terminators.go b/router/handler_ctrl/validate_terminators.go index 64f516d4b..aaec1a362 100644 --- a/router/handler_ctrl/validate_terminators.go +++ b/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) } diff --git a/router/router.go b/router/router.go index 7af570b55..1718da3c4 100644 --- a/router/router.go +++ b/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 } diff --git a/trace/channel_peekhandler.go b/trace/channel_peekhandler.go index 0d76f9ceb..bef8e3785 100644 --- a/trace/channel_peekhandler.go +++ b/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, diff --git a/trace/event_controller.go b/trace/event_controller.go deleted file mode 100644 index a68efffb5..000000000 --- a/trace/event_controller.go +++ /dev/null @@ -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) - } -} diff --git a/trace/event_handler.go b/trace/event_handler.go new file mode 100644 index 000000000..ee0674361 --- /dev/null +++ b/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) +} diff --git a/util/stackdump-tool/stackdump_tool.go b/util/stackdump-tool/stackdump_tool.go new file mode 100644 index 000000000..5f1d0effe --- /dev/null +++ b/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) + } + } + } +}