[ISSUE-1211]: working multiple volumes per pod with fake-attach (#1214)

tests/e2e-test-framework/conftest.py

@@ -113,8 +113,8 @@ def get_utils(request) -> Utils:
 
 def get_ssh_executors(request) -> dict[str, SSHCommandExecutor]:
     utils  = get_utils(request)
-    worker_ips = utils.get_worker_ips()
-    executors = {ip: SSHCommandExecutor(ip_address=ip, username=utils.vm_user, password=utils.vm_cred) for ip in worker_ips}
+    ips = utils.get_worker_ips() + utils.get_controlplane_ips()
+    executors = {ip: SSHCommandExecutor(ip_address=ip, username=utils.vm_user, password=utils.vm_cred) for ip in ips}
     return executors
 
 @pytest.fixture(scope="session")

tests/e2e-test-framework/framework/const.py

@@ -32,14 +32,28 @@
 STATUS_ONLINE = "ONLINE"
 STATUS_OFFLINE = "OFFLINE"
 
+# annotation keys
+DRIVE_HEALTH_ANNOTATION = "health"
+VOLUME_RELEASE_ANNOTATION = "release"
+FAKE_ATTACH_PVC_ANNOTATION_KEY = "pv.attach.kubernetes.io/ignore-if-inaccessible"
+
+# annotation values
+VOLUME_RELEASE_DONE_VALUE = "done"
+FAKE_ATTACH_PVC_ANNOTATION_VALUE = "yes"
+
 # health
 HEALTH_GOOD = "GOOD"
 HEALTH_BAD = "BAD"
 
-# fake attach
+# fake attach events
 FAKE_ATTACH_INVOLVED = "FakeAttachInvolved"
 FAKE_ATTACH_CLEARED = "FakeAttachCleared"
 
+# drive events
+DRIVE_HEALTH_FAILURE_EVENT = "DriveHealthFailure"
+DRIVE_READY_FOR_PHYSICAL_REMOVAL_EVENT = "DriveReadyForPhysicalRemoval"
+DRIVE_SUCCESSFULLY_REMOVED_EVENT = "DriveSuccessfullyRemoved"
+
 # plurals
 DRIVES_PLURAL = "drives"
 AC_PLURAL = "availablecapacities"

tests/e2e-test-framework/framework/utils.py

@@ -3,6 +3,7 @@
 
 from typing import Any, Callable, Dict, List, Optional
 from kubernetes.client.rest import ApiException
+from kubernetes import watch
 from kubernetes.client.models import (
     V1Pod,
     V1PersistentVolumeClaim,
@@ -566,6 +567,10 @@ def annotate_custom_resource(
                 custom_resource,
             )
 
+        logging.info(
+            f"{resource_type}/{resource_name} annotated with {annotation_key}: {annotation_value}"
+        )
+
     def annotate_pvc(
         self,
         resource_name: str,
@@ -595,6 +600,9 @@ def annotate_pvc(
         self.core_v1_api.patch_namespaced_persistent_volume_claim(
             name=resource_name, namespace=namespace, body=pvc
         )
+        logging.info(
+            f"pvc {resource_name} annotated with {annotation_key}: {annotation_value}"
+        )
 
     def clear_csi_resources(self, namespace: str) -> None:
         """
@@ -691,3 +699,59 @@ def recreate_pod(self, name: str, namespace: str) -> V1Pod:
         logging.info(f"pod {name} is ready")
 
         return pod
+
+    def wait_for_event_with_reason(
+        self, reason: str, timeout_seconds: int = 60
+    ) -> bool:
+        """
+        Wait for an event with a specified reason in the Kubernetes cluster.
+
+        Parameters:
+        - reason (str): The reason of the event to listen for.
+        - timeout_seconds (int): The time in seconds to wait for the event. Default is 60 seconds.
+
+        Returns:
+        - bool: True if the event with the specified reason is raised, False otherwise.
+        """
+        w = watch.Watch()
+        for event in w.stream(
+            self.core_v1_api.list_event_for_all_namespaces,
+            timeout_seconds=timeout_seconds,
+        ):
+            event_reason = event["object"].reason
+            if event_reason == reason:
+                logging.info(f"Event with reason '{reason}' found: {event}")
+                return True
+
+        logging.warning(
+            f"No event with reason '{reason}' found within {timeout_seconds} seconds."
+        )
+        return False
+
+    def clear_pvc_and_pod(
+        self, pod_name: str, pvc_name: str, volume_name: str, namespace: str
+    ) -> None:
+        """
+        Clears the PersistentVolumeClaim (PVC) and the Pod with the specified names in the Kubernetes cluster.
+
+        Args:
+            pod_name (str): The name of the Pod to be cleared.
+            pvc_name (str): The name of the PersistentVolumeClaim to be cleared.
+            volume_name (str): The name of the volume to be checked.
+            namespace (str): The namespace of the PersistentVolumeClaim and Pod.
+
+        Returns:
+            None: This function does not return anything.
+        """
+        logging.info(f"clearing pvc {pvc_name} and pod {pod_name}")
+        self.core_v1_api.delete_namespaced_persistent_volume_claim(
+            name=pvc_name,
+            namespace=namespace,
+        )
+
+        assert self.wait_volume(
+            name=volume_name,
+            expected_usage=const.USAGE_RELEASED,
+        ), f"Volume: {volume_name} failed to reach expected usage: {const.USAGE_RELEASED}"
+
+        self.recreate_pod(name=pod_name, namespace=namespace)

tests/e2e-test-framework/tests/test_fake_attach.py

@@ -50,8 +50,8 @@ def test_5808_fake_attach_without_dr(self):
 
         self.utils.annotate_pvc(
             resource_name=pvc.metadata.name,
-            annotation_key="pv.attach.kubernetes.io/ignore-if-inaccessible",
-            annotation_value="yes",
+            annotation_key=const.FAKE_ATTACH_PVC_ANNOTATION_KEY,
+            annotation_value=const.FAKE_ATTACH_PVC_ANNOTATION_VALUE,
             namespace=self.namespace,
         )
 

tests/e2e-test-framework/tests/test_fake_attach_dr.py

@@ -0,0 +1,142 @@
+import logging
+import time
+import pytest
+
+import framework.const as const
+
+from framework.sts import STS
+from framework.utils import Utils
+from framework.drive import DriveUtils
+
+
+class TestFakeAttachMultipleVolumesPerPod:
+    @classmethod
+    @pytest.fixture(autouse=True)
+    def setup_class(
+        cls,
+        namespace: str,
+        drive_utils_executors: dict[str, DriveUtils],
+        utils: Utils,
+    ):
+        cls.namespace = namespace
+        cls.name = "test-sts-fake-attach-dr"
+        cls.timeout = 120
+        cls.replicas = 1
+
+        cls.utils = utils
+
+        cls.drive_utils = drive_utils_executors
+        cls.sts = STS(cls.namespace, cls.name, cls.replicas)
+        cls.sts.delete()
+        cls.sts.create(storage_classes=[const.SSD_SC, const.HDD_SC])
+
+        yield
+
+        cls.sts.delete()
+
+    @pytest.mark.hal
+    def test_6281_multiple_volumes_per_pod_fake_attach(self):
+        assert (
+            self.sts.verify(self.timeout) is True
+        ), f"STS: {self.name} failed to reach desired number of replicas: {self.replicas}"
+        pod = self.utils.list_pods(
+            label="app=" + self.name, namespace=self.namespace
+        )[0]
+        node_ip = self.utils.get_pod_node_ip(
+            pod_name=pod.metadata.name, namespace=self.namespace
+        )
+        pvcs = self.utils.list_persistent_volume_claims(
+            namespace=self.namespace, pod_name=pod.metadata.name
+        )
+        pvc = [
+            pvc for pvc in pvcs if pvc.spec.storage_class_name == const.HDD_SC
+        ][0]
+        volume = self.utils.list_volumes(
+            name=pvc.spec.volume_name, storage_class=const.HDD_SC
+        )[0]
+        volume_name = volume["metadata"]["name"]
+
+        drive_cr = self.utils.get_drive_cr(
+            volume_name=volume["metadata"]["name"], namespace=self.namespace
+        )
+        drive_name = drive_cr["metadata"]["name"]
+        drive_path = drive_cr["spec"]["Path"]
+
+        self.utils.annotate_custom_resource(
+            resource_name=drive_name,
+            resource_type=const.DRIVES_PLURAL,
+            annotation_key=const.DRIVE_HEALTH_ANNOTATION,
+            annotation_value=const.HEALTH_BAD,
+        )
+
+        assert self.utils.wait_drive(
+            name=drive_name,
+            expected_health=const.HEALTH_BAD,
+            expected_status=const.STATUS_ONLINE,
+            expected_usage=const.USAGE_RELEASING,
+        ), f"Drive: {drive_name} failed to reach expected health: {const.HEALTH_BAD}"
+
+        assert self.utils.event_in(
+            resource_name=drive_name,
+            reason=const.DRIVE_HEALTH_FAILURE_EVENT,
+        )
+
+        self.utils.annotate_custom_resource(
+            resource_name=volume_name,
+            resource_type=const.VOLUMES_PLURAL,
+            annotation_key=const.VOLUME_RELEASE_ANNOTATION,
+            annotation_value=const.VOLUME_RELEASE_DONE_VALUE,
+            namespace=self.namespace,
+        )
+
+        assert self.utils.wait_volume(
+            name=volume_name,
+            expected_usage=const.USAGE_RELEASED,
+        ), f"Volume: {volume_name} failed to reach expected usage: {const.USAGE_RELEASED}"
+
+        self.utils.annotate_pvc(
+            resource_name=pvc.metadata.name,
+            annotation_key=const.FAKE_ATTACH_PVC_ANNOTATION_KEY,
+            annotation_value=const.FAKE_ATTACH_PVC_ANNOTATION_VALUE,
+            namespace=self.namespace,
+        )
+        logging.info(
+            f"PVC {pvc.metadata.name} annotated with {const.FAKE_ATTACH_PVC_ANNOTATION_KEY} = {const.FAKE_ATTACH_PVC_ANNOTATION_VALUE}"
+        )
+        time.sleep(5)
+
+        pod = self.utils.recreate_pod(
+            name=pod.metadata.name, namespace=self.namespace
+        )
+
+        assert self.utils.event_in(
+            resource_name=drive_name,
+            reason=const.DRIVE_READY_FOR_PHYSICAL_REMOVAL_EVENT,
+        )
+
+        assert self.utils.wait_drive(
+            name=drive_name,
+            expected_status=const.STATUS_ONLINE,
+            expected_usage=const.USAGE_REMOVED,
+        )
+
+        scsi_id = self.drive_utils[node_ip].get_scsi_id(drive_path)
+        assert scsi_id, "scsi_id not found"
+        logging.info(f"scsi_id: {scsi_id}")
+
+        self.drive_utils[node_ip].remove(scsi_id)
+        logging.info(f"drive {drive_path}, {scsi_id} removed")
+
+        assert self.utils.wait_for_event_with_reason(
+            reason=const.DRIVE_SUCCESSFULLY_REMOVED_EVENT
+        )
+
+        self.utils.clear_pvc_and_pod(
+            pod_name=pod.metadata.name,
+            pvc_name=pvc.metadata.name,
+            volume_name=pvc.spec.volume_name,
+            namespace=self.namespace,
+        )
+        assert self.utils.is_pod_running(
+            pod_name=pod.metadata.name, timeout=self.timeout
+        ), f"Pod: {pod.metadata.name} failed to reach running state"