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

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

.github/ISSUE_TEMPLATE/bug_report.yaml

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

.github/ISSUE_TEMPLATE/feature_request.yaml

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

.github/ISSUE_TEMPLATE/other.yaml

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

receiver/k8slogreceiver/Makefile

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

receiver/k8slogreceiver/config.go

@@ -0,0 +1,300 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package k8slogreceiver
+
+import (
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+	"go.uber.org/multierr"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/consumerretry"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
+)
+
+const (
+	ModeDaemonSetStdout = "daemonset-stdout"
+	ModeDaemonSetFile   = "daemonset-file"
+	ModeSidecar         = "sidecar"
+	StartAtBeginning    = "beginning"
+	StartAtEnd          = "end"
+)
+
+const (
+	DefaultFingerprintSize             = 1000
+	DefaultStartAt                     = StartAtEnd
+	DefaultEncoding                    = "utf-8"
+	DefaultMaxLogSize                  = 1024 * 1024
+	DefaultForceFlushPeriod            = 500 * time.Millisecond
+	DefaultPreserveLeadingWhitespaces  = true
+	DefaultPreserveTrailingWhitespaces = false
+	DefaultMaxReaders                  = 1000
+	DefaultIncludeFileName             = true
+	DefaultIncludeFilePath             = true
+	DefaultMode                        = ModeDaemonSetStdout
+	DefaultHostRoot                    = "/host_root"
+	DefaultNodeFromEnv                 = "K8S_NODE_NAME"
+)
+
+// Config is the configuration of a k8s input operator
+type Config struct {
+	Discovery K8sSourceConfig `mapstructure:"discovery"`
+	Extract   ExtractConfig   `mapstructure:"extract"`
+
+	// FingerprintSize represents the size of the fingerprint.
+	FingerprintSize int `mapstructure:"fingerprint_size"`
+
+	// Include represents a list of globs to include files.
+	Include []string `mapstructure:"include"`
+
+	// Exclude represents a list of globs to exclude files.
+	Exclude []string `mapstructure:"exclude"`
+
+	// IncludeRegex represents a list of regex to include files.
+	ExcludeRegex []string `mapstructure:"exclude_regex"`
+
+	// IncludeFileName represents whether to include file name in the emitted entries.
+	IncludeFileName bool `mapstructure:"include_file_name"`
+
+	// IncludeFilePath represents whether to include file path in the emitted entries.
+	IncludeFilePath bool `mapstructure:"include_file_path"`
+
+	// MaxReaders represents the maximum number of readers.
+	MaxReaders int `mapstructure:"max_readers"`
+
+	// PollInterval represents the interval to poll files.
+	PollInterval time.Duration `mapstructure:"poll_interval"`
+
+	// StartAt represents the starting point of the reader.
+	StartAt string `mapstructure:"start_at"`
+
+	// ForceFlushPeriod represents the period of force flushing even without linebreak.
+	ForceFlushPeriod time.Duration `mapstructure:"force_flush_period"`
+
+	// Encoding represents the encoding of the reader.
+	Encoding string `mapstructure:"encoding"`
+
+	// MaxLogSize represents the max size of the log.
+	MaxLogSize int `mapstructure:"max_log_size"`
+
+	// PreserveLeadingWhitespaces represents whether to preserve leading whitespaces.
+	PreserveLeadingWhitespaces bool `mapstructure:"preserve_leading_whitespaces"`
+
+	// PreserveTrailingWhitespaces represents whether to preserve trailing whitespaces.
+	PreserveTrailingWhitespaces bool `mapstructure:"preserve_trailing_whitespaces"`
+
+	// Resource represents Additional resource attributes
+	Resource map[string]string `mapstructure:"resource"`
+
+	// Attributes represents Additional attributes
+	Attributes map[string]string `mapstructure:"attributes"`
+
+	// Storage represents the component id of storage extension.
+	Storage *component.ID `mapstructure:"storage"`
+
+	// RetryOnFailure represents the retry on failure configuration.
+	RetryOnFailure consumerretry.Config `mapstructure:"retry_on_failure"`
+
+	// Operator represents the list of operators to apply to the input.
+	Operators []operator.Config `mapstructure:"operators"`
+}
+
+// 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 {
+	err := c.Discovery.Validate()
+	if c.Discovery.Mode != ModeDaemonSetStdout &&
+		len(c.Include) == 0 {
+		err = multierr.Append(err, fmt.Errorf("include must be specified when discovery mode is not daemonset_stdout"))
+	}
+	if c.StartAt != StartAtBeginning && c.StartAt != StartAtEnd {
+		err = multierr.Append(err, fmt.Errorf("invalid start_at: %s", c.StartAt))
+	}
+	return err
+}
+
+// K8sSourceConfig allows specifying how to discover containers to collect logs from.
+type K8sSourceConfig 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.
+	// - "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 K8sSourceConfig) Validate() error {
+	var err error
+	if c.Mode != ModeDaemonSetStdout && c.Mode != ModeDaemonSetFile && c.Mode != ModeSidecar {
+		return fmt.Errorf("invalid mode %q", c.Mode)
+	}
+	if c.Mode != ModeSidecar {
+		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"`
+
+	// Expr represents the rules to filter containers based on a custom expression.
+	// TODO: define the values passed to expr.
+	Expr string `mapstructure:"expr"`
+}
+
+// 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"`
+}