add mutex locks when multiple receivers setting callback functions (open-telemetry#41062)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
- In otel collector extensions are always started before receivers. This
can lead to a scenario where `k8sleaderelector` extension has already
started and became a leader, but the receiver has not finished setting
up of callback functions. This causes that desired callback functions
are not executed when the instance is leader.
- Fix this by executing the late registered callbacks immediately if we
are already a leader.
- Add mutex locks so that list of callbacks are updated atomically
    

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

open-telemetry#40346

<!--Describe what testing was performed and which tests were added.-->
#### Testing
 - Added additional unit test to cover this scenario
<!--Describe the documentation added.-->
#### Documentation

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

Author: rakesh-garimella

.chloggen/k8sleaderelector_handle_late_receiever_registration.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:
+  bug_fix
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component:
+  k8sleaderelector
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: handle late registration of receivers to k8sleaderelector
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40346]
+
+# (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]

extension/k8sleaderelector/extension.go

@@ -25,10 +25,27 @@ type LeaderElection interface {
 	SetCallBackFuncs(StartCallback, StopCallback)
 }
 
+type callBackFuncs struct {
+	onStartLeading StartCallback
+	onStopLeading  StopCallback
+}
+
 // SetCallBackFuncs set the functions that can be invoked when the leader wins or loss the election
 func (lee *leaderElectionExtension) SetCallBackFuncs(onStartLeading StartCallback, onStopLeading StopCallback) {
-	lee.onStartedLeading = append(lee.onStartedLeading, onStartLeading)
-	lee.onStoppedLeading = append(lee.onStoppedLeading, onStopLeading)
+	// Have a write lock while setting the callbacks.
+	lee.mu.Lock()
+	defer lee.mu.Unlock()
+	callBack := callBackFuncs{
+		onStartLeading: onStartLeading,
+		onStopLeading:  onStopLeading,
+	}
+
+	lee.callBackFuncs = append(lee.callBackFuncs, callBack)
+
+	if lee.isLeader {
+		// Immediately invoke the callback since we are already leader
+		onStartLeading(context.Background())
+	}
 }
 
 // leaderElectionExtension is the main struct implementing the extension's behavior.
@@ -40,21 +57,34 @@ type leaderElectionExtension struct {
 	cancel        context.CancelFunc
 	waitGroup     sync.WaitGroup
 
-	onStartedLeading []StartCallback
-	onStoppedLeading []StopCallback
+	callBackFuncs []callBackFuncs
+
+	isLeader bool
+
+	mu sync.Mutex
 }
 
 // If the receiver sets a callback function then it would be invoked when the leader wins the election
 func (lee *leaderElectionExtension) startedLeading(ctx context.Context) {
-	for _, callback := range lee.onStartedLeading {
-		callback(ctx)
+	// Have read lock so that we no new callbacks can be added while we are invoking the callbacks.
+	lee.mu.Lock()
+	defer lee.mu.Unlock()
+	lee.isLeader = true
+	for _, callback := range lee.callBackFuncs {
+		callback.onStartLeading(ctx)
 	}
 }
 
 // If the receiver sets a callback function then it would be invoked when the leader loss the election
 func (lee *leaderElectionExtension) stoppedLeading() {
-	for _, callback := range lee.onStoppedLeading {
-		callback()
+	// Have a read lock while stopping the receivers. This would make sure that if we have executed any onStartLeading callbacks
+	// after becoming leader, we would execute the onStopLeading callbacks for them as well.
+	lee.mu.Lock()
+	defer lee.mu.Unlock()
+
+	lee.isLeader = false
+	for _, callback := range lee.callBackFuncs {
+		callback.onStopLeading()
 	}
 }
 
@@ -64,6 +94,7 @@ func (lee *leaderElectionExtension) Start(_ context.Context, _ component.Host) e
 
 	ctx := context.Background()
 	ctx, lee.cancel = context.WithCancel(ctx)
+
 	// Create the K8s leader elector
 	leaderElector, err := newK8sLeaderElector(lee.config, lee.client, lee.startedLeading, lee.stoppedLeading, lee.leaseHolderID)
 	if err != nil {
@@ -77,7 +108,6 @@ func (lee *leaderElectionExtension) Start(_ context.Context, _ component.Host) e
 		defer lee.waitGroup.Done()
 		for {
 			leaderElector.Run(ctx)
-
 			if ctx.Err() != nil {
 				break
 			}

extension/k8sleaderelector/extension_test.go

@@ -72,3 +72,57 @@ func TestExtension(t *testing.T) {
 	require.True(t, onStartLeadingInvoked.Load())
 	require.NoError(t, leaderElection.Shutdown(ctx))
 }
+
+func TestExtension_WithDelay(t *testing.T) {
+	config := &Config{
+		LeaseName:      "foo",
+		LeaseNamespace: "default",
+		LeaseDuration:  15 * time.Second,
+		RenewDuration:  10 * time.Second,
+		RetryPeriod:    2 * time.Second,
+	}
+
+	ctx := context.TODO()
+	fakeClient := fake.NewClientset()
+	config.makeClient = func(_ k8sconfig.APIConfig) (kubernetes.Interface, error) {
+		return fakeClient, nil
+	}
+
+	observedZapCore, _ := observer.New(zap.WarnLevel)
+
+	leaderElection := leaderElectionExtension{
+		config:        config,
+		client:        fakeClient,
+		logger:        zap.New(observedZapCore),
+		leaseHolderID: "foo",
+	}
+
+	var onStartLeadingInvoked atomic.Bool
+
+	require.NoError(t, leaderElection.Start(ctx, componenttest.NewNopHost()))
+
+	// Simulate a delay of setting up callbacks after the leader has been elected.
+	expectedLeaseDurationSeconds := ptr.To(int32(15))
+	require.Eventually(t, func() bool {
+		lease, err := fakeClient.CoordinationV1().Leases("default").Get(ctx, "foo", metav1.GetOptions{})
+		require.NoError(t, err)
+		require.NotNil(t, lease)
+		require.NotNil(t, lease.Spec.AcquireTime)
+		require.NotNil(t, lease.Spec.HolderIdentity)
+		require.Equal(t, expectedLeaseDurationSeconds, lease.Spec.LeaseDurationSeconds)
+		return true
+	}, 10*time.Second, 100*time.Millisecond)
+
+	leaderElection.SetCallBackFuncs(
+		func(_ context.Context) {
+			onStartLeadingInvoked.Store(true)
+			fmt.Printf("%v: LeaderElection started leading\n", time.Now().String())
+		},
+		func() {
+			fmt.Printf("%v: LeaderElection stopped leading\n", time.Now().String())
+		},
+	)
+
+	require.True(t, onStartLeadingInvoked.Load())
+	require.NoError(t, leaderElection.Shutdown(ctx))
+}