Add LVMS support for Windows VM bootstorm workload

benchmark_runner/common/oc/oc.py

@@ -425,29 +425,33 @@ def is_odf_installed(self):
 
     def check_dv_status(self,
                         status: str,
-                        namespace: str = environment_variables.environment_variables_dict['namespace']):
+                        namespace: str = environment_variables.environment_variables_dict['namespace'],
+                        dv_name: str = ''):
         """
         This method checks dv status
         :return:
         """
-        namespace = f'-n {namespace}' if namespace else ''
-        verify_cmd = f"{self._cli} get dv {namespace} -o jsonpath='{{.items[].status.phase}}'"
+        namespace_flag = f'-n {namespace}' if namespace else ''
+        if dv_name:
+            verify_cmd = f"{self._cli} get dv {dv_name} {namespace_flag} -o jsonpath='{{.status.phase}}'"
+        else:
+            verify_cmd = f"{self._cli} get dv {namespace_flag} -o jsonpath='{{.items[].status.phase}}'"
         if status in self.run(verify_cmd):
             return True
         return False
 
-    @typechecked
     @logger_time_stamp
     def wait_for_dv_status(self,
                            status: str = 'Succeeded',
-                           timeout: int = int(environment_variables.environment_variables_dict['timeout'])):
+                           timeout: int = int(environment_variables.environment_variables_dict['timeout']),
+                           dv_name: str = ''):
         """
         This method waits for methods status
         @return: True/ False if reach to status
         """
         current_wait_time = 0
         while timeout <= 0 or current_wait_time <= timeout:
-            if self.check_dv_status(status=status):
+            if self.check_dv_status(status=status, dv_name=dv_name):
                 return True
             # sleep for x seconds
             time.sleep(OC.SLEEP_TIME)

benchmark_runner/common/ocp_resources/create_lvms.py

@@ -0,0 +1,54 @@
+
+import os
+
+from benchmark_runner.common.oc.oc import OC
+from benchmark_runner.common.logger.logger_time_stamp import logger_time_stamp, logger
+from benchmark_runner.common.ocp_resources.create_ocp_resource_operations import CreateOCPResourceOperations
+
+
+class CreateLVMS(CreateOCPResourceOperations):
+    """
+    This class creates LVMS (Logical Volume Manager Storage) operator on NVMe devices
+    """
+    def __init__(self, oc: OC, path: str, resource_list: list, lvms_version: str, lvms_devices: list):
+        super().__init__(oc)
+        self.__oc = oc
+        self.__path = path
+        self.__resource_list = resource_list
+        self.__lvms_version = lvms_version
+        self.__lvms_devices = lvms_devices
+
+    @logger_time_stamp
+    def create_lvms(self, upgrade_version: str = ''):
+        """
+        This method creates LVMS operator and LVMCluster
+        :param upgrade_version: if set, upgrade existing LVMS
+        :return: True if successful
+        """
+        if upgrade_version:
+            self.__oc.apply_async(yaml=os.path.join(self.__path, '01_subscription.yaml'))
+            logger.info(f'Wait till LVMS upgrade to version: {upgrade_version}')
+            self.verify_csv_installation(namespace='openshift-storage', operator='lvms', upgrade_version=upgrade_version)
+        else:
+            for resource in self.__resource_list:
+                logger.info(f'run {resource}')
+                self.__oc.create_async(yaml=os.path.join(self.__path, resource))
+
+                if '01_subscription.yaml' in resource:
+                    self.verify_csv_installation(namespace='openshift-storage', operator='lvms')
+
+                elif '02_lvmcluster.yaml' in resource:
+                    self.wait_for_ocp_resource_create(
+                        operator='lvms',
+                        verify_cmd="oc get lvmcluster lvms-nvme -n openshift-storage -o jsonpath='{.status.ready}'",
+                        status='true'
+                    )
+
+            # Verify StorageClass was created
+            sc_name = self.__oc.run("oc get sc -o jsonpath='{.items[?(@.provisioner==\"topolvm.io\")].metadata.name}'")
+            if sc_name:
+                logger.info(f'LVMS StorageClass created: {sc_name}')
+            else:
+                logger.warning('LVMS StorageClass not found after LVMCluster creation')
+
+            return True

