upgrade: retry when node image update fails due to conflict error (#2123)

* retry when node image update fails due to conflict error

* improve test

cli/internal/kubernetes/BUILD.bazel

@@ -38,6 +38,7 @@ go_library(
         "@io_k8s_client_go//dynamic",
         "@io_k8s_client_go//kubernetes",
         "@io_k8s_client_go//tools/clientcmd",
+        "@io_k8s_client_go//util/retry",
         "@io_k8s_kubernetes//cmd/kubeadm/app/apis/kubeadm/v1beta3",
         "@io_k8s_sigs_yaml//:yaml",
     ],
@@ -59,10 +60,13 @@ go_test(
         "//internal/versions/components",
         "//operators/constellation-node-operator/api/v1alpha1",
         "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//mock",
         "@com_github_stretchr_testify//require",
         "@io_k8s_api//core/v1:core",
+        "@io_k8s_apimachinery//pkg/api/errors",
         "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
         "@io_k8s_apimachinery//pkg/apis/meta/v1/unstructured",
         "@io_k8s_apimachinery//pkg/runtime",
+        "@io_k8s_apimachinery//pkg/runtime/schema",
     ],
 )

cli/internal/kubernetes/upgrade.go

@@ -44,6 +44,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/retry"
 	kubeadmv1beta3 "k8s.io/kubernetes/cmd/kubeadm/app/apis/kubeadm/v1beta3"
 	"sigs.k8s.io/yaml"
 )
@@ -426,23 +427,31 @@ func (u *Upgrader) applyComponentsCM(ctx context.Context, components *corev1.Con
 }
 
 func (u *Upgrader) applyNodeVersion(ctx context.Context, nodeVersion updatev1alpha1.NodeVersion) (updatev1alpha1.NodeVersion, error) {
-	raw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&nodeVersion)
-	if err != nil {
-		return updatev1alpha1.NodeVersion{}, fmt.Errorf("converting nodeVersion to unstructured: %w", err)
-	}
 	u.log.Debugf("Triggering NodeVersion upgrade now")
-	// Send the updated NodeVersion resource
+	var updatedNodeVersion updatev1alpha1.NodeVersion
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		newNode, err := u.getClusterStatus(ctx)
+		if err != nil {
+			return fmt.Errorf("retrieving current NodeVersion: %w", err)
+		}
+		cmd := newUpgradeVersionCmd(nodeVersion)
+		cmd.SetUpdatedVersions(&newNode)
+		raw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&newNode)
+		if err != nil {
+			return fmt.Errorf("converting nodeVersion to unstructured: %w", err)
+		}
 		updated, err := u.dynamicInterface.Update(ctx, &unstructured.Unstructured{Object: raw})
 		if err != nil {
-		return updatev1alpha1.NodeVersion{}, fmt.Errorf("updating NodeVersion: %w", err)
+			return err
 		}
 
-	var updatedNodeVersion updatev1alpha1.NodeVersion
 		if err := runtime.DefaultUnstructuredConverter.FromUnstructured(updated.UnstructuredContent(), &updatedNodeVersion); err != nil {
-		return updatev1alpha1.NodeVersion{}, fmt.Errorf("converting unstructured to NodeVersion: %w", err)
+			return fmt.Errorf("converting unstructured to NodeVersion: %w", err)
 		}
+		return nil
+	})
 
-	return updatedNodeVersion, nil
+	return updatedNodeVersion, err
 }
 
 func (u *Upgrader) getClusterStatus(ctx context.Context) (updatev1alpha1.NodeVersion, error) {
@@ -599,3 +608,34 @@ type imageFetcher interface {
 		image, region string,
 	) (string, error)
 }
+
+type upgradeVersionCmd struct {
+	imageVersion     string
+	imageRef         string
+	k8sComponentsRef string
+	k8sVersion       string
+}
+
+func newUpgradeVersionCmd(newNodeVersion updatev1alpha1.NodeVersion) upgradeVersionCmd {
+	return upgradeVersionCmd{
+		imageVersion:     newNodeVersion.Spec.ImageVersion,
+		imageRef:         newNodeVersion.Spec.ImageReference,
+		k8sComponentsRef: newNodeVersion.Spec.KubernetesComponentsReference,
+		k8sVersion:       newNodeVersion.Spec.KubernetesClusterVersion,
+	}
+}
+
+func (u upgradeVersionCmd) SetUpdatedVersions(node *updatev1alpha1.NodeVersion) {
+	if u.imageVersion != "" {
+		node.Spec.ImageVersion = u.imageVersion
+	}
+	if u.imageRef != "" {
+		node.Spec.ImageReference = u.imageRef
+	}
+	if u.k8sComponentsRef != "" {
+		node.Spec.KubernetesComponentsReference = u.k8sComponentsRef
+	}
+	if u.k8sVersion != "" {
+		node.Spec.KubernetesClusterVersion = u.k8sVersion
+	}
+}

cli/internal/kubernetes/upgrade_test.go

