Fix add_kubernetes_metadata matcher: support rotated logs when 'resource_type: pod'

CHANGELOG.next.asciidoc

@@ -53,6 +53,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...main[Check the HEAD dif
 - auditd: Prevent mapping explosion when truncated EXECVE records are ingested. {pull}30382[30382]
 - elasticsearch: fix duplicate ingest when using a common appender configuration {issue}30428[30428] {pull}30440[30440]
 - Fix compatibility with ECS by renaming `source` log key to `source_file` {issue}30667[30667]
+- Fix add_kubernetes_metadata matcher: support rotated logs when `resource_type: pod` {pull}30720[30720]
 
 *Filebeat*
 

filebeat/processor/add_kubernetes_metadata/matchers.go

@@ -100,7 +100,7 @@ func (f *LogPathMatcher) MetadataIndex(event common.MapStr) string {
 
 	if f.ResourceType == "pod" {
 		// Pod resource type will extract only the pod UID, which offers less granularity of metadata when compared to the container ID
-		if strings.HasSuffix(source, ".log") {
+		if strings.Contains(source, ".log") && !strings.HasSuffix(source, ".gz") {
 			// Specify a pod resource type when writting logs into manually mounted log volume,
 			// those logs apper under under "/var/lib/kubelet/pods/<pod_id>/volumes/..."
 			if strings.HasPrefix(f.LogsPath, podKubeletLogsPath()) {
@@ -130,27 +130,27 @@ func (f *LogPathMatcher) MetadataIndex(event common.MapStr) string {
 			f.logger.Error("Error extracting pod uid - source value does not contains matcher's logs_path")
 			return ""
 		}
-	}
-	// In case of the Kubernetes log path "/var/log/containers/",
-	// the container ID will be located right before the ".log" extension.
-	// file name example: /var/log/containers/<pod_name>_<namespace>_<container_name>-<continer_id>.log
-	if strings.HasPrefix(f.LogsPath, containerLogsPath()) && strings.HasSuffix(source, ".log") && sourceLen >= containerIdLen+4 {
-		containerIDEnd := sourceLen - 4
-		cid := source[containerIDEnd-containerIdLen : containerIDEnd]
-		f.logger.Debugf("Using container id: %s", cid)
-		return cid
-	}
+	} else {
+		// In case of the Kubernetes log path "/var/log/containers/",
+		// the container ID will be located right before the ".log" extension.
+		// file name example: /var/log/containers/<pod_name>_<namespace>_<container_name>-<continer_id>.log
+		if strings.HasPrefix(f.LogsPath, containerLogsPath()) && strings.HasSuffix(source, ".log") && sourceLen >= containerIdLen+4 {
+			containerIDEnd := sourceLen - 4
+			cid := source[containerIDEnd-containerIdLen : containerIDEnd]
+			f.logger.Debugf("Using container id: %s", cid)
+			return cid
+		}
 
-	// In any other case, we assume the container ID will follow right after the log path.
-	// However we need to check the length to prevent "slice bound out of range" runtime errors.
-	// for the default log path /var/lib/docker/containers/ container ID will follow right after the log path.
-	// file name example: /var/lib/docker/containers/<container_id>/<container_id>-json.log
-	if sourceLen >= logsPathLen+containerIdLen {
-		cid := source[logsPathLen : logsPathLen+containerIdLen]
-		f.logger.Debugf("Using container id: %s", cid)
-		return cid
+		// In any other case, we assume the container ID will follow right after the log path.
+		// However we need to check the length to prevent "slice bound out of range" runtime errors.
+		// for the default log path /var/lib/docker/containers/ container ID will follow right after the log path.
+		// file name example: /var/lib/docker/containers/<container_id>/<container_id>-json.log
+		if sourceLen >= logsPathLen+containerIdLen {
+			cid := source[logsPathLen : logsPathLen+containerIdLen]
+			f.logger.Debugf("Using container id: %s", cid)
+			return cid
+		}
 	}
-
 	f.logger.Error("Error extracting container id - source value contains matcher's logs_path, however it is too short to contain a Docker container ID.")
 	return ""
 }

filebeat/processor/add_kubernetes_metadata/matchers_test.go

@@ -125,6 +125,28 @@ func TestLogsPathMatcher_InvalidVarLogPodSource(t *testing.T) {
 	executeTestWithResourceType(t, cfgLogsPath, cfgResourceType, source, expectedResult)
 }
 
+func TestLogsPathMatcher_ValidVarLogPodSource(t *testing.T) {
+	cfgLogsPath := "/var/log/pods/"
+	cfgResourceType := "pod"
+	sourcePath := "/var/log/pods/namespace_pod-name_%s/container/0.log.20220221-210912"
+
+	if runtime.GOOS == "windows" {
+		cfgLogsPath = "C:\\var\\log\\pods\\"
+		sourcePath = "C:\\var\\log\\pods\\namespace_pod-name_%s\\container\\0.log.20220221-210912"
+	}
+	source := fmt.Sprintf(sourcePath, puid)
+	expectedResult := puid
+	executeTestWithResourceType(t, cfgLogsPath, cfgResourceType, source, expectedResult)
+}
+
+func TestLogsPathMatcher_InvalidVarLogPodSource2(t *testing.T) {
+	cfgLogsPath := "/var/log/pods/"
+	cfgResourceType := "pod"
+	source := fmt.Sprintf("/var/log/pods/namespace_pod-name_%s/container/0.log.20220221-210526.gz", puid)
+	expectedResult := ""
+	executeTestWithResourceType(t, cfgLogsPath, cfgResourceType, source, expectedResult)
+}
+
 func TestLogsPathMatcher_InvalidVarLogPodIDFormat(t *testing.T) {
 	cfgLogsPath := "/var/log/pods/"
 	cfgResourceType := "pod"