Add alarms and anomalies. Fix counters

integrations/influxunifi/events.go

@@ -6,14 +6,17 @@ import (
 	"github.com/unifi-poller/unifi"
 )
 
+const (
+	Tevent = item("Event")
+	TIDS   = item("IDS")
+)
+
 // batchIDS generates intrusion detection datapoints for InfluxDB.
-func (u *InfluxUnifi) batchIDS(r report, i *unifi.IDS) {
+func (u *InfluxUnifi) batchIDS(r report, i *unifi.IDS) { // nolint: godupl
 	if time.Since(i.Datetime) > u.Interval.Duration+time.Second {
 		return // The event is older than our interval, ignore it.
 	}
 
-	r.addIDS()
-
 	fields := map[string]interface{}{
 		"dest_port":            i.DestPort,
 		"src_port":             i.SrcPort,
@@ -41,23 +44,24 @@ func (u *InfluxUnifi) batchIDS(r report, i *unifi.IDS) {
 		"srcip_organization":   i.SourceIPGeo.Organization,
 	}
 
+	r.addCount(TIDS)
 	r.send(&metric{
 		Table:  "unifi_ids",
 		TS:     i.Datetime,
 		Fields: cleanFields(fields),
 		Tags: cleanTags(map[string]string{
-			"site_name":          i.SiteName,
+			"site_name":  i.SiteName,
-			"source":             i.SourceName,
+			"source":     i.SourceName,
-			"in_iface":           i.InIface,
+			"in_iface":   i.InIface,
-			"event_type":         i.EventType,
+			"event_type": i.EventType,
-			"subsystem":          i.Subsystem,
+			"subsystem":  i.Subsystem,
-			"archived":           i.Archived.Txt,
+			"archived":   i.Archived.Txt,
-			"usgip":              i.USGIP,
+			"usgip":      i.USGIP,
-			"proto":              i.Proto,
+			"proto":      i.Proto,
-			"key":                i.Key,
+			"key":        i.Key,
-			"catname":            i.Catname,
+			"catname":    i.Catname,
-			"app_proto":          i.AppProto,
+			"app_proto":  i.AppProto,
-			"inner_alert_action": i.InnerAlertAction,
+			"action":     i.InnerAlertAction,
 		}),
 	})
 }
@@ -68,8 +72,6 @@ func (u *InfluxUnifi) batchEvent(r report, i *unifi.Event) { // nolint: funlen
 		return // The event is older than our interval, ignore it.
 	}
 
-	r.addEvent()
-
 	fields := map[string]interface{}{
 		"msg":                  i.Msg,          // contains user[] or guest[] or admin[]
 		"duration":             i.Duration.Val, // probably microseconds?
@@ -103,6 +105,7 @@ func (u *InfluxUnifi) batchEvent(r report, i *unifi.Event) { // nolint: funlen
 		"srcip_organization":   i.SourceIPGeo.Organization,
 	}
 
+	r.addCount(Tevent)
 	r.send(&metric{
 		TS:     i.Datetime,
 		Table:  "unifi_events",

integrations/influxunifi/go.mod

@@ -6,6 +6,6 @@ require (
 	github.com/influxdata/influxdb1-client v0.0.0-20200515024757-02f0bf5dbca3
 	github.com/pkg/errors v0.9.1
 	github.com/unifi-poller/poller v0.0.8-0.20200621214016-5d1ed3324a46
-	github.com/unifi-poller/unifi v0.0.5-0.20200621075746-253ccae7e106
+	github.com/unifi-poller/unifi v0.0.6-0.20200625090439-421046871a37
 	golift.io/cnfg v0.0.5
 )

integrations/influxunifi/go.sum

@@ -112,6 +112,8 @@ github.com/unifi-poller/unifi v0.0.5-0.20200619092006-d24c776a42f5/go.mod h1:L1k
 github.com/unifi-poller/unifi v0.0.5-0.20200620103801-b927287ea1cd/go.mod h1:L1kMRH2buZhB31vZnRC1im7Tk/4uD3ET4biwl2faYy8=
 github.com/unifi-poller/unifi v0.0.5-0.20200621075746-253ccae7e106 h1:eKErSqWD656pLSWgxFwhDhHe/zfAXrm7F39Zn4R+si8=
 github.com/unifi-poller/unifi v0.0.5-0.20200621075746-253ccae7e106/go.mod h1:L1kMRH2buZhB31vZnRC1im7Tk/4uD3ET4biwl2faYy8=
+github.com/unifi-poller/unifi v0.0.6-0.20200625090439-421046871a37 h1:T2y8JWkjZd1vz2ZKu4vmmAk9s6PUwupuTldwhfww5xY=
+github.com/unifi-poller/unifi v0.0.6-0.20200625090439-421046871a37/go.mod h1:L1kMRH2buZhB31vZnRC1im7Tk/4uD3ET4biwl2faYy8=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=

integrations/influxunifi/influxdb.go

@@ -92,7 +92,7 @@ func (u *InfluxUnifi) PollController() {
 			continue
 		}
 
-		u.LogInfluxReport(report)
+		u.Collector.Logf("UniFi Metrics Recorded. %v", report)
 	}
 }
 
@@ -166,7 +166,13 @@ func (u *InfluxUnifi) getPassFromFile(filename string) string {
 // Call this after you've collected all the data you care about.
 // Returns an error if influxdb calls fail, otherwise returns a report.
 func (u *InfluxUnifi) ReportMetrics(m *poller.Metrics, e *poller.Events) (*Report, error) {
-	r := &Report{Metrics: m, Events: e, ch: make(chan *metric), Start: time.Now()}
+	r := &Report{
+		Metrics: m,
+		Events:  e,
+		ch:      make(chan *metric),
+		Start:   time.Now(),
+		Counts:  make(map[item]int),
+	}
 	defer close(r.ch)
 
 	var err error
@@ -248,16 +254,12 @@ func (u *InfluxUnifi) loopPoints(r report) {
 func (u *InfluxUnifi) switchExport(r report, v interface{}) {
 	switch v := v.(type) {
 	case *unifi.UAP:
-		r.addUAP()
 		u.batchUAP(r, v)
 	case *unifi.USW:
-		r.addUSW()
 		u.batchUSW(r, v)
 	case *unifi.USG:
-		r.addUSG()
 		u.batchUSG(r, v)
 	case *unifi.UDM:
-		r.addUDM()
 		u.batchUDM(r, v)
 	case *unifi.Site:
 		u.batchSite(r, v)
@@ -267,18 +269,11 @@ func (u *InfluxUnifi) switchExport(r report, v interface{}) {
 		u.batchEvent(r, v)
 	case *unifi.IDS:
 		u.batchIDS(r, v)
+	case *unifi.Alarm:
+		u.batchAlarms(r, v)
+	case *unifi.Anomaly:
+		u.batchAnomaly(r, v)
 	default:
 		u.Collector.LogErrorf("invalid export type: %T", v)
 	}
 }
-
-// LogInfluxReport writes a log message after exporting to influxdb.
-func (u *InfluxUnifi) LogInfluxReport(r *Report) {
-	m := r.Metrics
-	u.Collector.Logf("UniFi Metrics Recorded. Site: %d, Client: %d, "+
-		"UAP: %d, USG/UDM: %d, USW: %d, IDS/Events: %d/%d, DPI Site/Client: %d/%d, "+
-		"Point: %d, Field: %d, Err: %d, Dur: %v",
-		len(m.Sites), len(m.Clients), r.UAP, r.UDM+r.USG,
-		r.USW, r.IDS, r.Eve, len(m.SitesDPI), len(m.ClientsDPI), r.Total,
-		r.Fields, len(r.Errors), r.Elapsed.Round(time.Millisecond))
-}

integrations/influxunifi/report.go

@@ -1,6 +1,7 @@
 package influxunifi
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -13,14 +14,7 @@ type Report struct {
 	Metrics *poller.Metrics
 	Events  *poller.Events
 	Errors  []error
-	Total   int
+	Counts  map[item]int
-	Fields  int
-	USG     int // Total count of USG devices.
-	USW     int // Total count of USW devices.
-	UAP     int // Total count of UAP devices.
-	UDM     int // Total count of UDM devices.
-	Eve     int // Total count of Events.
-	IDS     int // Total count of IDS/IPS Events.
 	Start   time.Time
 	Elapsed time.Duration
 	ch      chan *metric
@@ -37,12 +31,7 @@ type report interface {
 	batch(m *metric, pt *influx.Point)
 	metrics() *poller.Metrics
 	events() *poller.Events
-	addUDM()
+	addCount(item, ...int)
-	addUSG()
-	addUAP()
-	addUSW()
-	addEvent()
-	addIDS()
 }
 
 func (r *Report) metrics() *poller.Metrics {
@@ -68,23 +57,16 @@ func (r *Report) send(m *metric) {
 
 /* The following methods are not thread safe. */
 
-func (r *Report) addUSW() {
+type item string
-	r.USW++
+
-}
+func (r *Report) addCount(name item, counts ...int) {
-func (r *Report) addUAP() {
+	if len(counts) == 0 {
-	r.UAP++
+		r.Counts[name]++
-}
+	}
-func (r *Report) addUSG() {
+
-	r.USG++
+	for _, c := range counts {
-}
+		r.Counts[name] += c
-func (r *Report) addUDM() {
+	}
-	r.UDM++
-}
-func (r *Report) addEvent() {
-	r.Eve++
-}
-func (r *Report) addIDS() {
-	r.IDS++
 }
 
 func (r *Report) error(err error) {
@@ -93,8 +75,24 @@ func (r *Report) error(err error) {
 	}
 }
 
+const (
+	Ttotal = item("Point")
+	Tfield = item("Fields")
+)
+
 func (r *Report) batch(m *metric, p *influx.Point) {
-	r.Total++
+	r.addCount(Ttotal)
-	r.Fields += len(m.Fields)
+	r.addCount(Tfield, len(m.Fields))
 	r.bp.AddPoint(p)
 }
+
+func (r *Report) String() string {
+	return fmt.Sprintf("Site: %d, Client: %d, "+
+		"%s: %d, %s/%s: %d, %s: %d, %s/%s/%s/%s: %d/%d/%d/%d, "+
+		"DPI Site/Client: %d/%d, %s: %d, %s: %d, Err: %d, Dur: %v",
+		len(r.Metrics.Sites), len(r.Metrics.Clients),
+		TUAP, r.Counts[TUAP], TUDM, TUSG, r.Counts[TUDM]+r.Counts[TUSG], TUSW, r.Counts[TUSW],
+		TIDS, Tevent, Talarm, Tanomaly, r.Counts[TIDS], r.Counts[Tevent], r.Counts[Talarm], r.Counts[Tanomaly],
+		len(r.Metrics.SitesDPI), len(r.Metrics.ClientsDPI), Ttotal, r.Counts[Ttotal],
+		Tfield, r.Counts[Tfield], len(r.Errors), r.Elapsed.Round(time.Millisecond))
+}

integrations/influxunifi/uap.go

@@ -4,6 +4,8 @@ import (
 	"github.com/unifi-poller/unifi"
 )
 
+const TUAP = item("UAP")
+
 // batchUAP generates Wireless-Access-Point datapoints for InfluxDB.
 // These points can be passed directly to influx.
 func (u *InfluxUnifi) batchUAP(r report, s *unifi.UAP) {
@@ -32,6 +34,7 @@ func (u *InfluxUnifi) batchUAP(r report, s *unifi.UAP) {
 	fields["guest-num_sta"] = int(s.GuestNumSta.Val)
 	fields["num_sta"] = s.NumSta.Val
 
+	r.addCount(TUAP)
 	r.send(&metric{Table: "uap", Tags: tags, Fields: fields})
 	u.processRadTable(r, tags, s.RadioTable, s.RadioTableStats)
 	u.processVAPTable(r, tags, s.VapTable)

integrations/influxunifi/udm.go

@@ -4,6 +4,8 @@ import (
 	"github.com/unifi-poller/unifi"
 )
 
+const TUDM = item("UDM")
+
 // Combine concatenates N maps. This will delete things if not used with caution.
 func Combine(in ...map[string]interface{}) map[string]interface{} {
 	out := make(map[string]interface{})
@@ -82,6 +84,7 @@ func (u *InfluxUnifi) batchUDM(r report, s *unifi.UDM) { // nolint: funlen
 		},
 	)
 
+	r.addCount(TUDM)
 	r.send(&metric{Table: "usg", Tags: tags, Fields: fields})
 	u.batchNetTable(r, tags, s.NetworkTable)
 	u.batchUSGwans(r, tags, s.Wan1, s.Wan2)

integrations/influxunifi/usg.go

@@ -4,6 +4,8 @@ import (
 	"github.com/unifi-poller/unifi"
 )
 
+const TUSG = item("USG")
+
 // batchUSG generates Unifi Gateway datapoints for InfluxDB.
 // These points can be passed directly to influx.
 func (u *InfluxUnifi) batchUSG(r report, s *unifi.USG) {
@@ -43,6 +45,7 @@ func (u *InfluxUnifi) batchUSG(r report, s *unifi.USG) {
 		},
 	)
 
+	r.addCount(TUSG)
 	r.send(&metric{Table: "usg", Tags: tags, Fields: fields})
 	u.batchNetTable(r, tags, s.NetworkTable)
 	u.batchUSGwans(r, tags, s.Wan1, s.Wan2)

integrations/influxunifi/usw.go

@@ -4,6 +4,8 @@ import (
 	"github.com/unifi-poller/unifi"
 )
 
+const TUSW = item("USW")
+
 // batchUSW generates Unifi Switch datapoints for InfluxDB.
 // These points can be passed directly to influx.
 func (u *InfluxUnifi) batchUSW(r report, s *unifi.USW) {
@@ -38,6 +40,7 @@ func (u *InfluxUnifi) batchUSW(r report, s *unifi.USW) {
 			"user-num_sta":        s.UserNumSta.Val,
 		})
 
+	r.addCount(TUSW)
 	r.send(&metric{Table: "usw", Tags: tags, Fields: fields})
 	u.batchPortTable(r, tags, s.PortTable)
 }