Longw/retina networkflow logs configmap telemetry (#1398)

* add telemetry for networkFlowLogsFlushedRateMetric networkFlowLogsRateMetric networkFlowLogsSizeMetric

Author: wanlonghenry

source/plugins/go/src/network_flow_logs.go

@@ -1,16 +1,16 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
 	"net"
+	"strconv"
 	"sync"
 	"time"
-	"strconv"
-	"encoding/json"
 
 	"github.com/fluent/fluent-bit-go/output"
-	"github.com/tinylib/msgp/msgp"
 	"github.com/google/uuid"
+	"github.com/tinylib/msgp/msgp"
 )
 
 // Stream name for retina networkflow logs
@@ -25,10 +25,10 @@ var (
 
 var (
 	// networkflow telemetry
-    NetworkFlowTelemetryMutex = &sync.Mutex{}
-    NetworkFlowLogsMDSDClientCreateErrors float64
-    MdsdNetworkFlowClient net.Conn
-    NetworkFlowTagRefreshTracker time.Time
+	NetworkFlowTelemetryMutex             = &sync.Mutex{}
+	NetworkFlowLogsMDSDClientCreateErrors float64
+	MdsdNetworkFlowClient                 net.Conn
+	NetworkFlowTagRefreshTracker          time.Time
 )
 
 // NetworkFlowMsgPackEntry represents the object corresponding to a single messagepack event in the messagepack stream
@@ -40,9 +40,10 @@ type NetworkFlowMsgPackEntry struct {
 // PostNetworkFlowRecords sends data to the mdsd and amacoreagent
 func PostNetworkFlowRecords(tailPluginRecords []map[interface{}]interface{}) int {
 	if IsNetworkFlowLogsEnabled {
-		Log(fmt.Sprintf("Debug: PostNetworkFlowRecords starting"))
 		start := time.Now()
 		var elapsed time.Duration
+		var bts int
+		var er error
 
 		var dataMap map[string]interface{}
 		var networkFlowLogsMsgPackEntries []NetworkFlowMsgPackEntry
@@ -93,7 +94,7 @@ func PostNetworkFlowRecords(tailPluginRecords []map[interface{}]interface{}) int
 				}
 			}
 
-			bts, er := writeNetworkFlowMsgPackEntries(MdsdNetworkFlowClient, MdsdNetworkFlowLogsStreamTagName, networkFlowLogsMsgPackEntries)
+			bts, er = writeNetworkFlowMsgPackEntries(MdsdNetworkFlowClient, MdsdNetworkFlowLogsStreamTagName, networkFlowLogsMsgPackEntries)
 			elapsed = time.Since(start)
 
 			if er != nil {
@@ -114,18 +115,14 @@ func PostNetworkFlowRecords(tailPluginRecords []map[interface{}]interface{}) int
 			}
 		}
 
-		//TODO Telemetry
-		// NetworkFlowTelemetryMutex.Lock()
-		// defer NetworkFlowTelemetryMutex.Unlock()
+		NetworkFlowTelemetryMutex.Lock()
+		defer NetworkFlowTelemetryMutex.Unlock()
 
-		// if numNetworkLogRecords > 0 {
-		// 	FlushedRecordsCount += float64(numNetworkLogRecords)
-		// 	FlushedRecordsTimeTaken += float64(elapsed / time.Millisecond)
-
-		// 	if maxLatency >= AgentLogProcessingMaxLatencyMs {
-		// 		AgentLogProcessingMaxLatencyMs = maxLatency
-		// 	}
-		// }
+		if numNetworkLogRecords > 0 {
+			NetworkFlowLogsFlushedCount += float64(numNetworkLogRecords)
+			NetworkFlowLogsFlushedSize += float64(bts)
+			NetworkFlowLogsFlushedTimeTaken += float64(elapsed / time.Millisecond)
+		}
 	}
 	return output.FLB_OK
 }

source/plugins/go/src/telemetry.go

