test(robot): migrate test_setting_concurrent_rebuild_limit

Signed-off-by: Yang Chiu <yang.chiu@suse.com>

e2e/keywords/setting.resource

@@ -0,0 +1,8 @@
+*** Settings ***
+Documentation    Setting Keywords
+
+Library    ../libs/keywords/setting_keywords.py
+
+*** Keywords ***
+Set setting ${setting_key} to ${setting_value}
+    update_setting    ${setting_key}    ${setting_value}

e2e/keywords/volume.resource

@@ -45,6 +45,10 @@ Write data to volume ${volume_id}
     ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
     write_volume_random_data   ${volume_name}    2048
 
+Write ${size} GB data to volume ${volume_id}
+    ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
+    write_volume_random_data   ${volume_name}    ${${size} * 1024}
+
 Keep writing data to volume ${volume_id}
     ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
     keep_writing_data    ${volume_name}
@@ -69,6 +73,24 @@ Wait until volume ${volume_id} replicas rebuilding completed
     ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
     wait_for_replica_rebuilding_to_complete   ${volume_name}
 
+Wait until volume ${volume_id} replica rebuilding stopped on ${replica_locality}
+    ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
+    wait_for_replica_rebuilding_to_stop_on_node    ${volume_name}    ${replica_locality}
+
+Only one replica rebuilding on ${replica_locality} will start at a time, either for volume ${volume_id_0} or volume ${volume_id_1}
+    ${volume_name_0} =    generate_name_with_suffix    volume    ${volume_id_0}
+    ${volume_name_1} =    generate_name_with_suffix    volume    ${volume_id_1}
+    only_one_replica_rebuilding_will_start_at_a_time_on_node    ${volume_name_0}    ${volume_name_1}    ${replica_locality}
+
+Both volume ${volume_id_0} and volume ${volume_id_1} replica rebuilding on ${replica_locality} will start at the same time
+    ${volume_name_0} =    generate_name_with_suffix    volume    ${volume_id_0}
+    ${volume_name_1} =    generate_name_with_suffix    volume    ${volume_id_1}
+    both_replica_rebuildings_will_start_at_the_same_time_on_node    ${volume_name_0}    ${volume_name_1}    ${replica_locality}
+
+Crash volume ${volume_id} replica processes
+    ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
+    crash_replica_processes    ${volume_name}
+
 Check volume ${volume_id} data is intact
     ${volume_name} =    generate_name_with_suffix    volume    ${volume_id}
     check_data_checksum    ${volume_name}

e2e/libs/backupstore/base.py

@@ -13,15 +13,6 @@ def is_backupTarget_s3(self, s):
     def is_backupTarget_nfs(self, s):
         return s.startswith("nfs://")
 
-    @classmethod
-    def get_backupstores(cls):
-        backupstore = os.environ['LONGHORN_BACKUPSTORES']
-        backupstore = backupstore.replace(" ", "")
-        backupstores = backupstore.split(",")
-        for i in range(len(backupstores)):
-            backupstores[i] = backupstores[i].split(":")[0]
-        return backupstores
-
     def backup_volume_path(self, volume_name):
         volume_name_sha512 = \
             hashlib.sha512(volume_name.encode('utf-8')).hexdigest()

e2e/libs/keywords/backupstore_keywords.py

@@ -1,27 +1,29 @@
 from backupstore import Nfs, Minio
 from utility.utility import get_longhorn_client
+from utility.utility import get_backupstore
 from kubernetes import client
 from setting import Setting
 
 
 class backupstore_keywords:
 
     def __init__(self):
-        backupstores = Minio.get_backupstores()
-        if backupstores[0] == "s3":
+        backupstore = get_backupstore()
+        if backupstore == "s3":
             self.backupstore = Minio()
-        else:
+        elif backupstore == "nfs":
             self.backupstore = Nfs()
         self.setting = Setting()
 
     def set_backupstore(self):
         self.setting.set_backupstore()
 
     def cleanup_backupstore(self):
-        client = get_longhorn_client()
-        self.backupstore.cleanup_system_backups(client)
-        self.backupstore.cleanup_backup_volumes(client)
-        self.setting.reset_backupstore_setting()
+        if get_backupstore():
+            client = get_longhorn_client()
+            self.backupstore.cleanup_system_backups(client)
+            self.backupstore.cleanup_backup_volumes(client)
+            self.setting.reset_backupstore_setting()
 
     def create_dummy_in_progress_backup(self, volume_name):
         client = get_longhorn_client()

