[receiver/k8s cluster receiver]add support for K8s leader extension into k8s cluster receiver

.chloggen/k8s_leader_support_for_k8s_cluster_receiver.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: k8sclusterreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "add support for k8s leader election in k8s cluster receiver"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38429]
+
+# (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: Allows multiple instances of the k8s cluster receiver to run in a HA mode in a single cluster.
+
+# 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: []

extension/k8sleaderelector/go.mod

@@ -11,6 +11,7 @@ require (
 	go.opentelemetry.io/collector/confmap v1.30.0
 	go.opentelemetry.io/collector/extension v1.30.0
 	go.opentelemetry.io/collector/extension/extensiontest v0.124.0
+	go.opentelemetry.io/collector/pipeline v0.124.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
 	k8s.io/apimachinery v0.32.3

extension/k8sleaderelector/k8sleaderelectortest/fake_extension.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package k8sleaderelectortest // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector/k8sleaderelectortest"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pipeline"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector"
+)
+
+type FakeHost struct {
+	FakeLeaderElection *FakeLeaderElection
+}
+
+func (fh *FakeHost) GetExtensions() map[component.ID]component.Component {
+	extID := component.MustNewID("k8s_leader_elector")
+	return map[component.ID]component.Component{
+		extID: fh.FakeLeaderElection,
+	}
+}
+
+func (fh *FakeHost) GetExporters() map[pipeline.Signal]map[component.ID]component.Component {
+	return nil
+}
+
+type FakeLeaderElection struct {
+	OnLeading  func(context.Context)
+	OnStopping func()
+}
+
+func (fle *FakeLeaderElection) SetCallBackFuncs(onLeading k8sleaderelector.StartCallback, onStopping k8sleaderelector.StopCallback) {
+	fle.OnLeading = onLeading
+	fle.OnStopping = onStopping
+}
+
+func (fle *FakeLeaderElection) InvokeOnLeading() {
+	if fle.OnLeading != nil {
+		fle.OnLeading(context.Background())
+	}
+}
+
+func (fle *FakeLeaderElection) Start(_ context.Context, _ component.Host) error { return nil }
+
+func (fle *FakeLeaderElection) Shutdown(_ context.Context) error { return nil }
+
+func (fle *FakeLeaderElection) InvokeOnStopping() {
+	if fle.OnStopping != nil {
+		fle.OnStopping()
+	}
+}

receiver/k8sclusterreceiver/README.md

@@ -67,6 +67,7 @@ Example:
 ```yaml
   k8s_cluster:
     auth_type: kubeConfig
+    k8s_leader_elector: <reference k8s leader elector extension>
     node_conditions_to_report: [Ready, MemoryPressure]
     allocatable_types_to_report: [cpu, memory]
     metrics:
@@ -80,6 +81,16 @@ Example:
 The full list of settings exposed for this receiver are documented in [config.go](./config.go)
 with detailed sample configurations in [testdata/config.yaml](./testdata/config.yaml).
 
+### k8s_leader_elector
+Provide name of the k8s leader elector extension defined in config. This allows multiple instances of k8s cluster
+receiver to be executed on a cluster. At a given time only the pod which has the is active.
+
+```yaml
+    k8s_cluster:
+        k8s_leader_elector: k8s_leader_elector
+...
+```
+
 ### node_conditions_to_report
 
 For example, with the config below the receiver will emit two metrics
@@ -96,6 +107,7 @@ k8s_cluster:
 ...
 ```
 
+
 ### metadata_exporters
 
 A list of metadata exporters to which metadata being collected by this receiver

receiver/k8sclusterreceiver/config.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"time"
 
+	"go.opentelemetry.io/collector/component"
+
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/metadata"
 )
