Repository: rancher/fleet
Pull request #3364 (2025-02-17 15:01:15)

Switch agent from StatefulSet to Deployment (SURE-9419)
Part of #3096

<!-- Specify the issue ID that this pull request is solving -->
<!-- Make sure that the referenced issue provides steps to reproduce it -->

<!-- Describe the changes introduced by this pull request -->

<!--
  Please provide a unit, integration (`./integrationtests/`) or e2e (`./e2e/`) test if possible.
-->


Patch:
diff --git a/.github/scripts/deploy-fleet.sh b/.github/scripts/deploy-fleet.sh
index 7aa7e297..d2865eeb 100755
--- a/.github/scripts/deploy-fleet.sh
+++ b/.github/scripts/deploy-fleet.sh
@@ -74,7 +74,7 @@ eventually helm upgrade --install fleet charts/fleet \
 # wait for controller and agent rollout
 kubectl -n cattle-fleet-system rollout status deploy/fleet-controller
 { grep -E -q -m 1 "fleet-agent-local.*1/1"; kill $!; } < <(kubectl get bundles -n fleet-local -w)
-kubectl -n cattle-fleet-system rollout status statefulset/fleet-agent
+kubectl -n cattle-fleet-system rollout status deployment/fleet-agent
 
 # label local cluster
 kubectl patch clusters.fleet.cattle.io -n fleet-local local --type=json -p '[{"op": "add", "path": "/metadata/labels/management.cattle.io~1cluster-display-name", "value": "local" }]'
diff --git a/.github/scripts/upgrade-rancher-fleet-to-dev-fleet.sh b/.github/scripts/upgrade-rancher-fleet-to-dev-fleet.sh
index fc36433e..7247017b 100755
--- a/.github/scripts/upgrade-rancher-fleet-to-dev-fleet.sh
+++ b/.github/scripts/upgrade-rancher-fleet-to-dev-fleet.sh
@@ -46,6 +46,6 @@ sleep 5
 { grep -E -q -m 1 "fleet-agent-c.*1/1"; kill $!; } < <(kubectl get bundles -n fleet-default -w)
 
 kubectl config use-context k3d-downstream
-kubectl -n cattle-fleet-system rollout status statefulset/fleet-agent
+kubectl -n cattle-fleet-system rollout status deployment/fleet-agent
 
 helm list -A
diff --git a/charts/fleet-agent/templates/deployment.yaml b/charts/fleet-agent/templates/deployment.yaml
index d689cd9f..61fe5675 100644
--- a/charts/fleet-agent/templates/deployment.yaml
+++ b/charts/fleet-agent/templates/deployment.yaml
@@ -1,9 +1,8 @@
 apiVersion: apps/v1
-kind: StatefulSet
+kind: Deployment
 metadata:
   name: fleet-agent
 spec:
-  serviceName: fleet-agent
   selector:
     matchLabels:
       app: fleet-agent
@@ -18,6 +17,18 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+        {{- if $.Values.leaderElection.leaseDuration }}
+        - name: CATTLE_ELECTION_LEASE_DURATION
+          value: {{$.Values.leaderElection.leaseDuration}}
+        {{- end }}
+        {{- if $.Values.leaderElection.retryPeriod }}
+        - name: CATTLE_ELECTION_RETRY_PERIOD
+          value: {{$.Values.leaderElection.retryPeriod}}
+        {{- end }}
+        {{- if $.Values.leaderElection.renewDeadline }}
+        - name: CATTLE_ELECTION_RENEW_DEADLINE
+          value: {{$.Values.leaderElection.renewDeadline}}
+        {{- end }}
         image: '{{ template "system_default_registry" . }}{{.Values.image.repository}}:{{.Values.image.tag}}'
         name: fleet-agent-register
         command:
@@ -51,6 +62,18 @@ spec:
         - name: DRIFT_RECONCILER_WORKERS
           value: {{ quote $.Values.agent.reconciler.workers.drift }}
         {{- end }}
+        {{- if $.Values.leaderElection.leaseDuration }}
+        - name: CATTLE_ELECTION_LEASE_DURATION
+          value: {{$.Values.leaderElection.leaseDuration}}
+        {{- end }}
+        {{- if $.Values.leaderElection.retryPeriod }}
+        - name: CATTLE_ELECTION_RETRY_PERIOD
+          value: {{$.Values.leaderElection.retryPeriod}}
+        {{- end }}
+        {{- if $.Values.leaderElection.renewDeadline }}
+        - name: CATTLE_ELECTION_RENEW_DEADLINE
+          value: {{$.Values.leaderElection.renewDeadline}}
+        {{- end }}
         image: '{{ template "system_default_registry" . }}{{.Values.image.repository}}:{{.Values.image.tag}}'
         name: fleet-agent
         command:
@@ -79,6 +102,18 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+        {{- if $.Values.leaderElection.leaseDuration }}
+        - name: CATTLE_ELECTION_LEASE_DURATION
+          value: {{$.Values.leaderElection.leaseDuration}}
+        {{- end }}
+        {{- if $.Values.leaderElection.retryPeriod }}
+        - name: CATTLE_ELECTION_RETRY_PERIOD
+          value: {{$.Values.leaderElection.retryPeriod}}
+        {{- end }}
+        {{- if $.Values.leaderElection.renewDeadline }}
+        - name: CATTLE_ELECTION_RENEW_DEADLINE
+          value: {{$.Values.leaderElection.renewDeadline}}
+        {{- end }}
         image: '{{ template "system_default_registry" . }}{{.Values.image.repository}}:{{.Values.image.tag}}'
         name: fleet-agent-clusterstatus
         command:
diff --git a/charts/fleet-agent/templates/service.yaml b/charts/fleet-agent/templates/service.yaml
deleted file mode 100644
index 36d35588..00000000
--- a/charts/fleet-agent/templates/service.yaml
+++ /dev/null
@@ -1,9 +0,0 @@
-apiVersion: v1
-kind: Service
-metadata:
-  name: fleet-agent
-spec:
-    type: ClusterIP
-    clusterIP: None
-    selector:
-      app: fleet-agent
diff --git a/charts/fleet-agent/values.yaml b/charts/fleet-agent/values.yaml
index f3ef3771..161bf50b 100644
--- a/charts/fleet-agent/values.yaml
+++ b/charts/fleet-agent/values.yaml
@@ -81,3 +81,9 @@ agent:
     workers:
       bundledeployment: "50"
       drift: "50"
+
+## Leader election configuration
+leaderElection:
+  leaseDuration: 30s
+  retryPeriod: 10s
+  renewDeadline: 25s
diff --git a/dev/setup-fleet b/dev/setup-fleet
index 81ce0ecb..da800938 100755
--- a/dev/setup-fleet
+++ b/dev/setup-fleet
@@ -50,7 +50,7 @@ helm -n cattle-fleet-system upgrade --install --create-namespace --wait --reset-
 # wait for controller and agent rollout
 kubectl -n cattle-fleet-system rollout status deployment/fleet-controller
 { grep -E -q -m 1 "fleet-agent-local.*1/1"; kill $!; } < <(kubectl get bundles -n fleet-local -w)
-kubectl -n cattle-fleet-system rollout status statefulset/fleet-agent
+kubectl -n cattle-fleet-system rollout status deployment/fleet-agent
 
 # label local cluster
 kubectl patch clusters.fleet.cattle.io -n fleet-local local --type=json -p '[{"op": "add", "path": "/metadata/labels/management.cattle.io~1cluster-display-name", "value": "local" }]'
diff --git a/internal/cmd/agent/clusterstatus.go b/internal/cmd/agent/clusterstatus.go
index e8b33098..1148e309 100644
--- a/internal/cmd/agent/clusterstatus.go
+++ b/internal/cmd/agent/clusterstatus.go
@@ -3,16 +3,22 @@ package agent
 import (
 	"context"
 	"fmt"
+	"os"
 	"time"
 
 	"github.com/spf13/cobra"
+
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/discovery"
 	memory "k8s.io/client-go/discovery/cached"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/restmapper"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/util/workqueue"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -87,75 +93,126 @@ func (cs *ClusterStatus) Run(cmd *cobra.Command, args []string) error {
 	localConfig := rest.CopyConfig(kc)
 	localConfig.RateLimiter = ratelimit.None
 
-	// cannot start without kubeconfig for upstream cluster
-	setupLog.Info("Fetching kubeconfig for upstream cluster from registration", "namespace", cs.Namespace)
-	agentInfo, _, err := register.Get(ctx, cs.Namespace, localConfig)
-	if err != nil {
-		setupLog.Error(err, "failed to get kubeconfig from upstream cluster")
-		return err
-	}
-
-	// set up factory for upstream cluster
-	fleetNamespace, _, err := agentInfo.ClientConfig.Namespace()
-	if err != nil {
-		setupLog.Error(err, "failed to get namespace from upstream cluster")
-		return err
-	}
-
-	fleetRESTConfig, err := agentInfo.ClientConfig.ClientConfig()
+	localClient, err := kubernetes.NewForConfig(kc)
 	if err != nil {
-		setupLog.Error(err, "failed to get kubeconfig from upstream cluster")
-		return err
+		return fmt.Errorf("failed to create local client: %w", err)
 	}
 
-	//  now we have both configs
-	fleetMapper, mapper, _, err := newMappers(ctx, fleetRESTConfig, clientConfig)
+	identifier, err := getUniqueIdentifier()
 	if err != nil {
-		setupLog.Error(err, "failed to get mappers")
-		return err
+		return fmt.Errorf("failed to get unique identifier: %w", err)
 	}
 
-	fleetSharedFactory, err := newSharedControllerFactory(fleetRESTConfig, fleetMapper, fleetNamespace)
-	if err != nil {
-		setupLog.Error(err, "failed to build shared controller factory")
-		return err
-	}
-
-	fleetFactory, err := fleet.NewFactoryFromConfigWithOptions(fleetRESTConfig, &fleet.FactoryOptions{
-		SharedControllerFactory: fleetSharedFactory,
-	})
-	if err != nil {
-		setupLog.Error(err, "failed to build fleet factory")
-		return err
+	lock := resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      leaseLockNameClusterStatus,
+			Namespace: cs.Namespace,
+		},
+		Client: localClient.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identifier,
+		},
 	}
 