@@ -25,6 +25,12 @@ var (
 	FlushedMetadataSize float64
 	// FlushedRecordsTimeTaken indicates the cumulative time taken to flush the records for the current period
 	FlushedRecordsTimeTaken float64
+	// NetworkFlowLogsFlushedCount indicates the number of flushed network flow logs in the current period
+	NetworkFlowLogsFlushedCount float64
+	// NetworkFlowLogsFlushedSize indicates the size of the flushed network flow logs in the current period
+	NetworkFlowLogsFlushedSize float64
+	// NetworkFlowLogsFlushedTimeTaken indicates the cumulative time taken to flush the network flow logs for the current period
+	NetworkFlowLogsFlushedTimeTaken float64
 	// This is telemetry for how old/latent logs we are processing in milliseconds (max over a period of time)
 	AgentLogProcessingMaxLatencyMs float64
 	// This is telemetry for which container logs were latent (max over a period of time)
@@ -105,6 +111,9 @@ const (
 	metricNameAvgLogGenerationRate                                    = "ContainerLogsGeneratedPerSec"
 	metricNameLogSize                                                 = "ContainerLogsSize"
 	metricNameMetadataSize                                            = "ContainerLogsMetadataSize"
+	metricNameNetworkFlowFlushRate                                    = "NetworkFlowLogsAvgRecordsFlushedPerSec"
+	metricNameNetworkFlowLogGenerationRate                            = "NetworkFlowLogsGeneratedPerSec"
+	metricNameNetworkFlowLogSize                                      = "NetworkFlowLogsSize"
 	metricNameAgentLogProcessingMaxLatencyMs                          = "ContainerLogsAgentSideLatencyMs"
 	metricNameNumberofTelegrafMetricsSentSuccessfully                 = "TelegrafMetricsSentCount"
 	metricNameNumberofSendErrorsTelegrafMetrics                       = "TelegrafMetricsSendErrorCount"
@@ -151,6 +160,9 @@ func SendContainerLogPluginMetrics(telemetryPushIntervalProperty string) {
 		logRate := FlushedRecordsCount / float64(elapsed/time.Second)
 		logSizeRate := FlushedRecordsSize / float64(elapsed/time.Second)
 		metadataSizeRate := FlushedMetadataSize / float64(elapsed/time.Second)
+		networkFlowLogsFlushedRate := NetworkFlowLogsFlushedCount / NetworkFlowLogsFlushedTimeTaken * 1000
+		networkFlowLogsRate := NetworkFlowLogsFlushedCount / float64(elapsed/time.Second)
+		networkFlowLogsSizeRate := NetworkFlowLogsFlushedSize / float64(elapsed/time.Second)
 		telegrafMetricsSentCount := TelegrafMetricsSentCount
 		telegrafMetricsSendErrorCount := TelegrafMetricsSendErrorCount
 		telegrafMetricsSend429ErrorCount := TelegrafMetricsSend429ErrorCount
@@ -184,6 +196,9 @@ func SendContainerLogPluginMetrics(telemetryPushIntervalProperty string) {
 		FlushedRecordsSize = 0.0
 		FlushedMetadataSize = 0.0
 		FlushedRecordsTimeTaken = 0.0
+		NetworkFlowLogsFlushedCount = 0.0
+		NetworkFlowLogsFlushedSize = 0.0
+		NetworkFlowLogsFlushedTimeTaken = 0.0
 		logLatencyMs := AgentLogProcessingMaxLatencyMs
 		logLatencyMsContainer := AgentLogProcessingMaxLatencyMsContainer
 		AgentLogProcessingMaxLatencyMs = 0
@@ -315,6 +330,15 @@ func SendContainerLogPluginMetrics(telemetryPushIntervalProperty string) {
 				logLatencyMetric := appinsights.NewMetricTelemetry(metricNameAgentLogProcessingMaxLatencyMs, logLatencyMs)
 				logLatencyMetric.Properties["Container"] = logLatencyMsContainer
 				TelemetryClient.Track(logLatencyMetric)
+
+				if IsNetworkFlowLogsEnabled {
+					networkFlowLogsFlushedRateMetric := appinsights.NewMetricTelemetry(metricNameNetworkFlowFlushRate, networkFlowLogsFlushedRate)
+					networkFlowLogsRateMetric := appinsights.NewMetricTelemetry(metricNameNetworkFlowLogGenerationRate, networkFlowLogsRate)
+					networkFlowLogsSizeMetric := appinsights.NewMetricTelemetry(metricNameNetworkFlowLogSize, networkFlowLogsSizeRate)
+					TelemetryClient.Track(networkFlowLogsFlushedRateMetric)
+					TelemetryClient.Track(networkFlowLogsRateMetric)
+					TelemetryClient.Track(networkFlowLogsSizeMetric)
+				}
 			}
 		}
 		telegrafConfig := make(map[string]string)