From 363c83709d86bea83bba79e1c5941308b75c84ec Mon Sep 17 00:00:00 2001
From: Christos Markou <chrismarkou92@gmail.com>
Date: Wed, 8 Jan 2025 00:40:09 +0200
Subject: [PATCH] [receiver/receiver_creator] Add support for enabling logs'
 collecting from K8s hints (#36581)

#### Description

This PR adds the logs part for
https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/34427
based on the design decided at
https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/34427#issuecomment-2436140051.

See the README docs for the description of this feature:
https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/35617/files#diff-4127365c4062a7510fb7fede0fa239e9232549732898303d94c12fef0433d39d

#### Link to tracking issue
Fixes
https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/34427

#### Testing
Added unit-tests

#### Documentation
Added README section

#### How to test this manually

1. Deploy the Collector helm chart:
```yaml
mode: daemonset

image:
  repository: otelcontribcol-dev
  tag: "latest"
  pullPolicy: IfNotPresent

command:
  name: otelcontribcol

clusterRole:
  create: true
  rules:
   - apiGroups:
     - ''
     resources:
     - 'pods'
     - 'nodes'
     verbs:
     - 'get'
     - 'list'
     - 'watch'
   - apiGroups: [ "" ]
     resources: [ "nodes/proxy"]
     verbs: [ "get" ]
   - apiGroups:
       - ""
     resources:
       - nodes/stats
     verbs:
       - get
   - nonResourceURLs:
       - "/metrics"
     verbs:
       - get

extraVolumeMounts:
 - name: varlogpods
   mountPath: /var/log/pods
   readOnly: true

extraVolumes:
  - name: varlogpods
    hostPath:
      path: /var/log/pods

config:
  extensions:
    k8s_observer:
      auth_type: serviceAccount
      node: ${env:K8S_NODE_NAME}
      observe_nodes: true
  exporters:
    debug:
      verbosity: detailed

  receivers:
    receiver_creator/metrics:
      watch_observers: [ k8s_observer ]
      discovery:
        enabled: true
        ignore_receivers:
          - nginx2
      receivers:

    receiver_creator/logs:
      watch_observers: [ k8s_observer ]
      discovery:
        enabled: true
        default_logs_discovery: false
      receivers:


  service:
    extensions: [health_check, k8s_observer]
    telemetry:
      logs:
        level: INFO
    pipelines:
      metrics:
        receivers: [ receiver_creator/metrics ]
        processors: [ batch ]
        exporters: [ debug ]
      logs/discovery:
        receivers: [ receiver_creator/logs ]
        #processors: [ batch ]
        exporters: [ debug ]
```
2. Then deploy a target Pod with 2 containers:
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: redis-deployment
  labels:
    app: redis
spec:
  replicas: 1
  selector:
    matchLabels:
      app: redis
  template:
    metadata:
      labels:
        app: redis
      annotations:
        io.opentelemetry.discovery.metrics.6379/enabled: "true"
        io.opentelemetry.discovery.metrics.6379/scraper: redis
        io.opentelemetry.discovery.metrics.6379/signals: metrics
        io.opentelemetry.discovery.metrics.6379/config: |
          collection_interval: "20s"
          timeout: "10s"

        io.opentelemetry.discovery.logs.busybox/enabled: "true"
        io.opentelemetry.discovery.logs.busybox/config: |
          operators:
            - id: some
              type: add
              field: attributes.tag
              value: hints
    spec:
      containers:
        - image: redis
          imagePullPolicy: IfNotPresent
          name: redis
          ports:
            - name: redis
              containerPort: 6379
              protocol: TCP
        - name: busybox
          image: busybox
          args:
            - /bin/sh
            - -c
            - while true; do echo "otel logs at $(date +%H:%M:%S)" && sleep 15s; done
```
3. Esnure that logs are collected from both containers and that Redis
metrics are collected from the Redis container:
```console
2024-11-28T11:04:14.921Z	info	receivercreator@v0.114.0/observerhandler.go:201	starting receiver	{"kind": "receiver", "name": "receiver_creator/metrics", "data_type": "metrics", "name": "redis/91ec7d5c-c6fb-4977-9dbb-c24a85101326_6379", "endpoint": "10.244.0.6:6379", "endpoint_id": "k8s_observer/91ec7d5c-c6fb-4977-9dbb-c24a85101326/redis(6379)", "config": {"collection_interval":"20s","endpoint":"10.244.0.6:6379","timeout":"10s"}}
2024-11-28T11:04:14.921Z	info	receivercreator@v0.114.0/observerhandler.go:201	starting receiver	{"kind": "receiver", "name": "receiver_creator/logs", "data_type": "logs", "name": "filelog/91ec7d5c-c6fb-4977-9dbb-c24a85101326_busybox", "endpoint": "10.244.0.6", "endpoint_id": "k8s_observer/91ec7d5c-c6fb-4977-9dbb-c24a85101326/busybox", "config": {"include":["/var/log/pods/default_redis-deployment-7777bf7db4-5rm6d_91ec7d5c-c6fb-4977-9dbb-c24a85101326/busybox/*.log"],"include_file_name":false,"include_file_path":true,"operators":[{"id":"container-parser","type":"container"},{"field":"attributes.tag","id":"some","type":"add","value":"hints"}]}}
2024-11-28T11:04:14.922Z	info	adapter/receiver.go:41	Starting stanza receiver	{"kind": "receiver", "name": "receiver_creator/logs", "data_type": "logs", "name": "filelog/91ec7d5c-c6fb-4977-9dbb-c24a85101326_busybox/receiver_creator/logs{endpoint=\"10.244.0.6\"}/k8s_observer/91ec7d5c-c6fb-4977-9dbb-c24a85101326/busybox"}
2024-11-28T11:04:15.122Z	info	fileconsumer/file.go:265	Started watching file	{"kind": "receiver", "name": "receiver_creator/logs", "data_type": "logs", "name": "filelog/91ec7d5c-c6fb-4977-9dbb-c24a85101326_busybox/receiver_creator/logs{endpoint=\"10.244.0.6\"}/k8s_observer/91ec7d5c-c6fb-4977-9dbb-c24a85101326/busybox", "component": "fileconsumer", "path": "/var/log/pods/default_redis-deployment-7777bf7db4-5rm6d_91ec7d5c-c6fb-4977-9dbb-c24a85101326/busybox/0.log"}
2024-11-28T11:04:15.979Z	info	Metrics	{"kind": "exporter", "data_type": "metrics", "name": "debug/2", "resource metrics": 1, "metrics": 26, "data points": 31}
```

### Follow-ups

1. File an issue for enhancing default behaviors:
https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/36581#discussion_r1864013358

---------

Signed-off-by: ChrsMark <chrismarkou92@gmail.com>
---
 .chloggen/f_hints_logs.yaml                   |  27 ++
 receiver/receivercreator/README.md            |  99 ++++-
 receiver/receivercreator/config_test.go       |  37 ++
 receiver/receivercreator/discovery.go         |  98 ++++-
 receiver/receivercreator/discovery_test.go    | 365 +++++++++++++++++-
 receiver/receivercreator/fixtures_test.go     |  16 +
 receiver/receivercreator/observerhandler.go   |  17 +-
 .../receivercreator/observerhandler_test.go   |  70 ++++
 8 files changed, 714 insertions(+), 15 deletions(-)
 create mode 100644 .chloggen/f_hints_logs.yaml

diff --git a/.chloggen/f_hints_logs.yaml b/.chloggen/f_hints_logs.yaml
new file mode 100644
index 000000000000..9e4b274aefbf
--- /dev/null
+++ b/.chloggen/f_hints_logs.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: receivercreator
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for starting logs' collection based on provided k8s annotations' hints
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [34427]
+
+# (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]
diff --git a/receiver/receivercreator/README.md b/receiver/receivercreator/README.md
index ac836eb98e02..4fe12d38bcbe 100644
--- a/receiver/receivercreator/README.md
+++ b/receiver/receivercreator/README.md
@@ -458,7 +458,8 @@ receiver_creator/metrics:
      # ignore_receivers: []
 ```
 
-Find bellow the supported annotations that user can define to automatically enable receivers to start collecting metrics signals from the target Pods/containers.
+Find bellow the supported annotations that user can define to automatically enable receivers to start
+collecting metrics and logs signals from the target Pods/containers.
 
 ### Supported metrics annotations
 
@@ -511,11 +512,76 @@ The current implementation relies on the implementation of `k8sobserver` extensi
 the [pod_endpoint](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.111.0/extension/observer/k8sobserver/pod_endpoint.go).
 The hints are evaluated per container by extracting the annotations from each [`Port` endpoint](#Port) that is emitted. 
 
+### Supported logs annotations
+
+This feature enables `filelog` receiver in order to collect logs from the discovered Pods.
+
+#### Enable/disable discovery
+
+`io.opentelemetry.discovery.logs/enabled` (Required. Example: `"true"`)
+
+By default `"false"`.
+
+#### Define configuration
+
+The default configuration for the `filelog` receiver is the following:
+
+```yaml
+include:
+  - /var/log/pods/`pod.namespace`_`pod.name`_`pod.uid`/`container_name`/*.log
+include_file_name: false
+include_file_path: true
+operators:
+  - id: container-parser
+    type: container
+```
+This default can be extended or overridden using the respective annotation:
+`io.opentelemetry.discovery.logs/config`
+
+**Example:**
+
+```yaml
+io.opentelemetry.discovery.logs/config: |
+  include_file_name: true
+  max_log_size: "2MiB"
+  operators:
+    - type: container
+      id: container-parser
+    - type: regex_parser
+      regex: "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"
+```
+
+`include` cannot be overridden and is fixed to discovered container's log file path.
+
+#### Support multiple target containers
+
+Users can target the annotation to a specific container by suffixing it with the name of that container:
+`io.opentelemetry.discovery.logs.<container_name>/endpoint`.
+For example:
+```yaml
+io.opentelemetry.discovery.logs.busybox/config: |
+  max_log_size: "3MiB"
+  operators:
+    - type: container
+      id: container-parser
+    - id: some
+      type: add
+      field: attributes.tag
+      value: hints
+```
+where `busybox` is the name of the target container.
+
+If a Pod is annotated with both container level hints and pod level hints the container level hints have priority and
+the Pod level hints are used as a fallback (see detailed example bellow).
+
+The current implementation relies on the implementation of `k8sobserver` extension and specifically
+the [pod_endpoint](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.111.0/extension/observer/k8sobserver/pod_endpoint.go).
+The hints are evaluated per container by extracting the annotations from each [`Pod Container` endpoint](#Pod Container) that is emitted.
 
 
 ### Examples
 
-#### Metrics example
+#### Metrics and Logs example
 
 Collector's configuration:
 ```yaml
@@ -525,12 +591,22 @@ receivers:
     discovery:
       enabled: true
     receivers:
+  
+  receiver_creator/logs:
+    watch_observers: [ k8s_observer ]
+    discovery:
+      enabled: true
+    receivers:
 
 service:
   extensions: [ k8s_observer]
   pipelines:
     metrics:
-      receivers: [ receiver_creator ]
+      receivers: [ receiver_creator/metrics ]
+      processors: []
+      exporters: [ debug ]
+    logs:
+      receivers: [ receiver_creator/logs ]
       processors: []
       exporters: [ debug ]
 ```
@@ -600,6 +676,23 @@ spec:
           endpoint: "http://`endpoint`/nginx_status"
           collection_interval: "30s"
           timeout: "20s"
+
+        # redis pod container logs hints
+        io.opentelemetry.discovery.logs.redis/enabled: "true"
+        io.opentelemetry.discovery.logs.redis/config: |
+          max_log_size: "4MiB"
+          operators:
+            - type: container
+              id: container-parser
+            - id: some
+              type: add
+              field: attributes.tag
+              value: logs_hints
+
+        # nginx pod container logs hints
+        io.opentelemetry.discovery.logs.webserver/enabled: "true"
+        io.opentelemetry.discovery.logs.webserver/config: |
+          max_log_size: "3MiB"
     spec:
       volumes:
         - name: nginx-conf
diff --git a/receiver/receivercreator/config_test.go b/receiver/receivercreator/config_test.go
index 618add4e174d..1bb751939140 100644
--- a/receiver/receivercreator/config_test.go
+++ b/receiver/receivercreator/config_test.go
@@ -299,3 +299,40 @@ func (*nopWithoutEndpointFactory) CreateTraces(
 		cfg:      cfg,
 	}, nil
 }
+
+type nopWithFilelogConfig struct {
+	Include         []string `mapstructure:"include"`
+	IncludeFileName bool     `mapstructure:"include_file_name"`
+	IncludeFilePath bool     `mapstructure:"include_file_path"`
+	Operators       []any    `mapstructure:"operators"`
+}
+
+type nopWithFilelogFactory struct {
+	rcvr.Factory
+}
+
+type nopWithFilelogReceiver struct {
+	mockComponent
+	consumer.Logs
+	consumer.Metrics
+	consumer.Traces
+	rcvr.Settings
+	cfg component.Config
+}
+
+func (*nopWithFilelogFactory) CreateDefaultConfig() component.Config {
+	return &nopWithFilelogConfig{}
+}
+
+func (*nopWithFilelogFactory) CreateLogs(
+	_ context.Context,
+	rcs rcvr.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Logs,
+) (rcvr.Logs, error) {
+	return &nopWithEndpointReceiver{
+		Logs:     nextConsumer,
+		Settings: rcs,
+		cfg:      cfg,
+	}, nil
+}
diff --git a/receiver/receivercreator/discovery.go b/receiver/receivercreator/discovery.go
index f8a694912751..1908be3da7f5 100644
--- a/receiver/receivercreator/discovery.go
+++ b/receiver/receivercreator/discovery.go
@@ -21,11 +21,15 @@ const (
 
 	// hint suffix for metrics
 	otelMetricsHints = otelHints + ".metrics"
+	otelLogsHints    = otelHints + ".logs"
 
 	// hints definitions
 	discoveryEnabledHint = "enabled"
 	scraperHint          = "scraper"
 	configHint           = "config"
+
+	logsReceiver          = "filelog"
+	defaultLogPathPattern = "/var/log/pods/%s_%s_%s/%s/*.log"
 )
 
 // k8sHintsBuilder creates configurations from hints provided as Pod's annotations.
@@ -57,7 +61,7 @@ func (builder *k8sHintsBuilder) createReceiverTemplateFromHints(env observer.End
 		return nil, fmt.Errorf("could not get endpoint type: %v", zap.Any("env", env))
 	}
 
-	if endpointType != string(observer.PortType) {
+	if endpointType != string(observer.PortType) && endpointType != string(observer.PodContainerType) {
 		return nil, nil
 	}
 
@@ -72,7 +76,14 @@ func (builder *k8sHintsBuilder) createReceiverTemplateFromHints(env observer.End
 		return nil, nil
 	}
 
-	return builder.createScraper(pod.Annotations, env)
+	switch endpointType {
+	case string(observer.PortType):
+		return builder.createScraper(pod.Annotations, env)
+	case string(observer.PodContainerType):
+		return builder.createLogsReceiver(pod.Annotations, env)
+	default:
+		return nil, nil
+	}
 }
 
 func (builder *k8sHintsBuilder) createScraper(
@@ -91,7 +102,7 @@ func (builder *k8sHintsBuilder) createScraper(
 	port = p.Port
 	pod := p.Pod
 
-	if !discoveryMetricsEnabled(annotations, otelMetricsHints, fmt.Sprint(port)) {
+	if !discoveryEnabled(annotations, otelMetricsHints, fmt.Sprint(port)) {
 		return nil, nil
 	}
 
@@ -118,6 +129,48 @@ func (builder *k8sHintsBuilder) createScraper(
 	return &recTemplate, err
 }
 
+func (builder *k8sHintsBuilder) createLogsReceiver(
+	annotations map[string]string,
+	env observer.EndpointEnv,
+) (*receiverTemplate, error) {
+	if _, ok := builder.ignoreReceivers[logsReceiver]; ok {
+		// receiver is ignored
+		return nil, nil
+	}
+
+	var containerName string
+	var c observer.PodContainer
+	err := mapstructure.Decode(env, &c)
+	if err != nil {
+		return nil, fmt.Errorf("could not extract pod's container: %v", zap.Any("env", env))
+	}
+	if c.Name == "" {
+		return nil, fmt.Errorf("could not extract container name: %v", zap.Any("container", c))
+	}
+	containerName = c.Name
+	pod := c.Pod
+
+	if !discoveryEnabled(annotations, otelLogsHints, containerName) {
+		return nil, nil
+	}
+
+	subreceiverKey := logsReceiver
+	builder.logger.Debug("handling added hinted receiver", zap.Any("subreceiverKey", subreceiverKey))
+
+	userConfMap := createLogsConfig(
+		annotations,
+		containerName,
+		pod.UID,
+		pod.Name,
+		pod.Namespace,
+		builder.logger)
+
+	recTemplate, err := newReceiverTemplate(fmt.Sprintf("%v/%v_%v", subreceiverKey, pod.UID, containerName), userConfMap)
+	recTemplate.signals = receiverSignals{metrics: false, logs: true, traces: false}
+
+	return &recTemplate, err
+}
+
 func getScraperConfFromAnnotations(
 	annotations map[string]string,
 	defaultEndpoint, scopeSuffix string,
@@ -149,6 +202,43 @@ func getScraperConfFromAnnotations(
 	return conf, nil
 }
 
+func createLogsConfig(
+	annotations map[string]string,
+	containerName, podUID, podName, namespace string,
+	logger *zap.Logger,
+) userConfigMap {
+	scopeSuffix := containerName
+	logPath := fmt.Sprintf(defaultLogPathPattern, namespace, podName, podUID, containerName)
+	cont := []any{map[string]any{"id": "container-parser", "type": "container"}}
+	defaultConfMap := userConfigMap{
+		"include":           []string{logPath},
+		"include_file_path": true,
+		"include_file_name": false,
+		"operators":         cont,
+	}
+
+	configStr, found := getHintAnnotation(annotations, otelLogsHints, configHint, scopeSuffix)
+	if !found || configStr == "" {
+		return defaultConfMap
+	}
+
+	userConf := make(map[string]any)
+	if err := yaml.Unmarshal([]byte(configStr), &userConf); err != nil {
+		logger.Debug("could not unmarshal configuration from hint", zap.Error(err))
+	}
+
+	for k, v := range userConf {
+		if k == "include" {
+			// path cannot be other than the one of the target container
+			logger.Warn("include setting cannot be set through annotation's hints")
+			continue
+		}
+		defaultConfMap[k] = v
+	}
+
+	return defaultConfMap
+}
+
 func getHintAnnotation(annotations map[string]string, hintBase string, hintKey string, suffix string) (string, bool) {
 	// try to scope the hint more on container level by suffixing
 	// with .<port> in case of Port event or # TODO: .<container_name> in case of Pod Container event
@@ -162,7 +252,7 @@ func getHintAnnotation(annotations map[string]string, hintBase string, hintKey s
 	return podLevelHint, ok
 }
 
-func discoveryMetricsEnabled(annotations map[string]string, hintBase string, scopeSuffix string) bool {
+func discoveryEnabled(annotations map[string]string, hintBase string, scopeSuffix string) bool {
 	enabledHint, found := getHintAnnotation(annotations, hintBase, discoveryEnabledHint, scopeSuffix)
 	if !found {
 		return false
diff --git a/receiver/receivercreator/discovery_test.go b/receiver/receivercreator/discovery_test.go
index 982f78de8396..fd94b6af80b2 100644
--- a/receiver/receivercreator/discovery_test.go
+++ b/receiver/receivercreator/discovery_test.go
@@ -215,6 +215,320 @@ password: "changeme"`
 	}
 }
 
+func TestK8sHintsBuilderLogs(t *testing.T) {
+	logger := zaptest.NewLogger(t, zaptest.Level(zap.InfoLevel))
+
+	id := component.ID{}
+	err := id.UnmarshalText([]byte("filelog/pod-2-UID_redis"))
+	assert.NoError(t, err)
+
+	idNginx := component.ID{}
+	err = idNginx.UnmarshalText([]byte("filelog/pod-2-UID_nginx"))
+	assert.NoError(t, err)
+
+	config := `
+include_file_name: true
+max_log_size: "2MiB"
+operators:
+- type: container
+  id: container-parser
+- type: regex_parser
+  regex: "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"`
+	configNginx := `
+include_file_name: true
+max_log_size: "4MiB"
+operators:
+- type: container
+  id: container-parser
+- type: add
+  field: attributes.tag
+  value: beta`
+	configReplaceContainer := `
+operators:
+- type: container
+  id: replaced
+- type: regex_parser
+  regex: "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"`
+	configReplaceInclude := `
+include:
+- /var/log/pod/x/foo.log`
+
+	tests := map[string]struct {
+		inputEndpoint    observer.Endpoint
+		expectedReceiver receiverTemplate
+		wantError        bool
+		ignoreReceivers  []string
+	}{
+		`logs_pod_level_hints_only`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:",
+				Details: &observer.PodContainer{
+					Name: "redis", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + "/enabled": "true",
+							otelLogsHints + "/config":  config,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{
+				receiverConfig: receiverConfig{
+					id: id,
+					config: userConfigMap{
+						"include":           []string{"/var/log/pods/default_pod-2_pod-2-UID/redis/*.log"},
+						"include_file_name": true,
+						"include_file_path": true,
+						"max_log_size":      "2MiB",
+						"operators": []any{
+							map[string]any{"id": "container-parser", "type": "container"},
+							map[string]any{"type": "regex_parser", "regex": "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"},
+						},
+					},
+				}, signals: receiverSignals{metrics: false, logs: true, traces: false},
+			},
+			wantError:       false,
+			ignoreReceivers: []string{},
+		}, `logs_pod_level_hints_only_ignore_receiver`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:",
+				Details: &observer.PodContainer{
+					Name: "redis", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + "/enabled": "true",
+							otelLogsHints + "/config":  config,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{},
+			wantError:        false,
+			ignoreReceivers:  []string{logsReceiver},
+		}, `logs_pod_level_hints_only_defaults`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:6379",
+				Details: &observer.PodContainer{
+					Name: "redis", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + "/enabled": "true",
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{
+				receiverConfig: receiverConfig{
+					id: id,
+					config: userConfigMap{
+						"include":           []string{"/var/log/pods/default_pod-2_pod-2-UID/redis/*.log"},
+						"include_file_name": false,
+						"include_file_path": true,
+						"operators": []any{
+							map[string]any{"id": "container-parser", "type": "container"},
+						},
+					},
+				}, signals: receiverSignals{metrics: false, logs: true, traces: false},
+			},
+			wantError:       false,
+			ignoreReceivers: []string{},
+		}, `logs_container_level_hints`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:6379",
+				Details: &observer.PodContainer{
+					Name: "redis", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + ".redis/enabled": "true",
+							otelLogsHints + ".redis/config":  config,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{
+				receiverConfig: receiverConfig{
+					id: id,
+					config: userConfigMap{
+						"include":           []string{"/var/log/pods/default_pod-2_pod-2-UID/redis/*.log"},
+						"include_file_name": true,
+						"include_file_path": true,
+						"max_log_size":      "2MiB",
+						"operators": []any{
+							map[string]any{"id": "container-parser", "type": "container"},
+							map[string]any{"type": "regex_parser", "regex": "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"},
+						},
+					},
+				}, signals: receiverSignals{metrics: false, logs: true, traces: false},
+			},
+			wantError:       false,
+			ignoreReceivers: []string{},
+		}, `logs_mix_level_hints`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:6379",
+				Details: &observer.PodContainer{
+					Name: "nginx", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + ".nginx/enabled": "true",
+							otelLogsHints + "/config":        config,
+							otelLogsHints + ".nginx/config":  configNginx,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{
+				receiverConfig: receiverConfig{
+					id: idNginx,
+					config: userConfigMap{
+						"include":           []string{"/var/log/pods/default_pod-2_pod-2-UID/nginx/*.log"},
+						"include_file_name": true,
+						"include_file_path": true,
+						"max_log_size":      "4MiB",
+						"operators": []any{
+							map[string]any{"id": "container-parser", "type": "container"},
+							map[string]any{"type": "add", "field": "attributes.tag", "value": "beta"},
+						},
+					},
+				}, signals: receiverSignals{metrics: false, logs: true, traces: false},
+			},
+			wantError:       false,
+			ignoreReceivers: []string{},
+		}, `logs_no_container_error`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4",
+				Details: &observer.PodContainer{
+					Name: "", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + ".nginx/enabled": "true",
+							otelLogsHints + "/config":        config,
+							otelLogsHints + ".nginx/config":  configNginx,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{},
+			wantError:        true,
+			ignoreReceivers:  []string{},
+		}, `logs_pod_level_hints_replace_container_operator`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:",
+				Details: &observer.PodContainer{
+					Name: "redis", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + "/enabled": "true",
+							otelLogsHints + "/config":  configReplaceContainer,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{
+				receiverConfig: receiverConfig{
+					id: id,
+					config: userConfigMap{
+						"include":           []string{"/var/log/pods/default_pod-2_pod-2-UID/redis/*.log"},
+						"include_file_name": false,
+						"include_file_path": true,
+						"operators": []any{
+							map[string]any{"id": "replaced", "type": "container"},
+							map[string]any{"type": "regex_parser", "regex": "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"},
+						},
+					},
+				}, signals: receiverSignals{metrics: false, logs: true, traces: false},
+			},
+			wantError:       false,
+			ignoreReceivers: []string{},
+		}, `logs_pod_level_hints_do_not_replace_path`: {
+			inputEndpoint: observer.Endpoint{
+				ID:     "namespace/pod-2-UID/filelog(redis)",
+				Target: "1.2.3.4:",
+				Details: &observer.PodContainer{
+					Name: "redis", Pod: observer.Pod{
+						Name:      "pod-2",
+						Namespace: "default",
+						UID:       "pod-2-UID",
+						Labels:    map[string]string{"env": "prod"},
+						Annotations: map[string]string{
+							otelLogsHints + "/enabled": "true",
+							otelLogsHints + "/config":  configReplaceInclude,
+						},
+					},
+				},
+			},
+			expectedReceiver: receiverTemplate{
+				receiverConfig: receiverConfig{
+					id: id,
+					config: userConfigMap{
+						"include":           []string{"/var/log/pods/default_pod-2_pod-2-UID/redis/*.log"},
+						"include_file_name": false,
+						"include_file_path": true,
+						"operators": []any{
+							map[string]any{"id": "container-parser", "type": "container"},
+						},
+					},
+				}, signals: receiverSignals{metrics: false, logs: true, traces: false},
+			},
+			wantError:       false,
+			ignoreReceivers: []string{},
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			builder := createK8sHintsBuilder(
+				DiscoveryConfig{
+					Enabled:         true,
+					IgnoreReceivers: test.ignoreReceivers,
+				},
+				logger)
+			env, err := test.inputEndpoint.Env()
+			require.NoError(t, err)
+			subreceiverTemplate, err := builder.createReceiverTemplateFromHints(env)
+			if subreceiverTemplate == nil {
+				require.Equal(t, receiverTemplate{}, test.expectedReceiver)
+				return
+			}
+			if !test.wantError {
+				require.NoError(t, err)
+				require.Equal(t, subreceiverTemplate.receiverConfig.config, test.expectedReceiver.receiverConfig.config)
+				require.Equal(t, subreceiverTemplate.signals, test.expectedReceiver.signals)
+				require.Equal(t, subreceiverTemplate.id, test.expectedReceiver.id)
+			} else {
+				require.Error(t, err)
+			}
+		})
+	}
+}
+
 func TestGetConfFromAnnotations(t *testing.T) {
 	config := `
 endpoint: "0.0.0.0:8080"
@@ -299,7 +613,54 @@ nested_example:
 	}
 }
 
-func TestDiscoveryMetricsEnabled(t *testing.T) {
+func TestCreateLogsConfig(t *testing.T) {
+	config := `
+include_file_name: true
+max_log_size: "2MiB"
+operators:
+- type: container
+  id: container-parser
+- type: regex_parser
+  regex: "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"`
+	tests := map[string]struct {
+		hintsAnn        map[string]string
+		expectedConf    userConfigMap
+		defaultEndpoint string
+	}{
+		"simple_annotation_case": {
+			hintsAnn: map[string]string{
+				"io.opentelemetry.discovery.logs/config": config,
+			}, expectedConf: userConfigMap{
+				"include":           []string{"/var/log/pods/my-ns_my-pod_my-uid/my-container/*.log"},
+				"include_file_name": true,
+				"include_file_path": true,
+				"max_log_size":      "2MiB",
+				"operators": []any{
+					map[string]any{"id": "container-parser", "type": "container"},
+					map[string]any{"type": "regex_parser", "regex": "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"},
+				},
+			}, defaultEndpoint: "1.2.3.4:8080",
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			assert.Equal(
+				t,
+				test.expectedConf,
+				createLogsConfig(
+					test.hintsAnn,
+					"my-container",
+					"my-uid",
+					"my-pod",
+					"my-ns",
+					zaptest.NewLogger(t, zaptest.Level(zap.InfoLevel))),
+			)
+		})
+	}
+}
+
+func TestDiscoveryEnabled(t *testing.T) {
 	config := `
 endpoint: "0.0.0.0:8080"`
 	tests := map[string]struct {
@@ -342,7 +703,7 @@ endpoint: "0.0.0.0:8080"`
 			assert.Equal(
 				t,
 				test.expected,
-				discoveryMetricsEnabled(test.hintsAnn, otelMetricsHints, test.scopeSuffix),
+				discoveryEnabled(test.hintsAnn, otelMetricsHints, test.scopeSuffix),
 			)
 		})
 	}
diff --git a/receiver/receivercreator/fixtures_test.go b/receiver/receivercreator/fixtures_test.go
index 7abffe5d986e..964ce437bf9e 100644
--- a/receiver/receivercreator/fixtures_test.go
+++ b/receiver/receivercreator/fixtures_test.go
@@ -78,6 +78,22 @@ var portEndpointWithHints = observer.Endpoint{
 	},
 }
 
+var podContainerEndpointWithHints = observer.Endpoint{
+	ID:     "namespace/pod-2-UID/redis(6379)",
+	Target: "1.2.3.4:6379",
+	Details: &observer.PodContainer{
+		Name: "redis", Pod: observer.Pod{
+			Name:      "pod-2",
+			Namespace: "default",
+			UID:       "pod-2-UID",
+			Labels:    map[string]string{"env": "prod"},
+			Annotations: map[string]string{
+				otelLogsHints + "/enabled": "true",
+			},
+		},
+	},
+}
+
 var hostportEndpoint = observer.Endpoint{
 	ID:     "port-1",
 	Target: "localhost:1234",
diff --git a/receiver/receivercreator/observerhandler.go b/receiver/receivercreator/observerhandler.go
index 40244c8e7bfe..e90d58f604de 100644
--- a/receiver/receivercreator/observerhandler.go
+++ b/receiver/receivercreator/observerhandler.go
@@ -143,12 +143,6 @@ func (obs *observerHandler) OnChange(changed []observer.Endpoint) {
 }
 
 func (obs *observerHandler) startReceiver(template receiverTemplate, env observer.EndpointEnv, e observer.Endpoint) {
-	obs.params.TelemetrySettings.Logger.Info("starting receiver",
-		zap.String("name", template.id.String()),
-		zap.String("endpoint", e.Target),
-		zap.String("endpoint_id", string(e.ID)),
-		zap.Any("config", template.config))
-
 	resolvedConfig, err := expandConfig(template.config, env)
 	if err != nil {
 		obs.params.TelemetrySettings.Logger.Error("unable to resolve template config", zap.String("receiver", template.id.String()), zap.Error(err))
@@ -199,6 +193,17 @@ func (obs *observerHandler) startReceiver(template receiverTemplate, env observe
 
 	filterConsumerSignals(consumer, template.signals)
 
+	// short-circuit if no consumers are set
+	if consumer.metrics == nil && consumer.logs == nil && consumer.traces == nil {
+		return
+	}
+
+	obs.params.TelemetrySettings.Logger.Info("starting receiver",
+		zap.String("name", template.id.String()),
+		zap.String("endpoint", e.Target),
+		zap.String("endpoint_id", string(e.ID)),
+		zap.Any("config", template.config))
+
 	var receiver component.Component
 	if receiver, err = obs.runner.start(
 		receiverConfig{
diff --git a/receiver/receivercreator/observerhandler_test.go b/receiver/receivercreator/observerhandler_test.go
index 8b91da5064a2..c6575d371c9e 100644
--- a/receiver/receivercreator/observerhandler_test.go
+++ b/receiver/receivercreator/observerhandler_test.go
@@ -285,6 +285,75 @@ func TestOnAddForLogs(t *testing.T) {
 	}
 }
 
+func TestOnAddForLogsWithHints(t *testing.T) {
+	for _, test := range []struct {
+		name                   string
+		expectedReceiverType   component.Component
+		expectedReceiverConfig component.Config
+		target                 observer.Endpoint
+		hintsConfig            DiscoveryConfig
+		expectedError          string
+	}{
+		{
+			name:                 "dynamically generated standard filelog receiver with explicit enablement",
+			target:               podContainerEndpointWithHints,
+			hintsConfig:          DiscoveryConfig{Enabled: true},
+			expectedReceiverType: &nopWithFilelogReceiver{},
+			expectedReceiverConfig: &nopWithFilelogConfig{
+				Include:         []string{"/var/log/pods/default_pod-2_pod-2-UID/redis/*.log"},
+				IncludeFileName: false,
+				IncludeFilePath: true,
+				Operators:       []any{map[string]any{"id": "container-parser", "type": "container"}},
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := createDefaultConfig().(*Config)
+			cfg.Discovery = test.hintsConfig
+
+			handler, mr := newObserverHandler(t, cfg, consumertest.NewNop(), nil, nil)
+			handler.OnAdd([]observer.Endpoint{
+				test.target,
+				unsupportedEndpoint,
+			})
+
+			if test.expectedError != "" {
+				assert.Equal(t, 0, handler.receiversByEndpointID.Size())
+				require.Error(t, mr.lastError)
+				require.ErrorContains(t, mr.lastError, test.expectedError)
+				require.Nil(t, mr.startedComponent)
+				return
+			}
+
+			assert.Equal(t, 1, handler.receiversByEndpointID.Size())
+			require.NoError(t, mr.lastError)
+			require.NotNil(t, mr.startedComponent)
+
+			wr, ok := mr.startedComponent.(*wrappedReceiver)
+			require.True(t, ok)
+
+			require.Nil(t, wr.metrics)
+			require.Nil(t, wr.traces)
+
+			var actualConfig component.Config
+			switch v := wr.logs.(type) {
+			case *nopWithEndpointReceiver:
+				require.NotNil(t, v)
+				actualConfig = v.cfg
+			case *nopWithoutEndpointReceiver:
+				require.NotNil(t, v)
+				actualConfig = v.cfg
+			case *nopWithFilelogReceiver:
+				require.NotNil(t, v)
+				actualConfig = v.cfg
+			default:
+				t.Fatalf("unexpected startedComponent: %T", v)
+			}
+			require.Equal(t, test.expectedReceiverConfig, actualConfig)
+		})
+	}
+}
+
 func TestOnAddForTraces(t *testing.T) {
 	for _, test := range []struct {
 		name                   string
@@ -516,6 +585,7 @@ func newMockHost(t *testing.T, host component.Host) *mockHost {
 	factories, err := otelcoltest.NopFactories()
 	require.NoError(t, err)
 	factories.Receivers[component.MustNewType("with_endpoint")] = &nopWithEndpointFactory{Factory: receivertest.NewNopFactory()}
+	factories.Receivers[component.MustNewType("filelog")] = &nopWithFilelogFactory{Factory: receivertest.NewNopFactory()}
 	factories.Receivers[component.MustNewType("without_endpoint")] = &nopWithoutEndpointFactory{Factory: receivertest.NewNopFactory()}
 	return &mockHost{t: t, factories: factories, Host: host}
 }