add ut

Author: Frapschen

receiver/k8sobjectsreceiver/go.mod

@@ -4,6 +4,7 @@ go 1.23.0
 
 require (
 	github.com/google/uuid v1.6.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector v0.125.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.125.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.125.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.125.0
@@ -82,7 +83,7 @@ require (
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/collector v0.125.1-0.20250505155216-829157cef7bb // indirect
+	go.opentelemetry.io/collector v0.125.0 // indirect
 	go.opentelemetry.io/collector/client v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/component/componentstatus v0.125.1-0.20250505155216-829157cef7bb // indirect
 	go.opentelemetry.io/collector/config/configauth v0.125.1-0.20250505155216-829157cef7bb // indirect
@@ -95,6 +96,7 @@ require (
 	go.opentelemetry.io/collector/config/configtls v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.125.1-0.20250505155216-829157cef7bb // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.125.1-0.20250505155216-829157cef7bb // indirect
+	go.opentelemetry.io/collector/extension v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/extension/extensionauth v1.31.1-0.20250505152726-56c7da210783 // indirect
 	go.opentelemetry.io/collector/extension/extensionmiddleware v0.125.1-0.20250505155216-829157cef7bb // indirect
 	go.opentelemetry.io/collector/featuregate v1.31.1-0.20250505152726-56c7da210783 // indirect

receiver/k8sobjectsreceiver/go.sum

@@ -82,6 +82,45 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, host component.Host) er
 		return err
 	}
 	kr.client = client
+
+	// Validate objects against K8s API
+	validObjects, err := kr.config.getValidObjects()
+	if err != nil {
+		return err
+	}
+
+	var validConfigs []*K8sObjectsConfig
+	for _, object := range kr.objects {
+		gvrs, ok := validObjects[object.Name]
+		if !ok {
+			availableResource := make([]string, 0, len(validObjects))
+			for k := range validObjects {
+				availableResource = append(availableResource, k)
+			}
+			err = fmt.Errorf("resource not found: %s. Available resources in cluster: %v", object.Name, availableResource)
+			if handlerErr := kr.handleError(err, ""); handlerErr != nil {
+				return handlerErr
+			}
+			continue
+		}
+
+		gvr := gvrs[0]
+		for i := range gvrs {
+			if gvrs[i].Group == object.Group {
+				gvr = gvrs[i]
+				break
+			}
+		}
+
+		object.gvr = gvr
+		validConfigs = append(validConfigs, object)
+	}
+
+	if len(validConfigs) == 0 {
+		err = errors.New("no valid Kubernetes objects found to watch")
+		return err
+	}
+
 	cctx, cancel := context.WithCancel(ctx)
 	kr.cancel = cancel
 
@@ -97,12 +136,6 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, host component.Host) er
 			return fmt.Errorf("the extension %T is not implement k8sleaderelector.LeaderElection", k8sLeaderElector)
 		}
 
-		validConfigs := Validate()
-		if len(validConfigs) == 0 {
-			err := errors.New("no valid Kubernetes objects found to watch")
-			return err
-		}
-
 		elector.SetCallBackFuncs(
 			func(_ context.Context) {
 				kr.setting.Logger.Info("Object Receiver started as leader")
@@ -117,13 +150,8 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, host component.Host) er
 					kr.setting.Logger.Error("shutdown receiver error:", zap.Error(err))
 				}
 			})
-		return nil
-	}
 
-	validConfigs := Validate()
-	if len(validConfigs) == 0 {
-		err := errors.New("no valid Kubernetes objects found to watch")
-		return err
+		return nil
 	}
 
 	kr.setting.Logger.Info("Object Receiver started")
@@ -175,49 +203,6 @@ func (kr *k8sobjectsreceiver) start(ctx context.Context, object *K8sObjectsConfi
 	}
 }
 