-	// set up factory for local cluster
-	localFactory, err := newSharedControllerFactory(localConfig, mapper, "")
+	leaderOpts, err := command.NewLeaderElectionOptions()
 	if err != nil {
-		setupLog.Error(err, "failed to build shared controller factory")
 		return err
 	}
 
-	coreFactory, err := core.NewFactoryFromConfigWithOptions(localConfig, &core.FactoryOptions{
-		SharedControllerFactory: localFactory,
-	})
-	if err != nil {
-		setupLog.Error(err, "failed to build core factory")
-		return err
+	leaderElectionConfig := leaderelection.LeaderElectionConfig{
+		Lock:          &lock,
+		LeaseDuration: *leaderOpts.LeaseDuration,
+		RetryPeriod:   *leaderOpts.RetryPeriod,
+		RenewDeadline: *leaderOpts.RenewDeadline,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				// cannot start without kubeconfig for upstream cluster
+				setupLog.Info("Fetching kubeconfig for upstream cluster from registration", "namespace", cs.Namespace)
+				agentInfo, _, err := register.Get(ctx, cs.Namespace, localConfig)
+				if err != nil {
+					setupLog.Error(err, "failed to get kubeconfig from upstream cluster")
+					return
+				}
+
+				// set up factory for upstream cluster
+				fleetNamespace, _, err := agentInfo.ClientConfig.Namespace()
+				if err != nil {
+					setupLog.Error(err, "failed to get namespace from upstream cluster")
+					return
+				}
+
+				fleetRESTConfig, err := agentInfo.ClientConfig.ClientConfig()
+				if err != nil {
+					setupLog.Error(err, "failed to get kubeconfig from upstream cluster")
+					return
+				}
+
+				//  now we have both configs
+				fleetMapper, mapper, _, err := newMappers(ctx, fleetRESTConfig, clientConfig)
+				if err != nil {
+					setupLog.Error(err, "failed to get mappers")
+					return
+				}
+
+				fleetSharedFactory, err := newSharedControllerFactory(fleetRESTConfig, fleetMapper, fleetNamespace)
+				if err != nil {
+					setupLog.Error(err, "failed to build shared controller factory")
+					return
+				}
+
+				fleetFactory, err := fleet.NewFactoryFromConfigWithOptions(fleetRESTConfig, &fleet.FactoryOptions{
+					SharedControllerFactory: fleetSharedFactory,
+				})
+				if err != nil {
+					setupLog.Error(err, "failed to build fleet factory")
+					return
+				}
+
+				// set up factory for local cluster
+				localFactory, err := newSharedControllerFactory(localConfig, mapper, "")
+				if err != nil {
+					setupLog.Error(err, "failed to build shared controller factory")
+					return
+				}
+
+				coreFactory, err := core.NewFactoryFromConfigWithOptions(localConfig, &core.FactoryOptions{
+					SharedControllerFactory: localFactory,
+				})
+				if err != nil {
+					setupLog.Error(err, "failed to build core factory")
+					return
+				}
+
+				setupLog.Info("Starting cluster status ticker", "checkin interval", checkinInterval.String(), "cluster namespace", agentInfo.ClusterNamespace, "cluster name", agentInfo.ClusterName)
+
+				go func() {
+					clusterstatus.Ticker(ctx,
+						cs.Namespace,
+						agentInfo.ClusterNamespace,
+						agentInfo.ClusterName,
+						checkinInterval,
+						coreFactory.Core().V1().Node(),
+						fleetFactory.Fleet().V1alpha1().Cluster(),
+					)
+
+					<-cmd.Context().Done()
+				}()
+			},
+			OnStoppedLeading: func() {
+				setupLog.Info("stopped leading")
+				os.Exit(1)
+			},
+			OnNewLeader: func(identity string) {
+				if identity == identifier {
+					setupLog.Info("renewed leader", "new identity", identity, "own identity", identifier)
+				} else {
+					setupLog.Info("new leader", "new identity", identity, "own identity", identifier)
+				}
+			},
+		},
 	}
 
