Add flowlog metrics

These metrics will allow to see how many flow records were published or
dropped and check if there are any issues such as flowlog queue congestion.

Signed-off-by: Milan Lenco <[email protected]>

Author: milan-zededa

api/go/metrics/metrics.pb.go

@@ -207,6 +207,7 @@ message deviceMetric {
   google.protobuf.Timestamp last_processed_config = 19;
 
   repeated CellularMetric cellular = 20;
+  FlowlogMetric flowlog = 21;
 }
 
 message AclMetric {
@@ -560,3 +561,31 @@ message vlanInfo {
   uint32 num_trunk_ports = 1; // Number of ports attached to this network instance that are designated trunk
   map <uint32, uint32> vlan_counts = 2; // vlan id to it's usage count map
 }
+
+// Flowlog stats.
+message FlowlogMetric {
+  // Counting FlowMessage instances.
+  // Note that FlowMessage is used to package and carry a list of flows and DNS requests.
+  FlowlogCounters messages = 1;
+  // Counting FlowMessage.Flows.
+  FlowlogCounters flows = 2;
+  // Counting FlowMessage.DnsReqs.
+  FlowlogCounters dns_requests = 3;
+}
+
+// Counters for published/dropped flowlog messages/flows or DNS requests.
+// Note that every record is eventually either successfully published or dropped.
+// In the process of publishing a flowlog record, one or more failed attempts can be made.
+// This means that the total number of fully processed records (i.e. not queued anymore)
+// equals the sum of "success" and "drops", while "failed_attempts" is an orthogonal metric.
+message FlowlogCounters {
+  // Flow records successfully published to zedcloud.
+  uint64 success = 1;
+  // Flow records dropped because the flowlog queue would not fit them.
+  // This is either because the queue was already full when the record was created,
+  // or the publish attempts would keep failing and the queue was nearing its capacity.
+  uint64 drops = 2;
+  // The number of failed attempts to publish a flow record.
+  uint64 failed_attempts = 3;
+}
+

api/images/devconfig.dot

@@ -616,6 +616,15 @@ func publishMetrics(ctx *zedagentContext, iteration int) {
 		}
 	}
 
+	// Report flowlog metrics.
+	ctx.flowLogMetrics.Lock()
+	ReportDeviceMetric.Flowlog = &metrics.FlowlogMetric{
+		Messages:    protoEncodeFlowlogCounters(ctx.flowLogMetrics.Messages),
+		Flows:       protoEncodeFlowlogCounters(ctx.flowLogMetrics.Flows),
+		DnsRequests: protoEncodeFlowlogCounters(ctx.flowLogMetrics.DNSReqs),
+	}
+	ctx.flowLogMetrics.Unlock()
+
 	ReportMetrics.MetricContent = new(metrics.ZMetricMsg_Dm)
 	if x, ok := ReportMetrics.GetMetricContent().(*metrics.ZMetricMsg_Dm); ok {
 		x.Dm = ReportDeviceMetric
@@ -1561,3 +1570,11 @@ func protoEncodeNetworkInstanceMetricProto(status types.NetworkInstanceMetrics)
 
 	return metric
 }
+
+func protoEncodeFlowlogCounters(counters types.FlowlogCounters) *metrics.FlowlogCounters {
+	return &metrics.FlowlogCounters{
+		Success:        counters.Success,
+		Drops:          counters.Drops,
+		FailedAttempts: counters.FailedAttempts,
+	}
+}

api/images/devconfig.dot.svg

@@ -444,6 +444,11 @@ func handleAppFlowMonitorImpl(ctxArg interface{}, key string,
 	case ctx.FlowlogQueue <- pflows:
 	default:
 		log.Errorf("Flowlog queue is full, dropping flowlog entry: %+v", pflows.Scope)
+		ctx.flowLogMetrics.Lock()
+		ctx.flowLogMetrics.Messages.Drops++
+		ctx.flowLogMetrics.Flows.Drops += uint64(len(pflows.Flows))
+		ctx.flowLogMetrics.DNSReqs.Drops += uint64(len(pflows.DnsReqs))
+		ctx.flowLogMetrics.Unlock()
 	}
 }
 
@@ -545,11 +550,26 @@ func flowlogTask(ctx *zedagentContext, flowlogQueue <-chan *flowlog.FlowMessage)
 		err := publishFlowMessage(msg, iteration)
 		if err == nil {
 			iteration++
+			ctx.flowLogMetrics.Lock()
+			ctx.flowLogMetrics.Messages.Success++
+			ctx.flowLogMetrics.Flows.Success += uint64(len(msg.Flows))
+			ctx.flowLogMetrics.DNSReqs.Success += uint64(len(msg.DnsReqs))
+			ctx.flowLogMetrics.Unlock()
 		} else {
 			log.Error(err)
+			ctx.flowLogMetrics.Lock()
+			ctx.flowLogMetrics.Messages.FailedAttempts++
+			ctx.flowLogMetrics.Flows.FailedAttempts += uint64(len(msg.Flows))
+			ctx.flowLogMetrics.DNSReqs.FailedAttempts += uint64(len(msg.DnsReqs))
+			ctx.flowLogMetrics.Unlock()
 			if (100*len(flowlogQueue))/cap(flowlogQueue) > 90 {
 				// More than 90% of the queue is used, start dropping instead of retrying.
 				log.Warnf("flowlogTask: dropped flow message: %+v", msg.Scope)
+				ctx.flowLogMetrics.Lock()
+				ctx.flowLogMetrics.Messages.Drops++
+				ctx.flowLogMetrics.Flows.Drops += uint64(len(msg.Flows))
+				ctx.flowLogMetrics.DNSReqs.Drops += uint64(len(msg.DnsReqs))
+				ctx.flowLogMetrics.Unlock()
 			} else {
 				retry = true
 			}

api/images/devconfig.png

@@ -148,6 +148,7 @@ type zedagentContext struct {
 	globalConfig            types.ConfigItemValueMap
 	specMap                 types.ConfigItemSpecMap
 	globalStatus            types.GlobalStatus
+	flowLogMetrics          types.FlowlogMetrics
 	appContainerStatsTime   time.Time // last time the App Container stats uploaded
 	// The MaintenanceMode can come from GlobalConfig and from the config
 	// API. Those are merged into maintenanceMode

api/proto/metrics/metrics.proto

@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2021. Zededa, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package types
+
+import "sync"
+
+// FlowlogCounters encapsulates counters for published/dropped flowlog
+// messages/flows or DNS requests.
+type FlowlogCounters struct {
+	Success        uint64
+	Drops          uint64
+	FailedAttempts uint64
+}
+
+// FlowlogMetrics contains flowlog metrics as collected by flowlogTask of zedagent.
+type FlowlogMetrics struct {
+	sync.Mutex
+	Messages FlowlogCounters
+	Flows    FlowlogCounters
+	DNSReqs  FlowlogCounters
+}