Repository: kubernetes/kubernetes (language: go)

Issue:
[CLE] Parallelize lease candidate ping order
In the leader election controller: https://github.com/kubernetes/kubernetes/blob/master/pkg/controlplane/controller/leaderelection/leaderelection_controller.go#L267-L293, candidates are iterated through sequentially and `LeaseCandidates.Update()` is a blocking operation that is also called sequentially. This can lead to cases where certain candidates do not have enough time to respond. We should parallelize this entire operation and iterate through the list in a random order to ensure every candidate has a fair chance of responding.

/cc @Henrywu573
/triage accepted
/sig api-machinery




Discussion / hints:
Sorry, linked this issue by accident when playing with some simple issues to fix in my spare time.

Do you really need to randomize the order? The concurrent requests don't block each other. I guess something like https://github.com/tchap/kubernetes/commit/415c889fcd3a99586ad9afc6aa1c4dfe159337c5 could suffice?
/assign
> Sorry, linked this issue by accident when playing with some simple issues to fix in my spare time.
> 
> Do you really need to randomize the order? The concurrent requests don't block each other. I guess something like [tchap@415c889](https://github.com/tchap/kubernetes/commit/415c889fcd3a99586ad9afc6aa1c4dfe159337c5) could suffice?

cc @sttts. I think it makes sense that randomization is not strictly necessary if we parallelize. We should be able to guarantee that each candidate has a fair chance of acking with just the parallelization. 

Fix patch:
diff --git a/pkg/controlplane/controller/leaderelection/leaderelection_controller.go b/pkg/controlplane/controller/leaderelection/leaderelection_controller.go
index cb53df92e8102..7e4ac20b40155 100644
--- a/pkg/controlplane/controller/leaderelection/leaderelection_controller.go
+++ b/pkg/controlplane/controller/leaderelection/leaderelection_controller.go
@@ -22,6 +22,7 @@ import (
 	"reflect"
 	"time"
 
+	"golang.org/x/sync/errgroup"
 	v1 "k8s.io/api/coordination/v1"
 	v1alpha1 "k8s.io/api/coordination/v1alpha1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -264,6 +265,7 @@ func (c *Controller) reconcileElectionStep(ctx context.Context, leaseNN types.Na
 
 	now := c.clock.Now()
 	canVoteYet := true
+	g, gCtx := errgroup.WithContext(ctx)
 	for _, candidate := range candidates {
 		if candidate.Spec.PingTime != nil && candidate.Spec.PingTime.Add(electionDuration).After(now) &&
 			candidate.Spec.RenewTime != nil && candidate.Spec.RenewTime.Before(candidate.Spec.PingTime) {
@@ -280,17 +282,18 @@ func (c *Controller) reconcileElectionStep(ctx context.Context, leaseNN types.Na
 			// If PingTime is outdated, send another PingTime only if it already acked the first one.
 			// This checks for pingTime <= renewTime because equality is possible in unit tests using a fake clock.
 			(candidate.Spec.PingTime.Add(electionDuration).Before(now) && !candidate.Spec.RenewTime.Before(candidate.Spec.PingTime)) {
-			// TODO(jefftree): We should randomize the order of sending pings and do them in parallel
-			// so that all candidates have equal opportunity to ack.
 			clone := candidate.DeepCopy()
 			clone.Spec.PingTime = &metav1.MicroTime{Time: now}
-			_, err := c.leaseCandidateClient.LeaseCandidates(clone.Namespace).Update(ctx, clone, metav1.UpdateOptions{})
-			if err != nil {
-				return defaultRequeueInterval, err
-			}
+			g.Go(func() error {
+				_, err := c.leaseCandidateClient.LeaseCandidates(clone.Namespace).Update(gCtx, clone, metav1.UpdateOptions{})
+				return err
+			})
 			canVoteYet = false
 		}
 	}
+	if err := g.Wait(); err != nil {
+		return defaultRequeueInterval, err
+	}
 	if !canVoteYet {
 		return defaultRequeueInterval, nil
 	}


Test patch: