feat: initial commit for k8slog receiver

.chloggen/k8slog_receiver_setup.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: k8slogreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add the skeleton for the new k8slogreceiver in development."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [23339]
+
+# (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:

.github/CODEOWNERS

@@ -210,6 +210,7 @@ receiver/jmxreceiver/                                    @open-telemetry/collect
 receiver/journaldreceiver/                               @open-telemetry/collector-contrib-approvers @sumo-drosiek @djaglowski
 receiver/k8sclusterreceiver/                             @open-telemetry/collector-contrib-approvers @dmitryax @TylerHelmuth
 receiver/k8seventsreceiver/                              @open-telemetry/collector-contrib-approvers @dmitryax @TylerHelmuth
+receiver/k8slogreceiver/                                 @open-telemetry/collector-contrib-approvers @h0cheung @TylerHelmuth
 receiver/k8sobjectsreceiver/                             @open-telemetry/collector-contrib-approvers @dmitryax @hvaghani221 @TylerHelmuth
 receiver/kafkametricsreceiver/                           @open-telemetry/collector-contrib-approvers @dmitryax
 receiver/kafkareceiver/                                  @open-telemetry/collector-contrib-approvers @pavolloffay @MovieStoreGuy

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -198,6 +198,7 @@ body:
       - receiver/journald
       - receiver/k8scluster
       - receiver/k8sevents
+      - receiver/k8slog
       - receiver/k8sobjects
       - receiver/kafka
       - receiver/kafkametrics

.github/ISSUE_TEMPLATE/feature_request.yaml

@@ -192,6 +192,7 @@ body:
       - receiver/journald
       - receiver/k8scluster
       - receiver/k8sevents
+      - receiver/k8slog
       - receiver/k8sobjects
       - receiver/kafka
       - receiver/kafkametrics

.github/ISSUE_TEMPLATE/other.yaml

@@ -192,6 +192,7 @@ body:
       - receiver/journald
       - receiver/k8scluster
       - receiver/k8sevents
+      - receiver/k8slog
       - receiver/k8sobjects
       - receiver/kafka
       - receiver/kafkametrics

cmd/githubgen/allowlist.txt

@@ -20,4 +20,5 @@ oded-dd
 shaochengwang
 svrakitin
 thepeterstone
-yiyang5055
+yiyang5055
+h0cheung

receiver/k8slogreceiver/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

receiver/k8slogreceiver/config.go

