Longw/metadata containerlogv2 kubernetes

.trivyignore

@@ -3,11 +3,15 @@ CVE-2023-39325 #same as CVE-2023-44487
 CVE-2023-3978
 CVE-2023-44487 #false positive according to Mariner team
 GHSA-jq35-85cj-fj4p
+CVE-2019-3826
 
 #telegraf HIGH
 GHSA-m425-mq94-257g
 CVE-2023-46129
 CVE-2023-47090 
 
-# ruby HIGH
+#ruby HIGH
 CVE-2017-10784
+
+#openssl
+CVE-2023-5678

build/common/installer/scripts/tomlparser.rb

@@ -215,7 +215,10 @@ def populateSettingValuesFromConfigMap(parsedConfig)
         @logEnableKubernetesMetadata = parsedConfig[:log_collection_settings][:metadata_collection][:enabled]
         if !parsedConfig[:log_collection_settings][:metadata_collection][:include_fields].nil?
           puts "config::Using config map setting for kubernetes metadata include fields"
-          @logKubernetesMetadataiIncludeFields = parsedConfig[:log_collection_settings][:metadata_collection][:include_fields]
+          include_fields = parsedConfig[:log_collection_settings][:metadata_collection][:include_fields]
+          if include_fields.kind_of?(Array)
+            @logKubernetesMetadataiIncludeFields = include_fields.join(",")
+          end
         end
       end
     end

build/linux/installer/conf/fluent-bit.conf

@@ -53,5 +53,6 @@
 #${KubernetesFilterEnabled}    Use_Kubelet         true
 #${KubernetesFilterEnabled}    Kubelet_Port        10250
 #${KubernetesFilterEnabled}    Kubelet_Host        ${NODE_IP}
+#${KubernetesFilterEnabled}    tls.verify          Off
 #${KubernetesFilterEnabled}    K8S-Logging.Exclude On
 #${KubernetesFilterEnabled}    Kube_Tag_Prefix     oms.container.log.la.var.log.containers.

source/plugins/go/src/oms.go

