Merge pull request #1275 from openziti/event-changes

Update for fabric events changes

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()
+
 }

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

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
+}

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

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)
 }

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

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=