diff --git a/internal/pkg/handler/upgrade.go b/internal/pkg/handler/upgrade.go index e84bed4e..b87a8598 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" ) @@ -203,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) @@ -216,10 +217,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(fetchResource bool) error { + return upgradeResource(clients, config, upgradeFuncs, collectors, recorder, strategy, item, fetchResource) }) - if err != nil { return err } @@ -228,16 +228,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..1d68a3ed 100644 --- a/internal/pkg/handler/upgrade_test.go +++ b/internal/pkg/handler/upgrade_test.go @@ -1441,6 +1441,18 @@ func TestRollingUpgradeForDeploymentWithConfigmapUsingArs(t *testing.T) { deploymentFuncs := GetDeploymentRollingUpgradeFuncs() collectors := getCollectors() + itemCalled := 0 + itemsCalled := 0 + + deploymentFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return callbacks.GetDeploymentItem(client, namespace, name) + } + deploymentFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return callbacks.GetDeploymentItems(client, namespace) + } + err := PerformAction(clients, config, deploymentFuncs, collectors, nil, invokeReloadStrategy) time.Sleep(5 * time.Second) if err != nil { @@ -1460,6 +1472,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 +1490,18 @@ func TestRollingUpgradeForDeploymentWithPatchAndRetryUsingArs(t *testing.T) { assert.True(t, deploymentFuncs.SupportsPatch) assert.NotEmpty(t, deploymentFuncs.PatchTemplatesFunc().AnnotationTemplate) + itemCalled := 0 + itemsCalled := 0 + + deploymentFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return callbacks.GetDeploymentItem(client, namespace, name) + } + deploymentFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return callbacks.GetDeploymentItems(client, namespace) + } + patchCalled := 0 deploymentFuncs.PatchFunc = func(client kube.Clients, namespace string, resource runtime.Object, patchType patchtypes.PatchType, bytes []byte) error { patchCalled++ @@ -1498,7 +1526,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 +2234,18 @@ func TestRollingUpgradeForDaemonSetWithConfigmapUsingArs(t *testing.T) { daemonSetFuncs := GetDaemonSetRollingUpgradeFuncs() collectors := getCollectors() + itemCalled := 0 + itemsCalled := 0 + + daemonSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return callbacks.GetDaemonSetItem(client, namespace, name) + } + daemonSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return callbacks.GetDaemonSetItems(client, namespace) + } + err := PerformAction(clients, config, daemonSetFuncs, collectors, nil, invokeReloadStrategy) time.Sleep(5 * time.Second) if err != nil { @@ -2224,6 +2266,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 +2280,18 @@ func TestRollingUpgradeForDaemonSetWithPatchAndRetryUsingArs(t *testing.T) { config := getConfigWithAnnotations(envVarPostfix, arsConfigmapName, shaData, options.ConfigmapUpdateOnChangeAnnotation, options.ConfigmapReloaderAutoAnnotation) daemonSetFuncs := GetDaemonSetRollingUpgradeFuncs() + itemCalled := 0 + itemsCalled := 0 + + daemonSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return callbacks.GetDaemonSetItem(client, namespace, name) + } + daemonSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return callbacks.GetDaemonSetItems(client, namespace) + } + assert.True(t, daemonSetFuncs.SupportsPatch) assert.NotEmpty(t, daemonSetFuncs.PatchTemplatesFunc().AnnotationTemplate) @@ -2263,7 +2320,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 +2465,18 @@ func TestRollingUpgradeForStatefulSetWithConfigmapUsingArs(t *testing.T) { statefulSetFuncs := GetStatefulSetRollingUpgradeFuncs() collectors := getCollectors() + itemCalled := 0 + itemsCalled := 0 + + statefulSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return callbacks.GetStatefulSetItem(client, namespace, name) + } + statefulSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return callbacks.GetStatefulSetItems(client, namespace) + } + err := PerformAction(clients, config, statefulSetFuncs, collectors, nil, invokeReloadStrategy) time.Sleep(5 * time.Second) if err != nil { @@ -2426,6 +2497,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 +2511,18 @@ func TestRollingUpgradeForStatefulSetWithPatchAndRetryUsingArs(t *testing.T) { config := getConfigWithAnnotations(envVarPostfix, arsConfigmapName, shaData, options.ConfigmapUpdateOnChangeAnnotation, options.ConfigmapReloaderAutoAnnotation) statefulSetFuncs := GetStatefulSetRollingUpgradeFuncs() + itemCalled := 0 + itemsCalled := 0 + + statefulSetFuncs.ItemFunc = func(client kube.Clients, namespace string, name string) (runtime.Object, error) { + itemCalled++ + return callbacks.GetStatefulSetItem(client, namespace, name) + } + statefulSetFuncs.ItemsFunc = func(client kube.Clients, namespace string) []runtime.Object { + itemsCalled++ + return callbacks.GetStatefulSetItems(client, namespace) + } + assert.True(t, statefulSetFuncs.SupportsPatch) assert.NotEmpty(t, statefulSetFuncs.PatchTemplatesFunc().AnnotationTemplate) @@ -2465,7 +2551,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)