agent (#270)

controller/metrics2/agent.go

@@ -0,0 +1,58 @@
+package metrics2
+
+import (
+	"github.com/openziti/zrok/controller/store"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+type Agent struct {
+	events  chan ZitiEventJson
+	src     ZitiEventJsonSource
+	srcJoin chan struct{}
+	cache   *cache
+	snk     UsageSink
+}
+
+func NewAgent(cfg *AgentConfig, str *store.Store, ifxCfg *InfluxConfig) (*Agent, error) {
+	a := &Agent{}
+	if v, ok := cfg.Source.(ZitiEventJsonSource); ok {
+		a.src = v
+	} else {
+		return nil, errors.New("invalid event json source")
+	}
+	a.cache = newShareCache(str)
+	a.snk = newInfluxWriter(ifxCfg)
+	return a, nil
+}
+
+func (a *Agent) Start() error {
+	a.events = make(chan ZitiEventJson)
+	srcJoin, err := a.src.Start(a.events)
+	if err != nil {
+		return err
+	}
+	a.srcJoin = srcJoin
+
+	go func() {
+		for {
+			select {
+			case event := <-a.events:
+				if usage, err := Ingest(event); err == nil {
+					if err := a.snk.Handle(usage); err != nil {
+						logrus.Error(err)
+					}
+				} else {
+					logrus.Error(err)
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (a *Agent) Stop() {
+	a.src.Stop()
+	close(a.events)
+}

controller/metrics2/config.go

@@ -1,5 +1,9 @@
 package metrics2
 
+type AgentConfig struct {
+	Source interface{}
+}
+
 type InfluxConfig struct {
 	Url    string
 	Bucket string

controller/metrics2/influxWriter.go

@@ -15,13 +15,13 @@ type influxWriter struct {
 	writeApi api.WriteAPIBlocking
 }
 
-func openInfluxWriter(cfg *InfluxConfig) *influxWriter {
+func newInfluxWriter(cfg *InfluxConfig) *influxWriter {
 	idb := influxdb2.NewClient(cfg.Url, cfg.Token)
 	writeApi := idb.WriteAPIBlocking(cfg.Org, cfg.Bucket)
 	return &influxWriter{idb, writeApi}
 }
 
-func (w *influxWriter) Write(u *Usage) error {
+func (w *influxWriter) Handle(u *Usage) error {
 	out := fmt.Sprintf("share: %v, circuit: %v", u.ShareToken, u.ZitiCircuitId)
 
 	envId := fmt.Sprintf("%d", u.EnvironmentId)

controller/metrics2/model.go

@@ -35,6 +35,10 @@ func (u Usage) String() string {
 	return out
 }
 
+type UsageSink interface {
+	Handle(u *Usage) error
+}
+
 type ZitiEventJson string
 
 type ZitiEventJsonSource interface {