@@ -161,6 +161,10 @@ var (
 	ContainerLogSchemaV2 bool
 	// container log schema version from config map
 	ContainerLogV2ConfigMap bool
+	// Kubernetes Metadata enabled through configmap flag
+	KubernetesMetadataConfigMap bool
+	// Kubernetes Metadata enabled exclude list
+	KubernetesMetadataIncludeList []string
 	//ADX Cluster URI
 	AdxClusterUri string
 	// ADX clientID
@@ -272,6 +276,8 @@ type DataItemLAv2 struct {
 	PodNamespace  string `json:"PodNamespace"`
 	LogMessage    string `json:"LogMessage"`
 	LogSource     string `json:"LogSource"`
+	KubernetesMetadata    string `json:"KubernetesMetadata"`
+	//LogLevel     string `json:"LogLevel"`
 	//PodLabels			  string `json:"PodLabels"`
 }
 
@@ -1105,6 +1111,62 @@ func UpdateNumTelegrafMetricsSentTelemetry(numMetricsSent int, numSendErrors int
 	ContainerLogTelemetryMutex.Unlock()
 }
 
+func processIncludes(kubernetesMetadataMap map[string]interface{}, includesList []string) map[string]interface{} {
+	includedMetadata := make(map[string]interface{})
+	for _, include := range includesList {
+		switch include {
+		case "podUid":
+			if val, ok := kubernetesMetadataMap["pod_id"]; ok {
+				includedMetadata["podUid"] = val
+			}
+		case "podLabels":
+			if val, ok := kubernetesMetadataMap["labels"]; ok {
+				includedMetadata["podLabels"] = val
+			}
+		case "podAnnotations":
+			if val, ok := kubernetesMetadataMap["annotations"]; ok {
+				includedMetadata["podAnnotations"] = val
+			}
+		case "image":
+			if hash, ok := kubernetesMetadataMap["container_hash"]; ok {
+				includedMetadata["image_hash"] = hash
+			}
+			if image, ok := kubernetesMetadataMap["container_image"]; ok {
+				includedMetadata["image"] = image
+			}
+		}
+	}
+	return includedMetadata
+}
+
+func convertKubernetesMetadata(kubernetesMetadataJson interface{}) (map[string]interface{}, error) {
+    m, ok := kubernetesMetadataJson.(map[interface{}]interface{})
+    if !ok {
+        return nil, fmt.Errorf("type assertion to map[interface{}]interface{} failed")
+    }
+
+    strMap := make(map[string]interface{})
+    for k, v := range m {
+        strKey, ok := k.(string)
+        if !ok {
+            continue
+        }
+        switch val := v.(type) {
+        case map[interface{}]interface{}:
+            convertedMap, err := convertKubernetesMetadata(val)
+            if err != nil {
+                return nil, err
+            }
+            strMap[strKey] = convertedMap
+        case []byte:
+            strMap[strKey] = string(val)
+        default:
+            strMap[strKey] = val
+        }
+    }
+    return strMap, nil
+}
+
 // PostDataHelper sends data to the ODS endpoint or oneagent or ADX
 func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 	start := time.Now()
@@ -1135,6 +1197,27 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 	for _, record := range tailPluginRecords {
 		containerID, k8sNamespace, k8sPodName, containerName := GetContainerIDK8sNamespacePodNameFromFileName(ToString(record["filepath"]))
 		logEntrySource := ToString(record["stream"])
+		kubernetesMetadata := ""
+		if KubernetesMetadataConfigMap {
+			if kubernetesMetadataJson, exists := record["kubernetes"]; exists {
+				kubernetesMetadataMap, err := convertKubernetesMetadata(kubernetesMetadataJson)
+				if err != nil {
+					Log(fmt.Sprintf("Error convertKubernetesMetadata: %v", err))
+				}
+				includedMetadata := processIncludes(kubernetesMetadataMap, KubernetesMetadataIncludeList)
+				kubernetesMetadataBytes, err := json.Marshal(includedMetadata)
+				if err != nil {
+					message := fmt.Sprintf("Error while Marshalling kubernetesMetadataBytes to json bytes: %s", err.Error())
+					Log(message)
+					SendException(message)
+				}
+				kubernetesMetadata = string(kubernetesMetadataBytes)
+			} else {
+				message := fmt.Sprintf("Error while fetching kubernetesMetadataJson")
+				Log(message)
+				continue
+			}
+		}
 
 		if strings.EqualFold(logEntrySource, "stdout") {
 			if containerID == "" || containsKey(StdoutIgnoreNsSet, k8sNamespace) {
@@ -1160,6 +1243,8 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 		}
 
 		logEntry := ToString(record["log"])
+		//filter loglevel and define here
+		//logLevel := ToString(record["LogLevel"])
 		logEntryTimeStamp := ToString(record["time"])
 
 		if !ContainerLogV2ConfigMap && IsAADMSIAuthMode == true && !IsGenevaLogsIntegrationEnabled {
@@ -1179,7 +1264,18 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 		}
 
 		//ADX Schema & LAv2 schema are almost the same (except resourceId)
-		if ContainerLogSchemaV2 == true || ContainerLogsRouteADX == true {
+		if ContainerLogSchemaV2 == true {
+			stringMap["Computer"] = Computer
+			stringMap["ContainerId"] = containerID
+			stringMap["ContainerName"] = containerName
+			stringMap["PodName"] = k8sPodName
+			stringMap["PodNamespace"] = k8sNamespace
+			stringMap["LogMessage"] = logEntry
+			stringMap["LogSource"] = logEntrySource
+			stringMap["TimeGenerated"] = logEntryTimeStamp
+			stringMap["KubernetesMetadata"] = kubernetesMetadata
+			//stringMap["LogLevel"] = logLevel
+		} else if ContainerLogsRouteADX == true {
 			stringMap["Computer"] = Computer
 			stringMap["ContainerId"] = containerID
 			stringMap["ContainerName"] = containerName
@@ -1251,6 +1347,8 @@ func PostDataHelper(tailPluginRecords []map[interface{}]interface{}) int {
 					PodNamespace:  stringMap["PodNamespace"],
 					LogMessage:    stringMap["LogMessage"],
 					LogSource:     stringMap["LogSource"],
+					KubernetesMetadata: stringMap["KubernetesMetadata"],
+					//LogLevel: stringMap["LogLevel"],
 				}
 				//ODS-v2 schema
 				dataItemsLAv2 = append(dataItemsLAv2, dataItemLAv2)
@@ -1870,6 +1968,17 @@ func InitializePlugin(pluginConfPath string, agentVersion string) {
 	ContainerLogSchemaV2 = false //default is v1 schema
 	ContainerLogV2ConfigMap = (strings.Compare(ContainerLogSchemaVersion, ContainerLogV2SchemaVersion) == 0)
 
+	KubernetesMetadataConfigMap = false
+	KubernetesMetadataConfigMap = (strings.Compare(strings.ToLower(os.Getenv("AZMON_KUBERNETES_METADATA_ENABLED")), "true") == 0)
+	metadataIncludeList := os.Getenv("AZMON_KUBERNETES_METADATA_INCLUDES_FIELDS")
+	Log(fmt.Sprintf("KubernetesMetadataIncludeList from configmap: %+v\n", metadataIncludeList))
+	KubernetesMetadataIncludeList = []string{"podLabels", "podAnnotations", "podUid", "image"}
+	if KubernetesMetadataConfigMap && len(metadataIncludeList) > 0 {
+		KubernetesMetadataIncludeList = strings.Split(metadataIncludeList, ",")
+	} else if KubernetesMetadataConfigMap {
+		KubernetesMetadataIncludeList = []string{}
+	}
+
 	if ContainerLogV2ConfigMap && ContainerLogsRouteADX != true {
 		ContainerLogSchemaV2 = true
 		Log("Container logs schema=%s", ContainerLogV2SchemaVersion)

test/scenario/containerlogv2_linux-resource-app.yml

@@ -0,0 +1,29 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: linux-resource-intensive-app
+  namespace: kube-system
+spec:
+  replicas: 10
+  selector:
+    matchLabels:
+      app: linux-resource-intensive-app
+  template:
+    metadata:
+      labels:
+        app: linux-resource-intensive-app
+    spec:
+      nodeSelector:
+        kubernetes.io/os: linux
+      containers:
+      - name: linux-resource-intensive-app
+        image: ubuntu
+        command: ["/bin/bash", "-c"]
+        args:
+          - |
+            while true; do
+              timestamp=$(date "+%Y/%m/%d %H:%M:%S.%3N")
+              logEntry=$(printf 'Test-%.0s' {1..200})
+              echo "$timestamp $logEntry"
+              sleep 0.001  # Sleep for 1 millisecond
+            done

test/scenario/containerlogv2_windows-resource-app.yml

@@ -0,0 +1,37 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: windows-resource-intensive-app
+  namespace: kube-system
+spec:
+  replicas: 10 # 50
+  selector:
+    matchLabels:
+      app: windows-resource-intensive-app
+  template:
+    metadata:
+      labels:
+        app: windows-resource-intensive-app
+        # more labels
+    spec:
+      nodeSelector:
+        kubernetes.io/os: windows
+      containers:
+      - name: windows-resource-intensive-app
+        image: mcr.microsoft.com/windows/servercore:ltsc2022
+        command: ["powershell", "-c"]
+        args:
+          - |
+            while ($true) {
+              $timestamp = Get-Date -Format "yyyy/MM/dd HH:mm:ss.fff"
+              $logEntry = 'Test-' * 200   # Adjusted to create a log entry of about 1000 bytes
+              Write-Output "$timestamp $logEntry"
+              Start-Sleep -Milliseconds 1  # Generates approximately 1000 logs per second
+            }
+
+## thousand logs per second
+## 10 pods in certain nodes combining logs each generate some logs  scale  performance  podlevel
+## multiple 100 pods
+
+## nodes level testing  both windows linux   if api then proabbly node level   kubelet
+## create a branch for current code branch for testing current stage