add validation for logs_path matchers config

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>
elastic · Nov 9, 2021 · ab3c821 · ab3c821
1 parent e8ab1fe
commit ab3c821
Show file tree

Hide file tree

Showing 3 changed files with 103 additions and 2 deletions.
diff --git a/filebeat/processor/add_kubernetes_metadata/matchers.go b/filebeat/processor/add_kubernetes_metadata/matchers.go
@@ -125,8 +125,7 @@ func (f *LogPathMatcher) MetadataIndex(event common.MapStr) string {
 				}
 			}
 
-			f.logger.Error(`Error extracting pod uid - source value does not contains matcher's logs_path,
-				supported log_path for 'pod' resource_type: '/var/lib/kubelet/pods/', '/var/log/pods/'.`)
+			f.logger.Error("Error extracting pod uid - source value does not contains matcher's logs_path")
 			return ""
 		}
 	}

diff --git a/libbeat/processors/add_kubernetes_metadata/config.go b/libbeat/processors/add_kubernetes_metadata/config.go
@@ -69,5 +69,34 @@ func (k *kubeAnnotatorConfig) Validate() error {
 		k.Host = ""
 	}
 
+	for _, matcher := range k.Matchers {
+		if matcherCfg, ok := matcher["logs_path"]; ok {
+			if matcherCfg.HasField("resource_type") {
+				logsPathMatcher := struct {
+					LogsPath     string `config:"logs_path"`
+					ResourceType string `config:"resource_type"`
+				}{}
+
+				err := matcherCfg.Unpack(&logsPathMatcher)
+				if err != nil {
+					return fmt.Errorf("fail to unpack the `logs_path` matcher configuration: %s", err)
+				}
+				if logsPathMatcher.LogsPath == "" {
+					return fmt.Errorf("invalid logs_path matcher configuration: when resource_type is defined, logs_path must be set as well")
+				}
+				if logsPathMatcher.ResourceType != "pod" && logsPathMatcher.ResourceType != "container" {
+					return fmt.Errorf("invalid resource_type %s, valid values include `pod`, `container`", logsPathMatcher.ResourceType)
+				}
+				if logsPathMatcher.ResourceType == "pod" && !(logsPathMatcher.LogsPath == "/var/lib/kubelet/pods/" || logsPathMatcher.LogsPath == "/var/log/pods/") {
+					return fmt.Errorf("invalid logs_path defined for resource_type: %s, valid values include `/var/lib/kubelet/pods/`, `/var/log/pods/`", logsPathMatcher.ResourceType)
+				}
+				if logsPathMatcher.ResourceType == "container" && logsPathMatcher.LogsPath != "/var/log/containers/" {
+					return fmt.Errorf("invalid logs_path defined for resource_type: %s, valid value is `/var/log/containers/`", logsPathMatcher.ResourceType)
+				}
+			}
+
+		}
+	}
+
 	return nil
 }
diff --git a/libbeat/processors/add_kubernetes_metadata/config_test.go b/libbeat/processors/add_kubernetes_metadata/config_test.go
@@ -60,3 +60,76 @@ func TestConfigValidate(t *testing.T) {
 		}
 	}
 }
+
+func TestConfigValidate_LogsPatchMatcher(t *testing.T) {
+	tests := []struct {
+		matcherName   string
+		matcherConfig map[string]interface{}
+		error         bool
+	}{
+		{
+			matcherName:   "",
+			matcherConfig: map[string]interface{}{},
+			error:         false,
+		},
+		{
+			matcherName: "logs_path",
+			matcherConfig: map[string]interface{}{
+				"resource_type": "pod",
+			},
+			error: true,
+		},
+		{
+			matcherName: "logs_path",
+			matcherConfig: map[string]interface{}{
+				"resource_type": "pod",
+				"invalid_field": "invalid_value",
+			},
+			error: true,
+		},
+		{
+			matcherName: "logs_path",
+			matcherConfig: map[string]interface{}{
+				"resource_type": "pod",
+				"logs_path":     "/var/log/invalid/path/",
+			},
+			error: true,
+		},
+		{
+			matcherName: "logs_path",
+			matcherConfig: map[string]interface{}{
+				"resource_type": "pod",
+				"logs_path":     "/var/log/pods/",
+			},
+			error: false,
+		},
+		{
+			matcherName: "logs_path",
+			matcherConfig: map[string]interface{}{
+				"resource_type": "container",
+				"logs_path":     "/var/log/containers/",
+			},
+			error: false,
+		},
+	}
+
+	for _, test := range tests {
+		cfg, _ := common.NewConfigFrom(test.matcherConfig)
+
+		c := defaultKubernetesAnnotatorConfig()
+		c.DefaultMatchers = Enabled{false}
+
+		err := cfg.Unpack(&c)
+		c.Matchers = PluginConfig{
+			{
+				test.matcherName: *cfg,
+			},
+		}
+		err = c.Validate()
+		if test.error {
+			require.NotNil(t, err)
+		} else {
+			require.Nil(t, err)
+		}
+	}
+}