Introduce staggered exit when caps change

This patch changes the behavior of how the pod is exited when the
capabilities have changed. Instead of all the replicas exiting at
the same time, the exits are staggered. If the pod is the leader
or no leader election is configured, the pod is exited after a
brief delay. Otherwise the pod is exited immediately.

This should allow the non-leaders to come back online before the
leader is exited, enabling one of the restarted pods to take over
as the leader when it exits.

.golangci.yml

@@ -85,7 +85,9 @@ linters-settings:
       - alias: pkgctx
         pkg: github.com/vmware-tanzu/vm-operator/pkg/context
       - alias: pkgerr
-        pkg: github.com/vmware-tanzu/vm-operator/pkg/pkgerr
+        pkg: github.com/vmware-tanzu/vm-operator/pkg/errors
+      - alias: pkgexit
+        pkg: github.com/vmware-tanzu/vm-operator/pkg/exit
       - alias: ctxop
         pkg: github.com/vmware-tanzu/vm-operator/pkg/context/operation
       - alias: pkgmgr

config/local/vmoperator/local_env_var_patch.yaml

@@ -13,6 +13,8 @@ spec:
           value: "true"
         - name: ASYNC_CREATE_ENABLED
           value: "true"
+        - name: EXIT_DELAY
+          value: "3s"
         - name: MEM_STATS_PERIOD
           value: "10m"
         - name: VSPHERE_NETWORKING

config/wcp/vmoperator/manager_env_var_patch.yaml

@@ -40,6 +40,12 @@
     name: PRIVILEGED_USERS
     value: "<COMMA_SEPARATED_LIST_OF_USERS>"
 
+- op: add
+  path: /spec/template/spec/containers/0/env/-
+  value:
+    name: EXIT_DELAY
+    value: "3s"
+
 - op: add
   path: /spec/template/spec/containers/0/env/-
   value:

controllers/infra/capability/configmap/configmap_capability_controller.go

@@ -20,10 +20,10 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
-	"github.com/vmware-tanzu/vm-operator/controllers/infra/capability/exit"
 	pkgcfg "github.com/vmware-tanzu/vm-operator/pkg/config"
 	"github.com/vmware-tanzu/vm-operator/pkg/config/capabilities"
 	pkgctx "github.com/vmware-tanzu/vm-operator/pkg/context"
+	pkgexit "github.com/vmware-tanzu/vm-operator/pkg/exit"
 	pkgmgr "github.com/vmware-tanzu/vm-operator/pkg/manager"
 	"github.com/vmware-tanzu/vm-operator/pkg/record"
 	kubeutil "github.com/vmware-tanzu/vm-operator/pkg/util/kube"
@@ -53,6 +53,7 @@ func AddToManager(ctx *pkgctx.ControllerManagerContext, mgr manager.Manager) err
 		cache,
 		ctrl.Log.WithName("controllers").WithName(controllerName),
 		record.New(mgr.GetEventRecorderFor(controllerNameLong)),
+		mgr.Elected(),
 	)
 
 	// This controller is also run on the non-leaders (webhooks) pods too
@@ -89,13 +90,15 @@ func NewReconciler(
 	ctx context.Context,
 	client ctrlclient.Reader,
 	logger logr.Logger,
-	recorder record.Recorder) *Reconciler {
+	recorder record.Recorder,
+	elected <-chan struct{}) *Reconciler {
 
 	return &Reconciler{
 		Context:  ctx,
 		Client:   client,
 		Logger:   logger,
 		Recorder: recorder,
+		Elected:  elected,
 	}
 }
 
@@ -104,6 +107,7 @@ type Reconciler struct {
 	Client   ctrlclient.Reader
 	Logger   logr.Logger
 	Recorder record.Recorder
+	Elected  <-chan struct{}
 }
 
 // +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch
@@ -120,8 +124,10 @@ func (r *Reconciler) Reconcile(
 	}
 
 	if capabilities.UpdateCapabilitiesFeatures(ctx, obj) {
-		r.Logger.Info("killing pod due to changed capabilities")
-		exit.Exit()
+		pkgexit.Exit(
+			logr.NewContext(ctx, r.Logger),
+			"capabilities have changed",
+			r.Elected)
 	}
 
 	return ctrl.Result{}, nil

controllers/infra/capability/configmap/configmap_capability_controller_suite_test.go

@@ -7,21 +7,26 @@ package capability_test
 import (
 	"sync/atomic"
 	"testing"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 
 	capability "github.com/vmware-tanzu/vm-operator/controllers/infra/capability/configmap"
-	"github.com/vmware-tanzu/vm-operator/controllers/infra/capability/exit"
 	pkgcfg "github.com/vmware-tanzu/vm-operator/pkg/config"
+	pkgexit "github.com/vmware-tanzu/vm-operator/pkg/exit"
 	"github.com/vmware-tanzu/vm-operator/pkg/manager"
 	"github.com/vmware-tanzu/vm-operator/test/builder"
 )
 