benchmark_runner/common/ocp_resources/create_ocp_resource.py

@@ -10,6 +10,7 @@
 from benchmark_runner.common.ocp_resources.create_cnv import CreateCNV
 from benchmark_runner.common.ocp_resources.create_nhc_far import CreateNHCFAR
 from benchmark_runner.common.ocp_resources.create_custom import CreateCustom
+from benchmark_runner.common.ocp_resources.create_lvms import CreateLVMS
 from benchmark_runner.common.ocp_resources.migrate_infra import MigrateInfra
 
 
@@ -105,6 +106,11 @@ def create_resource(self, resource: str, upgrade_version: str):
         elif 'infra' == resource:
             create_infra = MigrateInfra(self.__oc, path=os.path.join(self.__dir_path, resource), resource_list=resource_files)
             create_infra.migrate_infra()
+        elif 'lvms' == resource:
+            lvms_version = self.__environment_variables_dict.get('lvms_version', '')
+            lvms_devices = ast.literal_eval(self.__environment_variables_dict.get('lvms_devices', "['/dev/nvme0n1', '/dev/nvme1n1']"))
+            create_lvms = CreateLVMS(self.__oc, path=os.path.join(self.__dir_path, resource), resource_list=resource_files, lvms_version=lvms_version, lvms_devices=lvms_devices)
+            create_lvms.create_lvms(upgrade_version)
         elif 'custom' == resource:
             create_custom = CreateCustom(self.__oc, path=os.path.join(self.__dir_path, resource), resource_list=resource_files)
             create_custom.create_custom()

benchmark_runner/common/ocp_resources/lvms/template/01_subscription_template.yaml

@@ -0,0 +1,11 @@
+apiVersion: operators.coreos.com/v1alpha1
+kind: Subscription
+metadata:
+  name: lvms-operator
+  namespace: openshift-storage
+spec:
+  channel: "stable-{{ lvms_version }}"
+  installPlanApproval: Automatic
+  name: lvms-operator
+  source: redhat-operators
+  sourceNamespace: openshift-marketplace

benchmark_runner/common/ocp_resources/lvms/template/02_lvmcluster_template.yaml

@@ -0,0 +1,24 @@
+apiVersion: lvm.topolvm.io/v1alpha1
+kind: LVMCluster
+metadata:
+  name: lvms-nvme
+  namespace: openshift-storage
+spec:
+  storage:
+    deviceClasses:
+    - name: nvme
+      default: false
+      thinPoolConfig:
+        name: thin-pool-nvme
+        sizePercent: 90
+        overprovisionRatio: 10
+      deviceSelector:
+        paths:
+{%- for device in lvms_devices %}
+        - {{ device }}
+{%- endfor %}
+      nodeSelector:
+        nodeSelectorTerms:
+        - matchExpressions:
+          - key: node-role.kubernetes.io/worker
+            operator: Exists

benchmark_runner/common/template_operations/template_operations.py