-// Validate objects against K8s API
-func (kr *k8sobjectsreceiver) Validate() []*K8sObjectsConfig  {
-	validObjects, err := kr.config.getValidObjects()
-	if err != nil {
-		return err
-	}
-
-	var validConfigs []*K8sObjectsConfig
-	for _, object := range kr.objects {
-		gvrs, ok := validObjects[object.Name]
-		if !ok {
-			availableResource := make([]string, 0, len(validObjects))
-			for k := range validObjects {
-				availableResource = append(availableResource, k)
-			}
-			err = fmt.Errorf("resource not found: %s. Available resources in cluster: %v", object.Name, availableResource)
-			if handlerErr := kr.handleError(err, ""); handlerErr != nil {
-				return handlerErr
-			}
-
-			continue
-		}
-
-		gvr := gvrs[0]
-		for i := range gvrs {
-			if gvrs[i].Group == object.Group {
-				gvr = gvrs[i]
-				break
-			}
-		}
-
-		object.gvr = gvr
-		validConfigs = append(validConfigs, object)
-	}
-
-	if len(validConfigs) == 0 {
-		err = errors.New("no valid Kubernetes objects found to watch")
-		return err
-	}
-
-	return validConfigs
-}
-
 func (kr *k8sobjectsreceiver) startPull(ctx context.Context, config *K8sObjectsConfig, resource dynamic.ResourceInterface) {
 	stopperChan := make(chan struct{})
 	kr.mu.Lock()

receiver/k8sobjectsreceiver/receiver.go

@@ -10,11 +10,13 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	apiWatch "k8s.io/apimachinery/pkg/watch"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector/k8sleaderelectortest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/k8sobjectsreceiver/internal/metadata"
 )
 
@@ -294,3 +296,62 @@ func TestExcludeDeletedTrue(t *testing.T) {
 
 	assert.NoError(t, r.Shutdown(ctx))
 }
+
+func TestReceiverWithLeaderElection(t *testing.T) {
+	fakeLeaderElection := &k8sleaderelectortest.FakeLeaderElection{}
+	fakeHost := &k8sleaderelectortest.FakeHost{
+		FakeLeaderElection: fakeLeaderElection,
+	}
+	leaderElectorID := component.MustNewID("k8s_leader_elector")
+
+	mockClient := newMockDynamicClient()
+	rCfg := createDefaultConfig().(*Config)
+	rCfg.makeDynamicClient = mockClient.getMockDynamicClient
+	rCfg.makeDiscoveryClient = getMockDiscoveryClient
+	rCfg.ErrorMode = PropagateError
+	rCfg.Objects = []*K8sObjectsConfig{
+		{
+			Name: "pods",
+			Mode: PullMode,
+		},
+	}
+	rCfg.K8sLeaderElector = &leaderElectorID
+
+	r, err := newReceiver(
+		receivertest.NewNopSettings(metadata.Type),
+		rCfg,
+		consumertest.NewNop(),
+	)
+	require.NoError(t, err)
+	kr := r.(*k8sobjectsreceiver)
+	sink := new(consumertest.LogsSink)
+	kr.consumer = sink
+
+	// Setup k8s resources.
+	numPods := 2
+	mockClient.createPods(
+		generatePod("pod1", "default", map[string]any{
+			"environment": "production",
+		}, "1"),
+		generatePod("pod2", "default", map[string]any{
+			"environment": "production",
+		}, "1"),
+	)
+
+	err = kr.Start(context.Background(), fakeHost)
+	require.NoError(t, err)
+
+	// elected leader
+	fakeLeaderElection.InvokeOnLeading()
+
+	expectedNumMetrics := numPods
+	var initialLogRecordCount int
+	require.Eventually(t, func() bool {
+		initialLogRecordCount = sink.LogRecordCount()
+		return initialLogRecordCount == expectedNumMetrics
+	}, 20*time.Second, 100*time.Millisecond,
+		"logs not collected")
+
+	// lost election
+	fakeLeaderElection.InvokeOnStopping()
+}