Fix CSI node DaemonSet progressing

The Progression condition should be set only when the DaemonSet has a new
content and is actively rolling out its update. After all replicas are
updated, the Progressing condition should be false, even when some pods are
missing. E.g. because a node is drained, something evicted them or so on.

To find out if a DaemonSet has finished roll out (= stable), store the generation
of last stable DaemonSet in a DaemonSet annotation.

If the DaemonSet misses a pod, check its annotation. Is the current
DaemonSet generation equal to the last stable generation? Then it's not a
reconfiguration of the CSI driver and do not set Progressing condition.
Is the current DaemonSet generation different? The operator has changed the
DaemonSet spec, i.e. reconfiguration is in progress, i.e. set the
Progressing condition.

There is still a case when user changes the DaemonSet spec and thus bumps
the generation. Progressing condition will be set. We consider it
acceptable.

Author: jsafrane

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 := c.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 (c *CSIDriverNodeServiceController) 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(int64(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(int64(daemonSet.Generation), 10)
+	if lastStableGeneration == currentGeneration {
+		return daemonSet, nil
+	}
+
+	if isProgressing, _ := c.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 {