test(robot): migrate test_replica_auto_balance_disk_in_pressure

Signed-off-by: Yang Chiu <[email protected]>

e2e/keywords/common.resource

@@ -34,7 +34,7 @@ Set test environment
     ${host_provider}=    Get Environment Variable    HOST_PROVIDER
     ${disk_path}=    Set Variable If    "${host_provider}" == "harvester"    /dev/vdc    /dev/xvdh
     FOR    ${worker_node}    IN    @{worker_nodes}
-        add_disk    ${worker_node}    block    ${disk_path}
+        add_disk    block-disk    ${worker_node}    block    ${disk_path}
     END
 
 Cleanup test resources

e2e/keywords/node.resource

@@ -3,12 +3,13 @@ Documentation    Node Keywords
 
 Library    ../libs/keywords/common_keywords.py
 Library    ../libs/keywords/node_keywords.py
+Library    ../libs/keywords/volume_keywords.py
 
 *** Keywords ***
 Add ${disk_type} type disk ${disk_path} for all worker nodes
     ${worker_nodes}=    get_worker_nodes
     FOR    ${worker_node}    IN    @{worker_nodes}
-        add_disk    ${worker_node}    ${disk_type}    ${disk_path}
+        add_disk    ${disk_type}-disk    ${worker_node}    ${disk_type}    ${disk_path}
     END
 
 Set node ${node_id} with
@@ -31,3 +32,32 @@ Disable node ${node_id} default disk
 Enable node ${node_id} default disk
     ${node_name} =    get_node_by_index    ${node_id}
     enable_default_disk    ${node_name}
+
+Disable disk ${disk_id} scheduling on node ${node_id}
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    disable_disk    ${node_name}    ${disk_name}
+
+Enable disk ${disk_id} scheduling on node ${node_id}
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    enable_disk    ${node_name}    ${disk_name}
+
+Check node ${node_id} disk ${disk_id} is in pressure
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    wait_for_disk_in_pressure    ${node_name}    ${disk_name}
+
+Check node ${node_id} disk ${disk_id} is not in pressure
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    wait_for_disk_not_in_pressure    ${node_name}    ${disk_name}
+
+Create ${disk_size} Gi disk ${disk_id} on node ${node_id}
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    create_volume   ${disk_name}    size=${disk_size}Gi    numberOfReplicas=1
+    attach_volume    ${disk_name}    ${node_name}
+    wait_for_volume_healthy    ${disk_name}
+    ${mount_path} =    mount_disk    ${disk_name}    ${node_name}
+    add_disk    ${disk_name}    ${node_name}    filesystem    ${mount_path}

e2e/keywords/replica.resource

@@ -3,8 +3,16 @@ Documentation       Longhorn replica related keywords
 
 Library             ../libs/keywords/common_keywords.py
 Library             ../libs/keywords/replica_keywords.py
+Library             ../libs/keywords/node_keywords.py
 
 *** Keywords ***
 Volume ${volume_id} replica ${setting_name} should be ${setting_value}
     ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
     validate_replica_setting    ${volume_name}    ${setting_name}    ${setting_value}
+
+There should be replicas running on node ${node_id} disk ${disk_id}
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    ${disk_uuid} =    get_disk_uuid    ${node_name}    ${disk_name}
+    ${replicas} =    get_replicas    volume_name=    node_name=${node_name}    disk_uuid=${disk_uuid}
+    Should Be True    len(${replicas}) > 0

e2e/keywords/statefulset.resource

@@ -15,6 +15,14 @@ Create statefulset ${statefulset_id} using ${volume_type} volume with ${sc_name}
     ${statefulset_name} =    generate_name_with_suffix    statefulset    ${statefulset_id}
     create_statefulset    ${statefulset_name}    ${volume_type}    ${sc_name}
 
+Create statefulset ${statefulset_id} using ${volume_type} volume with ${sc_name} storageclass and size ${size} Mi
+    ${statefulset_name} =    generate_name_with_suffix    statefulset    ${statefulset_id}
+    create_statefulset    ${statefulset_name}    ${volume_type}    ${sc_name}    ${size}Mi
+
+Create statefulset ${statefulset_id} using ${volume_type} volume with ${sc_name} storageclass and size ${size} Gi
+    ${statefulset_name} =    generate_name_with_suffix    statefulset    ${statefulset_id}
+    create_statefulset    ${statefulset_name}    ${volume_type}    ${sc_name}    ${size}Gi
+
 Scale statefulset ${statefulset_id} to ${replicaset_size}
     ${statefulset_name} =    generate_name_with_suffix    statefulset    ${statefulset_id}
     scale_statefulset    ${statefulset_name}    ${replicaset_size}

