kubernetes processor use correct os path separator

filebeat/processor/add_kubernetes_metadata/matchers.go

@@ -19,6 +19,8 @@ package add_kubernetes_metadata
 
 import (
 	"fmt"
+	"os"
+	"runtime"
 	"strings"
 
 	"github.com/elastic/beats/libbeat/common"
@@ -38,6 +40,7 @@ func init() {
 }
 
 const LogPathMatcherName = "logs_path"
+const PathSeparator = string(os.PathSeparator)
 
 type LogPathMatcher struct {
 	LogsPath     string
@@ -49,7 +52,7 @@ func newLogsPathMatcher(cfg common.Config) (add_kubernetes_metadata.Matcher, err
 		LogsPath     string `config:"logs_path"`
 		ResourceType string `config:"resource_type"`
 	}{
-		LogsPath:     "/var/lib/docker/containers/",
+		LogsPath:     defaultLogPath(),
 		ResourceType: "container",
 	}
 
@@ -59,8 +62,8 @@ func newLogsPathMatcher(cfg common.Config) (add_kubernetes_metadata.Matcher, err
 	}
 
 	logPath := config.LogsPath
-	if logPath[len(logPath)-1:] != "/" {
-		logPath = logPath + "/"
+	if logPath[len(logPath)-1:] != PathSeparator {
+		logPath = logPath + PathSeparator
 	}
 	resourceType := config.ResourceType
 
@@ -92,10 +95,10 @@ func (f *LogPathMatcher) MetadataIndex(event common.MapStr) string {
 		if f.ResourceType == "pod" {
 			// Specify a pod resource type when manually mounting log volumes and they end up under "/var/lib/kubelet/pods/"
 			// This will extract only the pod UID, which offers less granularity of metadata when compared to the container ID
-			if strings.HasPrefix(f.LogsPath, "/var/lib/kubelet/pods/") && strings.HasSuffix(source, ".log") {
-				pathDirs := strings.Split(source, "/")
+			if strings.HasPrefix(f.LogsPath, PodLogsPath()) && strings.HasSuffix(source, ".log") {
+				pathDirs := strings.Split(source, PathSeparator)
 				if len(pathDirs) > podUIDPos {
-					podUID := strings.Split(source, "/")[podUIDPos]
+					podUID := strings.Split(source, PathSeparator)[podUIDPos]
 
 					logp.Debug("kubernetes", "Using pod uid: %s", podUID)
 					return podUID
@@ -106,7 +109,7 @@ func (f *LogPathMatcher) MetadataIndex(event common.MapStr) string {
 		} else {
 			// In case of the Kubernetes log path "/var/log/containers/",
 			// the container ID will be located right before the ".log" extension.
-			if strings.HasPrefix(f.LogsPath, "/var/log/containers/") && strings.HasSuffix(source, ".log") && sourceLen >= containerIdLen+4 {
+			if strings.HasPrefix(f.LogsPath, ContainerLogsPath()) && strings.HasSuffix(source, ".log") && sourceLen >= containerIdLen+4 {
 				containerIDEnd := sourceLen - 4
 				cid := source[containerIDEnd-containerIdLen : containerIDEnd]
 				logp.Debug("kubernetes", "Using container id: %s", cid)
@@ -127,3 +130,24 @@ func (f *LogPathMatcher) MetadataIndex(event common.MapStr) string {
 
 	return ""
 }
+
+func defaultLogPath() string {
+	if runtime.GOOS == "windows" {
+		return "C:\\ProgramData\\Docker\\containers"
+	}
+	return "/var/lib/docker/containers/"
+}
+
+func PodLogsPath() string {
+	if runtime.GOOS == "windows" {
+		return "C:\\var\\lib\\kubelet\\pods\\"
+	}
+	return "/var/lib/kubelet/pods/"
+}
+
+func ContainerLogsPath() string {
+	if runtime.GOOS == "windows" {
+		return "C:\\var\\log\\containers\\"
+	}
+	return "/var/log/containers/"
+}

filebeat/processor/add_kubernetes_metadata/matchers_test.go

@@ -19,6 +19,7 @@ package add_kubernetes_metadata
 
 import (
 	"fmt"
+	"runtime"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -55,29 +56,55 @@ func TestLogsPathMatcher_InvalidSource3(t *testing.T) {
 
 func TestLogsPathMatcher_VarLibDockerContainers(t *testing.T) {
 	cfgLogsPath := "" // use the default matcher configuration
-	source := fmt.Sprintf("/var/lib/docker/containers/%s/%s-json.log", cid, cid)
+
+	path := "/var/lib/docker/containers/%s/%s-json.log"
+	if runtime.GOOS == "windows" {
+		path = "C:\\ProgramData\\Docker\\containers\\%s\\%s-json.log"
+	}
+
+	source := fmt.Sprintf(path, cid, cid)
+
 	expectedResult := cid
 	executeTest(t, cfgLogsPath, source, expectedResult)
 }
 
 func TestLogsPathMatcher_VarLogContainers(t *testing.T) {
 	cfgLogsPath := "/var/log/containers/"
-	source := fmt.Sprintf("/var/log/containers/kube-proxy-4d7nt_kube-system_kube-proxy-%s.log", cid)
+	sourcePath := "/var/log/containers/kube-proxy-4d7nt_kube-system_kube-proxy-%s.log"
+	if runtime.GOOS == "windows" {
+		cfgLogsPath = "C:\\var\\log\\containers\\"
+		sourcePath = "C:\\var\\log\\containers\\kube-proxy-4d7nt_kube-system_kube-proxy-%s.log"
+	}
+
+	source := fmt.Sprintf(sourcePath, cid)
 	expectedResult := cid
 	executeTest(t, cfgLogsPath, source, expectedResult)
 }
 
 func TestLogsPathMatcher_AnotherLogDir(t *testing.T) {
 	cfgLogsPath := "/var/log/other/"
-	source := fmt.Sprintf("/var/log/other/%s.log", cid)
+	sourcePath := "/var/log/other/%s.log"
+	if runtime.GOOS == "windows" {
+		cfgLogsPath = "C:\\var\\log\\other\\"
+		sourcePath = "C:\\var\\log\\other\\%s.log"
+	}
+
+	source := fmt.Sprintf(sourcePath, cid)
 	expectedResult := cid
 	executeTest(t, cfgLogsPath, source, expectedResult)
 }
 
 func TestLogsPathMatcher_VarLibKubeletPods(t *testing.T) {
 	cfgLogsPath := "/var/lib/kubelet/pods/"
+	sourcePath := "/var/lib/kubelet/pods/%s/volumes/kubernetes.io~empty-dir/applogs/server.log"
 	cfgResourceType := "pod"
-	source := fmt.Sprintf("/var/lib/kubelet/pods/%s/volumes/kubernetes.io~empty-dir/applogs/server.log", puid)
+
+	if runtime.GOOS == "windows" {
+		cfgLogsPath = "C:\\var\\lib\\kubelet\\pods\\"
+		sourcePath = "C:\\var\\lib\\kubelet\\pods\\%s\\volumes\\kubernetes.io~empty-dir\\applogs\\server.log"
+	}
+
+	source := fmt.Sprintf(sourcePath, puid)
 	expectedResult := puid
 	executeTestWithResourceType(t, cfgLogsPath, cfgResourceType, source, expectedResult)
 }