Add backoff period to prevent rapid deployment updates

charts/wave/templates/deployment.yaml

@@ -35,6 +35,9 @@ spec:
           {{- if .Values.syncPeriod }}
             - --sync-period={{ .Values.syncPeriod }}
           {{- end }}
+          {{- if .Values.minUpdateInterval }}
+            - --min-update-interval={{ .Values.minUpdateInterval }}
+          {{- end }}
           {{- if .Values.webhooks.enabled }}
             - --enable-webhooks=true
           {{- end }}

charts/wave/values.yaml

@@ -53,6 +53,11 @@ webhooks:
 # Period for reconciliation
 # syncPeriod: 5m
 
+# Minimum time between updates to prevent rapid deployment churn (default: 10s)
+# This prevents Wave from updating deployments too frequently, which can
+# overwhelm the Kubernetes API server when secrets/configmaps change rapidly
+minUpdateInterval: 10s
+
 resources: {}
   # We usually recommend not to specify default resources and to leave this as a conscious
   # choice for the user. This also increases chances charts run on environments with little

cmd/manager/main.go

@@ -47,6 +47,7 @@ var (
 	leaderElectionID        = flag.String("leader-election-id", "", "Name of the configmap used by the leader election system")
 	leaderElectionNamespace = flag.String("leader-election-namespace", "", "Namespace for the configmap used by the leader election system")
 	syncPeriod              = flag.Duration("sync-period", 10*time.Hour, "Reconcile sync period")
+	minUpdateInterval       = flag.Duration("min-update-interval", core.DefaultMinUpdateInterval, "Minimum time between updates to prevent rapid deployment churn")
 	showVersion             = flag.Bool("version", false, "Show version and exit")
 	enableWebhooks          = flag.Bool("enable-webhooks", false, "Enable webhooks")
 	namespaces              = flag.String("namespaces", "", "Comma-separated list of namespaces to watch. Defaults to all namespaces.")
@@ -109,22 +110,25 @@ func main() {
 
 	// Setup all Controllers
 	setupLog.Info("Setting up controller")
-	if err := controller.AddToManager(mgr); err != nil {
+	controllerConfig := controller.Config{
+		MinUpdateInterval: *minUpdateInterval,
+	}
+	if err := controller.AddToManager(mgr, controllerConfig); err != nil {
 		setupLog.Error(err, "unable to register controllers to the manager")
 		os.Exit(1)
 	}
 	if *enableWebhooks {
-		if err := deployment.AddDeploymentWebhook(mgr); err != nil {
+		if err := deployment.AddDeploymentWebhook(mgr, *minUpdateInterval); err != nil {
 			setupLog.Error(err, "unable to create webhook", "webhook", "Deployment")
 			os.Exit(1)
 		}
 
-		if err := statefulset.AddStatefulSetWebhook(mgr); err != nil {
+		if err := statefulset.AddStatefulSetWebhook(mgr, *minUpdateInterval); err != nil {
 			setupLog.Error(err, "unable to create webhook", "webhook", "StatefulSet")
 			os.Exit(1)
 		}
 
-		if err := daemonset.AddDaemonSetWebhook(mgr); err != nil {
+		if err := daemonset.AddDaemonSetWebhook(mgr, *minUpdateInterval); err != nil {
 			setupLog.Error(err, "unable to create webhook", "webhook", "DaemonSet")
 			os.Exit(1)
 		}

hack/test-update-throttling.sh

@@ -0,0 +1,157 @@
+#!/usr/bin/env bash
+
+set -eu
+set -o pipefail
+
+helm --help > /dev/null 2>&1 || {
+  echo "helm is not installed"
+  exit 1
+}
+
+kubectl --help > /dev/null 2>&1 || {
+  echo "kubectl is not installed"
+  exit 1
+}
+
+MINIKUBE_ALREADY_RUNNING=0
+kubectl get node minikube >/dev/null 2>&1 && MINIKUBE_ALREADY_RUNNING=1
+
+minikube --help > /dev/null 2>&1 || {
+  echo "minikube is not installed"
+  exit 1
+}
+
+KUBERNETES_VERSION=${1:-v1.21}
+THROTTLE_INTERVAL=${2:-10s}
+
+[ $MINIKUBE_ALREADY_RUNNING  -eq 0 ] && {
+	echo Starting minikube...
+	minikube start --kubernetes-version "$KUBERNETES_VERSION"
+	trap "minikube delete" EXIT
+}
+
+eval $(minikube -p minikube docker-env)
+docker build -f ./Dockerfile -t wave-local:local .
+
+echo Installing wave with minUpdateInterval=${THROTTLE_INTERVAL}...
+helm install wave charts/wave --set image.name=wave-local --set image.tag=local --set minUpdateInterval=${THROTTLE_INTERVAL}
+
+while [ "$(kubectl get pods -n default | grep -cE 'wave-wave')" -gt 1 ]; do echo Waiting for \"wave\" to be scheduled; sleep 10; done
+
+while [ "$(kubectl get pods -A | grep -cEv 'Running|Completed')" -gt 1 ]; do echo Waiting for \"cluster\" to start; sleep 10; done
+
+echo Creating test resources...
+kubectl apply -f - <<'EOF'
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: throttle-test
+data:
+  counter: "0"
+  timestamp: "0"
+EOF
+
+kubectl apply -f - <<'EOF'
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: throttle-test
+  annotations:
+    wave.pusher.com/update-on-config-change: "true"
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: throttle-test
+  template:
+    metadata:
+      labels:
+        app: throttle-test
+    spec:
+      containers:
+      - name: test
+        image: nixery.dev/shell/bash
+        command:
+          - /bin/bash
+          - -c
+          - |
+            echo "Pod started at $(date +%s)"
+            echo "Counter: $(cat /etc/config/counter)"
+            sleep infinity
+        volumeMounts:
+        - name: config
+          mountPath: /etc/config
+      volumes:
+      - name: config
+        configMap:
+          name: throttle-test
+EOF
+
+echo Waiting for initial deployment to be ready...
+kubectl wait --for=condition=available --timeout=60s deployment/throttle-test
+
+# Get the initial pod name and creation time
+INITIAL_POD=$(kubectl get pods -l app=throttle-test -o jsonpath='{.items[0].metadata.name}')
+INITIAL_CREATION=$(kubectl get pod $INITIAL_POD -o jsonpath='{.metadata.creationTimestamp}')
+echo "Initial pod: $INITIAL_POD created at $INITIAL_CREATION"
+
+# Record start time
+START_TIME=$(date +%s)
+
+echo ""
+echo "Testing update throttling by rapidly updating the ConfigMap..."
+echo "Expected behavior: Updates should be throttled to minimum interval of ${THROTTLE_INTERVAL}"
+echo ""
+
+# Rapidly update the ConfigMap 10 times
+for i in 1 2 3 4 5 6 7 8 9 0; do
+  echo "Update $i at $(date +%H:%M:%S)"
+  kubectl patch configmap throttle-test --type merge -p "{\"data\":{\"counter\":\"$i\",\"timestamp\":\"$(date +%s)\"}}"
+  sleep 1
+done
+
+echo ""
+echo "Waiting 15 seconds to observe throttling behavior..."
+sleep 15
+
+# Check how many pod restarts/updates occurred
+echo ""
+echo "Checking deployment update history..."
+kubectl get replicasets -l app=throttle-test -o wide
+
+# Get all pods that were created (including terminated ones)
+echo ""
+echo "Checking pod creation times..."
+POD_COUNT=$(kubectl get pods -l app=throttle-test --show-all 2>/dev/null | grep -c throttle-test || kubectl get pods -l app=throttle-test | grep -c throttle-test)
+echo "Total pods created: $POD_COUNT"
+
+# Get the deployment's pod template hash changes
+HASH_CHANGES=$(kubectl get replicasets -l app=throttle-test -o jsonpath='{range .items[*]}{.metadata.creationTimestamp}{"\t"}{.spec.template.spec.containers[0].image}{"\t"}{.spec.replicas}{"\n"}{end}')
+echo ""
+echo "ReplicaSet history (creation time, image, replicas):"
+echo "$HASH_CHANGES"
+
+# Count how many distinct replicasets were created
+RS_COUNT=$(kubectl get replicasets -l app=throttle-test --no-headers | wc -l)
+echo ""
+echo "Number of ReplicaSets created: $RS_COUNT"
+
+# Check wave controller logs for throttling messages
+echo ""
+echo "Wave controller logs (throttling messages):"
+kubectl logs -l app=wave --tail=50 | grep -i "throttl\|delayed" || echo "No throttling messages found in logs"
+
+# Verify throttling worked
+echo ""
+echo "=== Test Results ==="
+if [ "$RS_COUNT" -le 6 ]; then
+  echo "✓ PASS: Throttling is working correctly"
+  echo "  - ConfigMap was updated 10 times rapidly"
+  echo "  - Only $RS_COUNT deployment updates occurred (expected ≤6 with ${THROTTLE_INTERVAL} throttling)"
+  exit 0
+else
+  echo "✗ FAIL: Throttling may not be working correctly"
+  echo "  - ConfigMap was updated 10 times rapidly"
+  echo "  - $RS_COUNT deployment updates occurred (expected ≤6 with ${THROTTLE_INTERVAL} throttling)"
+  exit 1
+fi

pkg/controller/add_daemonset.go

@@ -18,9 +18,12 @@ package controller
 
 import (
 	"github.com/wave-k8s/wave/pkg/controller/daemonset"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 func init() {
 	// AddToManagerFuncs is a list of functions to create controllers and add them to a manager.
-	AddToManagerFuncs = append(AddToManagerFuncs, daemonset.Add)
+	AddToManagerFuncs = append(AddToManagerFuncs, func(mgr manager.Manager, cfg Config) error {
+		return daemonset.Add(mgr, cfg.MinUpdateInterval)
+	})
 }

pkg/controller/add_deployment.go

@@ -18,9 +18,12 @@ package controller
 
 import (
 	"github.com/wave-k8s/wave/pkg/controller/deployment"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 func init() {
 	// AddToManagerFuncs is a list of functions to create controllers and add them to a manager.
-	AddToManagerFuncs = append(AddToManagerFuncs, deployment.Add)
+	AddToManagerFuncs = append(AddToManagerFuncs, func(mgr manager.Manager, cfg Config) error {
+		return deployment.Add(mgr, cfg.MinUpdateInterval)
+	})
 }

pkg/controller/add_statefulset.go

@@ -18,9 +18,12 @@ package controller
 
 import (
 	"github.com/wave-k8s/wave/pkg/controller/statefulset"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 func init() {
 	// AddToManagerFuncs is a list of functions to create controllers and add them to a manager.
-	AddToManagerFuncs = append(AddToManagerFuncs, statefulset.Add)
+	AddToManagerFuncs = append(AddToManagerFuncs, func(mgr manager.Manager, cfg Config) error {
+		return statefulset.Add(mgr, cfg.MinUpdateInterval)
+	})
 }

pkg/controller/controller.go

@@ -17,16 +17,23 @@ limitations under the License.
 package controller
 
 import (
+	"time"
+
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
+// Config holds configuration options for controllers
+type Config struct {
+	MinUpdateInterval time.Duration
+}
+
 // AddToManagerFuncs is a list of functions to add all Controllers to the Manager
-var AddToManagerFuncs []func(manager.Manager) error
+var AddToManagerFuncs []func(manager.Manager, Config) error
 
-// AddToManager adds all Controllers to the Manager
-func AddToManager(m manager.Manager) error {
+// AddToManager adds all Controllers to the Manager with the given configuration
+func AddToManager(m manager.Manager, cfg Config) error {
 	for _, f := range AddToManagerFuncs {
-		if err := f(m); err != nil {
+		if err := f(m, cfg); err != nil {
 			return err
 		}
 	}

pkg/controller/daemonset/daemonset_controller.go

@@ -18,6 +18,7 @@ package daemonset
 
 import (
 	"context"
+	"time"
 
 	"github.com/wave-k8s/wave/pkg/core"
 	appsv1 "k8s.io/api/apps/v1"
@@ -33,16 +34,16 @@ import (
 
 // Add creates a new DaemonSet Controller and adds it to the Manager with default RBAC. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
-func Add(mgr manager.Manager) error {
-	r := newReconciler(mgr)
+func Add(mgr manager.Manager, minUpdateInterval time.Duration) error {
+	r := newReconciler(mgr, minUpdateInterval)
 	return add(mgr, r, r.handler)
 }
 
 // newReconciler returns a new reconcile.Reconciler
-func newReconciler(mgr manager.Manager) *ReconcileDaemonSet {
+func newReconciler(mgr manager.Manager, minUpdateInterval time.Duration) *ReconcileDaemonSet {
 	return &ReconcileDaemonSet{
 		scheme:  mgr.GetScheme(),
-		handler: core.NewHandler[*appsv1.DaemonSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave")),
+		handler: core.NewHandler[*appsv1.DaemonSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), minUpdateInterval),
 	}
 }
 

pkg/controller/daemonset/daemonset_controller_suite_test.go

@@ -139,7 +139,7 @@ var _ = BeforeSuite(func() {
 	Expect(add(mgr, recFn, r.handler)).NotTo(HaveOccurred())
 
 	// register mutating pod webhook
-	err = AddDaemonSetWebhook(mgr)
+	err = AddDaemonSetWebhook(mgr, 10*time.Second)
 	Expect(err).ToNot(HaveOccurred())
 
 	testCtx, testCancel = context.WithCancel(context.Background())

pkg/controller/daemonset/daemonset_webhook.go

@@ -2,6 +2,7 @@ package daemonset
 
 import (
 	"context"
+	"time"
 
 	"github.com/wave-k8s/wave/pkg/core"
 	appsv1 "k8s.io/api/apps/v1"
@@ -28,11 +29,11 @@ func (a *DaemonSetWebhook) Default(ctx context.Context, obj runtime.Object) erro
 	return err
 }
 
-func AddDaemonSetWebhook(mgr manager.Manager) error {
+func AddDaemonSetWebhook(mgr manager.Manager, minUpdateInterval time.Duration) error {
 	err := builder.WebhookManagedBy(mgr).For(&appsv1.DaemonSet{}).WithDefaulter(
 		&DaemonSetWebhook{
 			Client:  mgr.GetClient(),
-			Handler: core.NewHandler[*appsv1.DaemonSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave")),
+			Handler: core.NewHandler[*appsv1.DaemonSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), minUpdateInterval),
 		}).Complete()
 
 	return err

pkg/controller/deployment/deployment_controller.go

@@ -18,6 +18,7 @@ package deployment
 
 import (
 	"context"
+	"time"
 
 	"github.com/wave-k8s/wave/pkg/core"
 	appsv1 "k8s.io/api/apps/v1"
@@ -33,16 +34,16 @@ import (
 
 // Add creates a new Deployment Controller and adds it to the Manager with default RBAC. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
-func Add(mgr manager.Manager) error {
-	r := newReconciler(mgr)
+func Add(mgr manager.Manager, minUpdateInterval time.Duration) error {
+	r := newReconciler(mgr, minUpdateInterval)
 	return add(mgr, r, r.handler)
 }
 
 // newReconciler returns a new reconcile.Reconciler
-func newReconciler(mgr manager.Manager) *ReconcileDeployment {
+func newReconciler(mgr manager.Manager, minUpdateInterval time.Duration) *ReconcileDeployment {
 	return &ReconcileDeployment{
 		scheme:  mgr.GetScheme(),
-		handler: core.NewHandler[*appsv1.Deployment](mgr.GetClient(), mgr.GetEventRecorderFor("wave")),
+		handler: core.NewHandler[*appsv1.Deployment](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), minUpdateInterval),
 	}
 }
 

pkg/controller/deployment/deployment_controller_suite_test.go

@@ -141,7 +141,7 @@ var _ = BeforeSuite(func() {
 	Expect(add(mgr, recFn, r.handler)).NotTo(HaveOccurred())
 
 	// register mutating pod webhook
-	err = AddDeploymentWebhook(mgr)
+	err = AddDeploymentWebhook(mgr, 10*time.Second)
 	Expect(err).ToNot(HaveOccurred())
 
 	testCtx, testCancel = context.WithCancel(context.Background())