@@ -13,6 +13,8 @@ import (
 	"io"
 	"testing"
 
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+
 	"github.com/edgelesssys/constellation/v2/internal/attestation/measurements"
 	"github.com/edgelesssys/constellation/v2/internal/attestation/variant"
 	"github.com/edgelesssys/constellation/v2/internal/cloud/cloudprovider"
@@ -24,11 +26,13 @@ import (
 	"github.com/edgelesssys/constellation/v2/internal/versions/components"
 	updatev1alpha1 "github.com/edgelesssys/constellation/v2/operators/constellation-node-operator/v2/api/v1alpha1"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 func TestUpgradeNodeVersion(t *testing.T) {
@@ -46,6 +50,7 @@ func TestUpgradeNodeVersion(t *testing.T) {
 		wantErr               bool
 		wantUpdate            bool
 		assertCorrectError    func(t *testing.T, err error) bool
+		customClientFn        func(nodeVersion updatev1alpha1.NodeVersion) DynamicInterface
 	}{
 		"success": {
 			conf: func() *config.Config {
@@ -252,6 +257,29 @@ func TestUpgradeNodeVersion(t *testing.T) {
 				return assert.ErrorAs(t, err, &upgradeErr)
 			},
 		},
+		"succeed after update retry when the updated node object is outdated": {
+			conf: func() *config.Config {
+				conf := config.Default()
+				conf.Image = "v1.2.3"
+				conf.KubernetesVersion = versions.SupportedK8sVersions()[1]
+				return conf
+			}(),
+			currentImageVersion:   "v1.2.2",
+			currentClusterVersion: versions.SupportedK8sVersions()[0],
+			stable: &stubStableClient{
+				configMaps: map[string]*corev1.ConfigMap{
+					constants.JoinConfigMap: newJoinConfigMap(`{"0":{"expected":"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA","warnOnly":false}}`),
+				},
+			},
+			wantUpdate: false, // because customClient is used
+			customClientFn: func(nodeVersion updatev1alpha1.NodeVersion) DynamicInterface {
+				fakeClient := &fakeDynamicClient{}
+				fakeClient.On("GetCurrent", mock.Anything, mock.Anything).Return(unstructedObjectWithGeneration(nodeVersion, 1), nil)
+				fakeClient.On("Update", mock.Anything, mock.Anything).Return(nil, kerrors.NewConflict(schema.GroupResource{Resource: nodeVersion.Name}, nodeVersion.Name, nil)).Once()
+				fakeClient.On("Update", mock.Anything, mock.Anything).Return(unstructedObjectWithGeneration(nodeVersion, 2), nil).Once()
+				return fakeClient
+			},
+		},
 	}
 
 	for name, tc := range testCases {
@@ -269,9 +297,6 @@ func TestUpgradeNodeVersion(t *testing.T) {
 				},
 			}
 
-			unstrNodeVersion, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&nodeVersion)
-			require.NoError(err)
-
 			var badUpdatedObject *unstructured.Unstructured
 			if tc.badImageVersion != "" {
 				nodeVersion.Spec.ImageVersion = tc.badImageVersion
@@ -280,6 +305,8 @@ func TestUpgradeNodeVersion(t *testing.T) {
 				badUpdatedObject = &unstructured.Unstructured{Object: unstrBadNodeVersion}
 			}
 
+			unstrNodeVersion, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&nodeVersion)
+			require.NoError(err)
 			dynamicClient := &stubDynamicClient{object: &unstructured.Unstructured{Object: unstrNodeVersion}, badUpdatedObject: badUpdatedObject, getErr: tc.getErr}
 			upgrader := Upgrader{
 				stableInterface:  tc.stable,
@@ -288,9 +315,11 @@ func TestUpgradeNodeVersion(t *testing.T) {
 				log:              logger.NewTest(t),
 				outWriter:        io.Discard,
 			}
+			if tc.customClientFn != nil {
+				upgrader.dynamicInterface = tc.customClientFn(nodeVersion)
+			}
 
 			err = upgrader.UpgradeNodeVersion(context.Background(), tc.conf, tc.force)
-
 			// Check upgrades first because if we checked err first, UpgradeImage may error due to other reasons and still trigger an upgrade.
 			if tc.wantUpdate {
 				assert.NotNil(dynamicClient.updatedObject)
@@ -550,6 +579,23 @@ func newJoinConfigMap(data string) *corev1.ConfigMap {
 	}
 }
 
+type fakeDynamicClient struct {
+	mock.Mock
+}
+
+func (u *fakeDynamicClient) GetCurrent(ctx context.Context, str string) (*unstructured.Unstructured, error) {
+	args := u.Called(ctx, str)
+	return args.Get(0).(*unstructured.Unstructured), args.Error(1)
+}
+
+func (u *fakeDynamicClient) Update(ctx context.Context, updatedObject *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	args := u.Called(ctx, updatedObject)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return updatedObject, args.Error(1)
+}
+
 type stubDynamicClient struct {
 	object           *unstructured.Unstructured
 	updatedObject    *unstructured.Unstructured
@@ -615,3 +661,10 @@ func (f *stubImageFetcher) FetchReference(_ context.Context,
 ) (string, error) {
 	return f.reference, f.fetchReferenceErr
 }
+
+func unstructedObjectWithGeneration(nodeVersion updatev1alpha1.NodeVersion, generation int64) *unstructured.Unstructured {
+	unstrNodeVersion, _ := runtime.DefaultUnstructuredConverter.ToUnstructured(&nodeVersion)
+	object := &unstructured.Unstructured{Object: unstrNodeVersion}
+	object.SetGeneration(generation)
+	return object
+}

go.mod

@@ -133,6 +133,7 @@ require (
 	github.com/go-sql-driver/mysql v1.7.1 // indirect
 	github.com/google/pprof v0.0.0-20221103000818-d260c55eee4c // indirect
 	github.com/google/s2a-go v0.1.4 // indirect
+	github.com/stretchr/objx v0.5.0 // indirect
 	go.opentelemetry.io/otel v1.14.0 // indirect
 	go.opentelemetry.io/otel/trace v1.14.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20230530153820-e85fd2cbaebc // indirect