-	setupLog.Info("Starting cluster status ticker", "checkin interval", checkinInterval.String(), "cluster namespace", agentInfo.ClusterNamespace, "cluster name", agentInfo.ClusterName)
-
-	clusterstatus.Ticker(ctx,
-		cs.Namespace,
-		agentInfo.ClusterNamespace,
-		agentInfo.ClusterName,
-		checkinInterval,
-		coreFactory.Core().V1().Node(),
-		fleetFactory.Fleet().V1alpha1().Cluster(),
-	)
-
-	<-cmd.Context().Done()
+	leaderelection.RunOrDie(ctx, leaderElectionConfig)
 
 	return nil
 }
diff --git a/internal/cmd/agent/register.go b/internal/cmd/agent/register.go
index fa878458..9ae7b78a 100644
--- a/internal/cmd/agent/register.go
+++ b/internal/cmd/agent/register.go
@@ -3,9 +3,15 @@ package agent
 import (
 	"context"
 	"fmt"
+	"os"
 
 	"github.com/spf13/cobra"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+
 	command "github.com/rancher/fleet/internal/cmd"
 	"github.com/rancher/fleet/internal/cmd/agent/register"
 
@@ -53,32 +59,82 @@ func (r *Register) Run(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
-	setupLog.Info("starting registration on upstream cluster", "namespace", r.Namespace)
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	// try to register with upstream fleet controller by obtaining
-	// a kubeconfig for the upstream cluster
-	agentInfo, err := register.Register(ctx, r.Namespace, kc)
+	localClient, err := kubernetes.NewForConfig(kc)
 	if err != nil {
-		setupLog.Error(err, "failed to register with upstream cluster")
-		return err
+		return fmt.Errorf("failed to create local client: %w", err)
 	}
 
-	ns, _, err := agentInfo.ClientConfig.Namespace()
+	identifier, err := getUniqueIdentifier()
 	if err != nil {
-		setupLog.Error(err, "failed to get namespace from upstream cluster")
-		return err
+		return fmt.Errorf("failed to get unique identifier: %w", err)
 	}
 
-	_, err = agentInfo.ClientConfig.ClientConfig()
+	lock := resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      leaseLockNameRegister,
+			Namespace: r.Namespace,
+		},
+		Client: localClient.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identifier,
+		},
+	}
+
+	leaderOpts, err := command.NewLeaderElectionOptions()
 	if err != nil {
-		setupLog.Error(err, "failed to get kubeconfig from upstream cluster")
 		return err
 	}
 
-	setupLog.Info("successfully registered with upstream cluster", "namespace", ns)
+	leaderElectionConfig := leaderelection.LeaderElectionConfig{
+		Lock:          &lock,
+		LeaseDuration: *leaderOpts.LeaseDuration,
+		RetryPeriod:   *leaderOpts.RetryPeriod,
+		RenewDeadline: *leaderOpts.RenewDeadline,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				setupLog.Info("starting registration on upstream cluster", "namespace", r.Namespace)
+
+				ctx, cancel := context.WithCancel(ctx)
+				defer cancel()
+
+				// try to register with upstream fleet controller by obtaining
+				// a kubeconfig for the upstream cluster
+				agentInfo, err := register.Register(ctx, r.Namespace, kc)
+				if err != nil {
+					setupLog.Error(err, "failed to register with upstream cluster")
+					return
+				}
+
+				ns, _, err := agentInfo.ClientConfig.Namespace()
+				if err != nil {
+					setupLog.Error(err, "failed to get namespace from upstream cluster")
+					return
+				}
+
+				_, err = agentInfo.ClientConfig.ClientConfig()
+				if err != nil {
+					setupLog.Error(err, "failed to get kubeconfig from upstream cluster")
+					return
+				}
+
+				setupLog.Info("successfully registered with upstream cluster", "namespace", ns)
+				os.Exit(0)
+			},
+			OnStoppedLeading: func() {
+				setupLog.Info("stopped leading")
+				os.Exit(1)
+			},
+			OnNewLeader: func(identity string) {
+				if identity == identifier {
+					setupLog.Info("renewed leader", "identity", identity)
+				} else {
+					setupLog.Info("new leader", "identity", identity)
+				}
+			},
+		},
+	}
+
+	leaderelection.RunOrDie(ctx, leaderElectionConfig)
 
 	return nil
 }
