wire in the new 'limits.Agent' infrastructure; extend the 'metrics.Agent' to support additional 'metrics.UsageSink' instances (#271)

controller/controller.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"github.com/openziti/zrok/controller/config"
+	"github.com/openziti/zrok/controller/limits"
 	"github.com/openziti/zrok/controller/metrics"
 	"github.com/sirupsen/logrus"
 
@@ -80,6 +81,18 @@ func Run(inCfg *config.Config) error {
 			return errors.Wrap(err, "error starting metrics agent")
 		}
 		defer func() { ma.Stop() }()
+
+		if cfg.Limits != nil && cfg.Limits.Enforcing {
+			la, err := limits.NewAgent(cfg.Limits, cfg.Metrics.Influx, cfg.Ziti, str)
+			if err != nil {
+				return errors.Wrap(err, "error creating limits agent")
+			}
+			ma.AddUsageSink(la)
+			if err := la.Start(); err != nil {
+				return errors.Wrap(err, "error starting limits agent")
+			}
+			defer func() { la.Stop() }()
+		}
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())

controller/limits/agent.go

@@ -0,0 +1,27 @@
+package limits
+
+import (
+	"github.com/openziti/zrok/controller/metrics"
+	"github.com/openziti/zrok/controller/store"
+	"github.com/openziti/zrok/controller/zrokEdgeSdk"
+	"github.com/sirupsen/logrus"
+)
+
+type Agent struct {
+}
+
+func NewAgent(cfg *Config, ifxCfg *metrics.InfluxConfig, zCfg *zrokEdgeSdk.Config, str *store.Store) (*Agent, error) {
+	return &Agent{}, nil
+}
+
+func (a *Agent) Start() error {
+	return nil
+}
+
+func (a *Agent) Stop() {
+}
+
+func (a *Agent) Handle(u *metrics.Usage) error {
+	logrus.Infof("handling: %v", u)
+	return nil
+}

controller/limits/config.go

@@ -8,7 +8,7 @@ type Config struct {
 	Environments int
 	Shares       int
 	Bandwidth    *BandwidthConfig
-	Cycle        time.Duration
+	Enforcing    bool
 }
 
 type BandwidthConfig struct {

controller/metrics/agent.go

@@ -11,7 +11,7 @@ type Agent struct {
 	src     ZitiEventJsonSource
 	srcJoin chan struct{}
 	cache   *cache
-	snk     UsageSink
+	snks    []UsageSink
 }
 
 func NewAgent(cfg *AgentConfig, str *store.Store, ifxCfg *InfluxConfig) (*Agent, error) {
@@ -22,10 +22,14 @@ func NewAgent(cfg *AgentConfig, str *store.Store, ifxCfg *InfluxConfig) (*Agent,
 		return nil, errors.New("invalid event json source")
 	}
 	a.cache = newShareCache(str)
-	a.snk = newInfluxWriter(ifxCfg)
+	a.snks = append(a.snks, newInfluxWriter(ifxCfg))
 	return a, nil
 }
 
+func (a *Agent) AddUsageSink(snk UsageSink) {
+	a.snks = append(a.snks, snk)
+}
+
 func (a *Agent) Start() error {
 	a.events = make(chan ZitiEventJson)
 	srcJoin, err := a.src.Start(a.events)
@@ -44,8 +48,10 @@ func (a *Agent) Start() error {
 					if err := a.cache.addZrokDetail(usage); err != nil {
 						logrus.Error(err)
 					}
-					if err := a.snk.Handle(usage); err != nil {
+					for _, snk := range a.snks {
-						logrus.Error(err)
+						if err := snk.Handle(usage); err != nil {
+							logrus.Error(err)
+						}
 					}
 				} else {
 					logrus.Error(err)