e2e/keywords/workload.resource

@@ -8,6 +8,7 @@ Library    ../libs/keywords/volume_keywords.py
 Library    ../libs/keywords/workload_keywords.py
 Library    ../libs/keywords/host_keywords.py
 Library    ../libs/keywords/k8s_keywords.py
+Library    ../libs/keywords/replica_keywords.py
 
 *** Keywords ***
 Create pod ${pod_id} using volume ${volume_id}
@@ -213,3 +214,12 @@ Delete Longhorn ${workload_kind} ${workload_name} pod
     ${pod_name} =    get_workload_pod_name    ${workload_name}    longhorn-system
     Log    ${pod_name}
     delete_pod    ${pod_name}     longhorn-system
+
+Check volume of ${workload_kind} ${workload_id} replica on node ${node_id} disk ${disk_id}
+    ${workload_name} =   generate_name_with_suffix    ${workload_kind}    ${workload_id}
+    ${volume_name} =    get_workload_volume_name    ${workload_name}
+    ${node_name} =    get_node_by_index    ${node_id}
+    ${disk_name} =    generate_name_with_suffix    disk    ${disk_id}
+    ${disk_uuid} =    get_disk_uuid    ${node_name}    ${disk_name}
+    ${replicas} =    get_replicas    volume_name=${volume_name}    node_name=${node_name}    disk_uuid=${disk_uuid}
+    Should Be True    len(${replicas}) > 0

e2e/libs/keywords/node_keywords.py

@@ -11,10 +11,14 @@ def __init__(self):
     def list_node_names_by_role(self, role):
         return self.node.list_node_names_by_role(role)
 
