Adding change for networkflow logs new stream (#1551)

* adding change for new stream

* update code logic to handle renaming as legacy fields do not have data

* update stream

Author: wanlonghenry

scripts/onboarding/aks/onboarding-msi-bicep-syslog/existingClusterParam.json

@@ -90,7 +90,7 @@
         "Microsoft-ContainerInventory",
         "Microsoft-ContainerNodeInventory",
         "Microsoft-Perf",
-        "Microsoft-RetinaNetworkFlowLogs"
+        "Microsoft-ContainerNetworkLogs"
       ]
     },
     "useAzureMonitorPrivateLinkScope": {

scripts/onboarding/aks/onboarding-msi-bicep/existingClusterParam.json

@@ -54,7 +54,7 @@
         "Microsoft-ContainerInventory",
         "Microsoft-ContainerNodeInventory",
         "Microsoft-Perf",
-        "Microsoft-RetinaNetworkFlowLogs"
+        "Microsoft-ContainerNetworkLogs"
       ]
     },
     "useAzureMonitorPrivateLinkScope": {

scripts/onboarding/aks/onboarding-using-msi-auth/existingClusterParam.json

@@ -93,7 +93,7 @@
         "Microsoft-ContainerInventory",
         "Microsoft-ContainerNodeInventory",
         "Microsoft-Perf",
-        "Microsoft-RetinaNetworkFlowLogs"
+        "Microsoft-ContainerNetworkLogs"
       ]
     },
     "useAzureMonitorPrivateLinkScope": {

source/plugins/go/src/network_flow_logs.go

@@ -14,8 +14,11 @@ import (
 	"github.com/tinylib/msgp/msgp"
 )
 
-// Stream name for retina networkflow logs
-const RetinaNetworkFlowLogsStreamName = "RETINA_NETWORK_FLOW_LOGS"
+// Stream names for network flow logs
+const (
+	ContainerNetworkLogsStreamName  = "CONTAINER_NETWORK_LOGS"   // New stream name
+	RetinaNetworkFlowLogsStreamName = "RETINA_NETWORK_FLOW_LOGS" // Legacy stream name
+)
 
 var (
 	// retina networkflow logs stream tag name
@@ -75,10 +78,15 @@ func PostNetworkFlowRecords(tailPluginRecords []map[interface{}]interface{}) int
 		}
 
 		if len(networkFlowLogsMsgPackEntries) > 0 {
-			MdsdNetworkFlowLogsStreamTagName = getOutputStreamIdTag(RetinaNetworkFlowLogsStreamName, MdsdNetworkFlowLogsStreamTagName, &NetworkFlowTagRefreshTracker)
+			// Try getting the stream tag with the new name first
+			MdsdNetworkFlowLogsStreamTagName = getOutputStreamIdTag(ContainerNetworkLogsStreamName, MdsdNetworkFlowLogsStreamTagName, &NetworkFlowTagRefreshTracker)
 			if MdsdNetworkFlowLogsStreamTagName == "" {
-				Log("Error::mdsd::Failed to get stream tag for networkflow logs. Will retry ...")
-				return output.FLB_RETRY
+				// If new stream name fails, try the legacy stream name
+				MdsdNetworkFlowLogsStreamTagName = getOutputStreamIdTag(RetinaNetworkFlowLogsStreamName, MdsdNetworkFlowLogsStreamTagName, &NetworkFlowTagRefreshTracker)
+				if MdsdNetworkFlowLogsStreamTagName == "" {
+					Log("Error::mdsd::Failed to get stream tag for networkflow logs. Will retry ...")
+					return output.FLB_RETRY
+				}
 			}
 			if MdsdNetworkFlowClient == nil {
 				Log("Error::mdsd::mdsd connection does not exist for networkflow mdsd client. re-connecting ...")
@@ -286,12 +294,17 @@ func mapNetworkFlowLogsToDataMap(dataMap map[string]interface{}, record map[stri
 	if traceObservationPoint := extractString(flow, "trace_observation_point"); traceObservationPoint != "" {
 		dataMap["TraceObservationPoint"] = traceObservationPoint
 	}
-	// Packets and Bytes
-	if packetsSent, ok := flow["packets_sent"]; ok {
-		dataMap["PacketsSent"] = safeToInt(packetsSent)
-	}
-	if packetsReceived, ok := flow["packets_received"]; ok {
-		dataMap["PacketsReceived"] = safeToInt(packetsReceived)
+	// Flow counts from extensions
+	if extensions, ok := flow["extensions"].(map[string]interface{}); ok {
+		if ingressCount, ok := extensions["ingress_flow_count"]; ok {
+			dataMap["IngressFlowCount"] = safeToInt(ingressCount)
+		}
+		if egressCount, ok := extensions["egress_flow_count"]; ok {
+			dataMap["EgressFlowCount"] = safeToInt(egressCount)
+		}
+		if unknownCount, ok := extensions["unknown_direction_flow_count"]; ok {
+			dataMap["UnknownDirectionFlowCount"] = safeToInt(unknownCount)
+		}
 	}
 	// Policies
 	policiesData := map[string]interface{}{}
@@ -329,8 +342,17 @@ func mapNetworkFlowLogsToDataMap(dataMap map[string]interface{}, record map[stri
 	if summary, ok := flow["Summary"]; ok {
 		additionalData["Summary"] = summary
 	}
-	if extensions, ok := flow["extensions"]; ok {
-		additionalData["Extensions"] = extensions
+	if extensions, ok := flow["extensions"].(map[string]interface{}); ok {
+		// Create a new map without the flow count fields
+		filteredExtensions := make(map[string]interface{})
+		for k, v := range extensions {
+			if k != "ingress_flow_count" && k != "egress_flow_count" && k != "unknown_direction_flow_count" {
+				filteredExtensions[k] = v
+			}
+		}
+		if len(filteredExtensions) > 0 {
+			additionalData["Extensions"] = filteredExtensions
+		}
 	}
 	if len(additionalData) > 0 {
 		dataMap["AdditionalFlowData"] = serializeToJSON(additionalData)