From b910988adfe20a75c00b21291d2649aff2f07279 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 19 Dec 2022 22:33:49 -0500 Subject: [PATCH] Update for fabric events changes --- events/dispatcher.go | 21 ++++++-- events/dispatcher_api_session.go | 8 +++- events/dispatcher_entity_counts.go | 57 ++++++++++++++++++---- events/dispatcher_session.go | 8 +++- events/formatter.go | 77 +++++++----------------------- go.mod | 8 ++-- go.sum | 12 ++--- 7 files changed, 106 insertions(+), 85 deletions(-) diff --git a/events/dispatcher.go b/events/dispatcher.go index c7683a92a..3346f3277 100644 --- a/events/dispatcher.go +++ b/events/dispatcher.go @@ -20,7 +20,9 @@ import ( "github.com/openziti/edge/controller/persistence" "github.com/openziti/fabric/controller/network" "github.com/openziti/fabric/event" + "github.com/openziti/fabric/events" "github.com/openziti/foundation/v2/concurrenz" + "io" ) func NewDispatcher(n *network.Network, dbProvider persistence.DbProvider, stores *persistence.Stores, closeNotify <-chan struct{}) *Dispatcher { @@ -34,9 +36,15 @@ func NewDispatcher(n *network.Network, dbProvider persistence.DbProvider, stores fabricDispatcher := n.GetEventDispatcher() - fabricDispatcher.RegisterEventType(ApiSessionEventNS, result.registerApiSessionEventHandler) - fabricDispatcher.RegisterEventType(EntityCountEventNS, result.registerEntityCountEventHandler) - fabricDispatcher.RegisterEventType(SessionEventNS, result.registerSessionEventHandler) + fabricDispatcher.RegisterEventTypeFunctions(ApiSessionEventNS, result.registerApiSessionEventHandler, result.unregisterApiSessionEventHandler) + fabricDispatcher.RegisterEventTypeFunctions(EntityCountEventNS, result.registerEntityCountEventHandler, result.unregisterEntityCountEventHandler) + fabricDispatcher.RegisterEventTypeFunctions(SessionEventNS, result.registerSessionEventHandler, result.unregisterSessionEventHandler) + + fabricDispatcher.RegisterFormatterFactory("json", event.FormatterFactoryF(func(sink event.FormattedEventSink) io.Closer { + return &EdgeJsonFormatter{ + JsonFormatter: *events.NewJsonFormatter(16, sink), + } + })) fabricDispatcher.RegisterEventHandlerFactory("file", edgeFileEventLoggerFactory{}) fabricDispatcher.RegisterEventHandlerFactory("stdout", edgeStdOutLoggerFactory{}) @@ -47,8 +55,9 @@ func NewDispatcher(n *network.Network, dbProvider persistence.DbProvider, stores } type Dispatcher struct { - apiSessionEventHandlers concurrenz.CopyOnWriteSlice[ApiSessionEventHandler] - sessionEventHandlers concurrenz.CopyOnWriteSlice[SessionEventHandler] + apiSessionEventHandlers concurrenz.CopyOnWriteSlice[ApiSessionEventHandler] + entityCountEventHandlers concurrenz.CopyOnWriteSlice[*entityCountState] + sessionEventHandlers concurrenz.CopyOnWriteSlice[SessionEventHandler] network *network.Network fabricDispatcher event.Dispatcher @@ -60,4 +69,6 @@ type Dispatcher struct { func (self *Dispatcher) InitializeEvents() { self.initApiSessionEvents(self.stores) self.initSessionEvents(self.stores) + self.initEntityEvents() + } diff --git a/events/dispatcher_api_session.go b/events/dispatcher_api_session.go index dd2b18acd..7805a19be 100644 --- a/events/dispatcher_api_session.go +++ b/events/dispatcher_api_session.go @@ -94,7 +94,7 @@ func (self *Dispatcher) apiSessionDeleted(args ...interface{}) { } } -func (self *Dispatcher) registerApiSessionEventHandler(val interface{}, config map[interface{}]interface{}) error { +func (self *Dispatcher) registerApiSessionEventHandler(val interface{}, config map[string]interface{}) error { handler, ok := val.(ApiSessionEventHandler) if !ok { @@ -132,6 +132,12 @@ func (self *Dispatcher) registerApiSessionEventHandler(val interface{}, config m return nil } +func (self *Dispatcher) unregisterApiSessionEventHandler(val interface{}) { + if handler, ok := val.(ApiSessionEventHandler); ok { + self.RemoveApiSessionEventHandler(handler) + } +} + type apiSessionEventAdapter struct { wrapped ApiSessionEventHandler includeList []string diff --git a/events/dispatcher_entity_counts.go b/events/dispatcher_entity_counts.go index def9fdc0c..4aec24408 100644 --- a/events/dispatcher_entity_counts.go +++ b/events/dispatcher_entity_counts.go @@ -22,21 +22,50 @@ import ( "time" ) -func (self *Dispatcher) entityCountEventGenerator(interval time.Duration, handler EntityCountEventHandler) { - ticker := time.NewTicker(interval) +func (self *Dispatcher) AddEntityCountEventHandler(handler EntityCountEventHandler, interval time.Duration) { + self.entityCountEventHandlers.Append(&entityCountState{ + handler: handler, + interval: interval, + nextRun: time.Now(), + }) +} + +func (self *Dispatcher) RemoveEntityCountEventHandler(handler EntityCountEventHandler) { + for _, state := range self.entityCountEventHandlers.Value() { + if state.handler == handler { + self.entityCountEventHandlers.Delete(state) + } + } +} + +func (self *Dispatcher) initEntityEvents() { + go self.generateEntityEvents() +} + +func (self *Dispatcher) generateEntityEvents() { + ticker := time.NewTicker(time.Second) defer ticker.Stop() for { select { - case <-ticker.C: - self.generateEntityCountEvent(handler) + case t := <-ticker.C: + var event *EntityCountEvent + for _, state := range self.entityCountEventHandlers.Value() { + if t.After(state.nextRun) { + if event == nil { + event = self.generateEntityCountEvent() + } + state.handler.AcceptEntityCountEvent(event) + state.nextRun = state.nextRun.Add(state.interval) + } + } case <-self.closeNotify: return } } } -func (self *Dispatcher) generateEntityCountEvent(handler EntityCountEventHandler) { +func (self *Dispatcher) generateEntityCountEvent() *EntityCountEvent { event := &EntityCountEvent{ Namespace: EntityCountEventNS, Timestamp: time.Now(), @@ -49,10 +78,10 @@ func (self *Dispatcher) generateEntityCountEvent(handler EntityCountEventHandler event.Counts = data } - handler.AcceptEntityCountEvent(event) + return event } -func (self *Dispatcher) registerEntityCountEventHandler(val interface{}, config map[interface{}]interface{}) error { +func (self *Dispatcher) registerEntityCountEventHandler(val interface{}, config map[string]interface{}) error { handler, ok := val.(EntityCountEventHandler) if !ok { @@ -73,7 +102,19 @@ func (self *Dispatcher) registerEntityCountEventHandler(val interface{}, config } } - go self.entityCountEventGenerator(interval, handler) + self.AddEntityCountEventHandler(handler, interval) return nil } + +func (self *Dispatcher) unregisterEntityCountEventHandler(val interface{}) { + if handler, ok := val.(EntityCountEventHandler); ok { + self.RemoveEntityCountEventHandler(handler) + } +} + +type entityCountState struct { + handler EntityCountEventHandler + interval time.Duration + nextRun time.Time +} diff --git a/events/dispatcher_session.go b/events/dispatcher_session.go index dd5000d55..74b901643 100644 --- a/events/dispatcher_session.go +++ b/events/dispatcher_session.go @@ -97,7 +97,7 @@ func (self *Dispatcher) sessionDeleted(args ...interface{}) { } } -func (self *Dispatcher) registerSessionEventHandler(val interface{}, config map[interface{}]interface{}) error { +func (self *Dispatcher) registerSessionEventHandler(val interface{}, config map[string]interface{}) error { handler, ok := val.(SessionEventHandler) if !ok { @@ -135,6 +135,12 @@ func (self *Dispatcher) registerSessionEventHandler(val interface{}, config map[ return nil } +func (self *Dispatcher) unregisterSessionEventHandler(val interface{}) { + if handler, ok := val.(SessionEventHandler); ok { + self.RemoveSessionEventHandler(handler) + } +} + type sessionEventAdapter struct { wrapped SessionEventHandler includeList []string diff --git a/events/formatter.go b/events/formatter.go index b84b80a9a..799a2dbd8 100644 --- a/events/formatter.go +++ b/events/formatter.go @@ -17,7 +17,6 @@ package events import ( - "encoding/json" "github.com/openziti/fabric/events" "github.com/pkg/errors" "io" @@ -29,15 +28,7 @@ type edgeFormatterFactory struct{} func (f edgeFormatterFactory) NewLoggingHandler(format string, buffer int, out io.WriteCloser) (interface{}, error) { if strings.EqualFold(format, "json") { result := &EdgeJsonFormatter{ - JsonFormatter: *events.NewJsonFormatter(buffer, out), - } - go result.Run() - return result, nil - } - - if strings.EqualFold(format, "plain") { - result := &EdgePlainTextFormatter{ - PlainTextFormatter: *events.NewPlainTextFormatter(buffer, out), + JsonFormatter: *events.NewJsonFormatter(buffer, events.NewWriterEventSink(out)), } go result.Run() return result, nil @@ -76,64 +67,30 @@ func (formatter *EdgeJsonFormatter) AcceptEntityCountEvent(event *EntityCountEve type JsonSessionEvent SessionEvent -func (event *JsonSessionEvent) WriteTo(output io.Writer) error { - return marshalJson(event, output) +func (event *JsonSessionEvent) GetEventType() string { + return "session" +} + +func (event *JsonSessionEvent) Format() ([]byte, error) { + return events.MarshalJson(event) } type JsonApiSessionEvent ApiSessionEvent -func (event *JsonApiSessionEvent) WriteTo(output io.Writer) error { - return marshalJson(event, output) +func (event *JsonApiSessionEvent) GetEventType() string { + return "apiSession" +} + +func (event *JsonApiSessionEvent) Format() ([]byte, error) { + return events.MarshalJson(event) } type JsonEntityCountEvent EntityCountEvent -func (event *JsonEntityCountEvent) WriteTo(output io.Writer) error { - return marshalJson(event, output) +func (event *JsonEntityCountEvent) GetEventType() string { + return "entityCount" } -func marshalJson(v interface{}, output io.Writer) error { - buf, err := json.Marshal(v) - if err != nil { - return err - } - _, err = output.Write(buf) - return err -} - -type EdgePlainTextFormatter struct { - events.PlainTextFormatter -} - -func (formatter *EdgePlainTextFormatter) AcceptApiSessionEvent(event *ApiSessionEvent) { - formatter.AcceptLoggingEvent((*PlainTextApiSessionEvent)(event)) -} - -func (formatter *EdgePlainTextFormatter) AcceptSessionEvent(event *SessionEvent) { - formatter.AcceptLoggingEvent((*PlainTextSessionEvent)(event)) -} - -func (formatter *EdgePlainTextFormatter) AcceptEntityCountEvent(event *EntityCountEvent) { - formatter.AcceptLoggingEvent((*PlainTextEntityCountEvent)(event)) -} - -type PlainTextApiSessionEvent ApiSessionEvent - -func (event *PlainTextApiSessionEvent) WriteTo(output io.Writer) error { - _, err := output.Write([]byte((*ApiSessionEvent)(event).String() + "\n")) - return err -} - -type PlainTextSessionEvent SessionEvent - -func (event *PlainTextSessionEvent) WriteTo(output io.Writer) error { - _, err := output.Write([]byte((*SessionEvent)(event).String() + "\n")) - return err -} - -type PlainTextEntityCountEvent EntityCountEvent - -func (event *PlainTextEntityCountEvent) WriteTo(output io.Writer) error { - _, err := output.Write([]byte((*EntityCountEvent)(event).String() + "\n")) - return err +func (event *JsonEntityCountEvent) Format() ([]byte, error) { + return events.MarshalJson(event) } diff --git a/go.mod b/go.mod index 3f11b2281..3b2fb259f 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/openziti/edge -go 1.18 +go 1.19 require ( github.com/AppsFlyer/go-sundheit v0.5.0 @@ -33,7 +33,7 @@ require ( github.com/mitchellh/mapstructure v1.5.0 github.com/netfoundry/secretstream v0.1.2 github.com/openziti/channel/v2 v2.0.24 - github.com/openziti/fabric v0.21.40 + github.com/openziti/fabric v0.22.0 github.com/openziti/foundation/v2 v2.0.9 github.com/openziti/identity v1.0.28 github.com/openziti/jwks v1.0.2 @@ -110,7 +110,7 @@ require ( github.com/opentracing/opentracing-go v1.2.0 // indirect github.com/parallaxsecond/parsec-client-go v0.0.0-20221025095442-f0a77d263cf9 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect - github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c // indirect + github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect github.com/shirou/gopsutil/v3 v3.22.11 // indirect github.com/speps/go-hashids v2.0.0+incompatible // indirect github.com/tklauser/go-sysconf v0.3.11 // indirect @@ -122,7 +122,7 @@ require ( go.mozilla.org/pkcs7 v0.0.0-20210826202110-33d05740a352 // indirect go.opentelemetry.io/otel v1.11.1 // indirect go.opentelemetry.io/otel/trace v1.11.1 // indirect - golang.org/x/exp v0.0.0-20221126150942-6ab00d035af9 // indirect + golang.org/x/exp v0.0.0-20221217163422-3c43f8badb15 // indirect golang.org/x/mod v0.6.0 // indirect golang.org/x/term v0.3.0 // indirect golang.org/x/tools v0.2.0 // indirect diff --git a/go.sum b/go.sum index cb4e79a24..165c09e4b 100644 --- a/go.sum +++ b/go.sum @@ -450,8 +450,8 @@ github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+ github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc= github.com/openziti/channel/v2 v2.0.24 h1:mg9/+VMyXKXfbTZ06quwjRcTQAqU/MEMm+KxAboRLEo= github.com/openziti/channel/v2 v2.0.24/go.mod h1:Y75U/q7l2U+Iib+rfSXO0IscYvksfwWjgdGDXcM/WwY= -github.com/openziti/fabric v0.21.40 h1:Qs1ptFJ+NiRzB8UYpLHy0xzACcZUcYTmea5hL1mMRuU= -github.com/openziti/fabric v0.21.40/go.mod h1:i17oyfaF24Kenvh5nEq0qkBIL4AUAoh08D6WQW1dVbE= +github.com/openziti/fabric v0.22.0 h1:LjLDfD6sUYuxWMkjWwkYaCItcc9l6KXGKrvJYQx5AjU= +github.com/openziti/fabric v0.22.0/go.mod h1:i17oyfaF24Kenvh5nEq0qkBIL4AUAoh08D6WQW1dVbE= github.com/openziti/foundation/v2 v2.0.9 h1:4Z53AzfCcI8XK46eeoSDabroVk1GS2An04WDjNfwmio= github.com/openziti/foundation/v2 v2.0.9/go.mod h1:Z0gnQsGJb/H/B+pixLtmCIpw1dBllD4QoT6LeD35WzA= github.com/openziti/identity v1.0.28 h1:0Rjss3aBT4a2iC8IEXxzj2kfld5LTSD3eEUQkx8Uy+E= @@ -488,8 +488,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= -github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c h1:NRoLoZvkBTKvR5gQLgA3e0hqjkY9u1wm+iOL45VN/qI= -github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig= +github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM= github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo= @@ -639,8 +639,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4= golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM= golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU= -golang.org/x/exp v0.0.0-20221126150942-6ab00d035af9 h1:yZNXmy+j/JpX19vZkVktWqAo7Gny4PBWYYK3zskGpx4= -golang.org/x/exp v0.0.0-20221126150942-6ab00d035af9/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc= +golang.org/x/exp v0.0.0-20221217163422-3c43f8badb15 h1:5oN1Pz/eDhCpbMbLstvIPa0b/BEQo6g6nwV3pLjfM6w= +golang.org/x/exp v0.0.0-20221217163422-3c43f8badb15/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc= golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=