leader election and unique par identity

Author: merchristK

cmd/cluster-agent/subcommands/start/command.go

@@ -551,7 +551,7 @@ func start(log log.Component,
 	}
 
 	if config.GetBool("private_action_runner.enabled") {
-		drain, err := startPrivateActionRunner(mainCtx, config, hostnameGetter, rcClient, log)
+		drain, err := startPrivateActionRunner(mainCtx, config, hostnameGetter, rcClient, le, log)
 		if err != nil {
 			log.Errorf("Cannot start private action runner: %v", err)
 		} else {
@@ -688,11 +688,19 @@ func startPrivateActionRunner(
 	config config.Component,
 	hostnameGetter hostnameinterface.Component,
 	rcClient *rcclient.Client,
+	le *leaderelection.LeaderEngine,
 	log log.Component,
 ) (func(), error) {
 	if rcClient == nil {
 		return nil, errors.New("Remote config is disabled or failed to initialize, remote config is a required dependency for private action runner")
 	}
+	if !config.GetBool("leader_election") {
+		return nil, errors.New("leader election is not enabled on the Cluster Agent. The private action runner needs leader election for identity coordination across replicas")
+	}
+	err := le.EnsureLeaderElectionRuns()
+	if err != nil {
+		return nil, err
+	}
 	app, err := privateactionrunner.NewPrivateActionRunner(ctx, config, hostnameGetter, rcClient, log)
 	if err != nil {
 		return nil, err

pkg/privateactionrunner/enrollment/k8s_secret.go

@@ -46,28 +46,26 @@ func getIdentityFromK8sSecret(ctx context.Context, cfg configModel.Reader) (*Per
 	ns := namespace.GetResourcesNamespace()
 	secretName := getSecretName(cfg)
 
-	// Check if we're a follower - if so, we may need to wait for leader to create the secret
 	le, err := leaderelection.GetLeaderEngine()
-	isFollower := err == nil && !le.IsLeader()
+	if err != nil {
+		return nil, err
+	}
 
 	secret, err := client.CoreV1().Secrets(ns).Get(ctx, secretName, metav1.GetOptions{})
 	if err != nil {
-		if k8serrors.IsNotFound(err) {
-			// If we're a follower and secret doesn't exist, wait for leader to create it
-			if isFollower {
-				log.Info("Follower replica: waiting for leader to create PAR identity secret")
-				secret, err = waitForSecretCreation(ctx, client, ns, secretName)
-				if err != nil {
-					return nil, err
-				}
-				// Continue to parse the secret below
-			} else {
-				// Leader or no leader election - return nil to trigger self-enrollment
-				return nil, nil
-			}
-		} else {
+		if !k8serrors.IsNotFound(err) {
 			return nil, fmt.Errorf("failed to get identity secret: %w", err)
 		}
+		if le.IsLeader() {
+			// Leader - return nil to trigger self-enrollment
+			return nil, nil
+		}
+		// Follower - since secret doesn't exist, wait for leader to create it
+		log.Info("Follower replica: waiting for leader to create PAR identity secret")
+		secret, err = waitForSecretCreation(ctx, client, ns, secretName)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	privateKey, ok := secret.Data[privateKeyField]
@@ -90,12 +88,11 @@ func getIdentityFromK8sSecret(ctx context.Context, cfg configModel.Reader) (*Per
 
 // persistIdentityToK8sSecret saves the enrollment result to a Kubernetes secret
 func persistIdentityToK8sSecret(ctx context.Context, cfg configModel.Reader, result *Result) error {
-	// Check if this replica is the leader before creating the secret
 	le, err := leaderelection.GetLeaderEngine()
 	if err != nil {
-		log.Warnf("Failed to get leader engine, proceeding without leader check: %v", err)
-		// Fall through to create secret anyway if leader election is not available
-	} else if !le.IsLeader() {
+		return err
+	}
+	if !le.IsLeader() {
 		log.Info("Not leader, skipping PAR identity secret creation (leader will handle it)")
 		return nil
 	}

pkg/util/kubernetes/apiserver/leaderelection/leaderelection.go

@@ -191,6 +191,7 @@ func (le *LeaderEngine) init() error {
 func (le *LeaderEngine) StartLeaderElectionRun() {
 	le.once.Do(
 		func() {
+			log.Infof("Starting leader election goroutine for %q", le.HolderIdentity)
 			go le.runLeaderElection()
 		},
 	)
@@ -209,14 +210,20 @@ func (le *LeaderEngine) EnsureLeaderElectionRuns() error {
 
 	le.StartLeaderElectionRun()
 
+	log.Infof("Waiting for leader election to determine a leader (timeout: %s)...", getLeaderTimeout)
 	timeout := time.After(getLeaderTimeout)
 	tick := time.NewTicker(time.Second)
 	defer tick.Stop()
 	for {
-		log.Tracef("Waiting for new leader identity...")
 		select {
 		case <-tick.C:
-			leaderIdentity := le.GetLeader()
+			// Query Kubernetes directly to see current leader, don't wait for callback
+			leaderIdentity, err := le.getCurrentLeader()
+			if err != nil {
+				log.Warnf("Error querying current leader: %v", err)
+				continue
+			}
+			log.Infof("Polling for leader: current leader identity = %q", leaderIdentity)
 			if leaderIdentity != "" {
 				log.Infof("Leader election running, current leader is %q", leaderIdentity)
 				le.running = true