@@ -0,0 +1,219 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package k8slogreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8slogreceiver"
+
+import (
+	"fmt"
+
+	"go.uber.org/multierr"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
+)
+
+const (
+	ModeDaemonSetStdout = "daemonset-stdout"
+)
+
+const (
+	DefaultMode        = ModeDaemonSetStdout
+	DefaultHostRoot    = "/host_root"
+	DefaultNodeFromEnv = "K8S_NODE_NAME"
+)
+
+// Config is the configuration of a k8slog receiver
+type Config struct {
+	Discovery SourceConfig  `mapstructure:"discovery"`
+	Extract   ExtractConfig `mapstructure:"extract"`
+
+	// TODO: refactor fileconsumer and add it's config of k8s implementation here.
+}
+
+// ExtractConfig allows specifying how to extract resource attributes from pod.
+type ExtractConfig struct {
+	// Metadata represents the list of metadata fields to extract from pod.
+	// TODO: supported metadata fields and default values.
+	Metadata []string `mapstructure:"metadata"`
+
+	// Annotations represents the rules to extract from pod annotations.
+	Annotations []FieldExtractConfig `mapstructure:"annotations"`
+
+	// Labels represents the rules to extract from pod labels.
+	Labels []FieldExtractConfig `mapstructure:"labels"`
+
+	// Env represents the rules to extract from container environment variables.
+	Env []FieldExtractConfig `mapstructure:"env"`
+
+	// OtelEnv represents the rules to extract from environment variables of otel itself.
+	OtelEnv []FieldExtractConfig `mapstructure:"otel_env"`
+}
+
+// FieldExtractConfig allows specifying an extraction rule to extract a resource attribute from pod (or namespace)
+// annotations (or labels).
+// This is a copy of the config from the k8sattributes processor.
+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.pod.annotations.<annotation key>
+	//   - k8s.pod.labels.<label key>
+	//   - k8s.pod.env.<env key>
+	//   - otel.env.<env key>
+	// For example, if tag_name is not specified and the key is git_sha,
+	// then the attribute name will be `k8s.pod.annotations.git_sha`.
+	// When key_regex is present, tag_name supports back reference to both named capturing and positioned capturing.
+	// For example, if your pod 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 key (annotation, label or etc.) name. It uses exact match.
+	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"`
+
+	// Regex is an optional field used to extract a sub-string from a complex field value.
+	// The supplied regular expression must contain one named parameter with the string "value"
+	// as the name. For example, if your pod spec contains the following annotation,
+	//
+	// kubernetes.io/change-cause: 2019-08-28T18:34:33Z APP_NAME=my-app GIT_SHA=58a1e39 CI_BUILD=4120
+	//
+	// and you'd like to extract the GIT_SHA and the CI_BUILD values as tags, then you must
+	// specify the following two extraction rules:
+	//
+	// extract:
+	//   annotations:
+	//     - tag_name: git.sha
+	//       key: kubernetes.io/change-cause
+	//       regex: GIT_SHA=(?P<value>\w+)
+	//     - tag_name: ci.build
+	//       key: kubernetes.io/change-cause
+	//       regex: JENKINS=(?P<value>[\w]+)
+	//
+	// this will add the `git.sha` and `ci.build` resource attributes.
+	Regex string `mapstructure:"regex"`
+}
+
+func (c Config) Validate() error {
+	return c.Discovery.Validate()
+}
+
+// SourceConfig allows specifying how to discover containers to collect logs from.
+type SourceConfig struct {
+	// Mode represents the mode of the k8slog receiver.
+	// Valid values are:
+	// - "daemonset-stdout": (default) otel is deployed as a daemonset and collects logs from stdout of containers.
+	//
+	// Will be supported in the future:
+	// - "daemonset-file": otel is deployed as a daemonset and collects logs from files inside containers.
+	// - "sidecar": otel is deployed as a sidecar and collects logs from files.
+	Mode string `mapstructure:"mode"`
+
+	// NodeFromEnv represents the environment variable which contains the node name.
+	NodeFromEnv string `mapstructure:"node_from_env"`
+
+	// HostRoot represents the path which is used to mount the host's root filesystem.
+	HostRoot string `mapstructure:"host_root"`
+
+	// K8sAPI represents the configuration for the k8s API.
+	K8sAPI k8sconfig.APIConfig `mapstructure:"k8s_api"`
+
+	// RuntimeAPIs represents the configuration for the runtime APIs.
+	RuntimeAPIs []RuntimeAPIConfig `mapstructure:"runtime_apis"`
+
+	Filter []FilterConfig `mapstructure:"filter"`
+}
+
+func (c SourceConfig) Validate() error {
+	var err error
+	if c.Mode != ModeDaemonSetStdout {
+		return fmt.Errorf("invalid mode %q", c.Mode)
+	}
+	if c.HostRoot == "" {
+		err = multierr.Append(err, fmt.Errorf("host_root must be specified when mode is %q", c.Mode))
+	}
+	err = multierr.Append(err, c.K8sAPI.Validate())
+	for _, r := range c.RuntimeAPIs {
+		err = multierr.Append(err, r.Validate())
+	}
+	return err
+}
+
+// FilterConfig allows specifying how to filter containers to collect logs from.
+// By default, all containers are collected from.
+type FilterConfig struct {
+	// Annotations represents the rules to filter containers based on pod annotations.
+	Annotations []MapFilterConfig `mapstructure:"annotations"`
+
+	// Labels represents the rules to filter containers based on pod labels.
+	Labels []MapFilterConfig `mapstructure:"labels"`
+
+	// Env represents the rules to filter containers based on pod environment variables.
+	Env []MapFilterConfig `mapstructure:"env"`
+
+	// Namespaces represents the rules to filter containers based on pod namespaces.
+	Namespaces []ValueFilterConfig `mapstructure:"namespaces"`
+
+	// Containers represents the rules to filter containers based on container names.
+	Containers []ValueFilterConfig `mapstructure:"containers"`
+
+	// Pods represents the rules to filter containers based on pod names.
+	Pods []ValueFilterConfig `mapstructure:"pods"`
+
+	// UIDs represents the rules to filter containers based on pod UIDs.
+	UIDs []ValueFilterConfig `mapstructure:"uids"`
+
+	// FilterStatements represents the ottl statements to filter containers.
+	// TODO: define the context.
+	FilterStatements []string `mapstructure:"filter_statements"`
+}
+
+// ValueFilterConfig allows specifying a filter rule to filter containers based on string values,
+// such as pod names, namespaces, container names or pod UIDs.
+// If any of the values match, this rule is considered to match.
+type ValueFilterConfig struct {
+	// Op represents how to compare the value.
+	// Valid values are:
+	// - "equals": (default) the value must be equal to the specified value.
+	// - "not-equals": the value must not be equal to the specified value.
+	// - "matches": the value must match the specified regular expression.
+	// - "not-matches": the value must not match the specified regular expression.
+	Op string `mapstructure:"op"`
+
+	// Value represents the value to compare against.
+	Value string `mapstructure:"value"`
+}
+
+// MapFilterConfig allows specifying a filter rule to filter containers based on key value pairs,
+// such as pod annotations, labels or environment variables.
+// Only if all the keys match, this rule is considered to match.
+type MapFilterConfig struct {
+	// Op represents how to compare the values.
+	// Valid values are:
+	// - "equals": (default) the value must be equal to the specified value.
+	// - "not-equals": the value must not be equal to the specified value.
+	// - "exists": the value must exist.
+	// - "not-exists": the value must not exist.
+	// - "matches": the value must match the specified regular expression.
+	// - "not-matches": the value must not match the specified regular expression.
+	Op string `mapstructure:"op"`
+
+	// Key represents the key to compare against.
+	Key string `mapstructure:"key"`
+
+	// Value represents the value to compare against.
+	// If Op is "exists" or "not-exists", this field is ignored.
+	// If any of the values match, this rule is considered to match.
+	Value string `mapstructure:"value"`
+}