@@ -153,6 +153,11 @@ def __generate_yamls_internal(self, scale: str = None, scale_num: str = None, sc
 
         render_data = self.__build_template_data(template_render_data, workload_data)
 
+        if render_data.get('per_node_dv') and scale_node:
+            render_data['dv_source_name'] = f'windows-clone-dv-{scale_node}'
+        else:
+            render_data['dv_source_name'] = 'windows-clone-dv'
+
         hammerdb_config = self.__environment_variables_dict.get('hammerdb_config', {})
         if hammerdb_config and ('hammerdb' in self.__workload_name or 'winmssql' in self.__workload_name):
             hammerdb_config = {k: v for k, v in hammerdb_config.items() if v != ''}
@@ -195,7 +200,8 @@ def __generate_yamls_internal(self, scale: str = None, scale_num: str = None, sc
                 answer['namespace.yaml'] = render_yaml_file(dir_path=self.__dir_path, yaml_file='namespace_template.yaml', environment_variable_dict=self.__environment_variables_dict)
             # windows workload
             if 'win' in self.__workload_name:
-                answer['windows_dv.yaml'] = render_yaml_file(dir_path=os.path.join(workload_dir_path, 'internal_data'), yaml_file='windows_dv_template.yaml', environment_variable_dict=render_data)
+                dv_filename = f'windows_dv_{scale_node}.yaml' if render_data.get('per_node_dv') and scale_node else 'windows_dv.yaml'
+                answer[dv_filename] = render_yaml_file(dir_path=os.path.join(workload_dir_path, 'internal_data'), yaml_file='windows_dv_template.yaml', environment_variable_dict=render_data)
             # vdbench scale
             if scale and redis and 'vdbench' in self.__workload_name:
                     answer['redis.yaml'] = render_yaml_file(dir_path=os.path.join(self.__dir_path, 'scale'), yaml_file='redis_template.yaml', environment_variable_dict=self.__environment_variables_dict)

benchmark_runner/common/template_operations/templates/windows/internal_data/windows_dv_template.yaml

@@ -3,7 +3,11 @@ kind: DataVolume
 metadata:
   annotations:
     cdi.kubevirt.io/storage.deleteAfterCompletion: "false"
-  name: windows-clone-dv
+    {%- if per_node_dv and scale_node %}
+    cdi.kubevirt.io/storage.bind.immediate.requested: "true"
+    volume.kubernetes.io/selected-node: {{ scale_node }}
+    {%- endif %}
+  name: {{ dv_source_name }}
   namespace: {{ namespace }}
 spec:
   source:
@@ -19,7 +23,7 @@ spec:
     {%- endif %}
   pvc:
     accessModes:
-      - ReadWriteMany
+      - {{ vm_access_mode }}
     resources:
       requests:
         storage: {{ storage }}

benchmark_runner/common/template_operations/templates/windows/internal_data/windows_vm_template.yaml

@@ -98,7 +98,7 @@ spec:
           limits:
             cpu: {{ limit_cpu }}
             memory: {{ limit_memory }}
-      evictionStrategy: LiveMigrate
+      evictionStrategy: {{ eviction_strategy }}
       networks:
       - name: nic-0
         pod: {}
@@ -119,6 +119,9 @@ spec:
   - metadata:
       annotations:
         descheduler.alpha.kubernetes.io/evict: "true"
+        {%- if per_node_dv and scale_node %}
+        volume.kubernetes.io/selected-node: {{ scale_node }}
+        {%- endif %}
       {% if scale -%}
       name: windows-{{ kind }}-root-disk-{{ trunc_uuid }}-{{ scale }}
       {%- else -%}
@@ -127,7 +130,7 @@ spec:
     spec:
       pvc:
         accessModes:
-        - ReadWriteMany
+        - {{ vm_access_mode }}
         resources:
           requests:
             storage: {{ storage }}
@@ -136,4 +139,4 @@ spec:
       source:
         pvc:
           namespace: {{ namespace }}
-          name: windows-clone-dv
+          name: {{ dv_source_name }}

benchmark_runner/common/template_operations/templates/windows/windows_data_template.yaml

@@ -9,6 +9,9 @@ template_data:
     cdi_source_type: {{ cdi_source_type }}
     cdi_source_s3_cred: {{ cdi_source_s3_cred }}
     vm_storage_class: {{ vm_storage_class }}
+    vm_access_mode: {{ vm_access_mode }}
+    eviction_strategy: {{ eviction_strategy }}
+    per_node_dv: {{ per_node_dv }}
   run_type:
     perf_ci:
       requests_memory: 2G

benchmark_runner/main/environment_variables.py

@@ -95,6 +95,9 @@ def __init__(self):
         self._environment_variables_dict['cdi_source_s3_cred'] = EnvironmentVariables.get_env('CDI_SOURCE_S3_CRED', '')
         # Storage class for all VM workloads PVCs (override for clusters with different ODF config)
         self._environment_variables_dict['vm_storage_class'] = EnvironmentVariables.get_env('VM_STORAGE_CLASS', 'ocs-storagecluster-ceph-rbd-virtualization')
+        self._environment_variables_dict['vm_access_mode'] = EnvironmentVariables.get_env('VM_ACCESS_MODE', 'ReadWriteMany')
+        self._environment_variables_dict['eviction_strategy'] = EnvironmentVariables.get_env('EVICTION_STRATEGY', 'LiveMigrate')
+        self._environment_variables_dict['per_node_dv'] = EnvironmentVariables.get_boolean_from_environment('PER_NODE_DV', False)
         # Delete all resources before and after the run, default True
         self._environment_variables_dict['delete_all'] = EnvironmentVariables.get_boolean_from_environment('DELETE_ALL', True)
         # RunStrategy: Always can be set to True or False (default: False). Set it to True for VMs that need to start in a running state
@@ -334,6 +337,10 @@ def __init__(self):
         self._environment_variables_dict['lso_version'] = EnvironmentVariables.get_env('LSO_VERSION', '')
         # odf version
         self._environment_variables_dict['odf_version'] = EnvironmentVariables.get_env('ODF_VERSION', '')
+        # lvms version
+        self._environment_variables_dict['lvms_version'] = EnvironmentVariables.get_env('LVMS_VERSION', '4.22')
+        # lvms NVMe device paths
+        self._environment_variables_dict['lvms_devices'] = EnvironmentVariables.get_env('LVMS_DEVICES', "['/dev/nvme0n1', '/dev/nvme1n1']")
         # custom kata version, if empty fetch auto latest version
         self._environment_variables_dict['kata_csv'] = EnvironmentVariables.get_env('KATA_CSV', '')
         # number of odf disk for discovery

benchmark_runner/workloads/windows_vm.py

@@ -1,4 +1,6 @@
 
+import glob
+import json
 import os
 import sys
 import time
@@ -17,6 +19,72 @@ def __init__(self):
         super().__init__()
         if not self._windows_url:
             raise ValueError('Missing Windows DV URL')
+        self._per_node_dv = self._environment_variables_dict.get('per_node_dv', False)
+        self._created_sc_name = ''
+
+    def _create_snapshot_clone_sc(self):
+        """Create a StorageClass with snapshot clone strategy for fast LVMS cloning"""
+        base_sc = self._environment_variables_dict['vm_storage_class']
+        new_sc = f'{base_sc}-bootstorm'
+        # Delete if leftover from a previous run
+        try:
+            self._oc.run(f'{self._oc._cli} delete sc {new_sc} --wait=false')
+        except Exception:
+            pass
+        time.sleep(2)
+        # Get base SC details via jsonpath
+        provisioner = self._oc.run(f"{self._oc._cli} get sc {base_sc} -o jsonpath='{{.provisioner}}'").strip().strip("'")
+        reclaim = self._oc.run(f"{self._oc._cli} get sc {base_sc} -o jsonpath='{{.reclaimPolicy}}'").strip().strip("'")
+        binding = self._oc.run(f"{self._oc._cli} get sc {base_sc} -o jsonpath='{{.volumeBindingMode}}'").strip().strip("'")
+        params_json = self._oc.run(f"{self._oc._cli} get sc {base_sc} -o jsonpath='{{.parameters}}'").strip().strip("'")
+        sc_yaml = (
+            f'apiVersion: storage.k8s.io/v1\n'
+            f'kind: StorageClass\n'
+            f'metadata:\n'
+            f'  name: {new_sc}\n'
+            f'  annotations:\n'
+            f'    cdi.kubevirt.io/clone-strategy: "snapshot"\n'
+            f'provisioner: {provisioner}\n'
+            f'reclaimPolicy: {reclaim or "Delete"}\n'
+            f'volumeBindingMode: {binding or "WaitForFirstConsumer"}\n'
+            f'allowVolumeExpansion: true\n'
+        )
+        if params_json and params_json != '{}':
+            try:
+                params = json.loads(params_json)
+                sc_yaml += 'parameters:\n'
+                for k, v in params.items():
+                    sc_yaml += f'  {k}: "{v}"\n'
+            except json.JSONDecodeError:
+                pass
+        sc_dir = os.path.join('/tmp', 'bootstorm-sc')
+        os.makedirs(sc_dir, exist_ok=True)
+        sc_file = os.path.join(sc_dir, 'bootstorm_sc.yaml')
+        with open(sc_file, 'w') as f:
+            f.write(sc_yaml)
+        self._oc.create_async(yaml=sc_file)
+        logger.info(f'Created StorageClass {new_sc} with snapshot clone strategy')
+        for _ in range(30):
+            time.sleep(1)
+            try:
+                result = self._oc.run(f"{self._oc._cli} get storageprofile {new_sc} -o jsonpath='{{.status.cloneStrategy}}'")
+                if result and result.strip().strip("'"):
+                    logger.info(f'StorageProfile {new_sc} ready')
+                    break
+            except Exception:
+                pass
+        return new_sc
+
+    def _delete_snapshot_clone_sc(self):
+        """Delete the auto-created StorageClass"""
+        if self._created_sc_name:
+            sc_file = os.path.join('/tmp', 'bootstorm-sc', 'bootstorm_sc.yaml')
+            if os.path.isfile(sc_file):
+                try:
+                    self._oc.delete_async(yaml=sc_file)
+                    logger.info(f'Deleted StorageClass {self._created_sc_name}')
+                except Exception:
+                    logger.warning(f'Failed to delete StorageClass {self._created_sc_name}')
 
     @logger_time_stamp
     def run(self):
@@ -29,17 +97,38 @@ def run(self):
                 self._es_index = f"windows-{self._run_type.replace('_', '-')}-results"
             else:
                 self._es_index = 'windows-results'
+            if self._per_node_dv:
+                old_sc = self._environment_variables_dict['vm_storage_class']
+                self._created_sc_name = self._create_snapshot_clone_sc()
+                self._environment_variables_dict['vm_storage_class'] = self._created_sc_name
             self._initialize_run()
+            if self._per_node_dv and self._created_sc_name:
+                # Patch already-rendered YAML files with the new SC name
+                for yaml_file in glob.glob(os.path.join(self._run_artifacts_path, '*.yaml')):
+                    with open(yaml_file, 'r') as f:
+                        content = f.read()
+                    if old_sc in content:
+                        with open(yaml_file, 'w') as f:
+                            f.write(content.replace(old_sc, self._created_sc_name))
             if not self._verification_only:
-                # create windows dv
-                self._oc.create_async(yaml=os.path.join(f'{self._run_artifacts_path}', 'windows_dv.yaml'))
-                self._oc.wait_for_dv_status(status='Succeeded')
+                if self._per_node_dv and self._scale_node_list:
+                    for node in self._scale_node_list:
+                        self._oc.create_async(yaml=os.path.join(self._run_artifacts_path, f'windows_dv_{node}.yaml'))
+                    for node in self._scale_node_list:
+                        self._oc.wait_for_dv_status(status='Succeeded', dv_name=f'windows-clone-dv-{node}')
+                else:
+                    self._oc.create_async(yaml=os.path.join(self._run_artifacts_path, 'windows_dv.yaml'))
+                    self._oc.wait_for_dv_status(status='Succeeded')
             self.run_vm_workload()
             if self._delete_all:
-                # delete windows dv
-                self._oc.delete_async(yaml=os.path.join(f'{self._run_artifacts_path}', 'windows_dv.yaml'))
+                if self._per_node_dv and self._scale_node_list:
+                    for node in self._scale_node_list:
+                        self._oc.delete_async(yaml=os.path.join(self._run_artifacts_path, f'windows_dv_{node}.yaml'))
+                else:
+                    self._oc.delete_async(yaml=os.path.join(self._run_artifacts_path, 'windows_dv.yaml'))
                 # delete namespace
                 self._oc.delete_async(yaml=os.path.join(f'{self._run_artifacts_path}', 'namespace.yaml'))
+                self._delete_snapshot_clone_sc()
         except ElasticSearchDataNotUploaded as err:
             self._oc.delete_vm_sync(
                 yaml=os.path.join(f'{self._run_artifacts_path}', f'{self._name}.yaml'),