From fafb5f45d6e87af4cd6aa8428afd8412b8a41ec0 Mon Sep 17 00:00:00 2001 From: Vladimir Videlov Date: Tue, 1 Jul 2025 14:27:47 +0200 Subject: [PATCH 1/3] Do not fetch resource on first attempt (already up-to-date from Items func) --- internal/pkg/handler/upgrade.go | 39 ++++++++-- internal/pkg/handler/upgrade_test.go | 106 ++++++++++++++++++++++++++- 2 files changed, 135 insertions(+), 10 deletions(-) diff --git a/internal/pkg/handler/upgrade.go b/internal/pkg/handler/upgrade.go index e84bed4e..9731a630 100644 --- a/internal/pkg/handler/upgrade.go +++ b/internal/pkg/handler/upgrade.go @@ -22,9 +22,11 @@ import ( "github.com/stakater/Reloader/internal/pkg/util" "github.com/stakater/Reloader/pkg/kube" v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/runtime" patchtypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/record" "k8s.io/client-go/util/retry" ) @@ -216,10 +218,9 @@ func PerformAction(clients kube.Clients, config util.Config, upgradeFuncs callba items := upgradeFuncs.ItemsFunc(clients, config.Namespace) for _, item := range items { - err := retry.RetryOnConflict(retry.DefaultRetry, func() error { - return upgradeResource(clients, config, upgradeFuncs, collectors, recorder, strategy, item) + err := retryOnConflict(retry.DefaultRetry, func(shouldRefresh bool) error { + return upgradeResource(clients, config, upgradeFuncs, collectors, recorder, strategy, item, shouldRefresh) }) - if err != nil { return err } @@ -228,16 +229,40 @@ func PerformAction(clients kube.Clients, config util.Config, upgradeFuncs callba return nil } -func upgradeResource(clients kube.Clients, config util.Config, upgradeFuncs callbacks.RollingUpgradeFuncs, collectors metrics.Collectors, recorder record.EventRecorder, strategy invokeStrategy, resource runtime.Object) error { +func retryOnConflict(backoff wait.Backoff, fn func(_ bool) error) error { + var lastError error + fetchResource := false // do not fetch resource on first attempt, already done by ItemsFunc + err := wait.ExponentialBackoff(backoff, func() (bool, error) { + err := fn(fetchResource) + fetchResource = true + switch { + case err == nil: + return true, nil + case apierrors.IsConflict(err): + lastError = err + return false, nil + default: + return false, err + } + }) + if wait.Interrupted(err) { + err = lastError + } + return err +} + +func upgradeResource(clients kube.Clients, config util.Config, upgradeFuncs callbacks.RollingUpgradeFuncs, collectors metrics.Collectors, recorder record.EventRecorder, strategy invokeStrategy, resource runtime.Object, fetchResource bool) error { accessor, err := meta.Accessor(resource) if err != nil { return err } resourceName := accessor.GetName() - resource, err = upgradeFuncs.ItemFunc(clients, resourceName, config.Namespace) - if err != nil { - return err + if fetchResource { + resource, err = upgradeFuncs.ItemFunc(clients, resourceName, config.Namespace) + if err != nil { + return err + } } // find correct annotation and update the resource diff --git a/internal/pkg/handler/upgrade_test.go b/internal/pkg/handler/upgrade_test.go index 9817f21e..3a57783a 100644 --- a/internal/pkg/handler/upgrade_test.go +++ b/internal/pkg/handler/upgrade_test.go @@ -1441,6 +1441,20 @@ func TestRollingUpgradeForDeploymentWithConfigmapUsingArs(t *testing.T) { deploymentFuncs := GetDeploymentRollingUpgradeFuncs() collectors := getCollectors() + orgItemFunc := deploymentFuncs.ItemFunc + orgItemsFunc := deploymentFuncs.ItemsFunc + itemCalled := 0 + itemsCalled := 0 + + deploymentFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return orgItemFunc(client, namespace, name) + } + deploymentFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return orgItemsFunc(client, namespace) + } + err := PerformAction(clients, config, deploymentFuncs, collectors, nil, invokeReloadStrategy) time.Sleep(5 * time.Second) if err != nil { @@ -1460,6 +1474,10 @@ func TestRollingUpgradeForDeploymentWithConfigmapUsingArs(t *testing.T) { if promtestutil.ToFloat64(collectors.ReloadedByNamespace.With(prometheus.Labels{"success": "true", "namespace": arsNamespace})) != 1 { t.Errorf("Counter by namespace was not increased") } + + assert.Equal(t, 0, itemCalled, "ItemFunc should not be called") + assert.Equal(t, 2, itemsCalled, "ItemsFunc should be called twice") + testRollingUpgradeInvokeDeleteStrategyArs(t, clients, config, deploymentFuncs, collectors, envVarPostfix) } @@ -1474,6 +1492,20 @@ func TestRollingUpgradeForDeploymentWithPatchAndRetryUsingArs(t *testing.T) { assert.True(t, deploymentFuncs.SupportsPatch) assert.NotEmpty(t, deploymentFuncs.PatchTemplatesFunc().AnnotationTemplate) + orgItemFunc := deploymentFuncs.ItemFunc + orgItemsFunc := deploymentFuncs.ItemsFunc + itemCalled := 0 + itemsCalled := 0 + + deploymentFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return orgItemFunc(client, namespace, name) + } + deploymentFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return orgItemsFunc(client, namespace) + } + patchCalled := 0 deploymentFuncs.PatchFunc = func(client kube.Clients, namespace string, resource runtime.Object, patchType patchtypes.PatchType, bytes []byte) error { patchCalled++ @@ -1498,7 +1530,9 @@ func TestRollingUpgradeForDeploymentWithPatchAndRetryUsingArs(t *testing.T) { t.Errorf("Rolling upgrade failed for Deployment with Configmap") } - assert.Equal(t, 2, patchCalled) + assert.Equal(t, 1, itemCalled, "ItemFunc should be called once") + assert.Equal(t, 1, itemsCalled, "ItemsFunc should be called once") + assert.Equal(t, 2, patchCalled, "PatchFunc should be called twice") deploymentFuncs = GetDeploymentRollingUpgradeFuncs() testRollingUpgradeWithPatchAndInvokeDeleteStrategyArs(t, clients, config, deploymentFuncs, collectors, envVarPostfix) @@ -2204,6 +2238,20 @@ func TestRollingUpgradeForDaemonSetWithConfigmapUsingArs(t *testing.T) { daemonSetFuncs := GetDaemonSetRollingUpgradeFuncs() collectors := getCollectors() + orgItemFunc := daemonSetFuncs.ItemFunc + orgItemsFunc := daemonSetFuncs.ItemsFunc + itemCalled := 0 + itemsCalled := 0 + + daemonSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return orgItemFunc(client, namespace, name) + } + daemonSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return orgItemsFunc(client, namespace) + } + err := PerformAction(clients, config, daemonSetFuncs, collectors, nil, invokeReloadStrategy) time.Sleep(5 * time.Second) if err != nil { @@ -2224,6 +2272,9 @@ func TestRollingUpgradeForDaemonSetWithConfigmapUsingArs(t *testing.T) { t.Errorf("Counter by namespace was not increased") } + assert.Equal(t, 0, itemCalled, "ItemFunc should not be called") + assert.Equal(t, 2, itemsCalled, "ItemsFunc should be called twice") + testRollingUpgradeInvokeDeleteStrategyArs(t, clients, config, daemonSetFuncs, collectors, envVarPostfix) } @@ -2235,6 +2286,20 @@ func TestRollingUpgradeForDaemonSetWithPatchAndRetryUsingArs(t *testing.T) { config := getConfigWithAnnotations(envVarPostfix, arsConfigmapName, shaData, options.ConfigmapUpdateOnChangeAnnotation, options.ConfigmapReloaderAutoAnnotation) daemonSetFuncs := GetDaemonSetRollingUpgradeFuncs() + orgItemFunc := daemonSetFuncs.ItemFunc + orgItemsFunc := daemonSetFuncs.ItemsFunc + itemCalled := 0 + itemsCalled := 0 + + daemonSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return orgItemFunc(client, namespace, name) + } + daemonSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return orgItemsFunc(client, namespace) + } + assert.True(t, daemonSetFuncs.SupportsPatch) assert.NotEmpty(t, daemonSetFuncs.PatchTemplatesFunc().AnnotationTemplate) @@ -2263,7 +2328,9 @@ func TestRollingUpgradeForDaemonSetWithPatchAndRetryUsingArs(t *testing.T) { t.Errorf("Rolling upgrade failed for DaemonSet with configmap") } - assert.Equal(t, 2, patchCalled) + assert.Equal(t, 1, itemCalled, "ItemFunc should be called once") + assert.Equal(t, 1, itemsCalled, "ItemsFunc should be called once") + assert.Equal(t, 2, patchCalled, "PatchFunc should be called twice") daemonSetFuncs = GetDeploymentRollingUpgradeFuncs() testRollingUpgradeWithPatchAndInvokeDeleteStrategyArs(t, clients, config, daemonSetFuncs, collectors, envVarPostfix) @@ -2406,6 +2473,20 @@ func TestRollingUpgradeForStatefulSetWithConfigmapUsingArs(t *testing.T) { statefulSetFuncs := GetStatefulSetRollingUpgradeFuncs() collectors := getCollectors() + orgItemFunc := statefulSetFuncs.ItemFunc + orgItemsFunc := statefulSetFuncs.ItemsFunc + itemCalled := 0 + itemsCalled := 0 + + statefulSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return orgItemFunc(client, namespace, name) + } + statefulSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return orgItemsFunc(client, namespace) + } + err := PerformAction(clients, config, statefulSetFuncs, collectors, nil, invokeReloadStrategy) time.Sleep(5 * time.Second) if err != nil { @@ -2426,6 +2507,9 @@ func TestRollingUpgradeForStatefulSetWithConfigmapUsingArs(t *testing.T) { t.Errorf("Counter by namespace was not increased") } + assert.Equal(t, 0, itemCalled, "ItemFunc should not be called") + assert.Equal(t, 2, itemsCalled, "ItemsFunc should be called twice") + testRollingUpgradeInvokeDeleteStrategyArs(t, clients, config, statefulSetFuncs, collectors, envVarPostfix) } @@ -2437,6 +2521,20 @@ func TestRollingUpgradeForStatefulSetWithPatchAndRetryUsingArs(t *testing.T) { config := getConfigWithAnnotations(envVarPostfix, arsConfigmapName, shaData, options.ConfigmapUpdateOnChangeAnnotation, options.ConfigmapReloaderAutoAnnotation) statefulSetFuncs := GetStatefulSetRollingUpgradeFuncs() + orgItemFunc := statefulSetFuncs.ItemFunc + orgItemsFunc := statefulSetFuncs.ItemsFunc + itemCalled := 0 + itemsCalled := 0 + + statefulSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return orgItemFunc(client, namespace, name) + } + statefulSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return orgItemsFunc(client, namespace) + } + assert.True(t, statefulSetFuncs.SupportsPatch) assert.NotEmpty(t, statefulSetFuncs.PatchTemplatesFunc().AnnotationTemplate) @@ -2465,7 +2563,9 @@ func TestRollingUpgradeForStatefulSetWithPatchAndRetryUsingArs(t *testing.T) { t.Errorf("Rolling upgrade failed for StatefulSet with configmap") } - assert.Equal(t, 2, patchCalled) + assert.Equal(t, 1, itemCalled, "ItemFunc should be called once") + assert.Equal(t, 1, itemsCalled, "ItemsFunc should be called once") + assert.Equal(t, 2, patchCalled, "PatchFunc should be called twice") statefulSetFuncs = GetDeploymentRollingUpgradeFuncs() testRollingUpgradeWithPatchAndInvokeDeleteStrategyArs(t, clients, config, statefulSetFuncs, collectors, envVarPostfix) From 8b0311f7990d20901afdc14c3d30cc60637054ff Mon Sep 17 00:00:00 2001 From: Vladimir Videlov Date: Tue, 1 Jul 2025 14:39:08 +0200 Subject: [PATCH 2/3] Rename --- internal/pkg/handler/upgrade.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/pkg/handler/upgrade.go b/internal/pkg/handler/upgrade.go index 9731a630..a7b9bdcd 100644 --- a/internal/pkg/handler/upgrade.go +++ b/internal/pkg/handler/upgrade.go @@ -218,8 +218,8 @@ func PerformAction(clients kube.Clients, config util.Config, upgradeFuncs callba items := upgradeFuncs.ItemsFunc(clients, config.Namespace) for _, item := range items { - err := retryOnConflict(retry.DefaultRetry, func(shouldRefresh bool) error { - return upgradeResource(clients, config, upgradeFuncs, collectors, recorder, strategy, item, shouldRefresh) + err := retryOnConflict(retry.DefaultRetry, func(fetchResource bool) error { + return upgradeResource(clients, config, upgradeFuncs, collectors, recorder, strategy, item, fetchResource) }) if err != nil { return err From f896e1462d15b35f6eeecdc392a4c3c669ceb90e Mon Sep 17 00:00:00 2001 From: Vladimir Videlov Date: Wed, 2 Jul 2025 12:16:06 +0200 Subject: [PATCH 3/3] Update --- internal/pkg/handler/upgrade.go | 1 - internal/pkg/handler/upgrade_test.go | 36 ++++++++++------------------ 2 files changed, 12 insertions(+), 25 deletions(-) diff --git a/internal/pkg/handler/upgrade.go b/internal/pkg/handler/upgrade.go index a7b9bdcd..b87a8598 100644 --- a/internal/pkg/handler/upgrade.go +++ b/internal/pkg/handler/upgrade.go @@ -205,7 +205,6 @@ func doRollingUpgrade(config util.Config, collectors metrics.Collectors, recorde } func rollingUpgrade(clients kube.Clients, config util.Config, upgradeFuncs callbacks.RollingUpgradeFuncs, collectors metrics.Collectors, recorder record.EventRecorder, strategy invokeStrategy) error { - err := PerformAction(clients, config, upgradeFuncs, collectors, recorder, strategy) if err != nil { logrus.Errorf("Rolling upgrade for '%s' failed with error = %v", config.ResourceName, err) diff --git a/internal/pkg/handler/upgrade_test.go b/internal/pkg/handler/upgrade_test.go index 3a57783a..1d68a3ed 100644 --- a/internal/pkg/handler/upgrade_test.go +++ b/internal/pkg/handler/upgrade_test.go @@ -1441,18 +1441,16 @@ func TestRollingUpgradeForDeploymentWithConfigmapUsingArs(t *testing.T) { deploymentFuncs := GetDeploymentRollingUpgradeFuncs() collectors := getCollectors() - orgItemFunc := deploymentFuncs.ItemFunc - orgItemsFunc := deploymentFuncs.ItemsFunc itemCalled := 0 itemsCalled := 0 deploymentFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { itemCalled++ - return orgItemFunc(client, namespace, name) + return callbacks.GetDeploymentItem(client, namespace, name) } deploymentFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { itemsCalled++ - return orgItemsFunc(client, namespace) + return callbacks.GetDeploymentItems(client, namespace) } err := PerformAction(clients, config, deploymentFuncs, collectors, nil, invokeReloadStrategy) @@ -1492,18 +1490,16 @@ func TestRollingUpgradeForDeploymentWithPatchAndRetryUsingArs(t *testing.T) { assert.True(t, deploymentFuncs.SupportsPatch) assert.NotEmpty(t, deploymentFuncs.PatchTemplatesFunc().AnnotationTemplate) - orgItemFunc := deploymentFuncs.ItemFunc - orgItemsFunc := deploymentFuncs.ItemsFunc itemCalled := 0 itemsCalled := 0 deploymentFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { itemCalled++ - return orgItemFunc(client, namespace, name) + return callbacks.GetDeploymentItem(client, namespace, name) } deploymentFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { itemsCalled++ - return orgItemsFunc(client, namespace) + return callbacks.GetDeploymentItems(client, namespace) } patchCalled := 0 @@ -2238,18 +2234,16 @@ func TestRollingUpgradeForDaemonSetWithConfigmapUsingArs(t *testing.T) { daemonSetFuncs := GetDaemonSetRollingUpgradeFuncs() collectors := getCollectors() - orgItemFunc := daemonSetFuncs.ItemFunc - orgItemsFunc := daemonSetFuncs.ItemsFunc itemCalled := 0 itemsCalled := 0 daemonSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { itemCalled++ - return orgItemFunc(client, namespace, name) + return callbacks.GetDaemonSetItem(client, namespace, name) } daemonSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { itemsCalled++ - return orgItemsFunc(client, namespace) + return callbacks.GetDaemonSetItems(client, namespace) } err := PerformAction(clients, config, daemonSetFuncs, collectors, nil, invokeReloadStrategy) @@ -2286,18 +2280,16 @@ func TestRollingUpgradeForDaemonSetWithPatchAndRetryUsingArs(t *testing.T) { config := getConfigWithAnnotations(envVarPostfix, arsConfigmapName, shaData, options.ConfigmapUpdateOnChangeAnnotation, options.ConfigmapReloaderAutoAnnotation) daemonSetFuncs := GetDaemonSetRollingUpgradeFuncs() - orgItemFunc := daemonSetFuncs.ItemFunc - orgItemsFunc := daemonSetFuncs.ItemsFunc itemCalled := 0 itemsCalled := 0 daemonSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { itemCalled++ - return orgItemFunc(client, namespace, name) + return callbacks.GetDaemonSetItem(client, namespace, name) } daemonSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { itemsCalled++ - return orgItemsFunc(client, namespace) + return callbacks.GetDaemonSetItems(client, namespace) } assert.True(t, daemonSetFuncs.SupportsPatch) @@ -2473,18 +2465,16 @@ func TestRollingUpgradeForStatefulSetWithConfigmapUsingArs(t *testing.T) { statefulSetFuncs := GetStatefulSetRollingUpgradeFuncs() collectors := getCollectors() - orgItemFunc := statefulSetFuncs.ItemFunc - orgItemsFunc := statefulSetFuncs.ItemsFunc itemCalled := 0 itemsCalled := 0 statefulSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { itemCalled++ - return orgItemFunc(client, namespace, name) + return callbacks.GetStatefulSetItem(client, namespace, name) } statefulSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { itemsCalled++ - return orgItemsFunc(client, namespace) + return callbacks.GetStatefulSetItems(client, namespace) } err := PerformAction(clients, config, statefulSetFuncs, collectors, nil, invokeReloadStrategy) @@ -2521,18 +2511,16 @@ func TestRollingUpgradeForStatefulSetWithPatchAndRetryUsingArs(t *testing.T) { config := getConfigWithAnnotations(envVarPostfix, arsConfigmapName, shaData, options.ConfigmapUpdateOnChangeAnnotation, options.ConfigmapReloaderAutoAnnotation) statefulSetFuncs := GetStatefulSetRollingUpgradeFuncs() - orgItemFunc := statefulSetFuncs.ItemFunc - orgItemsFunc := statefulSetFuncs.ItemsFunc itemCalled := 0 itemsCalled := 0 statefulSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { itemCalled++ - return orgItemFunc(client, namespace, name) + return callbacks.GetStatefulSetItem(client, namespace, name) } statefulSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { itemsCalled++ - return orgItemsFunc(client, namespace) + return callbacks.GetStatefulSetItems(client, namespace) } assert.True(t, statefulSetFuncs.SupportsPatch)