receiver/k8slogreceiver/config_test.go

@@ -0,0 +1,102 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package k8slogreceiver
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8slogreceiver/internal/metadata"
+)
+
+func TestLoadConfig(t *testing.T) {
+	t.Parallel()
+
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
+	require.NoError(t, err)
+
+	tests := []struct {
+		id          component.ID
+		expected    component.Config
+		expectedErr error
+	}{
+		{
+			id:       component.NewIDWithName(metadata.Type, "default"),
+			expected: createDefaultConfig(),
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "ds-stdout"),
+			expected: &Config{
+				Discovery: SourceConfig{
+					K8sAPI:      k8sconfig.APIConfig{AuthType: "serviceAccount"},
+					Mode:        ModeDaemonSetStdout,
+					NodeFromEnv: DefaultNodeFromEnv,
+					HostRoot:    "/host_root",
+					RuntimeAPIs: []RuntimeAPIConfig{
+						{
+							&DockerConfig{
+								baseRuntimeAPIConfig: baseRuntimeAPIConfig{
+									Type: "docker",
+								},
+								Addr:           "unix:///host_root/var/run/docker.sock",
+								ContainerdAddr: "unix:///host_root/run/containerd/containerd.sock",
+							},
+						},
+						{
+							&CRIConfig{
+								baseRuntimeAPIConfig: baseRuntimeAPIConfig{
+									Type: "cri",
+								},
+								Addr:            "unix:///host_root/run/containerd/containerd.sock",
+								ContainerdState: "/host_root/run/containerd",
+							},
+						},
+					},
+					Filter: []FilterConfig{
+						{
+							Annotations: []MapFilterConfig{
+								{
+									Op:  "exists",
+									Key: "io.opentelemetry.collectlog",
+								},
+							},
+							Namespaces: []ValueFilterConfig{
+								{
+									Op:    "not-equals",
+									Value: "kube-system",
+								},
+							},
+						},
+					},
+				},
+				Extract: ExtractConfig{
+					Metadata: []string{
+						"k8s.pod.name",
+						"k8s.pod.uid",
+						"k8s.container.name",
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.id.String(), func(t *testing.T) {
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig().(*Config)
+
+			sub, err := cm.Sub(tt.id.String())
+			require.NoError(t, err)
+			require.NoError(t, component.UnmarshalConfig(sub, cfg))
+			assert.NoError(t, component.ValidateConfig(cfg))
+			assert.Equal(t, tt.expected, cfg)
+		})
+	}
+}