Merge branch 'main' into mysql-fix-log-resource-attributes

Author: cjksplunk

.chloggen/configure-sync-period.yaml

@@ -0,0 +1,6 @@
+change_type: enhancement
+component: processor/k8s_attributes
+note: Add `watch_sync_period` config option to configure informer cache resync period.
+issues: [48111]
+subtext: The `watch_sync_period` config option defaults to `5m` to match the previously hardcoded behavior.
+change_logs: [user]

processor/k8sattributesprocessor/README.md

@@ -226,6 +226,14 @@ wait_for_metadata: true
 wait_for_metadata_timeout: 10s
 ```
 
+## Informer Cache Resync Period
+
+Reprocessing the informer cache periodically (resyncing) enqueues all cached K8s objects back into event handlers. In large clusters (e.g., 100K pods), this causes significant CPU spikes, memory churn, and garbage collection overhead.
+Because resource state modifications are already pushed immediately via Kubernetes watch events, a resync period is almost entirely unnecessary.
+
+- `watch_sync_period` (`default: 5m`): The resync period for K8s informers. You may set this to `0s` to disable resyncing completely (recommended for large clusters).
+
+
 ## Extracting attributes from pod labels and annotations
 
 The k8sattributesprocessor can also set resource attributes from k8s labels and annotations of pods, namespaces, deployments, statefulsets, daemonsets, jobs and nodes.
@@ -700,6 +708,10 @@ k8s_attributes:
   # Default: 10s
   wait_for_metadata_timeout: 10s
   
+  # Informer resync period. Setting this to 0s disables resyncing completely.
+  # Default: 5m
+  watch_sync_period: 5m
+  
   # Extract configuration - defines what metadata to extract
   extract:
     # Metadata fields to extract as resource attributes
@@ -870,6 +882,7 @@ k8s_attributes:
 | `passthrough` | bool | `false` | Only add pod IP without extracting metadata (no K8s API calls) |
 | `wait_for_metadata` | bool | `false` | Block collector startup until metadata is synced |
 | `wait_for_metadata_timeout` | duration | `10s` | Max wait time for metadata sync on startup |
+| `watch_sync_period` | duration | `5m` | Resync period for K8s informers (`0s` disables resync completely) |
 
 #### Extract Options
 

processor/k8sattributesprocessor/client_test.go

@@ -45,7 +45,7 @@ func selectors() (labels.Selector, fields.Selector) {
 }
 
 // newFakeClient instantiates a new FakeClient object and satisfies the ClientProvider type
-func newFakeClient(_ component.TelemetrySettings, _ k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, associations []kube.Association, _ kube.Excludes, _ kube.APIClientsetProvider, _ kube.InformersFactoryList, _ bool, _ time.Duration) (kube.Client, error) {
+func newFakeClient(_ component.TelemetrySettings, _ k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, associations []kube.Association, _ kube.Excludes, _ kube.APIClientsetProvider, _ kube.InformersFactoryList, _ bool, _, _ time.Duration) (kube.Client, error) {
 	cs := fake.NewClientset()
 
 	ls, fs := selectors()

processor/k8sattributesprocessor/config.go

@@ -4,6 +4,7 @@
 package k8sattributesprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor"
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"regexp"
@@ -46,13 +47,22 @@ type Config struct {
 
 	// WaitForMetadataTimeout is the maximum time the processor will wait for the k8s metadata to be synced.
 	WaitForMetadataTimeout time.Duration `mapstructure:"wait_for_metadata_timeout"`
+
+	// WatchSyncPeriod determines the resync period for K8s informers.
+	// Reprocessing the informer cache periodically can cause significant memory churn and CPU spikes.
+	// Setting this to 0 disables resync.
+	WatchSyncPeriod time.Duration `mapstructure:"watch_sync_period"`
 }
 
 func (cfg *Config) Validate() error {
 	if err := cfg.APIConfig.Validate(); err != nil {
 		return err
 	}
 
+	if cfg.WatchSyncPeriod < 0 {
+		return errors.New("watch_sync_period must be greater than or equal to 0")
+	}
+
 	for _, assoc := range cfg.Association {
 		if len(assoc.Sources) > kube.PodIdentifierMaxLength {
 			return fmt.Errorf("too many association sources. limit is %v", kube.PodIdentifierMaxLength)

processor/k8sattributesprocessor/config.schema.yaml

@@ -135,5 +135,9 @@ properties:
     description: WaitForMetadataTimeout is the maximum time the processor will wait for the k8s metadata to be synced.
     type: string
     format: duration
+  watch_sync_period:
+    description: WatchSyncPeriod determines the resync period for K8s informers. Reprocessing the informer cache periodically can cause significant memory churn and CPU spikes. Setting this to 0 disables resync.
+    type: string
+    format: duration
 allOf:
   - $ref: /internal/k8sconfig.api_config

processor/k8sattributesprocessor/config_test.go

@@ -34,6 +34,7 @@ func TestLoadConfig(t *testing.T) {
 					DeploymentNameFromReplicaSet: false,
 				},
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -107,6 +108,7 @@ func TestLoadConfig(t *testing.T) {
 					},
 				},
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -131,6 +133,7 @@ func TestLoadConfig(t *testing.T) {
 					},
 				},
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -143,6 +146,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -183,6 +187,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -196,6 +201,7 @@ func TestLoadConfig(t *testing.T) {
 				Exclude:                defaultExcludes,
 				WaitForMetadata:        true,
 				WaitForMetadataTimeout: 30 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -209,6 +215,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -226,6 +233,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -243,6 +251,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -261,6 +270,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -276,6 +286,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -291,6 +302,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -306,6 +318,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -321,6 +334,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -336,6 +350,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -357,6 +372,7 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
@@ -370,11 +386,39 @@ func TestLoadConfig(t *testing.T) {
 				},
 				Exclude:                defaultExcludes,
 				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        5 * time.Minute,
 			},
 		},
 		{
 			id: component.NewIDWithName(metadata.Type, "bad_metadata_field"),
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "custom_intervals"),
+			expected: &Config{
+				APIConfig: k8sconfig.APIConfig{AuthType: k8sconfig.AuthTypeServiceAccount},
+				Extract: ExtractConfig{
+					Metadata: enabledAttributes(),
+				},
+				Exclude:                defaultExcludes,
+				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        20 * time.Second,
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "bad_watch_sync_period"),
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "disable_watch_sync"),
+			expected: &Config{
+				APIConfig: k8sconfig.APIConfig{AuthType: k8sconfig.AuthTypeServiceAccount},
+				Extract: ExtractConfig{
+					Metadata: enabledAttributes(),
+				},
+				Exclude:                defaultExcludes,
+				WaitForMetadataTimeout: 10 * time.Second,
+				WatchSyncPeriod:        0,
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -396,6 +440,7 @@ func TestLoadConfig(t *testing.T) {
 				assert.Error(t, err)
 				return
 			}
+
 			assert.NoError(t, xconfmap.Validate(cfg))
 			assert.Equal(t, tt.expected, cfg)
 		})

processor/k8sattributesprocessor/factory.go

@@ -50,6 +50,7 @@ func createDefaultConfig() component.Config {
 			Metadata: enabledAttributes(),
 		},
 		WaitForMetadataTimeout: 10 * time.Second,
+		WatchSyncPeriod:        5 * time.Minute,
 	}
 }
 
@@ -278,7 +279,8 @@ func createProcessorOpts(cfg component.Config) []option {
 		withAPIConfig(oCfg.APIConfig),
 		withExtractPodAssociations(oCfg.Association...),
 		withExcludes(oCfg.Exclude),
-		withWaitForMetadataTimeout(oCfg.WaitForMetadataTimeout))
+		withWaitForMetadataTimeout(oCfg.WaitForMetadataTimeout),
+		withWatchSyncPeriod(oCfg.WatchSyncPeriod))
 
 	if oCfg.WaitForMetadata {
 		opts = append(opts, withWaitForMetadata(true))

processor/k8sattributesprocessor/internal/kube/client.go

@@ -52,6 +52,7 @@ type WatchClient struct {
 	stopCh                 chan struct{}
 	waitForMetadata        bool
 	waitForMetadataTimeout time.Duration
+	watchSyncPeriod        time.Duration
 
 	// A map containing Pod related data, used to associate them with resources.
 	// Key can be either an IP address or Pod UID
@@ -123,6 +124,7 @@ func New(
 	informersFactory InformersFactoryList,
 	waitForMetadata bool,
 	waitForMetadataTimeout time.Duration,
+	watchSyncPeriod time.Duration,
 ) (Client, error) {
 	telemetryBuilder, err := metadata.NewTelemetryBuilder(set)
 	if err != nil {
@@ -140,6 +142,7 @@ func New(
 		telemetryBuilder:       telemetryBuilder,
 		waitForMetadata:        waitForMetadata,
 		waitForMetadataTimeout: waitForMetadataTimeout,
+		watchSyncPeriod:        watchSyncPeriod,
 	}
 
 	c.Pods = map[PodIdentifier]*Pod{}
@@ -171,19 +174,25 @@ func New(
 		zap.String("fieldSelector", fieldSelector.String()),
 	)
 	if informersFactory.newInformer == nil {
-		informersFactory.newInformer = newSharedInformer
+		informersFactory.newInformer = func(client kubernetes.Interface, ns string, ls labels.Selector, fs fields.Selector) cache.SharedInformer {
+			return newSharedInformer(client, ns, ls, fs, watchSyncPeriod)
+		}
 	}
 
 	if informersFactory.newNamespaceInformer == nil {
 		switch {
 		case c.extractNamespaceLabelsAnnotations():
 			// if rules to extract metadata from namespace is configured use namespace shared informer containing
 			// all namespaces including kube-system which contains cluster uid information (kube-system-uid)
-			informersFactory.newNamespaceInformer = newNamespaceSharedInformer
+			informersFactory.newNamespaceInformer = func(client clientmeta.Interface) cache.SharedInformer {
+				return newNamespaceSharedInformer(client, watchSyncPeriod)
+			}
 		case rules.ClusterUID:
 			// use kube-system shared informer to only watch kube-system namespace
 			// reducing overhead of watching all the namespaces
-			informersFactory.newNamespaceInformer = newKubeSystemSharedInformer
+			informersFactory.newNamespaceInformer = func(client clientmeta.Interface) cache.SharedInformer {
+				return newKubeSystemSharedInformer(client, watchSyncPeriod)
+			}
 		default:
 			informersFactory.newNamespaceInformer = NewNoOpInformer
 		}
@@ -217,7 +226,9 @@ func New(
 
 	if needReplicaSetInformer {
 		if informersFactory.newReplicaSetInformer == nil {
-			informersFactory.newReplicaSetInformer = newReplicaSetSharedInformer
+			informersFactory.newReplicaSetInformer = func(client clientmeta.Interface, namespace string) cache.SharedInformer {
+				return newReplicaSetSharedInformer(client, namespace, watchSyncPeriod)
+			}
 		}
 
 		c.replicasetInformer = informersFactory.newReplicaSetInformer(c.mc, c.Filters.Namespace)
@@ -227,23 +238,23 @@ func New(
 	}
 
 	if c.extractNodeLabelsAnnotations() || c.extractNodeUID() {
-		c.nodeInformer = newNodeSharedInformer(c.mc, c.Filters.Node)
+		c.nodeInformer = newNodeSharedInformer(c.mc, c.Filters.Node, watchSyncPeriod)
 	}
 
 	if c.extractDeploymentLabelsAnnotations() {
-		c.deploymentInformer = newDeploymentSharedInformer(c.mc, c.Filters.Namespace)
+		c.deploymentInformer = newDeploymentSharedInformer(c.mc, c.Filters.Namespace, watchSyncPeriod)
 	}
 
 	if c.extractStatefulSetLabelsAnnotations() {
-		c.statefulsetInformer = newStatefulSetSharedInformer(c.mc, c.Filters.Namespace)
+		c.statefulsetInformer = newStatefulSetSharedInformer(c.mc, c.Filters.Namespace, watchSyncPeriod)
 	}
 
 	if c.extractDaemonSetLabelsAnnotations() {
-		c.daemonsetInformer = newDaemonSetSharedInformer(c.mc, c.Filters.Namespace)
+		c.daemonsetInformer = newDaemonSetSharedInformer(c.mc, c.Filters.Namespace, watchSyncPeriod)
 	}
 
 	if c.extractJobLabelsAnnotations() || rules.CronJobUID {
-		c.jobInformer = newJobSharedInformer(c.mc, c.Filters.Namespace)
+		c.jobInformer = newJobSharedInformer(c.mc, c.Filters.Namespace, watchSyncPeriod)
 	}
 	return c, err
 }