diff --git a/charts/wave/templates/deployment.yaml b/charts/wave/templates/deployment.yaml index ed83043..802a309 100644 --- a/charts/wave/templates/deployment.yaml +++ b/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 }} diff --git a/charts/wave/values.yaml b/charts/wave/values.yaml index 45d2507..defc5a5 100644 --- a/charts/wave/values.yaml +++ b/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 diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 8bde92f..c8b3ad1 100644 --- a/cmd/manager/main.go +++ b/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) } diff --git a/hack/test-update-throttling.sh b/hack/test-update-throttling.sh new file mode 100755 index 0000000..6700dd5 --- /dev/null +++ b/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 diff --git a/pkg/controller/add_daemonset.go b/pkg/controller/add_daemonset.go index 401c4f1..76004f3 100644 --- a/pkg/controller/add_daemonset.go +++ b/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) + }) } diff --git a/pkg/controller/add_deployment.go b/pkg/controller/add_deployment.go index 409c39b..865ccca 100644 --- a/pkg/controller/add_deployment.go +++ b/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) + }) } diff --git a/pkg/controller/add_statefulset.go b/pkg/controller/add_statefulset.go index dc99d5e..d236749 100644 --- a/pkg/controller/add_statefulset.go +++ b/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) + }) } diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index 3e1b348..fca7f6c 100644 --- a/pkg/controller/controller.go +++ b/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 } } diff --git a/pkg/controller/daemonset/daemonset_controller.go b/pkg/controller/daemonset/daemonset_controller.go index d462b9e..1c0372e 100644 --- a/pkg/controller/daemonset/daemonset_controller.go +++ b/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), } } diff --git a/pkg/controller/daemonset/daemonset_controller_suite_test.go b/pkg/controller/daemonset/daemonset_controller_suite_test.go index bc73a29..23e6cd3 100644 --- a/pkg/controller/daemonset/daemonset_controller_suite_test.go +++ b/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()) diff --git a/pkg/controller/daemonset/daemonset_webhook.go b/pkg/controller/daemonset/daemonset_webhook.go index 8d6a55a..c581ed3 100644 --- a/pkg/controller/daemonset/daemonset_webhook.go +++ b/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 diff --git a/pkg/controller/deployment/deployment_controller.go b/pkg/controller/deployment/deployment_controller.go index 3366d5a..ebade3f 100644 --- a/pkg/controller/deployment/deployment_controller.go +++ b/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), } } diff --git a/pkg/controller/deployment/deployment_controller_suite_test.go b/pkg/controller/deployment/deployment_controller_suite_test.go index 0b87a22..ad7bd54 100644 --- a/pkg/controller/deployment/deployment_controller_suite_test.go +++ b/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()) diff --git a/pkg/controller/deployment/deployment_webhook.go b/pkg/controller/deployment/deployment_webhook.go index 277ffa4..5a2133b 100644 --- a/pkg/controller/deployment/deployment_webhook.go +++ b/pkg/controller/deployment/deployment_webhook.go @@ -2,6 +2,7 @@ package deployment import ( "context" + "time" "github.com/wave-k8s/wave/pkg/core" appsv1 "k8s.io/api/apps/v1" @@ -28,11 +29,11 @@ func (a *DeploymentWebhook) Default(ctx context.Context, obj runtime.Object) err return err } -func AddDeploymentWebhook(mgr manager.Manager) error { +func AddDeploymentWebhook(mgr manager.Manager, minUpdateInterval time.Duration) error { err := builder.WebhookManagedBy(mgr).For(&appsv1.Deployment{}).WithDefaulter( &DeploymentWebhook{ Client: mgr.GetClient(), - Handler: core.NewHandler[*appsv1.Deployment](mgr.GetClient(), mgr.GetEventRecorderFor("wave")), + Handler: core.NewHandler[*appsv1.Deployment](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), minUpdateInterval), }).Complete() return err diff --git a/pkg/controller/statefulset/statefulset_controller.go b/pkg/controller/statefulset/statefulset_controller.go index 09fecba..7f69f7a 100644 --- a/pkg/controller/statefulset/statefulset_controller.go +++ b/pkg/controller/statefulset/statefulset_controller.go @@ -18,6 +18,7 @@ package statefulset 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 StatefulSet 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) *ReconcileStatefulSet { +func newReconciler(mgr manager.Manager, minUpdateInterval time.Duration) *ReconcileStatefulSet { return &ReconcileStatefulSet{ scheme: mgr.GetScheme(), - handler: core.NewHandler[*appsv1.StatefulSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave")), + handler: core.NewHandler[*appsv1.StatefulSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), minUpdateInterval), } } diff --git a/pkg/controller/statefulset/statefulset_controller_suite_test.go b/pkg/controller/statefulset/statefulset_controller_suite_test.go index 1a5aff1..dc35b19 100644 --- a/pkg/controller/statefulset/statefulset_controller_suite_test.go +++ b/pkg/controller/statefulset/statefulset_controller_suite_test.go @@ -141,7 +141,7 @@ var _ = BeforeSuite(func() { Expect(add(mgr, recFn, r.handler)).NotTo(HaveOccurred()) // register mutating pod webhook - err = AddStatefulSetWebhook(mgr) + err = AddStatefulSetWebhook(mgr, 10*time.Second) Expect(err).ToNot(HaveOccurred()) testCtx, testCancel = context.WithCancel(context.Background()) diff --git a/pkg/controller/statefulset/statefulset_webhook.go b/pkg/controller/statefulset/statefulset_webhook.go index d187443..6b0f443 100644 --- a/pkg/controller/statefulset/statefulset_webhook.go +++ b/pkg/controller/statefulset/statefulset_webhook.go @@ -2,6 +2,7 @@ package statefulset import ( "context" + "time" "github.com/wave-k8s/wave/pkg/core" appsv1 "k8s.io/api/apps/v1" @@ -28,11 +29,11 @@ func (a *StatefulSetWebhook) Default(ctx context.Context, obj runtime.Object) er return err } -func AddStatefulSetWebhook(mgr manager.Manager) error { +func AddStatefulSetWebhook(mgr manager.Manager, minUpdateInterval time.Duration) error { err := builder.WebhookManagedBy(mgr).For(&appsv1.StatefulSet{}).WithDefaulter( &StatefulSetWebhook{ Client: mgr.GetClient(), - Handler: core.NewHandler[*appsv1.StatefulSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave")), + Handler: core.NewHandler[*appsv1.StatefulSet](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), minUpdateInterval), }).Complete() return err diff --git a/pkg/core/backoff.go b/pkg/core/backoff.go new file mode 100644 index 0000000..5915d65 --- /dev/null +++ b/pkg/core/backoff.go @@ -0,0 +1,112 @@ +/* +Copyright 2018 Pusher Ltd. and Wave Contributors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package core + +import ( + "sync" + "time" + + "k8s.io/apimachinery/pkg/types" +) + +const ( + // DefaultMinUpdateInterval is the default minimum time between updates (10 seconds) + DefaultMinUpdateInterval = 10 * time.Second +) + +// updateState tracks the last update time for a single instance +type updateState struct { + lastUpdateTime time.Time +} + +// UpdateThrottler manages update throttling for multiple instances +type UpdateThrottler struct { + states map[types.NamespacedName]*updateState + mutex sync.RWMutex + minUpdateInterval time.Duration +} + +// NewUpdateThrottler creates a new UpdateThrottler with the specified minimum interval +func NewUpdateThrottler(minUpdateInterval time.Duration) *UpdateThrottler { + if minUpdateInterval <= 0 { + minUpdateInterval = DefaultMinUpdateInterval + } + return &UpdateThrottler{ + states: make(map[types.NamespacedName]*updateState), + minUpdateInterval: minUpdateInterval, + } +} + +// ShouldUpdate checks if an instance should be updated based on the minimum interval +// Returns (shouldUpdate, requeueAfter) +func (ut *UpdateThrottler) ShouldUpdate(name types.NamespacedName) (bool, time.Duration) { + ut.mutex.Lock() + defer ut.mutex.Unlock() + + state, exists := ut.states[name] + now := time.Now() + + // First update or no state + if !exists { + ut.states[name] = &updateState{ + lastUpdateTime: now, + } + return true, 0 + } + + // Check if enough time has passed since last update + elapsed := now.Sub(state.lastUpdateTime) + + if elapsed < ut.minUpdateInterval { + // Still within minimum interval - delay update + remaining := ut.minUpdateInterval - elapsed + return false, remaining + } + + // Minimum interval has passed, allow update + state.lastUpdateTime = now + return true, 0 +} + +// RecordUpdate records a successful update (updates the timestamp) +func (ut *UpdateThrottler) RecordUpdate(name types.NamespacedName) { + ut.mutex.Lock() + defer ut.mutex.Unlock() + + if state, exists := ut.states[name]; exists { + state.lastUpdateTime = time.Now() + } +} + +// Reset removes the throttling state for an instance +func (ut *UpdateThrottler) Reset(name types.NamespacedName) { + ut.mutex.Lock() + defer ut.mutex.Unlock() + + delete(ut.states, name) +} + +// GetLastUpdateTime returns the last update time for debugging/testing +func (ut *UpdateThrottler) GetLastUpdateTime(name types.NamespacedName) (lastUpdate time.Time, exists bool) { + ut.mutex.RLock() + defer ut.mutex.RUnlock() + + if state, ok := ut.states[name]; ok { + return state.lastUpdateTime, true + } + return time.Time{}, false +} diff --git a/pkg/core/backoff_test.go b/pkg/core/backoff_test.go new file mode 100644 index 0000000..5d0a2e8 --- /dev/null +++ b/pkg/core/backoff_test.go @@ -0,0 +1,277 @@ +/* +Copyright 2018 Pusher Ltd. and Wave Contributors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package core + +import ( + "testing" + "time" + + "k8s.io/apimachinery/pkg/types" +) + +const testInterval = 100 * time.Millisecond + +func TestUpdateThrottler_FirstUpdate(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + shouldUpdate, requeueAfter := ut.ShouldUpdate(name) + + if !shouldUpdate { + t.Error("First update should be allowed") + } + if requeueAfter != 0 { + t.Errorf("First update should not require requeue, got %v", requeueAfter) + } + + // Verify state was created + _, exists := ut.GetLastUpdateTime(name) + if !exists { + t.Error("State should exist after first update") + } +} + +func TestUpdateThrottler_SecondUpdateThrottled(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // First update + shouldUpdate, _ := ut.ShouldUpdate(name) + if !shouldUpdate { + t.Fatal("First update should be allowed") + } + + // Immediate second update should be blocked + shouldUpdate, requeueAfter := ut.ShouldUpdate(name) + if shouldUpdate { + t.Error("Second immediate update should be blocked by throttling") + } + if requeueAfter <= 0 { + t.Error("Should return positive requeue duration") + } + if requeueAfter > testInterval { + t.Errorf("Requeue duration should be <= %v, got %v", testInterval, requeueAfter) + } +} + +func TestUpdateThrottler_AllUpdatesSubjectToThrottling(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // First update + ut.ShouldUpdate(name) + + // Multiple rapid updates should all be blocked + for i := 0; i < 5; i++ { + shouldUpdate, requeueAfter := ut.ShouldUpdate(name) + if shouldUpdate { + t.Errorf("Rapid update %d should be blocked by throttling", i+2) + } + if requeueAfter <= 0 { + t.Errorf("Update %d should have positive requeue duration", i+2) + } + } +} + +func TestUpdateThrottler_RapidUpdatesAreThrottled(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // Simulate the issue #182 scenario: rapid secret changes causing rapid reconciliations + // First update allowed + shouldUpdate, _ := ut.ShouldUpdate(name) + if !shouldUpdate { + t.Fatal("First update should be allowed") + } + + // Simulate multiple rapid reconciliations (e.g., buggy controller updating secrets) + for i := 0; i < 5; i++ { + shouldUpdate, requeueAfter := ut.ShouldUpdate(name) + if shouldUpdate { + t.Errorf("Rapid update %d should be blocked by throttling", i+2) + } + if requeueAfter <= 0 { + t.Errorf("Update %d should have positive requeue duration", i+2) + } + } + + // After throttle period, update should be allowed + ut.states[name].lastUpdateTime = time.Now().Add(-testInterval - time.Millisecond) + shouldUpdate, requeueAfter := ut.ShouldUpdate(name) + if !shouldUpdate { + t.Error("Update should be allowed after throttle period") + } + if requeueAfter != 0 { + t.Error("Should not require requeue after throttle period") + } +} + +func TestUpdateThrottler_AfterInterval(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // First update + shouldUpdate, _ := ut.ShouldUpdate(name) + if !shouldUpdate { + t.Fatal("First update should be allowed") + } + + // Wait for the interval to pass + time.Sleep(testInterval + 10*time.Millisecond) + + // Second update should now be allowed + shouldUpdate, requeueAfter := ut.ShouldUpdate(name) + if !shouldUpdate { + t.Error("Update should be allowed after interval has passed") + } + if requeueAfter != 0 { + t.Errorf("Should not require requeue, got %v", requeueAfter) + } +} + +func TestUpdateThrottler_Reset(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // Create some state + ut.ShouldUpdate(name) + + // Verify state exists + _, exists := ut.GetLastUpdateTime(name) + if !exists { + t.Fatal("State should exist") + } + + // Reset + ut.Reset(name) + + // Verify state is gone + _, exists = ut.GetLastUpdateTime(name) + if exists { + t.Error("State should not exist after reset") + } +} + +func TestUpdateThrottler_RecordUpdate(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // Create initial state + ut.ShouldUpdate(name) + + // Get initial time + initialTime, _ := ut.GetLastUpdateTime(name) + + // Small delay + time.Sleep(10 * time.Millisecond) + + // Record update + ut.RecordUpdate(name) + + // Get updated time + updatedTime, exists := ut.GetLastUpdateTime(name) + if !exists { + t.Fatal("State should still exist") + } + + if !updatedTime.After(initialTime) { + t.Error("Update time should be after initial time") + } +} + +func TestUpdateThrottler_MultipleInstances(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name1 := types.NamespacedName{Namespace: "test", Name: "deployment1"} + name2 := types.NamespacedName{Namespace: "test", Name: "deployment2"} + + // Update both instances + shouldUpdate1, _ := ut.ShouldUpdate(name1) + shouldUpdate2, _ := ut.ShouldUpdate(name2) + + if !shouldUpdate1 || !shouldUpdate2 { + t.Error("Both first updates should be allowed") + } + + // Verify both have independent state + _, exists1 := ut.GetLastUpdateTime(name1) + _, exists2 := ut.GetLastUpdateTime(name2) + + if !exists1 || !exists2 { + t.Error("Both instances should have state") + } + + // Block second update for name1 + shouldUpdate1, requeueAfter1 := ut.ShouldUpdate(name1) + if shouldUpdate1 { + t.Error("Second update for name1 should be blocked") + } + if requeueAfter1 <= 0 { + t.Error("Should have positive requeue duration for name1") + } + + // name2 should still be independent + shouldUpdate2, requeueAfter2 := ut.ShouldUpdate(name2) + if shouldUpdate2 { + t.Error("Second update for name2 should also be blocked") + } + if requeueAfter2 <= 0 { + t.Error("Should have positive requeue duration for name2") + } +} + +func TestUpdateThrottler_DefaultInterval(t *testing.T) { + // Test with zero interval (should use default) + ut := NewUpdateThrottler(0) + if ut.minUpdateInterval != DefaultMinUpdateInterval { + t.Errorf("Expected default interval %v, got %v", DefaultMinUpdateInterval, ut.minUpdateInterval) + } + + // Test with negative interval (should use default) + ut = NewUpdateThrottler(-1 * time.Second) + if ut.minUpdateInterval != DefaultMinUpdateInterval { + t.Errorf("Expected default interval %v, got %v", DefaultMinUpdateInterval, ut.minUpdateInterval) + } + + // Test with valid interval + customInterval := 5 * time.Second + ut = NewUpdateThrottler(customInterval) + if ut.minUpdateInterval != customInterval { + t.Errorf("Expected custom interval %v, got %v", customInterval, ut.minUpdateInterval) + } +} + +func TestUpdateThrottler_ConsistentThrottling(t *testing.T) { + ut := NewUpdateThrottler(testInterval) + name := types.NamespacedName{Namespace: "test", Name: "deployment"} + + // First update + ut.ShouldUpdate(name) + + // Check that throttling remains consistent regardless of config hash changes + // This addresses issue #182 where changing hashes shouldn't bypass throttling + shouldUpdate, _ := ut.ShouldUpdate(name) + if shouldUpdate { + t.Error("Update should be throttled regardless of hash changes") + } + + // After interval, update should be allowed + ut.states[name].lastUpdateTime = time.Now().Add(-testInterval - time.Millisecond) + shouldUpdate, _ = ut.ShouldUpdate(name) + if !shouldUpdate { + t.Error("Update should be allowed after interval") + } +} diff --git a/pkg/core/children_test.go b/pkg/core/children_test.go index 7fd6c3c..91fc082 100644 --- a/pkg/core/children_test.go +++ b/pkg/core/children_test.go @@ -70,7 +70,7 @@ var _ = Describe("Wave children Suite", func() { Expect(cerr).NotTo(HaveOccurred()) c = mgr.GetClient() // h = NewHandler(c, mgr.GetEventRecorderFor("wave")) - h = NewHandler[*appsv1.Deployment](mgr.GetClient(), mgr.GetEventRecorderFor("wave")) + h = NewHandler[*appsv1.Deployment](mgr.GetClient(), mgr.GetEventRecorderFor("wave"), 10*time.Second) m = utils.Matcher{Client: c} diff --git a/pkg/core/delete_test.go b/pkg/core/delete_test.go index a4cea58..0298cf8 100644 --- a/pkg/core/delete_test.go +++ b/pkg/core/delete_test.go @@ -55,7 +55,7 @@ var _ = Describe("Wave migration Suite", func() { var cerr error c, cerr = client.New(cfg, client.Options{Scheme: scheme.Scheme}) Expect(cerr).NotTo(HaveOccurred()) - h = NewHandler[*appsv1.Deployment](c, mgr.GetEventRecorderFor("wave")) + h = NewHandler[*appsv1.Deployment](c, mgr.GetEventRecorderFor("wave"), 10*time.Second) m = utils.Matcher{Client: c} // Create some configmaps and secrets diff --git a/pkg/core/handler.go b/pkg/core/handler.go index d7bf6bf..e1850b3 100644 --- a/pkg/core/handler.go +++ b/pkg/core/handler.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "sync" + "time" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" @@ -36,10 +37,11 @@ type Handler[I InstanceType] struct { recorder record.EventRecorder watchedConfigmaps WatcherList watchedSecrets WatcherList + updateThrottler *UpdateThrottler } // NewHandler constructs a new instance of Handler -func NewHandler[I InstanceType](c client.Client, r record.EventRecorder) *Handler[I] { +func NewHandler[I InstanceType](c client.Client, r record.EventRecorder, minUpdateInterval time.Duration) *Handler[I] { return &Handler[I]{Client: c, recorder: r, watchedConfigmaps: WatcherList{ watchers: make(map[types.NamespacedName]map[types.NamespacedName]bool), @@ -48,7 +50,9 @@ func NewHandler[I InstanceType](c client.Client, r record.EventRecorder) *Handle watchedSecrets: WatcherList{ watchers: make(map[types.NamespacedName]map[types.NamespacedName]bool), watchersMutex: &sync.RWMutex{}, - }} + }, + updateThrottler: NewUpdateThrottler(minUpdateInterval), + } } // HandleWebhook is called by the webhook @@ -125,6 +129,16 @@ func (h *Handler[I]) handlePodController(instance I) (reconcile.Result, error) { // If the desired state doesn't match the existing state, update it if hash != oldHash || schedulingChange { + instanceName := GetNamespacedNameFromObject(instance) + + // Check throttling before updating + shouldUpdate, requeueAfter := h.updateThrottler.ShouldUpdate(instanceName) + + if !shouldUpdate { + log.V(0).Info("Update delayed due to throttling", "hash", hash, "requeueAfter", requeueAfter) + return reconcile.Result{RequeueAfter: requeueAfter}, nil + } + log.V(0).Info("Updating instance hash", "hash", hash) h.recorder.Eventf(instance, corev1.EventTypeNormal, "ConfigChanged", "Configuration hash updated to %s", hash) @@ -132,6 +146,9 @@ func (h *Handler[I]) handlePodController(instance I) (reconcile.Result, error) { if err != nil { return reconcile.Result{}, fmt.Errorf("error updating instance %s/%s: %v", instance.GetNamespace(), instance.GetName(), err) } + + // Record successful update + h.updateThrottler.RecordUpdate(instanceName) } return reconcile.Result{}, nil } diff --git a/pkg/core/handler_test.go b/pkg/core/handler_test.go index d2b4564..2555342 100644 --- a/pkg/core/handler_test.go +++ b/pkg/core/handler_test.go @@ -76,7 +76,7 @@ var _ = Describe("Wave controller Suite", func() { c, cerr = client.New(cfg, client.Options{Scheme: scheme.Scheme}) Expect(cerr).NotTo(HaveOccurred()) - h = NewHandler[*appsv1.Deployment](c, mgr.GetEventRecorderFor("wave")) + h = NewHandler[*appsv1.Deployment](c, mgr.GetEventRecorderFor("wave"), 10*time.Second) m = utils.Matcher{Client: c} stopMgr, mgrStopped = StartTestManager(mgr) diff --git a/pkg/core/owner_references_test.go b/pkg/core/owner_references_test.go index d71a8e3..6cb522a 100644 --- a/pkg/core/owner_references_test.go +++ b/pkg/core/owner_references_test.go @@ -63,7 +63,7 @@ var _ = Describe("Wave owner references Suite", func() { var cerr error c, cerr = client.New(cfg, client.Options{Scheme: scheme.Scheme}) Expect(cerr).NotTo(HaveOccurred()) - h = NewHandler[*appsv1.Deployment](c, mgr.GetEventRecorderFor("wave")) + h = NewHandler[*appsv1.Deployment](c, mgr.GetEventRecorderFor("wave"), 10*time.Second) m = utils.Matcher{Client: c} // Create some configmaps and secrets