e2e/libs/keywords/setting_keywords.py

@@ -0,0 +1,9 @@
+from setting import Setting
+
+class setting_keywords:
+
+    def __init__(self):
+        self.setting = Setting()
+
+    def update_setting(self, key, value):
+        self.setting.update_setting(key, value)

e2e/libs/keywords/volume_keywords.py

@@ -1,11 +1,13 @@
+import asyncio
+import time
 from node import Node
 from node.utility import check_replica_locality
 
 from utility.constant import ANNOT_CHECKSUM
 from utility.constant import LABEL_TEST
 from utility.constant import LABEL_TEST_VALUE
 from utility.utility import logging
-
+from utility.utility import get_retry_count_and_interval
 from volume import Volume
 
 
@@ -118,23 +120,23 @@ def delete_replica_on_node(self, volume_name, replica_locality):
     def set_annotation(self, volume_name, annotation_key, annotation_value):
         self.volume.set_annotation(volume_name, annotation_key, annotation_value)
 
-    def wait_for_replica_rebuilding_start(self, volume_name, replica_node):
+    async def wait_for_replica_rebuilding_start(self, volume_name, replica_node):
         if str(replica_node).isdigit():
             replica_node = self.node.get_node_by_index(replica_node)
 
         logging(f"Waiting for volume {volume_name}'s replica on node {replica_node} rebuilding started")