diff --git a/internal/cmd/agent/root.go b/internal/cmd/agent/root.go
index 2030c02d..9558f4e3 100644
--- a/internal/cmd/agent/root.go
+++ b/internal/cmd/agent/root.go
@@ -1,6 +1,7 @@
 package agent
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	glog "log"
@@ -10,6 +11,11 @@ import (
 
 	"github.com/spf13/cobra"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -34,6 +40,12 @@ type AgentReconcilerWorkers struct {
 	Drift            int
 }
 
+const (
+	leaseLockName              = "fleet-agent"
+	leaseLockNameClusterStatus = "fleet-agent-clusterstatus"
+	leaseLockNameRegister      = "fleet-agent-register"
+)
+
 var (
 	setupLog = ctrl.Log.WithName("setup")
 	zopts    = &zap.Options{
@@ -49,41 +61,107 @@ func (a *FleetAgent) PersistentPre(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
+func getUniqueIdentifier() (string, error) {
+	// For in-cluster deployments.
+	if podName, ok := os.LookupEnv("POD_NAME"); ok {
+		return podName, nil
+	}
+
+	// For local development, combine hostname with process ID
+	hostname, err := os.Hostname()
+	if err != nil {
+		return "", fmt.Errorf("failed to get hostname: %w", err)
+	}
+
+	pid := os.Getpid()
+	return fmt.Sprintf("%s-%d", hostname, pid), nil
+}
+
 func (a *FleetAgent) Run(cmd *cobra.Command, args []string) error {
-	ctrl.SetLogger(zap.New(zap.UseFlagOptions(zopts)))
+	logger := zap.New(zap.UseFlagOptions(zopts))
+	ctrl.SetLogger(logger)
+
+	if a.Namespace == "" {
+		return fmt.Errorf("--namespace or env NAMESPACE is required to be set")
+	}
+
 	ctx := log.IntoContext(cmd.Context(), ctrl.Log)
 
 	localConfig := ctrl.GetConfigOrDie()
-	workersOpts := AgentReconcilerWorkers{}
-
-	if d := os.Getenv("BUNDLEDEPLOYMENT_RECONCILER_WORKERS"); d != "" {
-		w, err := strconv.Atoi(d)
-		if err != nil {
-			setupLog.Error(err, "failed to parse BUNDLEDEPLOYMENT_RECONCILER_WORKERS", "value", d)
-		}
-		workersOpts.BundleDeployment = w
+	localClient, err := kubernetes.NewForConfig(localConfig)
+	if err != nil {
+		return fmt.Errorf("failed to create local client: %w", err)
+	}
+
+	identifier, err := getUniqueIdentifier()
+	if err != nil {
+		return fmt.Errorf("failed to get unique identifier: %w", err)
+	}
+
+	lock := resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{
+			Name:      leaseLockName,
+			Namespace: a.Namespace,
+		},
+		Client: localClient.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identifier,
+		},
 	}
 
-	if d := os.Getenv("DRIFT_RECONCILER_WORKERS"); d != "" {
-		w, err := strconv.Atoi(d)
-		if err != nil {
-			setupLog.Error(err, "failed to parse DRIFT_RECONCILER_WORKERS", "value", d)
-		}
-		workersOpts.Drift = w
+	leaderOpts, err := command.NewLeaderElectionOptions()
+	if err != nil {
+		return err
 	}
 
 	go func() {
 		glog.Println(http.ListenAndServe("localhost:6060", nil)) // nolint:gosec // Debugging only
 	}()
 
-	if a.Namespace == "" {
-		return fmt.Errorf("--namespace or env NAMESPACE is required to be set")
-	}
-	if err := start(ctx, localConfig, a.Namespace, a.AgentScope, workersOpts); err != nil {
-		return err
+	leaderElectionConfig := leaderelection.LeaderElectionConfig{
+		Lock:          &lock,
+		LeaseDuration: *leaderOpts.LeaseDuration,
+		RetryPeriod:   *leaderOpts.RetryPeriod,
+		RenewDeadline: *leaderOpts.RenewDeadline,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				workersOpts := AgentReconcilerWorkers{}
+
+				if d := os.Getenv("BUNDLEDEPLOYMENT_RECONCILER_WORKERS"); d != "" {
+					w, err := strconv.Atoi(d)
+					if err != nil {
+						setupLog.Error(err, "failed to parse BUNDLEDEPLOYMENT_RECONCILER_WORKERS", "value", d)
+					}
+					workersOpts.BundleDeployment = w
+				}
+
+				if d := os.Getenv("DRIFT_RECONCILER_WORKERS"); d != "" {
+					w, err := strconv.Atoi(d)
+					if err != nil {
+						setupLog.Error(err, "failed to parse DRIFT_RECONCILER_WORKERS", "value", d)
+					}
+					workersOpts.Drift = w
+				}
+				if err := start(ctx, localConfig, a.Namespace, a.AgentScope, workersOpts); err != nil {
+					setupLog.Error(err, "failed to start agent")
+				}
+			},
+			OnStoppedLeading: func() {
+				setupLog.Info("stopped leading")
+				os.Exit(1)
+			},
+			OnNewLeader: func(identity string) {
+				if identity == identifier {
+					setupLog.Info("renewed leader", "identity", identity)
+				} else {
+					setupLog.Info("new leader", "identity", identity)
+				}
+			},
+		},
 	}
 
-	<-cmd.Context().Done()
+	leaderelection.RunOrDie(ctx, leaderElectionConfig)
+
 	return nil
 }
 
diff --git a/internal/cmd/controller/agentmanagement/agent/manifest.go b/internal/cmd/controller/agentmanagement/agent/manifest.go
index b879b253..ce3ac99c 100644
--- a/internal/cmd/controller/agentmanagement/agent/manifest.go
+++ b/internal/cmd/controller/agentmanagement/agent/manifest.go
@@ -25,7 +25,10 @@ var (
 )
 
 const (
-	DefaultName = "fleet-agent"
+	DefaultName           = "fleet-agent"
+	electionLeaseDuration = "30s"
+	electionRetryPeriod   = "10s"
+	electionRenewDeadline = "25s"
 )
 
 type ManifestOptions struct {
@@ -119,24 +122,9 @@ func Manifest(namespace string, agentScope string, opts ManifestOptions) []runti
 		},
 	}
 
-	// StatefulSets require a headless service. The service is not used.
-	service := &corev1.Service{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      DefaultName,
-			Namespace: namespace,
-		},
-		Spec: corev1.ServiceSpec{
-			Selector: map[string]string{
-				"app": DefaultName,
-			},
-			Type:      corev1.ServiceTypeClusterIP,
-			ClusterIP: "None",
-		},
-	}
-
 	var objs []runtime.Object
 	objs = append(objs, clusterRole...)
-	objs = append(objs, admin, defaultSa, service, agent, networkPolicy)
+	objs = append(objs, admin, defaultSa, agent, networkPolicy)
 
 	return objs
 }
