[receiver/k8sobjectsreceiver] Switch to standby mode when leader lease is lost (#43054)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Similar to what we did in #42330, this PR ensures similarity and puts
the receiver in stand-by instead of shutdown when `k8sleaderelector` is
used.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #42706

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Tested locally, and tests were not changed to ensure the current
receiver is not affected at all.

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Signed-off-by: Paulo Dias <[email protected]>

Author: paulojmdias

.chloggen/feat_42706_wg.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: "receiver/k8sobjectsreceiver"
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Switch to standby mode when leader lease is lost instead of shutdown"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [42706]
+
+# (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]

receiver/k8sobjectsreceiver/receiver.go

@@ -38,6 +38,7 @@ type k8sobjectsreceiver struct {
 	obsrecv         *receiverhelper.ObsReport
 	mu              sync.Mutex
 	cancel          context.CancelFunc
+	wg              sync.WaitGroup
 }
 
 func newReceiver(params receiver.Settings, config *Config, consumer consumer.Logs) (receiver.Logs, error) {
@@ -141,13 +142,12 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, host component.Host) er
 				}
 				kr.setting.Logger.Info("Object Receiver started as leader")
 			},
+			// onStoppedLeading: stop watches, but DO NOT shut the whole receiver down
 			func() {
-				kr.setting.Logger.Info("no longer leader, stopping")
-				err = kr.Shutdown(context.Background())
-				if err != nil {
-					kr.setting.Logger.Error("shutdown receiver error:", zap.Error(err))
-				}
-			})
+				kr.setting.Logger.Info("no longer leader, stopping watches")
+				kr.stopWatches()
+			},
+		)
 	} else {
 		cctx, cancel := context.WithCancel(ctx)
 		kr.cancel = cancel
@@ -160,17 +160,37 @@ func (kr *k8sobjectsreceiver) Start(ctx context.Context, host component.Host) er
 }
 
 func (kr *k8sobjectsreceiver) Shutdown(context.Context) error {
+	// Stop informers and wait for them to exit.
 	kr.setting.Logger.Info("Object Receiver stopped")
+	kr.stopWatches()
+
 	if kr.cancel != nil {
 		kr.cancel()
+		kr.cancel = nil
 	}
+	return nil
+}
 
+// stopWatches closes all informer stop channels (idempotently) and waits for their goroutines to exit.
+func (kr *k8sobjectsreceiver) stopWatches() {
 	kr.mu.Lock()
-	for _, stopperChan := range kr.stopperChanList {
-		close(stopperChan)
-	}
+	// Copy and clear the list under lock to avoid races on restart
+	chans := kr.stopperChanList
+	kr.stopperChanList = nil
 	kr.mu.Unlock()
-	return nil
+
+	if len(chans) == 0 {
+		return
+	}
+	for _, ch := range chans {
+		select {
+		case <-ch: // already closed
+		default:
+			close(ch)
+		}
+	}
+	// Now wait for all WG-tracked loops (both pull & watch) to exit
+	kr.wg.Wait()
 }
 
 func (kr *k8sobjectsreceiver) start(ctx context.Context, object *K8sObjectsConfig) {
@@ -205,7 +225,9 @@ func (kr *k8sobjectsreceiver) startPull(ctx context.Context, config *K8sObjectsC
 	stopperChan := make(chan struct{})
 	kr.mu.Lock()
 	kr.stopperChanList = append(kr.stopperChanList, stopperChan)
+	kr.wg.Add(1)
 	kr.mu.Unlock()
+	defer kr.wg.Done()
 	ticker := newTicker(ctx, config.Interval)
 	listOption := metav1.ListOptions{
 		FieldSelector: config.FieldSelector,
@@ -226,15 +248,21 @@ func (kr *k8sobjectsreceiver) startPull(ctx context.Context, config *K8sObjectsC
 				kr.setting.Logger.Error("error in pulling object",
 					zap.String("resource", config.gvr.String()),
 					zap.Error(err))
-			} else if len(objects.Items) > 0 {
-				logs := pullObjectsToLogData(objects, time.Now(), config, kr.setting.BuildInfo.Version)
-				obsCtx := kr.obsrecv.StartLogsOp(ctx)
-				logRecordCount := logs.LogRecordCount()
-				err = kr.consumer.ConsumeLogs(obsCtx, logs)
-				kr.obsrecv.EndLogsOp(obsCtx, metadata.Type.String(), logRecordCount, err)
+				continue
 			}
+			if len(objects.Items) == 0 {
+				continue
+			}
+			logs := pullObjectsToLogData(objects, time.Now(), config, kr.setting.BuildInfo.Version)
+			obsCtx := kr.obsrecv.StartLogsOp(ctx)
+			logRecordCount := logs.LogRecordCount()
+			err = kr.consumer.ConsumeLogs(obsCtx, logs)
+			kr.obsrecv.EndLogsOp(obsCtx, metadata.Type.String(), logRecordCount, err)
+
 		case <-stopperChan:
 			return
+		case <-ctx.Done():
+			return
 		}
 	}
 }
@@ -243,7 +271,9 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 	stopperChan := make(chan struct{})
 	kr.mu.Lock()
 	kr.stopperChanList = append(kr.stopperChanList, stopperChan)
+	kr.wg.Add(1)
 	kr.mu.Unlock()
+	defer kr.wg.Done()
 
 	if kr.config.IncludeInitialState {
 		kr.sendInitialState(ctx, config, resource)
@@ -256,6 +286,7 @@ func (kr *k8sobjectsreceiver) startWatch(ctx context.Context, config *K8sObjects
 	}
 
 	cancelCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	cfgCopy := *config
 	wait.UntilWithContext(cancelCtx, func(newCtx context.Context) {
 		resourceVersion, err := getResourceVersion(newCtx, &cfgCopy, resource)
@@ -343,6 +374,10 @@ func (kr *k8sobjectsreceiver) doWatch(ctx context.Context, config *K8sObjectsCon
 	res := watcher.ResultChan()
 	for {
 		select {
+		case <-ctx.Done():
+			kr.setting.Logger.Info("context canceled, stopping watch",
+				zap.String("resource", config.gvr.String()))
+			return true
 		case data, ok := <-res:
 			if data.Type == apiWatch.Error {
 				errObject := apierrors.FromObject(data.Object)
@@ -372,8 +407,9 @@ func (kr *k8sobjectsreceiver) doWatch(ctx context.Context, config *K8sObjectsCon
 				kr.setting.Logger.Error("error converting objects to log data", zap.Error(err))
 			} else {
 				obsCtx := kr.obsrecv.StartLogsOp(ctx)
+				cnt := logs.LogRecordCount()
 				err := kr.consumer.ConsumeLogs(obsCtx, logs)
-				kr.obsrecv.EndLogsOp(obsCtx, metadata.Type.String(), 1, err)
+				kr.obsrecv.EndLogsOp(obsCtx, metadata.Type.String(), cnt, err)
 			}
 		case <-stopperChan:
 			watcher.Stop()