test: add options for additional resources and verify volume detach to node drain test (#11526)

Author: chrischdi

test/e2e/node_drain.go

@@ -64,6 +64,16 @@ type NodeDrainTimeoutSpecInput struct {
 	// Allows to inject a function to be run after test namespace is created.
 	// If not specified, this is a no-op.
 	PostNamespaceCreated func(managementClusterProxy framework.ClusterProxy, workloadClusterNamespace string)
+
+	// Enables additional verification for volumes blocking machine deletion.
+	// Requires to add appropriate resources via CreateAdditionalResources.
+	VerifyNodeVolumeDetach bool
+
+	// Allows to overwrite the default function used for unblocking volume detachments.
+	UnblockNodeVolumeDetachment func(ctx context.Context, bootstrapClusterProxy framework.ClusterProxy, cluster *clusterv1.Cluster)
+
+	// Allows to create additional resources.
+	CreateAdditionalResources func(ctx context.Context, clusterProxy framework.ClusterProxy, cluster *clusterv1.Cluster)
 }
 
 // NodeDrainTimeoutSpec goes through the following steps:
@@ -72,13 +82,16 @@ type NodeDrainTimeoutSpecInput struct {
 // * Deploy MachineDrainRules
 // * Deploy Deployment with unevictable Pods on CP & MD Nodes
 // * Deploy Deployment with evictable Pods with finalizer on CP & MD Nodes
+// * Deploy additional resources if defined in input
 // * Trigger Node drain by scaling down the control plane to 1 and MachineDeployments to 0
 // * Get draining control plane and MachineDeployment Machines
 // * Verify drain of Deployments with order 1
 // * Verify drain of Deployments with order 5
 // * Verify skipped Pods are still there and don't have a deletionTimestamp
 // * Verify Node drains for control plane and MachineDeployment Machines are blocked (only by PDBs)
 // * Set NodeDrainTimeout to 1s to unblock Node drain
+// * Verify machine deletion is blocked by waiting for volume detachment (only if VerifyNodeVolumeDetach is enabled)
+// * Unblocks waiting for volume detachment (only if VerifyNodeVolumeDetach is enabled)
 // * Verify scale down succeeded because Node drains were unblocked.
 func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeoutSpecInput) {
 	var (
@@ -100,6 +113,10 @@ func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeo
 		Expect(input.E2EConfig.GetIntervals(specName, "wait-deployment-available")).ToNot(BeNil())
 		Expect(input.E2EConfig.GetIntervals(specName, "wait-machine-deleted")).ToNot(BeNil())
 
+		if input.VerifyNodeVolumeDetach && input.UnblockNodeVolumeDetachment == nil {
+			input.UnblockNodeVolumeDetachment = unblockNodeVolumeDetachmentFunc(input.E2EConfig.GetIntervals(specName, "wait-control-plane"), input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"))
+		}
+
 		// Setup a Namespace where to host objects for this spec and create a watcher for the namespace events.
 		namespace, cancelWatches = framework.SetupSpecNamespace(ctx, specName, input.BootstrapClusterProxy, input.ArtifactFolder, input.PostNamespaceCreated)
 		clusterResources = new(clusterctl.ApplyClusterTemplateAndWaitResult)
@@ -147,6 +164,9 @@ func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeo
 			Cluster:      cluster,
 			ModifyControlPlaneTopology: func(topology *clusterv1.ControlPlaneTopology) {
 				topology.NodeDrainTimeout = &metav1.Duration{Duration: time.Duration(0)}
+				if input.VerifyNodeVolumeDetach {
+					topology.NodeVolumeDetachTimeout = &metav1.Duration{Duration: time.Duration(0)}
+				}
 				if topology.Metadata.Labels == nil {
 					topology.Metadata.Labels = map[string]string{}
 				}
@@ -159,6 +179,9 @@ func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeo
 			Cluster:      cluster,
 			ModifyMachineDeploymentTopology: func(topology *clusterv1.MachineDeploymentTopology) {
 				topology.NodeDrainTimeout = &metav1.Duration{Duration: time.Duration(0)}
+				if input.VerifyNodeVolumeDetach {
+					topology.NodeVolumeDetachTimeout = &metav1.Duration{Duration: time.Duration(0)}
+				}
 				if topology.Metadata.Labels == nil {
 					topology.Metadata.Labels = map[string]string{}
 				}
@@ -174,12 +197,14 @@ func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeo
 		workloadClusterProxy := input.BootstrapClusterProxy.GetWorkloadCluster(ctx, cluster.Namespace, cluster.Name)
 
 		By("Deploy MachineDrainRules.")
-		Expect(input.BootstrapClusterProxy.GetClient().Create(ctx,
-			generateMachineDrainRule(namespace.Name, clusterName, "drain-order-1", 1))).To(Succeed())
-		Expect(input.BootstrapClusterProxy.GetClient().Create(ctx,
-			generateMachineDrainRule(namespace.Name, clusterName, "drain-order-5", 5))).To(Succeed())
-		Expect(input.BootstrapClusterProxy.GetClient().Create(ctx,
-			generateMachineDrainRule(namespace.Name, clusterName, "drain-order-10", 10))).To(Succeed())
+		machineDrainRules := []*clusterv1.MachineDrainRule{
+			generateMachineDrainRule(namespace.Name, clusterName, "drain-order-1", 1),
+			generateMachineDrainRule(namespace.Name, clusterName, "drain-order-5", 5),
+			generateMachineDrainRule(namespace.Name, clusterName, "drain-order-10", 10),
+		}
+		for _, rule := range machineDrainRules {
+			Expect(input.BootstrapClusterProxy.GetClient().Create(ctx, rule)).To(Succeed())
+		}
 
 		By("Deploy Deployment with unevictable Pods on control plane and MachineDeployment Nodes.")
 		framework.DeployUnevictablePod(ctx, framework.DeployUnevictablePodInput{
@@ -248,6 +273,10 @@ func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeo
 			}
 		}
 
+		if input.CreateAdditionalResources != nil {
+			input.CreateAdditionalResources(ctx, input.BootstrapClusterProxy, cluster)
+		}
+
 		By("Trigger Node drain by scaling down the control plane to 1 and MachineDeployments to 0.")
 		modifyControlPlaneViaClusterAndWait(ctx, modifyControlPlaneViaClusterAndWaitInput{
 			ClusterProxy: input.BootstrapClusterProxy,
@@ -432,7 +461,35 @@ func NodeDrainTimeoutSpec(ctx context.Context, inputGetter func() NodeDrainTimeo
 			WaitForMachineDeployments: input.E2EConfig.GetIntervals(specName, "wait-worker-nodes"),
 		})
 
-		By("Verify scale down succeeded because Node drains were unblocked")
+		if input.VerifyNodeVolumeDetach {
+			By("Verify Node removal for control plane and MachineDeployment Machines are blocked (only by volume detachments)")
+			Eventually(func(g Gomega) {
+				waitingCPMachine := &clusterv1.Machine{}
+				g.Expect(input.BootstrapClusterProxy.GetClient().Get(ctx, drainingCPMachineKey, waitingCPMachine)).To(Succeed())
+
+				condition := conditions.Get(waitingCPMachine, clusterv1.VolumeDetachSucceededCondition)
+				g.Expect(condition).ToNot(BeNil())
+				g.Expect(condition.Status).To(Equal(corev1.ConditionFalse))
+				// Deletion still not be blocked because of the volume.
+				g.Expect(condition.Message).To(ContainSubstring("Waiting for node volumes to be detached"))
+			}, input.E2EConfig.GetIntervals(specName, "wait-machine-deleted")...).Should(Succeed())
+			for _, machineKey := range drainingMDMachineKeys {
+				Eventually(func(g Gomega) {
+					drainedMDMachine := &clusterv1.Machine{}
+					g.Expect(input.BootstrapClusterProxy.GetClient().Get(ctx, machineKey, drainedMDMachine)).To(Succeed())
+
+					condition := conditions.Get(drainedMDMachine, clusterv1.VolumeDetachSucceededCondition)
+					g.Expect(condition).ToNot(BeNil())
+					g.Expect(condition.Status).To(Equal(corev1.ConditionFalse)) // Deletion still not be blocked because of the volume.
+					g.Expect(condition.Message).To(ContainSubstring("Waiting for node volumes to be detached"))
+				}, input.E2EConfig.GetIntervals(specName, "wait-machine-deleted")...).Should(Succeed())
+			}
+
+			By("Executing input.UnblockNodeVolumeDetachment to unblock waiting for volume detachments")
+			input.UnblockNodeVolumeDetachment(ctx, input.BootstrapClusterProxy, cluster)
+		}
+
+		By("Verify scale down succeeded because Node drains and Volume detachments were unblocked")
 		// When we scale down the KCP, controlplane machines are deleted one by one, so it requires more time
 		// MD Machine deletion is done in parallel and will be faster.
 		nodeDrainTimeoutKCPInterval := getDrainAndDeleteInterval(input.E2EConfig.GetIntervals(specName, "wait-machine-deleted"), drainTimeout, controlPlaneReplicas)
@@ -641,3 +698,28 @@ func getDrainAndDeleteInterval(deleteInterval []interface{}, drainTimeout *metav
 	res := []interface{}{intervalDuration.String(), deleteInterval[1]}
 	return res
 }
+
+func unblockNodeVolumeDetachmentFunc(waitControlPlaneIntervals, waitWorkerNodeIntervals []interface{}) func(ctx context.Context, bootstrapClusterProxy framework.ClusterProxy, cluster *clusterv1.Cluster) {
+	return func(ctx context.Context, bootstrapClusterProxy framework.ClusterProxy, cluster *clusterv1.Cluster) {
+		By("Set NodeVolumeDetachTimeout to 1s to unblock waiting for volume detachments")
+		// Note: This also verifies that KCP & MachineDeployments are still propagating changes to NodeVolumeDetachTimeout down to
+		// Machines that already have a deletionTimestamp.
+		nodeVolumeDetachTimeout := &metav1.Duration{Duration: time.Duration(1) * time.Second}
+		modifyControlPlaneViaClusterAndWait(ctx, modifyControlPlaneViaClusterAndWaitInput{
+			ClusterProxy: bootstrapClusterProxy,
+			Cluster:      cluster,
+			ModifyControlPlaneTopology: func(topology *clusterv1.ControlPlaneTopology) {
+				topology.NodeVolumeDetachTimeout = nodeVolumeDetachTimeout
+			},
+			WaitForControlPlane: waitControlPlaneIntervals,
+		})
+		modifyMachineDeploymentViaClusterAndWait(ctx, modifyMachineDeploymentViaClusterAndWaitInput{
+			ClusterProxy: bootstrapClusterProxy,
+			Cluster:      cluster,
+			ModifyMachineDeploymentTopology: func(topology *clusterv1.MachineDeploymentTopology) {
+				topology.NodeVolumeDetachTimeout = nodeVolumeDetachTimeout
+			},
+			WaitForMachineDeployments: waitWorkerNodeIntervals,
+		})
+	}
+}

test/e2e/node_drain_test.go

@@ -20,8 +20,18 @@ limitations under the License.
 package e2e
 
 import (
+	"context"
+	"fmt"
+
 	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
+
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/test/framework"
 )
 
 var _ = Describe("When testing Node drain", func() {
@@ -34,6 +44,40 @@ var _ = Describe("When testing Node drain", func() {
 			SkipCleanup:            skipCleanup,
 			Flavor:                 ptr.To("topology"),
 			InfrastructureProvider: ptr.To("docker"),
+			VerifyNodeVolumeDetach: true,
+			CreateAdditionalResources: func(ctx context.Context, clusterProxy framework.ClusterProxy, cluster *clusterv1.Cluster) {
+				workloadClusterClient := clusterProxy.GetWorkloadCluster(ctx, cluster.Namespace, cluster.Name).GetClient()
+
+				nodeList := &corev1.NodeList{}
+				Expect(workloadClusterClient.List(ctx, nodeList)).To(Succeed())
+
+				// Create a fake VolumeAttachment object for each Node without having a real backing csi driver.
+				for _, node := range nodeList.Items {
+					va := generateVolumeAttachment(node)
+					Expect(workloadClusterClient.Create(ctx, va)).To(Succeed())
+					// Set .Status.Attached to true to make the VolumeAttachment blocking for machine deletions.
+					va.Status.Attached = true
+					Expect(workloadClusterClient.Status().Update(ctx, va)).To(Succeed())
+				}
+			},
 		}
 	})
 })
+
+func generateVolumeAttachment(node corev1.Node) *storagev1.VolumeAttachment {
+	return &storagev1.VolumeAttachment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("va-%s", node.GetName()),
+			Finalizers: []string{
+				"test.cluster.x-k8s.io/block",
+			},
+		},
+		Spec: storagev1.VolumeAttachmentSpec{
+			Attacher: "manual",
+			NodeName: node.GetName(),
+			Source: storagev1.VolumeAttachmentSource{
+				PersistentVolumeName: ptr.To("foo"),
+			},
+		},
+	}
+}

test/framework/convenience.go

@@ -24,6 +24,7 @@ import (
 	coordinationv1 "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	storagev1 "k8s.io/api/storage/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextensionsv1beta "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -88,6 +89,9 @@ func TryAddDefaultSchemes(scheme *runtime.Scheme) {
 	// Add coordination to the schema
 	// Note: This is e.g. used to trigger kube-controller-manager restarts by stealing its lease.
 	_ = coordinationv1.AddToScheme(scheme)
+
+	// Add storagev1 to the scheme
+	_ = storagev1.AddToScheme(scheme)
 }
 
 // ObjectToKind returns the Kind without the package prefix. Pass in a pointer to a struct