@@ -152,23 +140,22 @@ func Resolve(global, prefix, image string) string {
 	return image
 }
 
-func agentApp(namespace string, agentScope string, opts ManifestOptions) *appsv1.StatefulSet {
+func agentApp(namespace string, agentScope string, opts ManifestOptions) *appsv1.Deployment {
 	name := DefaultName
 	serviceAccount := DefaultName
 	image := Resolve(opts.SystemDefaultRegistry, opts.PrivateRepoURL, opts.AgentImage)
 
-	app := &appsv1.StatefulSet{
+	app := &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: namespace,
 			Name:      name,
 		},
-		Spec: appsv1.StatefulSetSpec{
+		Spec: appsv1.DeploymentSpec{
 			Selector: &metav1.LabelSelector{
 				MatchLabels: map[string]string{
 					"app": name,
 				},
 			},
-			ServiceName: name,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
@@ -191,6 +178,9 @@ func agentApp(namespace string, agentScope string, opts ManifestOptions) *appsv1
 										},
 									},
 								},
+								{Name: "CATTLE_ELECTION_LEASE_DURATION", Value: electionLeaseDuration},
+								{Name: "CATTLE_ELECTION_RETRY_PERIOD", Value: electionRetryPeriod},
+								{Name: "CATTLE_ELECTION_RENEW_DEADLINE", Value: electionRenewDeadline},
 							},
 							Command: []string{
 								"fleetagent",
@@ -221,6 +211,9 @@ func agentApp(namespace string, agentScope string, opts ManifestOptions) *appsv1
 									},
 								},
 								{Name: "AGENT_SCOPE", Value: agentScope},
+								{Name: "CATTLE_ELECTION_LEASE_DURATION", Value: electionLeaseDuration},
+								{Name: "CATTLE_ELECTION_RETRY_PERIOD", Value: electionRetryPeriod},
+								{Name: "CATTLE_ELECTION_RENEW_DEADLINE", Value: electionRenewDeadline},
 							},
 							Command: []string{
 								"fleetagent",
@@ -250,6 +243,9 @@ func agentApp(namespace string, agentScope string, opts ManifestOptions) *appsv1
 									},
 								},
 								{Name: "CHECKIN_INTERVAL", Value: opts.CheckinInterval},