-var numExits int32
+var (
+	numExits int32
+	exited   = make(chan struct{})
+)
 
 func init() {
-	exit.Exit = func() {
+	pkgexit.ExitFn = func(_ time.Duration) {
 		atomic.AddInt32(&numExits, 1)
+		exited <- struct{}{}
 	}
 }
 
@@ -30,6 +35,7 @@ var suite = builder.NewTestSuiteForControllerWithContext(
 		pkgcfg.NewContextWithDefaultConfig(),
 		func(config *pkgcfg.Config) {
 			config.Features.SVAsyncUpgrade = false
+			config.ExitDelay = time.Millisecond * 100
 		},
 	),
 	capability.AddToManager,

controllers/infra/capability/configmap/configmap_capability_controller_test.go

@@ -28,6 +28,7 @@ var _ = Describe(
 		testlabels.EnvTest,
 		testlabels.V1Alpha3,
 	),
+	Ordered,
 	func() {
 
 		var (
@@ -61,6 +62,12 @@ var _ = Describe(
 		})
 
 		When("the capabilities have changed", func() {
+			JustBeforeEach(func() {
+				Eventually(exited).Should(Receive())
+			})
+			AfterEach(func() {
+				Consistently(exited).ShouldNot(Receive())
+			})
 			When("some capabilities are enabled", func() {
 				BeforeEach(func() {
 					pkgcfg.SetContext(
@@ -127,12 +134,20 @@ var _ = Describe(
 			})
 
 			JustBeforeEach(func() {
+				Eventually(exited).Should(Receive())
+
 				Eventually(func(g Gomega) {
 					g.Expect(pkgcfg.FromContext(suite.Context).Features.TKGMultipleCL).To(BeTrue())
 				}, time.Second*5).Should(Succeed())
 
 				obj.Data[capabilities.CapabilityKeyTKGMultipleContentLibraries] = "false"
 				Expect(ctx.Client.Update(ctx, obj)).To(Succeed())
+
+				Eventually(exited).Should(Receive())
+			})
+
+			AfterEach(func() {
+				Consistently(exited).ShouldNot(Receive())
 			})
 
 			Specify("the pod was exited once on create and once on update", func() {
@@ -162,6 +177,9 @@ var _ = Describe(
 					capabilities.CapabilityKeyWorkloadIsolation:           "false",
 				}
 			})
+			JustBeforeEach(func() {
+				Consistently(exited).ShouldNot(Receive())
+			})
 			Specify("the pod was not exited and features were not updated", func() {
 				Consistently(func(g Gomega) {
 					g.Expect(atomic.LoadInt32(&numExits)).To(Equal(int32(0)))

controllers/infra/capability/crd/crd_capability_controller.go

@@ -16,11 +16,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
-	"github.com/vmware-tanzu/vm-operator/controllers/infra/capability/exit"
 	capv1 "github.com/vmware-tanzu/vm-operator/external/capabilities/api/v1alpha1"
 	pkgcfg "github.com/vmware-tanzu/vm-operator/pkg/config"
 	"github.com/vmware-tanzu/vm-operator/pkg/config/capabilities"
 	pkgctx "github.com/vmware-tanzu/vm-operator/pkg/context"
+	pkgexit "github.com/vmware-tanzu/vm-operator/pkg/exit"
 	"github.com/vmware-tanzu/vm-operator/pkg/record"
 	"github.com/vmware-tanzu/vm-operator/pkg/util/ptr"
 )
@@ -39,6 +39,7 @@ func AddToManager(ctx *pkgctx.ControllerManagerContext, mgr manager.Manager) err
 		mgr.GetClient(),
 		ctrl.Log.WithName("controllers").WithName(controllerName),
 		record.New(mgr.GetEventRecorderFor(controllerNameLong)),
+		mgr.Elected(),
 	)
 
 	return ctrl.NewControllerManagedBy(mgr).
@@ -66,13 +67,15 @@ func NewReconciler(
 	ctx context.Context,
 	client ctrlclient.Client,
 	logger logr.Logger,
-	recorder record.Recorder) *Reconciler {
+	recorder record.Recorder,
+	elected <-chan struct{}) *Reconciler {
 
 	return &Reconciler{
 		Context:  ctx,
 		Client:   client,
 		Logger:   logger,
 		Recorder: recorder,
+		Elected:  elected,
 	}
 }
 
@@ -81,6 +84,7 @@ type Reconciler struct {
 	Client   ctrlclient.Client
 	Logger   logr.Logger
 	Recorder record.Recorder
+	Elected  <-chan struct{}
 }
 
 // +kubebuilder:rbac:groups=iaas.vmware.com,resources=capabilities,verbs=get;list;watch
@@ -98,8 +102,10 @@ func (r *Reconciler) Reconcile(
 	}
 
 	if capabilities.UpdateCapabilitiesFeatures(ctx, obj) {
-		r.Logger.Info("killing pod due to changed capabilities")
-		exit.Exit()
+		pkgexit.Exit(
+			logr.NewContext(ctx, r.Logger),
+			"capabilities have changed",
+			r.Elected)
 	}
 
 	return ctrl.Result{}, nil

controllers/infra/capability/crd/crd_capability_controller_suite_test.go

@@ -7,21 +7,26 @@ package capability_test
 import (
 	"sync/atomic"
 	"testing"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 
 	capability "github.com/vmware-tanzu/vm-operator/controllers/infra/capability/crd"
-	"github.com/vmware-tanzu/vm-operator/controllers/infra/capability/exit"
 	pkgcfg "github.com/vmware-tanzu/vm-operator/pkg/config"
+	pkgexit "github.com/vmware-tanzu/vm-operator/pkg/exit"
 	"github.com/vmware-tanzu/vm-operator/pkg/manager"
 	"github.com/vmware-tanzu/vm-operator/test/builder"
 )
 
-var numExits int32
+var (
+	numExits int32
+	exited   = make(chan struct{})
+)
 
 func init() {
-	exit.Exit = func() {
+	pkgexit.ExitFn = func(_ time.Duration) {
 		atomic.AddInt32(&numExits, 1)
+		exited <- struct{}{}
 	}
 }
 
@@ -30,6 +35,7 @@ var suite = builder.NewTestSuiteForControllerWithContext(
 		pkgcfg.NewContextWithDefaultConfig(),
 		func(config *pkgcfg.Config) {
 			config.Features.SVAsyncUpgrade = true
+			config.ExitDelay = time.Millisecond * 100
 		},
 	),
 	capability.AddToManager,

controllers/infra/capability/crd/crd_capability_controller_test.go

@@ -28,6 +28,7 @@ var _ = Describe(
 		testlabels.EnvTest,
 		testlabels.V1Alpha3,
 	),
+	Ordered,
 	func() {
 
 		var (
@@ -64,6 +65,13 @@ var _ = Describe(
 		})
 
 		When("the capabilities have changed", func() {
+			JustBeforeEach(func() {
+				Eventually(exited).Should(Receive())
+			})
+			AfterEach(func() {
+				Consistently(exited).ShouldNot(Receive())
+			})
+
 			When("some capabilities are enabled", func() {
 				BeforeEach(func() {
 					pkgcfg.SetContext(
@@ -164,6 +172,8 @@ var _ = Describe(
 			})
 
 			JustBeforeEach(func() {
+				Eventually(exited).Should(Receive())
+
 				Eventually(func(g Gomega) {
 					g.Expect(pkgcfg.FromContext(suite.Context).Features.BringYourOwnEncryptionKey).To(BeTrue())
 					g.Expect(pkgcfg.FromContext(suite.Context).Features.TKGMultipleCL).To(BeTrue())
@@ -174,6 +184,12 @@ var _ = Describe(
 					Activated: false,
 				}
 				Expect(ctx.Client.Status().Update(ctx, obj)).To(Succeed())
+
+				Eventually(exited).Should(Receive())
+			})
+
+			AfterEach(func() {
+				Consistently(exited).ShouldNot(Receive())
 			})
 
 			Specify("the pod was exited once on create and once on update", func() {
@@ -214,6 +230,9 @@ var _ = Describe(
 					},
 				}
 			})
+			JustBeforeEach(func() {
+				Consistently(exited).ShouldNot(Receive())
+			})
 			Specify("the pod was not exited and features were not updated", func() {
 				Consistently(func(g Gomega) {
 					g.Expect(atomic.LoadInt32(&numExits)).To(Equal(int32(0)))

main.go

@@ -34,6 +34,7 @@ import (
 	pkgcfg "github.com/vmware-tanzu/vm-operator/pkg/config"
 	"github.com/vmware-tanzu/vm-operator/pkg/config/capabilities"
 	pkgctx "github.com/vmware-tanzu/vm-operator/pkg/context"
+	pkgexit "github.com/vmware-tanzu/vm-operator/pkg/exit"
 	pkgmgr "github.com/vmware-tanzu/vm-operator/pkg/manager"
 	pkgmgrinit "github.com/vmware-tanzu/vm-operator/pkg/manager/init"
 	"github.com/vmware-tanzu/vm-operator/pkg/mem"
@@ -72,6 +73,8 @@ func main() {
 		"buildtype", pkg.BuildType,
 		"commit", pkg.BuildCommit)
 
+	initExitFn()
+
 	initContext()
 
 	initFlags()
@@ -128,6 +131,10 @@ func initMemStats() {
 		metrics.Registry.MustRegister)
 }
 
+func initExitFn() {
+	pkgexit.ExitFn = func(_ time.Duration) { os.Exit(1) }
+}
+
 func initContext() {
 	ctx = pkgcfg.WithConfig(defaultConfig)
 	ctx = cource.WithContext(ctx)

pkg/config/config.go

@@ -106,6 +106,14 @@ type Config struct {
 	// such as passwords. Defaults to false.
 	LogSensitiveData bool
 
+	// ExitDelay may be set to a time.Duration value and is the maximum delay
+	// used when exiting the pod due to an event that requires the pod be
+	// restarted, ex. the capabilities have changed. The exit will be delayed
+	// by some duration between 0-ExitDelay.
+	//
+	// Defaults to 3s.
+	ExitDelay time.Duration
+
 	// AsyncSignalEnabled may be set to false to disable the vm-watcher service
 	// used to reconcile VirtualMachine objects if their backend state has
 	// changed.