[receiver/k8seventsreceiver] add extraction rules for labels and annotations

Inspired by k8sattributesprocessor introduces the similar configuration
syntax to support labels and annotations extraction from event

Author: gmlexx

.chloggen/k8s-events-receiver-extraction-rules.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enchancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: k8seventsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add labels and annotations extraction from event
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: []
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: ["user"]

receiver/k8seventsreceiver/README.md

@@ -81,6 +81,26 @@ data:
 EOF
 ```
 
+## Extracting attributes from event labels and annotations
+
+The k8seventsreceiver can also set resource attributes from k8s labels and annotations of event.
+This config represents a list of annotations/labels that are extracted from event and added to log.
+Each item is specified as a config of an optional tag_name (representing the tag name to tag the spans with),
+key (representing the key used to extract value) or key_regex (representing the key regex used to extract the value).
+
+A few examples to use this config are as follows:
+
+```yaml
+extract:
+  annotations: # same can be applied for labels
+    - tag_name: a1 # extracts value of annotation from event with key `annotation-one` and inserts it as a tag with key `a1`
+      key: annotation-one
+    - tag_name: k8s.event.annotations.$1 # extracts value of annotation from event with key matching the regex
+      key_regex: topology.kubernetes.io/(.*)
+  labels:
+    - key: label1 # extracts value of label from event with key `label1` and inserts it as a tag with key `k8s.event.labels.label1`
+```
+
 ### Service Account
 
 Create a service account that the collector should use.

receiver/k8seventsreceiver/config.go

@@ -4,6 +4,9 @@
 package k8seventsreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8seventsreceiver"
 
 import (
+	"fmt"
+	"regexp"
+
 	k8s "k8s.io/client-go/kubernetes"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
@@ -16,12 +19,33 @@ type Config struct {
 	// List of ‘namespaces’ to collect events from.
 	Namespaces []string `mapstructure:"namespaces"`
 
+	// Extract section allows specifying extraction rules to extract
+	// data from k8s event specs
+	Extract ExtractConfig `mapstructure:"extract"`
+
 	// For mocking
 	makeClient func(apiConf k8sconfig.APIConfig) (k8s.Interface, error)
 }
 
 func (cfg *Config) Validate() error {
-	return cfg.APIConfig.Validate()
+	if err := cfg.APIConfig.Validate(); err != nil {
+		return err
+	}
+
+	for _, f := range append(cfg.Extract.Labels, cfg.Extract.Annotations...) {
+		if f.Key != "" && f.KeyRegex != "" {
+			return fmt.Errorf("Out of Key or KeyRegex only one option is expected to be configured at a time, currently Key:%s and KeyRegex:%s", f.Key, f.KeyRegex)
+		}
+
+		if f.KeyRegex != "" {
+			_, err := regexp.Compile("^(?:" + f.KeyRegex + ")$")
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
 }
 
 func (cfg *Config) getK8sClient() (k8s.Interface, error) {
@@ -30,3 +54,52 @@ func (cfg *Config) getK8sClient() (k8s.Interface, error) {
 	}
 	return cfg.makeClient(cfg.APIConfig)
 }
+
+// ExtractConfig section allows specifying extraction rules to extract
+// data from k8s event specs.
+type ExtractConfig struct {
+	// Annotations allows extracting data from event annotations and record it
+	// as resource attributes.
+	// It is a list of FieldExtractConfig type. See FieldExtractConfig
+	// documentation for more details.
+	Annotations []FieldExtractConfig `mapstructure:"annotations"`
+
+	// Labels allows extracting data from event labels and record it
+	// as resource attributes.
+	// It is a list of FieldExtractConfig type. See FieldExtractConfig
+	// documentation for more details.
+	Labels []FieldExtractConfig `mapstructure:"labels"`
+}
+
+// FieldExtractConfig allows specifying an extraction rule to extract a resource attribute from event
+// annotations (or labels).
+type FieldExtractConfig struct {
+	// TagName represents the name of the resource attribute that will be added to logs, metrics or spans.
+	// When not specified, a default tag name will be used of the format:
+	//   - k8s.event.annotations.<annotation key>
+	//   - k8s.event.labels.<label key>
+	// For example, if tag_name is not specified and the key is git_sha,
+	// then the attribute name will be `k8s.event.annotations.git_sha`.
+	// When key_regex is present, tag_name supports back reference to both named capturing and positioned capturing.
+	// For example, if your event spec contains the following labels,
+	//
+	// app.kubernetes.io/component: mysql
+	// app.kubernetes.io/version: 5.7.21
+	//
+	// and you'd like to add tags for all labels with prefix app.kubernetes.io/ and also trim the prefix,
+	// then you can specify the following extraction rules:
+	//
+	// extract:
+	//   labels:
+	//     - tag_name: $$1
+	//       key_regex: kubernetes.io/(.*)
+	//
+	// this will add the `component` and `version` tags to the spans or metrics.
+	TagName string `mapstructure:"tag_name"`
+
+	// Key represents the annotation (or label) name. This must exactly match an annotation (or label) name.
+	Key string `mapstructure:"key"`
+	// KeyRegex is a regular expression used to extract a Key that matches the regex.
+	// Out of Key or KeyRegex, only one option is expected to be configured at a time.
+	KeyRegex string `mapstructure:"key_regex"`
+}

receiver/k8seventsreceiver/config_test.go

@@ -39,6 +39,16 @@ func TestLoadConfig(t *testing.T) {
 				APIConfig: k8sconfig.APIConfig{
 					AuthType: k8sconfig.AuthTypeServiceAccount,
 				},
+				Extract: ExtractConfig{
+					Annotations: []FieldExtractConfig{
+						{TagName: "a1", Key: "annotation-one"},
+						{TagName: "a2", KeyRegex: "annotation-two"},
+					},
+					Labels: []FieldExtractConfig{
+						{TagName: "l1", Key: "label1"},
+						{TagName: "l2", KeyRegex: "label2"},
+					},
+				},
 			},
 		},
 	}

receiver/k8seventsreceiver/factory.go

@@ -40,3 +40,14 @@ func createLogsReceiver(
 
 	return newReceiver(params, rCfg, consumer)
 }
+
+func createReceiverOpts(cfg component.Config) []option {
+	oCfg := cfg.(*Config)
+	var opts []option
+
+	// extraction rules
+	opts = append(opts, withExtractLabels(oCfg.Extract.Labels...))
+	opts = append(opts, withExtractAnnotations(oCfg.Extract.Annotations...))
+
+	return opts
+}

receiver/k8seventsreceiver/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.124.1
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v1.30.0
+	go.opentelemetry.io/collector/component/componentstatus v0.124.0
 	go.opentelemetry.io/collector/component/componenttest v0.124.0
 	go.opentelemetry.io/collector/confmap v1.30.0
 	go.opentelemetry.io/collector/confmap/xconfmap v0.124.0

receiver/k8seventsreceiver/go.sum

@@ -0,0 +1,66 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package kube // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8seventsreceiver/internal/kube"
+
+import (
+	"fmt"
+	"regexp"
+)
+
+// ExtractionRules is used to specify the information that needs to be extracted
+// from event and added to the spans as tags.
+type ExtractionRules struct {
+	Annotations []FieldExtractionRule
+	Labels      []FieldExtractionRule
+}
+
+// FieldExtractionRule is used to specify which fields to extract from event fields
+// and inject into spans as attributes.
+type FieldExtractionRule struct {
+	// Name is used to as the Span tag name.
+	Name string
+	// Key is used to lookup k8s event fields.
+	Key string
+	// KeyRegex is a regular expression(full length match) used to extract a Key that matches the regex.
+	KeyRegex             *regexp.Regexp
+	HasKeyRegexReference bool
+	// Regex is a regular expression used to extract a sub-part of a field value.
+	// Full value is extracted when no regexp is provided.
+	Regex *regexp.Regexp
+}
+
+func (r *FieldExtractionRule) ExtractTagsFromMetadata(metadata map[string]string, formatter string) map[string]string {
+	tags := make(map[string]string)
+	if r.KeyRegex != nil {
+		for k, v := range metadata {
+			if r.KeyRegex.MatchString(k) && v != "" {
+				var name string
+				if r.HasKeyRegexReference {
+					var result []byte
+					name = string(r.KeyRegex.ExpandString(result, r.Name, k, r.KeyRegex.FindStringSubmatchIndex(k)))
+				} else {
+					name = fmt.Sprintf(formatter, k)
+				}
+				tags[name] = v
+			}
+		}
+	} else if v, ok := metadata[r.Key]; ok {
+		tags[r.Name] = r.extractField(v)
+	}
+	return tags
+}
+
+func (r *FieldExtractionRule) extractField(v string) string {
+	// Check if a subset of the field should be extracted with a regular expression
+	// instead of the whole field.
+	if r.Regex == nil {
+		return v
+	}
+
+	matches := r.Regex.FindStringSubmatch(v)
+	if len(matches) == 2 {
+		return matches[1]
+	}
+	return ""
+}

receiver/k8seventsreceiver/internal/kube/kube.go

@@ -11,6 +11,8 @@ import (
 	semconv "go.opentelemetry.io/collector/semconv/v1.27.0"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8seventsreceiver/internal/kube"
 )
 
 const (
@@ -29,7 +31,7 @@ var severityMap = map[string]plog.SeverityNumber{
 }
 
 // k8sEventToLogRecord converts Kubernetes event to plog.LogRecordSlice and adds the resource attributes.
-func k8sEventToLogData(logger *zap.Logger, ev *corev1.Event) plog.Logs {
+func k8sEventToLogData(logger *zap.Logger, ev *corev1.Event, extractionRules ...*kube.ExtractionRules) plog.Logs {
 	ld := plog.NewLogs()
 	rl := ld.ResourceLogs().AppendEmpty()
 	sl := rl.ScopeLogs().AppendEmpty()
@@ -73,6 +75,19 @@ func k8sEventToLogData(logger *zap.Logger, ev *corev1.Event) plog.Logs {
 	attrs.PutStr("k8s.event.uid", string(ev.UID))
 	attrs.PutStr(semconv.AttributeK8SNamespaceName, ev.InvolvedObject.Namespace)
 
+	for _, rules := range extractionRules {
+		for _, r := range rules.Labels {
+			for k, v := range r.ExtractTagsFromMetadata(ev.Labels, "k8s.event.labels.%s") {
+				attrs.PutStr(k, v)
+			}
+		}
+		for _, r := range rules.Annotations {
+			for k, v := range r.ExtractTagsFromMetadata(ev.Annotations, "k8s.event.annotations.%s") {
+				attrs.PutStr(k, v)
+			}
+		}
+	}
+
 	// "Count" field of k8s event will be '0' in case it is
 	// not present in the collected event from k8s.
 	if ev.Count != 0 {

receiver/k8seventsreceiver/k8s_event_to_logdata.go

@@ -9,6 +9,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8seventsreceiver/internal/kube"
 )
 
 func TestK8sEventToLogData(t *testing.T) {
@@ -51,6 +53,68 @@ func TestK8sEventToLogDataWithApiAndResourceVersion(t *testing.T) {
 	assert.Equal(t, "7387066320", attr.AsString())
 }
 
+func TestK8sEventToLogDataWithExtractionRules(t *testing.T) {
+	k8sEvent := getEvent()
+	k8sEvent.Labels["testlabel"] = "test-key-label"
+	k8sEvent.Labels["app.kubernetes.io/by-key"] = "test-key-label"
+	k8sEvent.Labels["app.kubernetes.io/by-regex"] = "test-regex-label"
+	k8sEvent.Annotations["topology.kubernetes.io/by-key"] = "test-key-annotation"
+	k8sEvent.Annotations["topology.kubernetes.io/by-regex"] = "test-regex-annotation"
+
+	labelsRules, err := ExtractFieldRules("labels",
+		FieldExtractConfig{
+			Key: "testlabel",
+		},
+		FieldExtractConfig{
+			TagName: "k8s.event.labels.label_by_key",
+			Key:     "app.kubernetes.io/by-key",
+		},
+		FieldExtractConfig{
+			TagName:  "k8s.event.labels.$1",
+			KeyRegex: "app.kubernetes.io/(.*)",
+		})
+	assert.NoError(t, err)
+
+	annotationsRules, err := ExtractFieldRules("annotations", FieldExtractConfig{
+		TagName: "k8s.event.annotations.annotation_by_key",
+		Key:     "topology.kubernetes.io/by-key",
+	}, FieldExtractConfig{
+		TagName:  "k8s.event.annotations.$1",
+		KeyRegex: "topology.kubernetes.io/(.*)",
+	})
+	assert.NoError(t, err)
+
+	ld := k8sEventToLogData(zap.NewNop(), k8sEvent, &kube.ExtractionRules{
+		Labels:      labelsRules,
+		Annotations: annotationsRules,
+	})
+	assert.NoError(t, err)
+
+	rl := ld.ResourceLogs().At(0)
+	lr := rl.ScopeLogs().At(0)
+	attrs := lr.LogRecords().At(0).Attributes()
+
+	attr, ok := attrs.Get("k8s.event.labels.testlabel")
+	assert.True(t, ok)
+	assert.Equal(t, "test-key-label", attr.AsString())
+
+	attr, ok = attrs.Get("k8s.event.labels.label_by_key")
+	assert.True(t, ok)
+	assert.Equal(t, "test-key-label", attr.AsString())
+
+	attr, ok = attrs.Get("k8s.event.labels.by-regex")
+	assert.True(t, ok)
+	assert.Equal(t, "test-regex-label", attr.AsString())
+
+	attr, ok = attrs.Get("k8s.event.annotations.annotation_by_key")
+	assert.True(t, ok)
+	assert.Equal(t, "test-key-annotation", attr.AsString())
+
+	attr, ok = attrs.Get("k8s.event.annotations.by-regex")
+	assert.True(t, ok)
+	assert.Equal(t, "test-regex-annotation", attr.AsString())
+}
+
 func TestUnknownSeverity(t *testing.T) {
 	k8sEvent := getEvent()
 	k8sEvent.Type = "Unknown"