tweaks for non-zrok traffic

controller/limits/agent.go

@@ -209,6 +209,7 @@ mainLoop:
 	for {
 		select {
 		case usage := <-a.queue:
+			if usage.ShareToken != "" {
 				if err := a.enforce(usage); err != nil {
 					logrus.Errorf("error running enforcement: %v", err)
 				}
@@ -218,6 +219,9 @@ mainLoop:
 					}
 					lastCycle = time.Now()
 				}
+			} else {
+				logrus.Warn("not enforcing for usage with no share token: %v", usage.String())
+			}
 
 		case <-time.After(a.cfg.Cycle):
 			if err := a.relax(); err != nil {

controller/metrics/agent.go

@@ -46,12 +46,12 @@ func (a *Agent) Start() error {
 			case event := <-a.events:
 				if usage, err := Ingest(event.Data()); err == nil {
 					if err := a.cache.addZrokDetail(usage); err != nil {
-						logrus.Error(err)
+						logrus.Errorf("unable to add zrok detail for: %v: %v", usage.String(), err)
 					}
 					shouldAck := true
 					for _, snk := range a.snks {
 						if err := snk.Handle(usage); err != nil {
-							logrus.Error(err)
+							logrus.Errorf("error handling usage: %v", err)
 							if shouldAck {
 								shouldAck = false
 							}
@@ -59,11 +59,14 @@ func (a *Agent) Start() error {
 					}
 					if shouldAck {
 						if err := event.Ack(); err != nil {
-							logrus.Error("unable to Ack message", err)
+							logrus.Errorf("unable to ack handled message: %v", err)
 						}
 					}
 				} else {
-					logrus.Error(err)
+					logrus.Errorf("unable to ingest '%v': %v", event.Data(), err)
+					if err := event.Ack(); err != nil {
+						logrus.Errorf("unable to ack unparseable message: %v", err)
+					}
 				}
 			}
 		}

controller/metrics/amqpSource.go

@@ -83,14 +83,6 @@ mainLoop:
 	msgLoop:
 		for {
 			select {
-			case event := <-s.msgs:
-				if event.Body != nil {
-					s.events <- &ZitiEventAMQP{
-						data: ZitiEventJson(event.Body),
-						msg:  event,
-					}
-				}
-
 			case err, ok := <-s.errs:
 				if err != nil || !ok {
 					logrus.Error(err)
@@ -99,6 +91,21 @@ mainLoop:
 
 			case <-s.close:
 				break mainLoop
+
+			case event, ok := <-s.msgs:
+				if !ok {
+					logrus.Warn("selecting on msg !ok")
+					break msgLoop
+				}
+				if event.Body != nil {
+					s.events <- &ZitiEventAMQP{
+						data: ZitiEventJson(event.Body),
+						msg:  event,
+					}
+				} else {
+					logrus.Warn("event body was nil!")
+					break msgLoop
+				}
 			}
 		}
 	}

controller/metrics/influxWriter.go

@@ -22,6 +22,7 @@ func newInfluxWriter(cfg *InfluxConfig) *influxWriter {
 }
 
 func (w *influxWriter) Handle(u *Usage) error {
+	if u.ShareToken != "" {
 		out := fmt.Sprintf("share: %v, circuit: %v", u.ShareToken, u.ZitiCircuitId)
 
 		envId := fmt.Sprintf("%d", u.EnvironmentId)
@@ -56,6 +57,7 @@ func (w *influxWriter) Handle(u *Usage) error {
 		} else {
 			return err
 		}
+	}
 
 	return nil
 }