Merge pull request #2033 from jsafrane/progressing-set-version-annotation

OCPBUGS-62634: Fix CSI node DaemonSet progressing

Author: openshift-merge-bot[bot]

pkg/operator/csi/csidrivernodeservicecontroller/csi_driver_node_service_controller.go

@@ -34,6 +34,7 @@ const (
 	nodeDriverRegistrarImageEnvName = "NODE_DRIVER_REGISTRAR_IMAGE"
 	livenessProbeImageEnvName       = "LIVENESS_PROBE_IMAGE"
 	kubeRBACProxyImageEnvName       = "KUBE_RBAC_PROXY_IMAGE"
+	stableGenerationAnnotationName  = "storage.openshift.io/stable-generation"
 )
 
 // DaemonSetHookFunc is a hook function to modify the DaemonSet.
@@ -181,6 +182,11 @@ func (c *CSIDriverNodeServiceController) syncManaged(ctx context.Context, opSpec
 		return err
 	}
 
+	daemonSet, err = c.storeLastStableGeneration(ctx, syncContext, daemonSet)
+	if err != nil {
+		return err
+	}
+
 	// Create an OperatorStatusApplyConfiguration with generations
 	status := applyoperatorv1.OperatorStatus().
 		WithGenerations(&applyoperatorv1.GenerationStatusApplyConfiguration{
@@ -217,7 +223,7 @@ func (c *CSIDriverNodeServiceController) syncManaged(ctx context.Context, opSpec
 		WithMessage("DaemonSet is not progressing").
 		WithReason("AsExpected")
 
-	if ok, msg := isProgressing(opStatus, daemonSet); ok {
+	if ok, msg := isProgressing(daemonSet); ok {
 		progressingCondition = progressingCondition.
 			WithStatus(opv1.ConditionTrue).
 			WithMessage(msg).
@@ -253,7 +259,22 @@ func (c *CSIDriverNodeServiceController) getDaemonSet(opSpec *opv1.OperatorSpec)
 	return required, nil
 }
 
-func isProgressing(status *opv1.OperatorStatus, daemonSet *appsv1.DaemonSet) (bool, string) {
+func isProgressing(daemonSet *appsv1.DaemonSet) (bool, string) {
+	// Progressing means "[the component] is actively rolling out new code, propagating config
+	// changes (e.g, a version change), or otherwise moving from one steady state to another."
+	// This controller expects that all "config changes" result in increased DaemonSet generation
+	// (i.e. DaemonSet .spec changes)
+	// The controller stores the last "stable" DS generation in an annotation.
+	// Stable means that all DS replicas are available and at the latest version.
+	// Any subsequent missing replicas must be caused by a node added / removed / rebooted /
+	// pod manually killed, which then does not result in Progressing=true.
+	lastStableGeneration := daemonSet.Annotations[stableGenerationAnnotationName]
+	currentGeneration := strconv.FormatInt(daemonSet.Generation, 10)
+	if lastStableGeneration == currentGeneration {
+		// The previous reconfiguration has completed in the past.
+		return false, ""
+	}
+
 	switch {
 	case daemonSet.Generation != daemonSet.Status.ObservedGeneration:
 		return true, "Waiting for DaemonSet to act on changes"
@@ -315,3 +336,20 @@ func (c *CSIDriverNodeServiceController) syncDeleting(ctx context.Context, opSpe
 	// All removed, remove the finalizer as the last step
 	return v1helpers.RemoveFinalizer(ctx, c.operatorClient, c.instanceName)
 }
+
+func (c *CSIDriverNodeServiceController) storeLastStableGeneration(ctx context.Context, syncContext factory.SyncContext, daemonSet *appsv1.DaemonSet) (*appsv1.DaemonSet, error) {
+	lastStableGeneration := daemonSet.Annotations[stableGenerationAnnotationName]
+	currentGeneration := strconv.FormatInt(daemonSet.Generation, 10)
+	if lastStableGeneration == currentGeneration {
+		return daemonSet, nil
+	}
+
+	if isProgressing, _ := isProgressing(daemonSet); isProgressing {
+		return daemonSet, nil
+	}
+
+	klog.V(2).Infof("DaemonSet %s/%s generation %d is stable", daemonSet.Namespace, daemonSet.Name, daemonSet.Generation)
+	daemonSet.Annotations[stableGenerationAnnotationName] = currentGeneration
+	daemonSet, _, err := resourceapply.ApplyDaemonSet(ctx, c.kubeClient.AppsV1(), syncContext.Recorder(), daemonSet, daemonSet.Generation)
+	return daemonSet, err
+}

pkg/operator/csi/csidrivernodeservicecontroller/csi_driver_node_service_controller_test.go

@@ -3,7 +3,6 @@ package csidrivernodeservicecontroller
 import (
 	"context"
 	"fmt"
-	clocktesting "k8s.io/utils/clock/testing"
 	"os"
 	"sort"
 	"strings"
@@ -19,6 +18,8 @@ import (
 	coreinformers "k8s.io/client-go/informers"
 	fakecore "k8s.io/client-go/kubernetes/fake"
 	core "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
+	clocktesting "k8s.io/utils/clock/testing"
 
 	"github.com/google/go-cmp/cmp"
 	opv1 "github.com/openshift/api/operator/v1"
@@ -100,7 +101,7 @@ func newTestContext(test testCase, t *testing.T) *testContext {
 	}
 
 	// Add global reactors
-	addGenerationReactor(coreClient)
+	addGenerationReactor(coreClient, coreInformerFactory.Apps().V1().DaemonSets().Informer().GetStore())
 
 	// fakeDriverInstance also fulfils the OperatorClient interface
 	fakeOperatorClient := v1helpers.NewFakeOperatorClientWithObjectMeta(
@@ -331,8 +332,18 @@ func withDaemonSetTLSConfig() daemonSetModifier {
 	}
 }
 
+func withDaemonSetAnnotation(annotation string, value string) daemonSetModifier {
+	return func(instance *appsv1.DaemonSet) *appsv1.DaemonSet {
+		if instance.Annotations == nil {
+			instance.Annotations = map[string]string{}
+		}
+		instance.Annotations[annotation] = value
+		return instance
+	}
+}
+
 // This reactor is always enabled and bumps DaemonSet generation when they get updated.
-func addGenerationReactor(client *fakecore.Clientset) {
+func addGenerationReactor(client *fakecore.Clientset, store cache.Store) {
 	client.PrependReactor("*", "daemonsets", func(action core.Action) (handled bool, ret runtime.Object, err error) {
 		switch a := action.(type) {
 		case core.CreateActionImpl:
@@ -343,7 +354,20 @@ func addGenerationReactor(client *fakecore.Clientset) {
 		case core.UpdateActionImpl:
 			object := a.GetObject()
 			ds := object.(*appsv1.DaemonSet)
-			ds.Generation++
+
+			// Bump generation only when the spec changes.
+			// Explicitly, do not bump it on annotation change - the API server does not do it either.
+			oldDSObject, exists, err := store.GetByKey(ds.Namespace + "/" + ds.Name)
+			if err != nil {
+				return false, nil, err
+			}
+			if !exists {
+				return false, nil, fmt.Errorf("DaemonSet %s not found", cache.MetaObjectToName(ds))
+			}
+			oldDS := oldDSObject.(*appsv1.DaemonSet)
+			if !equality.Semantic.DeepEqual(oldDS.Spec, ds.Spec) {
+				ds.Generation++
+			}
 			return false, ds, nil
 		}
 		return false, nil, nil
@@ -426,6 +450,7 @@ func TestSync(t *testing.T) {
 			},
 		},
 		{
+			name: "no change",
 			// DaemonSet is fully deployed and its status is synced to CR
 			manifestFunc: makeFakeManifest,
 			images:       defaultImages(),
@@ -434,15 +459,45 @@ func TestSync(t *testing.T) {
 					argsLevel2,
 					defaultImages(),
 					withDaemonSetGeneration(1, 1),
-					withDaemonSetStatus(replica1, replica1, replica1, replica0)),
+					withDaemonSetStatus(replica1, replica1, replica1, replica0),
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "1")),
 				driver: makeFakeDriverInstance(withGenerations(1)),
 			},
 			expectedObjects: testObjects{
 				daemonSet: getDaemonSet(
 					argsLevel2,
 					defaultImages(),
 					withDaemonSetGeneration(1, 1),
-					withDaemonSetStatus(replica1, replica1, replica1, replica0)),
+					withDaemonSetStatus(replica1, replica1, replica1, replica0),
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "1")),
+				driver: makeFakeDriverInstance(
+					// withStatus(replica1),
+					withGenerations(1),
+					withTrueConditions(conditionAvailable),
+					withFalseConditions(conditionProgressing)),
+			},
+		},
+		{
+			name: "finished progressing",
+			// DaemonSet is fully deployed and its status is synced to CR
+			manifestFunc: makeFakeManifest,
+			images:       defaultImages(),
+			initialObjects: testObjects{
+				daemonSet: getDaemonSet(
+					argsLevel2,
+					defaultImages(),
+					withDaemonSetGeneration(1, 1),
+					withDaemonSetStatus(replica1, replica1, replica1, replica0),
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "0")),
+				driver: makeFakeDriverInstance(withGenerations(1)),
+			},
+			expectedObjects: testObjects{
+				daemonSet: getDaemonSet(
+					argsLevel2,
+					defaultImages(),
+					withDaemonSetGeneration(1, 1),
+					withDaemonSetStatus(replica1, replica1, replica1, replica0),
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "1")), // The current generation has reached stability
 				driver: makeFakeDriverInstance(
 					// withStatus(replica1),
 					withGenerations(1),
@@ -490,7 +545,8 @@ func TestSync(t *testing.T) {
 					argsLevel2,
 					defaultImages(),
 					withDaemonSetGeneration(1, 1),
-					withDaemonSetStatus(replica0, replica0, replica1, replica1)), // the DaemonSet is updating 1 pod
+					withDaemonSetStatus(replica0, replica0, replica1, replica1),   // the DaemonSet is updating 1 pod
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "0")), // and the current generation was not stable yet
 				driver: makeFakeDriverInstance(
 					// withStatus(replica1),
 					withGenerations(1),
@@ -502,7 +558,8 @@ func TestSync(t *testing.T) {
 					argsLevel2,
 					defaultImages(),
 					withDaemonSetGeneration(1, 1),
-					withDaemonSetStatus(replica0, replica0, replica1, replica1)), // no change to the DaemonSet
+					withDaemonSetStatus(replica0, replica0, replica1, replica1),
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "0")), // and the current generation was not stable yet
 				driver: makeFakeDriverInstance(
 					// withStatus(replica0),
 					withGenerations(1),
@@ -658,6 +715,38 @@ func TestSync(t *testing.T) {
 					withFalseConditions(conditionAvailable)), // Degraded is set later on
 			},
 		},
+		{
+			// DaemonSet is updating pods
+			name:         "missing pods - not progressing",
+			manifestFunc: makeFakeManifest,
+			images:       defaultImages(),
+			initialObjects: testObjects{
+				daemonSet: getDaemonSet(
+					argsLevel2,
+					defaultImages(),
+					withDaemonSetGeneration(1, 1),
+					withDaemonSetStatus(replica0, replica0, replica1, replica1),   // the DaemonSet is updating 1 pod
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "1")), // But the current generation was stable in the past
+				driver: makeFakeDriverInstance(
+					// withStatus(replica1),
+					withGenerations(1),
+					withTrueConditions(conditionAvailable),
+					withFalseConditions(conditionProgressing)),
+			},
+			expectedObjects: testObjects{
+				daemonSet: getDaemonSet(
+					argsLevel2,
+					defaultImages(),
+					withDaemonSetGeneration(1, 1),
+					withDaemonSetStatus(replica0, replica0, replica1, replica1),
+					withDaemonSetAnnotation(stableGenerationAnnotationName, "1")), // no change to the DaemonSet
+				driver: makeFakeDriverInstance(
+					// withStatus(replica0),
+					withGenerations(1), // But the current generation was stable in the past
+					withTrueConditions(conditionAvailable),
+					withFalseConditions(conditionProgressing)), // The operator is not Progressing
+			},
+		},
 	}
 
 	for _, test := range testCases {