@@ -44,6 +46,10 @@ type Config struct {
 	// will not be able to be observed. Setting this option is recommended in environments where due to security restrictions
 	// the collector cannot be granted cluster-wide permissions.
 	Namespace string `mapstructure:"namespace"`
+
+	// K8sLeaderElector defines the reference to the k8s leader elector extension
+	// use this when k8s cluster receiver needs to be deployed in HA mode
+	K8sLeaderElector *component.ID `mapstructure:"k8s_leader_elector"`
 }
 
 func (cfg *Config) Validate() error {

receiver/k8sclusterreceiver/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/iancoleman/strcase v0.3.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.124.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.124.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.124.0
@@ -99,6 +100,7 @@ require (
 	go.opentelemetry.io/collector/config/configtls v1.30.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.124.0 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.124.0 // indirect
+	go.opentelemetry.io/collector/extension v1.30.0 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.30.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.30.0 // indirect
 	go.opentelemetry.io/collector/internal/sharedcomponent v0.124.0 // indirect
@@ -137,7 +139,7 @@ require (
 	k8s.io/kube-openapi v0.0.0-20241105132330-32ad38e42d3f // indirect
 	k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 // indirect
 	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
-	sigs.k8s.io/structured-merge-diff/v4 v4.4.2 // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.4.3 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect
 )
 
@@ -165,3 +167,5 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/xk8stest =
 replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent => ../../internal/sharedcomponent
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector => ../../extension/k8sleaderelector

receiver/k8sclusterreceiver/receiver.go

@@ -14,7 +14,9 @@ import (
 	"go.opentelemetry.io/collector/pipeline"
 	"go.opentelemetry.io/collector/receiver"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
+	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/collection"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sclusterreceiver/internal/metadata"
 )
@@ -42,9 +44,7 @@ type getExporters interface {
 	GetExporters() map[pipeline.Signal]map[component.ID]component.Component
 }
 
-func (kr *kubernetesReceiver) Start(ctx context.Context, host component.Host) error {
-	ctx, kr.cancel = context.WithCancel(ctx)
-
+func (kr *kubernetesReceiver) startReceiver(ctx context.Context, host component.Host) error {
 	if err := kr.resourceWatcher.initialize(); err != nil {
 		return err
 	}
@@ -97,15 +97,58 @@ func (kr *kubernetesReceiver) Start(ctx context.Context, host component.Host) er
 			}
 		}
 	}()
+	return nil
+}
+
+func (kr *kubernetesReceiver) Start(ctx context.Context, host component.Host) error {
+	ctx, kr.cancel = context.WithCancel(ctx)
+
+	// if extension is defined start with k8s leader elector
+	if kr.config.K8sLeaderElector != nil {
+		kr.settings.Logger.Info("Starting k8sClusterReceiver with leader election")
+		extList := host.GetExtensions()
+		if extList == nil {
+			return errors.New("extension list is empty")
+		}
+
+		ext := extList[*kr.config.K8sLeaderElector]
+		if ext == nil {
+			return errors.New("extension k8s leader elector not found")
+		}
+
+		leaderElectorExt, ok := ext.(k8sleaderelector.LeaderElection)
+		if !ok {
+			return errors.New("referenced extension is not k8s leader elector")
+		}
+
+		leaderElectorExt.SetCallBackFuncs(
+			func(ctx context.Context) {
+				if err := kr.startReceiver(ctx, host); err != nil {
+					kr.settings.Logger.Error("Failed to start receiver", zap.Error(err))
+				}
+			}, func() {
+				kr.stopReceiver()
+			},
+		)
+	} else {
+		kr.settings.Logger.Info("Starting k8sClusterReceiver without leader election")
+		if err := kr.startReceiver(ctx, host); err != nil {
+			return err
+		}
+	}
 
 	return nil
 }
 
-func (kr *kubernetesReceiver) Shutdown(context.Context) error {
-	if kr.cancel == nil {
-		return nil
+func (kr *kubernetesReceiver) stopReceiver() {
+	kr.settings.Logger.Info("Stopping the receiver")
+	if kr.cancel != nil {
+		kr.cancel()
 	}
-	kr.cancel()
+}
+
+func (kr *kubernetesReceiver) Shutdown(context.Context) error {
+	kr.stopReceiver()
 	return nil
 }