-        self.volume.wait_for_replica_rebuilding_start(
+        await self.volume.wait_for_replica_rebuilding_start(
             volume_name,
             replica_node
         )
 
-    def wait_for_replica_rebuilding_to_start_on_node(self, volume_name, replica_locality):
+    async def wait_for_replica_rebuilding_to_start_on_node(self, volume_name, replica_locality):
         check_replica_locality(replica_locality)
 
         node_id = self.get_node_id_by_replica_locality(volume_name, replica_locality)
 
         logging(f"Waiting for volume {volume_name}'s replica on node {node_id} rebuilding started")
-        self.volume.wait_for_replica_rebuilding_start(volume_name, node_id)
+        await self.volume.wait_for_replica_rebuilding_start(volume_name, node_id)
 
     def wait_for_replica_rebuilding_complete(self, volume_name, replica_node):
         if str(replica_node).isdigit():
@@ -159,6 +161,62 @@ def wait_for_replica_rebuilding_to_complete(self, volume_name):
             logging(f"Waiting for volume {volume_name}'s replica on node {node_id} rebuilding completed")
             self.volume.wait_for_replica_rebuilding_complete(volume_name, node_id)
 
+    async def only_one_replica_rebuilding_will_start_at_a_time_on_node(self, volume_name_0, volume_name_1, replica_locality):
+
+        node_id = self.get_node_id_by_replica_locality(volume_name_0, replica_locality)
+
+        first_replica_rebuilding = None
+        not_start_replica_rebuilding = None
+
+        async def find_first_replica_rebuilding():
+            tasks = [
+                asyncio.create_task(self.volume.wait_for_replica_rebuilding_start(volume_name_0, node_id), name=volume_name_0),
+                asyncio.create_task(self.volume.wait_for_replica_rebuilding_start(volume_name_1, node_id), name=volume_name_1)
+            ]
+
+            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+            for pending_task in pending:
+                pending_task.cancel()
+            return done.pop().get_name(), pending.pop().get_name()
+
+        first_replica_rebuilding, not_start_replica_rebuilding = await find_first_replica_rebuilding()
+        logging(f"Observed {first_replica_rebuilding} started replica rebuilding first")
+
+        while self.volume.is_replica_rebuilding_in_progress(first_replica_rebuilding, node_id):
+            logging(f"Checking volume {not_start_replica_rebuilding} replica rebuilding won't start \
+                if volume {first_replica_rebuilding} replica rebuilding is still in progress")
+            assert not self.volume.is_replica_rebuilding_in_progress(not_start_replica_rebuilding, node_id)
+            time.sleep(1)
+
+    async def both_replica_rebuildings_will_start_at_the_same_time_on_node(self, volume_name_0, volume_name_1, replica_locality):
+
+        node_id = self.get_node_id_by_replica_locality(volume_name_0, replica_locality)
+
+        async def wait_for_both_replica_rebuildings():
+            tasks = [
+                asyncio.create_task(self.volume.wait_for_replica_rebuilding_start(volume_name_0, node_id), name=volume_name_0),
+                asyncio.create_task(self.volume.wait_for_replica_rebuilding_start(volume_name_1, node_id), name=volume_name_1)
+            ]
+
+            done, pending = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
+            logging(f"Observed {done.pop().get_name()} and {done.pop().get_name()} started replica rebuilding first")
+
+        await wait_for_both_replica_rebuildings()
+
+        assert self.volume.is_replica_rebuilding_in_progress(volume_name_0, node_id) and self.volume.is_replica_rebuilding_in_progress(volume_name_1, node_id), \
+            f"Expect {volume_name_0} and {volume_name_1} replica rebuilding at the same time"
+
+    def crash_replica_processes(self, volume_name):
+        self.volume.crash_replica_processes(volume_name)
+
+    def wait_for_replica_rebuilding_to_stop_on_node(self, volume_name, replica_locality):
+        node_id = self.get_node_id_by_replica_locality(volume_name, replica_locality)
+        retry_count, retry_interval = get_retry_count_and_interval()
+        for i in range(retry_count):
+            if not self.volume.is_replica_rebuilding_in_progress(volume_name, node_id):
+                break
+            time.sleep(retry_interval)
+
     def wait_for_volume_attached(self, volume_name):
         logging(f'Waiting for volume {volume_name} to be attached')
         self.volume.wait_for_volume_attached(volume_name)

e2e/libs/node/utility.py

@@ -1,2 +1,2 @@
 def check_replica_locality(replica_locality):
-    assert replica_locality in ["replica node", "test pod node", "volume node"]
+    assert replica_locality in ["replica node", "test pod node", "volume node"], f"Unknown replica locality: {replica_locality}: "

e2e/libs/replica/rest.py

@@ -31,7 +31,7 @@ def wait_for_replica_rebuilding_start(self, volume_name, node_name):
                 if rebuilding_replica_name:
                     break
             except Exception as e:
-                logging(f"Failed to get volume {e}")
+                logging(f"Failed to get volume {volume_name} with error: {e}")
             time.sleep(RETRY_INTERVAL)
         assert rebuilding_replica_name != None, f'failed to get rebuilding replica name'
 
@@ -47,7 +47,7 @@ def wait_for_replica_rebuilding_start(self, volume_name, node_name):
                 if started:
                     break
             except Exception as e:
-                logging(f"Failed to get volume {e}")
+                logging(f"Failed to get volume {volume_name} with error: {e}")
             time.sleep(RETRY_INTERVAL)
         assert started, f'replica {rebuilding_replica_name} rebuilding starting failed'
 
@@ -69,7 +69,7 @@ def wait_for_replica_rebuilding_complete(self, volume_name, node_name):
                         completed = True
                         break
             except Exception as e:
-                logging(f"Failed to get volume {e}")
+                logging(f"Failed to get volume {volume_name} with error: {e}")
             if completed:
                 break
             time.sleep(RETRY_INTERVAL)

e2e/libs/setting/setting.py

@@ -1,73 +1,57 @@
 import os
+import time
+from utility.utility import logging
 from utility.utility import get_longhorn_client
-from setting.constant import SETTING_BACKUP_TARGET
-from setting.constant import SETTING_BACKUP_TARGET_CREDENTIAL_SECRET
-from setting.constant import SETTING_BACKUPSTORE_POLL_INTERVAL
+from utility.utility import get_retry_count_and_interval
 
 class Setting:
 
+    SETTING_BACKUP_TARGET = "backup-target"
+    SETTING_BACKUP_TARGET_CREDENTIAL_SECRET = "backup-target-credential-secret"
+    SETTING_BACKUPSTORE_POLL_INTERVAL = "backupstore-poll-interval"
+
     def __init__(self):
         self.longhorn_client = get_longhorn_client()
+        self.retry_count, self.retry_interval = get_retry_count_and_interval()
+
+    def update_setting(self, key, value):
+        for i in range(self.retry_count):
+            try:
+                logging(f"Trying to update setting {key} to {value} ... ({i})")
+                setting = self.longhorn_client.by_id_setting(key)
+                self.longhorn_client.update(setting, value=value)
+                break
+            except Exception as e:
+                logging(e)
+            time.sleep(self.retry_interval)
+
+    def get_setting(self, key):
+        return self.longhorn_client.by_id_setting(key).value
 
     def get_backupstore_url(self):
-        backupstore = os.environ['LONGHORN_BACKUPSTORES']
-        backupstore = backupstore.replace(" ", "")
-        backupstores = backupstore.split(",")
-        assert len(backupstores) != 0
-        return backupstores
+        return os.environ.get('LONGHORN_BACKUPSTORE')
 
     def get_backupstore_poll_interval(self):
-        poll_interval = os.environ['LONGHORN_BACKUPSTORE_POLL_INTERVAL']
-        assert len(poll_interval) != 0
-        return poll_interval
+        return os.environ.get('LONGHORN_BACKUPSTORE_POLL_INTERVAL')
 
     def set_backupstore(self):
-        backupstores = self.get_backupstore_url()
-        poll_interval = self.get_backupstore_poll_interval()
-        for backupstore in backupstores:
+        backupstore = self.get_backupstore_url()
+        if backupstore:
             backupsettings = backupstore.split("$")
-            self.set_backupstore_url(backupsettings[0])
-            self.set_backupstore_credential_secret(backupsettings[1])
-            self.set_backupstore_poll_interval(poll_interval)
-            break
-
-    def reset_backupstore_setting(self):
-        backup_target_setting = self.longhorn_client.by_id_setting(SETTING_BACKUP_TARGET)
-        self.longhorn_client.update(backup_target_setting, value="")
-
-        backup_target_credential_setting = self.longhorn_client.by_id_setting(
-            SETTING_BACKUP_TARGET_CREDENTIAL_SECRET)
-        self.longhorn_client.update(backup_target_credential_setting, value="")
-
-        backup_store_poll_interval = self.longhorn_client.by_id_setting(
-            SETTING_BACKUPSTORE_POLL_INTERVAL)
-        self.longhorn_client.update(backup_store_poll_interval, value="300")
-
-    def set_backupstore_url(self, url):
-        backup_target_setting = self.longhorn_client.by_id_setting(SETTING_BACKUP_TARGET)
-        backup_target_setting = self.longhorn_client.update(backup_target_setting,
-                                                            value=url)
-        assert backup_target_setting.value == url
+            self.update_setting(self.SETTING_BACKUP_TARGET, backupsettings[0])
+            if len(backupsettings) > 1:
+                self.update_setting(self.SETTING_BACKUP_TARGET_CREDENTIAL_SECRET, backupsettings[1])
 
-    def set_backupstore_credential_secret(self, credential_secret):
-        backup_target_credential_setting = self.longhorn_client.by_id_setting(
-            SETTING_BACKUP_TARGET_CREDENTIAL_SECRET)
-        backup_target_credential_setting = self.longhorn_client.update(
-            backup_target_credential_setting, value=credential_secret)
-        assert backup_target_credential_setting.value == credential_secret
+            poll_interval = self.get_backupstore_poll_interval()
+            self.update_setting(self.SETTING_BACKUPSTORE_POLL_INTERVAL, poll_interval)
 
-    def set_backupstore_poll_interval(self, poll_interval):
-        backup_store_poll_interval_setting = self.longhorn_client.by_id_setting(
-            SETTING_BACKUPSTORE_POLL_INTERVAL)
-        backup_target_poll_interal_setting = self.longhorn_client.update(
-            backup_store_poll_interval_setting, value=poll_interval)
-        assert backup_target_poll_interal_setting.value == poll_interval
+    def reset_backupstore(self):
+        self.update_setting(self.SETTING_BACKUP_TARGET, "")
+        self.update_setting(self.SETTING_BACKUP_TARGET_CREDENTIAL_SECRET, "")
+        self.update_setting(self.SETTING_BACKUPSTORE_POLL_INTERVAL, "300")
 
     def get_backup_target(self):
-        backup_target_setting = self.longhorn_client.by_id_setting(SETTING_BACKUP_TARGET)
-        return backup_target_setting.value
+        return self.get_setting(self.SETTING_BACKUP_TARGET)
 
     def get_secret(self):
-        backup_target_credential_setting = self.longhorn_client.by_id_setting(
-            SETTING_BACKUP_TARGET_CREDENTIAL_SECRET)
-        return backup_target_credential_setting.value
+        return self.get_setting(self.SETTING_BACKUP_TARGET_CREDENTIAL_SECRET)

e2e/libs/utility/constant.py

@@ -8,3 +8,7 @@
 ANNOT_EXPANDED_SIZE = f'{LABEL_TEST}/last-recorded-expanded-size'
 
 NAME_PREFIX = 'e2e-test'
+
+STREAM_EXEC_TIMEOUT = 300
+
+LONGHORN_NAMESPACE = 'longhorn-system'