feat(sourceprocessor): use namespace annotations (#1471)

use namespace annotations to include/exclude namespace from collection or set sourceCategory, sourceHost, and sourceName

.changelog/1471.added.txt

@@ -0,0 +1 @@
+feat(sourceprocessor): use namespace annotations to include/exclude namespace from collection or set sourceCategory, sourceHost, and sourceName

pkg/processor/sourceprocessor/README.md

@@ -41,10 +41,18 @@ processors:
       <attribute_key_1>: <attribute_value_regex_1>
       <attribute_key_2>: <attribute_value_regex_2>
 
-    # Prefix which allows to find given annotation; it is used for including/excluding pods, among other attributes.
+    # The processor assumes that pod annotations will be present as resource attributes,
+    # one attribute per annotation, and that these attributes have a common prefix.
+    # This setting controls the prefix.
     # default: "k8s.pod.annotation."
     annotation_prefix: <annotation_prefix>
 
+    # The processor assumes that namespace annotations will be present as resource attributes,
+    # one attribute per annotation, and that these attributes have a common prefix.
+    # This setting controls the prefix.
+    # default: "k8s.namespace.annotation."
+    namespace_annotation_prefix: <namespace_annotation_prefix>
+
     # Name of the attribute that contains the full name of the pod.
     # default: "k8s.pod.name"
     pod_key: <pod_key>
@@ -114,17 +122,17 @@ processors:
       pod: "custom-pod-.*"
 ```
 
-## Pod annotations
+## Pod and namespace annotations
 
-The following [Kubernetes annotations][k8s_annotations_doc] can be used on pods:
+The following [Kubernetes annotations][k8s_annotations_doc] can be used on pods or namespace:
 
-- `sumologic.com/exclude` - records from a pod that has this annotation set to
+- `sumologic.com/exclude` - records from a pod/namespace that has this annotation set to
   `true` are dropped,
 
   **NOTE**: this has precedence over `sumologic.com/include` if both are set at
-  the same time for one pod.
+  the same time for one pod/namespace.
 
-- `sumologic.com/include` - records from a pod that has this annotation set to
+- `sumologic.com/include` - records from a pod/namespace that has this annotation set to
   `true` are not checked against exclusion regexes from `exclude` processor settings
 
 - `sumologic.com/sourceCategory` - overrides `source_category` config option
@@ -142,6 +150,8 @@ This can be achieved with the [Kubernetes processor](../k8sprocessor).
 For example, if a resource has the `k8s.pod.annotation.sumologic.com/exclude`
 attribute set to `true`, the resource will be dropped.
 
+*Note:** Pod annotations take precedence over namespace annotations.
+
 [k8s_annotations_doc]: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
 
 ### Container-level pod annotations
@@ -151,7 +161,7 @@ it is possible to set pod annotations that are container-specific.
 
 The following rules apply:
 
-- Container-level annotations take precendence over other forms of setting the source category.
+- Container-level annotations take precedence over other forms of setting the source category.
 - No other transformations are applied to the source categories retrieved from
   container-level annotations, like adding source category prefix or replacing the dash.
 

pkg/processor/sourceprocessor/attribute_filler.go

@@ -69,17 +69,27 @@ func createSourceNameFiller(cfg *Config) attributeFiller {
 	return filler
 }
 
-func (f *attributeFiller) fillResourceOrUseAnnotation(atts *pcommon.Map, annotationKey string) bool {
+func (f *attributeFiller) fillResourceOrUseAnnotation(atts *pcommon.Map, annotationKey string, namespaceAnnotationKey string) bool {
 	val, found := atts.Get(annotationKey)
 	if found {
-		annotationFiller := extractFormat(val.Str(), f.name)
-		annotationFiller.dashReplacement = f.dashReplacement
-		annotationFiller.compiledFormat = f.prefix + annotationFiller.compiledFormat
-		return annotationFiller.fillAttributes(atts)
+		return f.useAnnotation(atts, val)
 	}
+
+	val, found = atts.Get(namespaceAnnotationKey)
+	if found {
+		return f.useAnnotation(atts, val)
+	}
+
 	return f.fillAttributes(atts)
 }
 
+func (f *attributeFiller) useAnnotation(atts *pcommon.Map, annotation pcommon.Value) bool {
+	annotationFiller := extractFormat(annotation.Str(), f.name)
+	annotationFiller.dashReplacement = f.dashReplacement
+	annotationFiller.compiledFormat = f.prefix + annotationFiller.compiledFormat
+	return annotationFiller.fillAttributes(atts)
+}
+
 func (f *attributeFiller) fillAttributes(atts *pcommon.Map) bool {
 	if len(f.compiledFormat) == 0 {
 		return false

pkg/processor/sourceprocessor/config.go

@@ -29,10 +29,11 @@ type Config struct {
 	// the processed entry is dropped.
 	Exclude map[string]string `mapstructure:"exclude"`
 
-	AnnotationPrefix   string `mapstructure:"annotation_prefix"`
-	PodKey             string `mapstructure:"pod_key"`
-	PodNameKey         string `mapstructure:"pod_name_key"`
-	PodTemplateHashKey string `mapstructure:"pod_template_hash_key"`
+	AnnotationPrefix          string `mapstructure:"annotation_prefix"`
+	NamespaceAnnotationPrefix string `mapstructure:"namespace_annotation_prefix"`
+	PodKey                    string `mapstructure:"pod_key"`
+	PodNameKey                string `mapstructure:"pod_name_key"`
+	PodTemplateHashKey        string `mapstructure:"pod_template_hash_key"`
 
 	ContainerAnnotations ContainerAnnotationsConfig `mapstructure:"container_annotations"`
 }

pkg/processor/sourceprocessor/config_test.go

@@ -58,10 +58,11 @@ func TestLoadConfig(t *testing.T) {
 			"_SYSTEMD_UNIT":      "excluded_systemd_unit_regex",
 		},
 
-		AnnotationPrefix:   "pod_annotation_",
-		PodKey:             "k8s.pod.name",
-		PodNameKey:         "k8s.pod.pod_name",
-		PodTemplateHashKey: "pod_labels_pod-template-hash",
+		AnnotationPrefix:          "pod_annotation_",
+		NamespaceAnnotationPrefix: "namespace_annotation_",
+		PodKey:                    "k8s.pod.name",
+		PodNameKey:                "k8s.pod.pod_name",
+		PodTemplateHashKey:        "pod_labels_pod-template-hash",
 
 		ContainerAnnotations: ContainerAnnotationsConfig{
 			Enabled:          false,

pkg/processor/sourceprocessor/factory.go

@@ -35,11 +35,12 @@ const (
 	defaultSourceCategoryPrefix      = "kubernetes/"
 	defaultSourceCategoryReplaceDash = "/"
 
-	defaultAnnotationPrefix   = "k8s.pod.annotation."
-	defaultPodKey             = "k8s.pod.name"
-	defaultPodNameKey         = "k8s.pod.pod_name"
-	defaultPodTemplateHashKey = "k8s.pod.label.pod-template-hash"
-	defaultContainerNameKey   = "k8s.container.name"
+	defaultAnnotationPrefix          = "k8s.pod.annotation."
+	defaultNamespaceAnnotationPrefix = "k8s.namespace.annotation."
+	defaultPodKey                    = "k8s.pod.name"
+	defaultPodNameKey                = "k8s.pod.pod_name"
+	defaultPodTemplateHashKey        = "k8s.pod.label.pod-template-hash"
+	defaultContainerNameKey          = "k8s.container.name"
 
 	stabilityLevel = component.StabilityLevelBeta
 )
@@ -67,10 +68,11 @@ func createDefaultConfig() component.Config {
 		SourceCategoryPrefix:      defaultSourceCategoryPrefix,
 		SourceCategoryReplaceDash: defaultSourceCategoryReplaceDash,
 
-		AnnotationPrefix:   defaultAnnotationPrefix,
-		PodKey:             defaultPodKey,
-		PodNameKey:         defaultPodNameKey,
-		PodTemplateHashKey: defaultPodTemplateHashKey,
+		AnnotationPrefix:          defaultAnnotationPrefix,
+		NamespaceAnnotationPrefix: defaultNamespaceAnnotationPrefix,
+		PodKey:                    defaultPodKey,
+		PodNameKey:                defaultPodNameKey,
+		PodTemplateHashKey:        defaultPodTemplateHashKey,
 
 		ContainerAnnotations: ContainerAnnotationsConfig{
 			Enabled:          false,

pkg/processor/sourceprocessor/source_category_filler.go

@@ -30,6 +30,7 @@ type sourceCategoryFiller struct {
 	prefix                       string
 	dashReplacement              string
 	annotationPrefix             string
+	namespaceAnnotationPrefix    string
 	containerAnnotationsEnabled  bool
 	containerNameKey             string
 	containerAnnotationsPrefixes []string
@@ -47,6 +48,7 @@ func newSourceCategoryFiller(cfg *Config, logger *zap.Logger) sourceCategoryFill
 		prefix:                       cfg.SourceCategoryPrefix,
 		dashReplacement:              cfg.SourceCategoryReplaceDash,
 		annotationPrefix:             cfg.AnnotationPrefix,
+		namespaceAnnotationPrefix:    cfg.NamespaceAnnotationPrefix,
 		containerAnnotationsEnabled:  cfg.ContainerAnnotations.Enabled,
 		containerNameKey:             cfg.ContainerAnnotations.ContainerNameKey,
 		containerAnnotationsPrefixes: cfg.ContainerAnnotations.Prefixes,
@@ -64,9 +66,10 @@ func extractTemplateAttributes(template string) []string {
 
 // fill takes a collection of attributes for a record and adds to it a new attribute with the source category for the record.
 //
-// The source category is retrieved from one of three places (in the following precedence):
+// The source category is retrieved from one of the following locations, listed in descending order of precedence:
 // - the source category container-level annotation (e.g. "k8s.pod.annotation.sumologic.com/container-name.sourceCategory"),
 // - the source category pod-level annotation (e.g. "k8s.pod.annotation.sumologic.com/sourceCategory"),
+// - the source category namespace-level annotation (e.g. "k8s.namespace.annotation.sumologic.com/sourceCategory"),
 // - the source category configured in the processor's "source_category" configuration option.
 func (f *sourceCategoryFiller) fill(attributes *pcommon.Map) {
 	containerSourceCategory := f.getSourceCategoryFromContainerAnnotation(attributes)
@@ -78,20 +81,13 @@ func (f *sourceCategoryFiller) fill(attributes *pcommon.Map) {
 	var templateAttributes []string
 	doesUseAnnotation := false
 
-	valueTemplate := getAnnotationAttributeValue(f.annotationPrefix, sourceCategorySpecialAnnotation, attributes)
-	if valueTemplate == "" {
-		valueTemplate = f.valueTemplate
-	} else {
-		doesUseAnnotation = true
-	}
+	// get sourceCategory and sourceCategoryPrefix from pod annotation
+	valueTemplate, doesUseAnnotation := f.getSourceCategoryFromAnnotation(f.annotationPrefix, attributes)
 
-	prefix := getAnnotationAttributeValue(f.annotationPrefix, sourceCategoryPrefixAnnotation, attributes)
-	if prefix == "" {
-		prefix = f.prefix
-	} else {
-		doesUseAnnotation = true
+	if !doesUseAnnotation {
+		// get sourceCategory and sourceCategoryPrefix from namespace annotation
+		valueTemplate, doesUseAnnotation = f.getSourceCategoryFromAnnotation(f.namespaceAnnotationPrefix, attributes)
 	}
-	valueTemplate = prefix + valueTemplate
 
 	if doesUseAnnotation {
 		templateAttributes = extractTemplateAttributes(valueTemplate)
@@ -101,15 +97,45 @@ func (f *sourceCategoryFiller) fill(attributes *pcommon.Map) {
 
 	sourceCategoryValue := f.replaceTemplateAttributes(valueTemplate, templateAttributes, attributes)
 
-	dashReplacement := getAnnotationAttributeValue(f.annotationPrefix, sourceCategoryReplaceDashAnnotation, attributes)
-	if dashReplacement == "" {
-		dashReplacement = f.dashReplacement
-	}
+	dashReplacement := f.getSourceCategoryDashReplacement(attributes)
 	sourceCategoryValue = strings.ReplaceAll(sourceCategoryValue, "-", dashReplacement)
 
 	attributes.PutStr(sourceCategoryKey, sourceCategoryValue)
 }
 
+func (f *sourceCategoryFiller) getSourceCategoryDashReplacement(attributes *pcommon.Map) string {
+	dashReplacement := getAnnotationAttributeValue(f.annotationPrefix, sourceCategoryReplaceDashAnnotation, attributes)
+	if dashReplacement != "" {
+		return dashReplacement
+	}
+
+	dashReplacement = getAnnotationAttributeValue(f.namespaceAnnotationPrefix, sourceCategoryReplaceDashAnnotation, attributes)
+	if dashReplacement != "" {
+		return dashReplacement
+	}
+	return f.dashReplacement
+}
+
+func (f *sourceCategoryFiller) getSourceCategoryFromAnnotation(annotationPrefix string, attributes *pcommon.Map) (string, bool) {
+	doesUseAnnotation := false
+
+	valueTemplate := getAnnotationAttributeValue(annotationPrefix, sourceCategorySpecialAnnotation, attributes)
+	if valueTemplate == "" {
+		valueTemplate = f.valueTemplate
+	} else {
+		doesUseAnnotation = true
+	}
+
+	prefix := getAnnotationAttributeValue(annotationPrefix, sourceCategoryPrefixAnnotation, attributes)
+	if prefix == "" {
+		prefix = f.prefix
+	} else {
+		doesUseAnnotation = true
+	}
+	valueTemplate = prefix + valueTemplate
+	return valueTemplate, doesUseAnnotation
+}
+
 func (f *sourceCategoryFiller) getSourceCategoryFromContainerAnnotation(attributes *pcommon.Map) string {
 	if !f.containerAnnotationsEnabled {
 		return ""

pkg/processor/sourceprocessor/source_category_filler_test.go

@@ -66,6 +66,23 @@ func TestFillWithAnnotations(t *testing.T) {
 	assertAttribute(t, attrs, "_sourceCategory", "ABC#123asd#Prefix:sc#from#annot#ns#1#123asd")
 }
 
+func TestFillWithNamespaceAnnotations(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+
+	attrs := pcommon.NewMap()
+	attrs.PutStr("k8s.namespace.name", "ns-1")
+	attrs.PutStr("k8s.pod.uid", "123asd")
+	attrs.PutStr("k8s.pod.name", "ABC")
+	attrs.PutStr("k8s.namespace.annotation.sumologic.com/sourceCategory", "sc-from-annot-%{k8s.namespace.name}-%{k8s.pod.uid}")
+	attrs.PutStr("k8s.namespace.annotation.sumologic.com/sourceCategoryPrefix", "%{k8s.pod.name}-%{k8s.pod.uid}-Prefix:")
+	attrs.PutStr("k8s.namespace.annotation.sumologic.com/sourceCategoryReplaceDash", "#")
+
+	filler := newSourceCategoryFiller(cfg, zap.NewNop())
+	filler.fill(&attrs)
+
+	assertAttribute(t, attrs, "_sourceCategory", "ABC#123asd#Prefix:sc#from#annot#ns#1#123asd")
+}
+
 func TestFillWithContainerAnnotations(t *testing.T) {
 	t.Run("container annotations are disabled by default", func(t *testing.T) {
 		cfg := createDefaultConfig().(*Config)