-    def add_disk(self, node_name, type, path):
-        logging(f"Adding {type} type disk {path} to node {node_name}")
+    def mount_disk(self, disk_name, node_name):
+        logging(f"Mount device /dev/longhorn/{disk_name} on node {node_name}")
+        return self.node.mount_disk(disk_name, node_name)
+
+    def add_disk(self, disk_name, node_name, type, path):
+        logging(f"Adding {type} type disk {disk_name} {path} to node {node_name}")
         disk = {
-            f"{type}-disk": {
+            f"{disk_name}": {
                 "diskType": type,
                 "path": path,
                 "allowScheduling": True
@@ -38,6 +42,13 @@ def set_node(self, node_name, allowScheduling=True, evictionRequested=False):
         logging(f"Setting node {node_name}; scheduling={allowScheduling}; evictionRequested={evictionRequested}")
         self.node.set_node(node_name, allowScheduling, evictionRequested)
 
+    def disable_disk(self, node_name, disk_name):
+        self.node.set_disk_scheduling(node_name, disk_name, allowScheduling=False)
+
+    def enable_disk(self, node_name, disk_name):
+        self.node.set_disk_scheduling(node_name, disk_name, allowScheduling=True)
+
+
     def disable_node_scheduling(self, node_name):
         self.node.set_node_scheduling(node_name, allowScheduling=False)
 
@@ -52,3 +63,15 @@ def reset_node_schedule(self):
 
     def check_node_is_not_schedulable(self, node_name):
         self.node.check_node_schedulable(node_name, schedulable="False")
+
+    def is_disk_in_pressure(self, node_name, disk_name):
+        return self.node.is_disk_in_pressure(node_name, disk_name)
+
+    def wait_for_disk_in_pressure(self, node_name, disk_name):
+        self.node.wait_for_disk_in_pressure(node_name, disk_name)
+
+    def wait_for_disk_not_in_pressure(self, node_name, disk_name):
+        self.node.wait_for_disk_not_in_pressure(node_name, disk_name)
+
+    def get_disk_uuid(self, node_name, disk_name):
+        return self.node.get_disk_uuid(node_name, disk_name)

e2e/libs/keywords/replica_keywords.py

@@ -8,3 +8,6 @@ def __init__(self):
 
     def validate_replica_setting(self, volume_name, setting_name, value):
         return self.replica.validate_replica_setting(volume_name, setting_name, value)
+
+    def get_replicas(self, volume_name=None, node_name=None, disk_uuid=None):
+        return self.replica.get(volume_name, node_name, disk_uuid)

e2e/libs/keywords/statefulset_keywords.py

@@ -28,9 +28,9 @@ def cleanup_statefulsets(self):
         for statefulset in statefulsets.items:
             self.delete_statefulset(statefulset.metadata.name)
 
-    def create_statefulset(self, name, volume_type="RWO", sc_name="longhorn"):
-        logging(f'Creating {volume_type} statefulset {name} with {sc_name} storageclass')
-        create_statefulset(name, volume_type, sc_name)
+    def create_statefulset(self, name, volume_type="RWO", sc_name="longhorn", size=None):
+        logging(f'Creating {volume_type} statefulset {name} with {sc_name} storageclass and size = {size}')
+        create_statefulset(name, volume_type, sc_name, size)
 
     def delete_statefulset(self, name):
         logging(f'Deleting statefulset {name}')

e2e/libs/node/node.py

@@ -1,5 +1,6 @@
 import time
 import re
+import os
 
 from kubernetes import client
 from robot.libraries.BuiltIn import BuiltIn
@@ -9,15 +10,27 @@
 from utility.utility import get_longhorn_client
 from utility.utility import get_retry_count_and_interval
 from utility.utility import logging
-
+from node_exec import NodeExec
 
 class Node:
 
     DEFAULT_DISK_PATH = "/var/lib/longhorn/"
+    DEFAULT_VOLUME_PATH = "/dev/longhorn/"
 
     def __init__(self):
         self.retry_count, self.retry_interval = get_retry_count_and_interval()
 
+    def mount_disk(self, disk_name, node_name):
+        mount_path = os.path.join(self.DEFAULT_DISK_PATH, disk_name)
+        device_path = os.path.join(self.DEFAULT_VOLUME_PATH, disk_name)
+        cmd = f"mkdir -p {mount_path}"
+        res = NodeExec(node_name).issue_cmd(cmd)
+        cmd = f"mkfs.ext4 {device_path}"
+        res = NodeExec(node_name).issue_cmd(cmd)
+        cmd = f"mount {device_path} {mount_path}"
+        res = NodeExec(node_name).issue_cmd(cmd)
+        return mount_path
+
     def update_disks(self, node_name, disks):
         node = get_longhorn_client().by_id_node(node_name)
         for _ in range(self.retry_count):
@@ -37,9 +50,9 @@ def wait_for_disk_update(self, node_name, disk_num):
                 disks = node.disks
                 for d in disks:
                     if disks[d]["diskUUID"] == "" or \
-                        not disks[d]["conditions"] or \
-                        disks[d]["conditions"]["Ready"]["status"] != "True" or \
-                        disks[d]["conditions"]["Schedulable"]["status"] != "True":
+                        (disks[d]["allowScheduling"] and \
+                        (not disks[d]["conditions"] or \
+                        disks[d]["conditions"]["Ready"]["status"] != "True")):
                         all_updated = False
                         break
                 if all_updated:
@@ -59,15 +72,20 @@ def reset_disks(self, node_name):
         for disk_name, disk in iter(node.disks.items()):
             if disk.path != self.DEFAULT_DISK_PATH:
                 disk.allowScheduling = False
+                logging(f"Disabling scheduling disk {disk_name} on node {node_name}")
+            else:
+                disk.allowScheduling = True
+                logging(f"Enabling scheduling disk {disk_name} on node {node_name}")
         self.update_disks(node_name, node.disks)
 
         disks = {}
         for disk_name, disk in iter(node.disks.items()):
             if disk.path == self.DEFAULT_DISK_PATH:
                 disks[disk_name] = disk
                 disk.allowScheduling = True
+                logging(f"Keeping disk {disk_name} on node {node_name}")
             else:
-                logging(f"Try to remove disk {disk_name} from node {node_name}")
+                logging(f"Removing disk {disk_name} from node {node_name}")
         self.update_disks(node_name, disks)
 
     def is_accessing_node_by_index(self, node):
@@ -183,6 +201,14 @@ def set_default_disk_scheduling(self, node_name, allowScheduling):
                 disk.allowScheduling = allowScheduling
         self.update_disks(node_name, node.disks)
 
+    def set_disk_scheduling(self, node_name, disk_name, allowScheduling):
+        node = get_longhorn_client().by_id_node(node_name)
+
+        for name, disk in iter(node.disks.items()):
+            if name == disk_name:
+                disk.allowScheduling = allowScheduling
+        self.update_disks(node_name, node.disks)
+
     def check_node_schedulable(self, node_name, schedulable):
         node = get_longhorn_client().by_id_node(node_name)
         for _ in range(self.retry_count):
@@ -194,3 +220,29 @@ def check_node_schedulable(self, node_name, schedulable):
     def is_node_schedulable(self, node_name):
         node = get_longhorn_client().by_id_node(node_name)
         return node["conditions"]["Schedulable"]["status"]
+
+    def is_disk_in_pressure(self, node_name, disk_name):
+        node = get_longhorn_client().by_id_node(node_name)
+        return node["disks"][disk_name]["conditions"]["Schedulable"]["reason"] == "DiskPressure"
+
+    def wait_for_disk_in_pressure(self, node_name, disk_name):
+        for i in range(self.retry_count):
+            is_in_pressure = self.is_disk_in_pressure(node_name, disk_name)
+            logging(f"Waiting for disk {disk_name} on node {node_name} in pressure ... ({i})")
+            if is_in_pressure:
+                break
+            time.sleep(self.retry_interval)
+        assert self.is_disk_in_pressure(node_name, disk_name), f"Waiting for node {node_name} disk {disk_name} in pressure failed: {get_longhorn_client().by_id_node(node_name)}"
+
+    def wait_for_disk_not_in_pressure(self, node_name, disk_name):
+        for i in range(self.retry_count):
+            is_in_pressure = self.is_disk_in_pressure(node_name, disk_name)
+            logging(f"Waiting for disk {disk_name} on node {node_name} not in pressure ... ({i})")
+            if not is_in_pressure:
+                break
+            time.sleep(self.retry_interval)
+        assert not self.is_disk_in_pressure(node_name, disk_name), f"Waiting for node {node_name} disk {disk_name} not in pressure failed: {get_longhorn_client().by_id_node(node_name)}"
+
+    def get_disk_uuid(self, node_name, disk_name):
+        node = get_longhorn_client().by_id_node(node_name)
+        return node["disks"][disk_name]["diskUUID"]

e2e/libs/replica/base.py

@@ -4,7 +4,7 @@
 class Base(ABC):
 
     @abstractmethod
-    def get(self, volume_name, node_name):
+    def get(self, volume_name, node_name, disk_uuid):
         return NotImplemented
 
     @abstractmethod

e2e/libs/replica/crd.py

@@ -10,12 +10,14 @@ class CRD(Base):
     def __init__(self):
         self.obj_api = client.CustomObjectsApi()
 
-    def get(self, volume_name, node_name=None):
+    def get(self, volume_name=None, node_name=None, disk_uuid=None):
         label_selector = []
-        if volume_name != "":
+        if volume_name:
             label_selector.append(f"longhornvolume={volume_name}")
         if node_name:
             label_selector.append(f"longhornnode={node_name}")
+        if disk_uuid:
+            label_selector.append(f"longhorndiskuuid={disk_uuid}")
 
         replicas = self.obj_api.list_namespaced_custom_object(
             group="longhorn.io",

e2e/libs/replica/replica.py

@@ -16,8 +16,8 @@ def __init__(self):
     def delete(self, volume_name="", node_name=""):
         return self.replica.delete(volume_name, node_name)
 
-    def get(self, volume_name, node_name):
-        return self.replica.get(volume_name, node_name)
+    def get(self, volume_name, node_name, disk_uuid=None):
+        return self.replica.get(volume_name, node_name, disk_uuid)
 
     def wait_for_rebuilding_start(self, volume_name, node_name):
         return self.replica.wait_for_rebuilding_start(volume_name,node_name)

e2e/libs/replica/rest.py

@@ -12,7 +12,7 @@ class Rest(Base):
     def __init__(self):
         pass
 
-    def get(self, volume_name, node_name):
+    def get(self, volume_name, node_name, disk_uuid):
         return NotImplemented
 
     def delete(self, volume_name, node_name):

e2e/libs/workload/statefulset.py

@@ -10,7 +10,7 @@
 from utility.utility import logging
 
 
-def create_statefulset(statefulset_name, volume_type, sc_name):
+def create_statefulset(statefulset_name, volume_type, sc_name, size):
     filepath = "./templates/workload/statefulset.yaml"
     with open(filepath, 'r') as f:
         namespace = 'default'
@@ -30,6 +30,10 @@ def create_statefulset(statefulset_name, volume_type, sc_name):
         if volume_type == 'RWX':
             manifest_dict['spec']['volumeClaimTemplates'][0]['spec']['accessModes'][0] = 'ReadWriteMany'
 
+        # correct request storage size
+        if size:
+            manifest_dict['spec']['volumeClaimTemplates'][0]['spec']['resources']['requests']['storage'] = size
+
         api = client.AppsV1Api()
         statefulset = api.create_namespaced_stateful_set(
             body=manifest_dict,