+								{Name: "CATTLE_ELECTION_LEASE_DURATION", Value: electionLeaseDuration},
+								{Name: "CATTLE_ELECTION_RETRY_PERIOD", Value: electionRetryPeriod},
+								{Name: "CATTLE_ELECTION_RENEW_DEADLINE", Value: electionRenewDeadline},
 							},
 							Command: []string{
 								"fleetagent",


Test patch:
diff --git a/e2e/installation/verify_test.go b/e2e/installation/verify_test.go
index 20f996db..d981a6d2 100644
--- a/e2e/installation/verify_test.go
+++ b/e2e/installation/verify_test.go
@@ -63,7 +63,7 @@ var _ = Describe("Fleet Installation", func() {
 	Context("Verify agents are updated to new image", func() {
 		It("has the expected fleet images", func() {
 			Eventually(func() string {
-				out, _ := k.Namespace(localAgentNamespace).Get("statefulsets", "-owide")
+				out, _ := k.Namespace(localAgentNamespace).Get("deployments", "-owide")
 				return out
 			}).Should(ContainSubstring("rancher/fleet-agent:" + version))
 		})
@@ -71,7 +71,7 @@ var _ = Describe("Fleet Installation", func() {
 		It("has the expected fleet-agent image in the downstream cluster", Label("multi-cluster"), func() {
 			kd := env.Kubectl.Context(env.Downstream)
 			Eventually(func() string {
-				out, _ := kd.Namespace(agentNamespace).Get("statefulsets", "-owide")
+				out, _ := kd.Namespace(agentNamespace).Get("deployments", "-owide")
 				return out
 			}).Should(ContainSubstring("rancher/fleet-agent:" + version))
 		})
diff --git a/e2e/multi-cluster/installation/agent_test.go b/e2e/multi-cluster/installation/agent_test.go
index 5bea51fd..95b6cc70 100644
--- a/e2e/multi-cluster/installation/agent_test.go
+++ b/e2e/multi-cluster/installation/agent_test.go
@@ -5,18 +5,24 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"github.com/onsi/gomega/matchers"
-	"github.com/rancher/fleet/e2e/testenv/kubectl"
 )
 
 var (
 	agentMode string
-	kd        kubectl.Command
 )
 
 var _ = Describe("Fleet installation with TLS agent modes", func() {
 	BeforeEach(func() {
 		kd = env.Kubectl.Context(env.Downstream)
+
+		_, err := kd.Delete(
+			"pod",
+			"-n",
+			"cattle-fleet-system",
+			"-l",
+			"app=fleet-agent",
+		)
+		Expect(err).NotTo(HaveOccurred())
 	})
 
 	JustBeforeEach(func() {
@@ -33,7 +39,6 @@ var _ = Describe("Fleet installation with TLS agent modes", func() {
 			),
 		)
 		Expect(err).ToNot(HaveOccurred(), string(out))
-
 	})
 
 	Context("with non-strict agent TLS mode", func() {
@@ -43,7 +48,7 @@ var _ = Describe("Fleet installation with TLS agent modes", func() {
 			})
 
 			It("reaches the server without cert issues", func() {
-				Eventually(func() bool {
+				Eventually(func(g Gomega) error {
 					logs, err := kd.Namespace("cattle-fleet-system").Logs(
 						"-l",
 						"app=fleet-agent",
@@ -52,19 +57,13 @@ var _ = Describe("Fleet installation with TLS agent modes", func() {
 						"--tail=-1",
 					)
 					if err != nil {
-						return false
+						return err
 					}
 
-					regexMatcher := matchers.MatchRegexpMatcher{
-						Regexp: "Failed to register agent.*could not find the requested resource",
-					}
-					reachesServerWithoutCertIssue, err := regexMatcher.Match(logs)
-					if err != nil {
-						return false
-					}
+					g.Expect(logs).To(MatchRegexp("Failed to register agent.*could not find the requested resource"))
 
-					return reachesServerWithoutCertIssue
-				}).Should(BeTrue())
+					return nil
+				}).ShouldNot(HaveOccurred())
 			})
 		})
 	})
@@ -76,7 +75,7 @@ var _ = Describe("Fleet installation with TLS agent modes", func() {
 			})
 
 			It("cannot reach the server because the cert is signed by an unknown authority", func() {
-				Eventually(func() bool {
+				Eventually(func(g Gomega) error {
 					logs, err := kd.Namespace("cattle-fleet-system").Logs(
 						"-l",
 						"app=fleet-agent",
@@ -85,19 +84,16 @@ var _ = Describe("Fleet installation with TLS agent modes", func() {
 						"--tail=-1",
 					)
 					if err != nil {
-						return false
+						return err
 					}
 
-					regexMatcher := matchers.MatchRegexpMatcher{
-						Regexp: "Failed to register agent.*signed by unknown authority",
-					}
-					reachesServerWithoutCertIssue, err := regexMatcher.Match(logs)
+					g.Expect(logs).To(MatchRegexp("Failed to register agent.*signed by unknown authority"))
 					if err != nil {
-						return false
+						return err
 					}
 
-					return reachesServerWithoutCertIssue
-				}).Should(BeTrue())
+					return nil
+				}).ShouldNot(HaveOccurred())
 			})
 		})
 	})
diff --git a/e2e/multi-cluster/installation/suite_test.go b/e2e/multi-cluster/installation/suite_test.go
index e591d9df..26d1e3fc 100644
--- a/e2e/multi-cluster/installation/suite_test.go
+++ b/e2e/multi-cluster/installation/suite_test.go
@@ -20,9 +20,11 @@ func TestE2E(t *testing.T) {
 }
 
 var (
-	env    *testenv.Env
-	ku     kubectl.Command
-	config string
+	env      *testenv.Env
+	ku       kubectl.Command
+	kd       kubectl.Command
+	config   string
+	strategy string
 )
 
 var _ = BeforeSuite(func() {
@@ -31,6 +33,7 @@ var _ = BeforeSuite(func() {
 
 	env = testenv.New()
 	ku = env.Kubectl.Context(env.Upstream)
+	kd = env.Kubectl.Context(env.Downstream)
 
 	// Save initial state of `fleet-controller` config map
 	cfg, err := ku.Get(
@@ -42,6 +45,29 @@ var _ = BeforeSuite(func() {
 		"jsonpath={.data.config}")
 	Expect(err).ToNot(HaveOccurred(), cfg)
 
+	// Save initial state of `fleet-agent` deployment
+	strategy, err = kd.Get(
+		"deployment",
+		"fleet-agent",
+		"-n",
+		"cattle-fleet-system",
+		"-o",
+		"jsonpath={.spec.strategy}",
+	)
+	Expect(err).ToNot(HaveOccurred(), cfg)
+
+	// Patch `fleet-agent` deployment to use Recreate strategy
+	out, err := kd.Patch(
+		"deployment",
+		"fleet-agent",
+		"-n",
+		"cattle-fleet-system",
+		"--type=merge",
+		"-p",
+		`{"spec":{"strategy":{"type":"Recreate", "rollingUpdate":null}}}`,
+	)
+	Expect(err).ToNot(HaveOccurred(), string(out))
+
 	cfg = strings.ReplaceAll(cfg, `"`, `\"`)
 	config = strings.ReplaceAll(cfg, "\n", "")
 })
@@ -58,4 +84,16 @@ var _ = AfterSuite(func() {
 		fmt.Sprintf(`{"data":{"config":"%s"}}`, config),
 	)
 	Expect(err).ToNot(HaveOccurred(), string(out))
+
+	// Restore initial state of deployment
+	out, err = kd.Patch(
+		"deployment",
+		"fleet-agent",
+		"-n",
+		"cattle-fleet-system",
+		"--type=merge",
+		"-p",
+		fmt.Sprintf(`{"spec":{"strategy":%s}}`, strategy),
+	)
+	Expect(err).ToNot(HaveOccurred(), string(out))
 })
diff --git a/internal/cmd/controller/agentmanagement/agent/manifest_test.go b/internal/cmd/controller/agentmanagement/agent/manifest_test.go
index da6ecd5c..985d0ffa 100644
--- a/internal/cmd/controller/agentmanagement/agent/manifest_test.go
+++ b/internal/cmd/controller/agentmanagement/agent/manifest_test.go
@@ -36,10 +36,10 @@ func TestImageResolve(t *testing.T) {
 	}
 }
 
-func getAgentFromManifests(scope string, opts agent.ManifestOptions) *appsv1.StatefulSet {
+func getAgentFromManifests(scope string, opts agent.ManifestOptions) *appsv1.Deployment {
 	objects := agent.Manifest(namespace, scope, opts)
 	for _, obj := range objects {
-		dep, ok := obj.(*appsv1.StatefulSet)
+		dep, ok := obj.(*appsv1.Deployment)
 		if ok {
 			return dep
 		}