adding no scenario type (#869)

Signed-off-by: Paige Patton <prubenda@redhat.com>

config/config.yaml

@@ -119,3 +119,10 @@ health_checks:                                              # Utilizing health c
           bearer_token:                                     # Bearer token for authentication if any
           auth:                                             # Provide authentication credentials (username , password) in tuple format if any, ex:("admin","secretpassword")
           exit_on_failure:                                  # If value is True exits when health check failed for application, values can be True/False
+
+kubevirt_checks:                                            # Utilizing virt check endpoints to observe ssh ability to VMI's during chaos injection.
+    interval: 2                                             # Interval in seconds to perform virt checks, default value is 2 seconds
+    namespace:                                              # Namespace where to find VMI's
+    name:                                                   # Regex Name style of VMI's to watch, optional, will watch all VMI names in the namespace if left blank
+    only_failures: False                                    # Boolean of whether to show all VMI's failures and successful ssh connection (False), or only failure status' (True) 
+    disconnected: False                                     # Boolean of how to try to connect to the VMIs; if True will use the ip_address to try ssh from within a node, if false will use the name and uses virtctl to try to connect; Default is False

containers/Dockerfile.template

@@ -31,6 +31,11 @@ RUN dnf update && dnf install -y --setopt=install_weak_deps=False \
     git python39 jq yq gettext wget which ipmitool &&\
     dnf clean all
 
+# Virtctl 
+RUN export VERSION=$(curl https://storage.googleapis.com/kubevirt-prow/release/kubevirt/kubevirt/stable.txt) && \
+    wget https://github.com/kubevirt/kubevirt/releases/download/${VERSION}/virtctl-${VERSION}-linux-amd64 && \
+    chmod +x virtctl-${VERSION}-linux-amd64 && sudo mv virtctl-${VERSION}-linux-amd64 /usr/local/bin/virtctl
+
 # copy oc client binary from oc-build image
 COPY --from=oc-build /tmp/oc/oc /usr/bin/oc
 

containers/krknctl-input.json

@@ -425,6 +425,55 @@
     "default": "False",
     "required": "false"
   },
+  {
+    "name": "kubevirt-check-interval",
+    "short_description": "Kube Virt check interval",
+    "description": "How often to check the kube virt check Vms ssh status",
+    "variable": "KUBE_VIRT_CHECK_INTERVAL",
+    "type": "number",
+    "default": "2",
+    "required": "false"
+  },
+  {
+    "name": "kubevirt-namespace",
+    "short_description": "KubeVirt namespace to check",
+    "description": "KubeVirt namespace to check the health of",
+    "variable": "KUBE_VIRT_NAMESPACE",
+    "type": "string",
+    "default": "",
+    "required": "false"
+  },
+  {
+    "name": "kubevirt-namespace",
+    "short_description": "KubeVirt regex names to watch",
+    "description": "KubeVirt regex names to check VMs",
+    "variable": "KUBE_VIRT_NAME",
+    "type": "string",
+    "default": "",
+    "required": "false"
+  },
+  {
+    "name": "kubevirt-only-failures",
+    "short_description": "KubeVirt checks only report if failure occurs",
+    "description": "KubeVirt checks only report if failure occurs",
+    "variable": "KUBE_VIRT_FAILURES",
+    "type": "enum",
+    "allowed_values": "True,False,true,false",
+    "separator": ",",
+    "default": "False",
+    "required": "false"
+  },
+  {
+    "name": "kubevirt-disconnected",
+    "short_description": "KubeVirt checks in disconnected mode",
+    "description": "KubeVirt checks in disconnected mode, bypassing the clusters Api",
+    "variable": "KUBE_VIRT_DISCONNECTED",
+    "type": "enum",
+    "allowed_values": "True,False,true,false",
+    "separator": ",",
+    "default": "False",
+    "required": "false"
+  },
   {
     "name": "krkn-debug",
     "short_description": "Krkn debug mode",

krkn/invoke/command.py

@@ -18,9 +18,8 @@ def invoke(command, timeout=None):
 def invoke_no_exit(command, timeout=None):
     output = ""
     try:
-        output = subprocess.check_output(command, shell=True, universal_newlines=True, timeout=timeout)
+        output = subprocess.check_output(command, shell=True, universal_newlines=True, timeout=timeout, stderr=subprocess.DEVNULL)
     except Exception as e:
-        logging.error("Failed to run %s, error: %s" % (command, e))
         return str(e)
     return output
 

krkn/scenario_plugins/abstract_scenario_plugin.py

@@ -15,7 +15,7 @@ from krkn.rollback.serialization import Serializer
 
 class AbstractScenarioPlugin(ABC):
 
-    def __init__(self, scenario_type: str):
+    def __init__(self, scenario_type: str = "placeholder_scenario_type"):
         """Initializes the AbstractScenarioPlugin with the scenario type and rollback configuration.
         
         :param scenario_type: the scenario type defined in the config.yaml
@@ -149,4 +149,4 @@ class AbstractScenarioPlugin(ABC):
             time.sleep(wait_duration)
         return failed_scenarios, scenario_telemetries
 
-    
+    

krkn/scenario_plugins/kubevirt_vm_outage/kubevirt_vm_outage_scenario_plugin.py

@@ -20,7 +20,8 @@ class KubevirtVmOutageScenarioPlugin(AbstractScenarioPlugin):
     This plugin simulates a VM crash or outage scenario and supports automated or manual recovery.
     """
 
-    def __init__(self, scenario_type: str):
+    def __init__(self, scenario_type: str = None):
+        scenario_type = self.get_scenario_types()[0]
         super().__init__(scenario_type)
         self.k8s_client = None
         self.original_vmi = None
@@ -150,6 +151,9 @@ class KubevirtVmOutageScenarioPlugin(AbstractScenarioPlugin):
                 logging.error("vm_name parameter is required")
                 return 1
             vmis_list = self.get_vmis(vm_name,namespace)
+            if len(vmis_list) == 0:
+                logging.error(f"No matching VMs with name {vm_name} in namespace {namespace}")
+                return 1
             rand_int = random.randint(0, len(vmis_list) - 1)
             vmi = vmis_list[rand_int]
                 

krkn/utils/VirtChecker.py

@@ -0,0 +1,136 @@
+
+import time
+import logging
+import queue
+from datetime import datetime
+from krkn_lib.models.telemetry.models import VirtCheck
+from krkn.invoke.command import invoke_no_exit
+from krkn.scenario_plugins.kubevirt_vm_outage.kubevirt_vm_outage_scenario_plugin import KubevirtVmOutageScenarioPlugin
+from krkn_lib.k8s import KrknKubernetes
+import threading
+from krkn_lib.utils.functions import get_yaml_item_value
+
+
+class VirtChecker:
+    current_iterations: int = 0
+    ret_value = 0
+    def __init__(self, kubevirt_check_config, iterations, krkn_lib: KrknKubernetes, threads_limt=20):
+        self.iterations = iterations
+        self.namespace = get_yaml_item_value(kubevirt_check_config, "namespace", "")
+        self.vm_list = []
+        self.threads = []
+        self.threads_limit = threads_limt
+        if self.namespace == "":
+            logging.info("kube virt checks config is not defined, skipping them")
+            return
+        vmi_name_match = get_yaml_item_value(kubevirt_check_config, "name", ".*")
+        self.krkn_lib = krkn_lib
+        self.disconnected =  get_yaml_item_value(kubevirt_check_config, "disconnected", False)
+        self.only_failures =  get_yaml_item_value(kubevirt_check_config, "only_failures", False)
+        self.interval = get_yaml_item_value(kubevirt_check_config, "interval", 2)
+        try:
+            self.kube_vm_plugin = KubevirtVmOutageScenarioPlugin()
+            self.kube_vm_plugin.init_clients(k8s_client=krkn_lib)
+        
+        except Exception as e:
+            logging.error('Virt Check init exception: ' + str(e))
+            return 
+        vmis = self.kube_vm_plugin.get_vmis(vmi_name_match,self.namespace)
+        
+        for vmi in vmis:
+            node_name = vmi.get("status",{}).get("nodeName")
+            vmi_name = vmi.get("metadata",{}).get("name")
+            ip_address = vmi.get("status",{}).get("interfaces",[])[0].get("ipAddress")
+            self.vm_list.append(VirtCheck({'vm_name':vmi_name, 'ip_address': ip_address, 'namespace':self.namespace, 'node_name':node_name}))
+
+    def check_disconnected_access(self, ip_address: str, worker_name:str = ''):
+
+        virtctl_vm_cmd = f"ssh core@{worker_name} 'ssh -o BatchMode=yes -o ConnectTimeout=2 -o StrictHostKeyChecking=no root@{ip_address} 2>&1 | grep Permission' && echo 'True' || echo 'False'"
+        if 'True' in invoke_no_exit(virtctl_vm_cmd):
+            return True
+        else:
+            return False
+
+    def get_vm_access(self, vm_name: str = '', namespace: str = ''):
+        """
+        This method returns True when the VM is access and an error message when it is not, using virtctl protocol
+        :param vm_name:
+        :param namespace:
+        :return: virtctl_status 'True' if successful, or an error message if it fails.
+        """
+        virtctl_vm_cmd = f"virtctl ssh --local-ssh-opts='-o BatchMode=yes' --local-ssh-opts='-o PasswordAuthentication=no' --local-ssh-opts='-o ConnectTimeout=2' root@{vm_name} -n {namespace}"
+        check_virtctl_vm_cmd = f"virtctl ssh --local-ssh-opts='-o BatchMode=yes' --local-ssh-opts='-o PasswordAuthentication=no' --local-ssh-opts='-o ConnectTimeout=2' root@{vm_name} -n {namespace} 2>&1 |egrep 'denied|verification failed'  && echo 'True' || echo 'False'"
+        if 'True' in invoke_no_exit(check_virtctl_vm_cmd):
+            return True
+        else:
+            second_invoke = invoke_no_exit(virtctl_vm_cmd)
+            if 'True' in second_invoke:
+                return True
+            return False
+    
+    def thread_join(self):
+        for thread in self.threads:
+            thread.join()
+
+    def batch_list(self,  queue: queue.Queue, batch_size=20):
+        # Provided prints to easily visualize how the threads are processed.    
+        for i in range (0, len(self.vm_list),batch_size):
+            sub_list = self.vm_list[i: i+batch_size]
+            index = i
+            t = threading.Thread(target=self.run_virt_check,name=str(index), args=(sub_list,queue))
+            self.threads.append(t)
+            t.start()
+
+    
+    def run_virt_check(self, vm_list_batch, virt_check_telemetry_queue: queue.Queue):
+        
+        virt_check_telemetry = []
+        virt_check_tracker = {}
+        while self.current_iterations < self.iterations:
+            for vm in vm_list_batch:
+                try: 
+                    if not self.disconnected: 
+                        vm_status = self.get_vm_access(vm.vm_name, vm.namespace)
+                    else:
+                        vm_status = self.check_disconnected_access(vm.ip_address, vm.node_name)
+                except Exception:
+                    vm_status = False
+                
+                if vm.vm_name not in virt_check_tracker:
+                    start_timestamp = datetime.now()
+                    virt_check_tracker[vm.vm_name] = {
+                        "vm_name": vm.vm_name,
+                        "ip_address": vm.ip_address,
+                        "namespace": vm.namespace,
+                        "node_name": vm.node_name,
+                        "status": vm_status,
+                        "start_timestamp": start_timestamp
+                    }
+                else:
+                    if vm_status != virt_check_tracker[vm.vm_name]["status"]:
+                        end_timestamp = datetime.now()
+                        start_timestamp = virt_check_tracker[vm.vm_name]["start_timestamp"]
+                        duration = (end_timestamp - start_timestamp).total_seconds()
+                        virt_check_tracker[vm.vm_name]["end_timestamp"] = end_timestamp.isoformat()
+                        virt_check_tracker[vm.vm_name]["duration"] = duration
+                        virt_check_tracker[vm.vm_name]["start_timestamp"] = start_timestamp.isoformat()
+                        if self.only_failures: 
+                            if not virt_check_tracker[vm.vm_name]["status"]:
+                                virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm.vm_name]))
+                        else:
+                            virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm.vm_name]))
+                        del virt_check_tracker[vm.vm_name]
+            time.sleep(self.interval)
+        virt_check_end_time_stamp = datetime.now()
+        for vm in virt_check_tracker.keys():
+            final_start_timestamp = virt_check_tracker[vm]["start_timestamp"]
+            final_duration = (virt_check_end_time_stamp - final_start_timestamp).total_seconds()
+            virt_check_tracker[vm]["end_timestamp"] = virt_check_end_time_stamp.isoformat()
+            virt_check_tracker[vm]["duration"] = final_duration
+            virt_check_tracker[vm]["start_timestamp"] = final_start_timestamp.isoformat()
+            if self.only_failures:
+                if not virt_check_tracker[vm]["status"]:
+                    virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm]))
+            else:
+                virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm]))
+        virt_check_telemetry_queue.put(virt_check_telemetry)

requirements.txt

@@ -16,7 +16,7 @@ google-cloud-compute==1.22.0
 ibm_cloud_sdk_core==3.18.0
 ibm_vpc==0.20.0
 jinja2==3.1.6
-krkn-lib==5.1.0
+krkn-lib==5.1.1
 lxml==5.1.0
 kubernetes==28.1.0
 numpy==1.26.4

run_kraken.py

@@ -29,6 +29,7 @@ from krkn_lib.utils.functions import get_yaml_item_value, get_junit_test_case
 
 from krkn.utils import TeeLogHandler
 from krkn.utils.HealthChecker import HealthChecker
+from krkn.utils.VirtChecker import VirtChecker
 from krkn.scenario_plugins.scenario_plugin_factory import (
     ScenarioPluginFactory,
     ScenarioPluginNotFound,
@@ -130,7 +131,8 @@ def main(options, command: Optional[str]) -> int:
             config["performance_monitoring"], "check_critical_alerts", False
         )
         telemetry_api_url = config["telemetry"].get("api_url")
-        health_check_config = config["health_checks"]
+        health_check_config = get_yaml_item_value(config, "health_checks",{})
+        kubevirt_check_config = get_yaml_item_value(config, "kubevirt_checks", {})
 
         # Initialize clients
         if not os.path.isfile(kubeconfig_path) and not os.path.isfile(
@@ -324,6 +326,10 @@ def main(options, command: Optional[str]) -> int:
                                                args=(health_check_config, health_check_telemetry_queue))
         health_check_worker.start()
 
+        kubevirt_check_telemetry_queue = queue.Queue()
+        kubevirt_checker = VirtChecker(kubevirt_check_config, iterations=iterations, krkn_lib=kubecli)
+        kubevirt_checker.batch_list(kubevirt_check_telemetry_queue)
+
         # Loop to run the chaos starts here
         while int(iteration) < iterations and run_signal != "STOP":
             # Inject chaos scenarios specified in the config
@@ -385,6 +391,7 @@ def main(options, command: Optional[str]) -> int:
 
             iteration += 1
             health_checker.current_iterations += 1
+            kubevirt_checker.current_iterations += 1
 
         # telemetry
         # in order to print decoded telemetry data even if telemetry collection
@@ -396,6 +403,17 @@ def main(options, command: Optional[str]) -> int:
             chaos_telemetry.health_checks = health_check_telemetry_queue.get_nowait()
         except queue.Empty:
             chaos_telemetry.health_checks = None
+        
+        kubevirt_checker.thread_join()
+        kubevirt_check_telem = []
+        i =0
+        while i <= kubevirt_checker.threads_limit:
+            if not kubevirt_check_telemetry_queue.empty():
+                kubevirt_check_telem.extend(kubevirt_check_telemetry_queue.get_nowait())
+            else:
+                break
+            i+= 1
+        chaos_telemetry.virt_checks = kubevirt_check_telem
 
         # if platform is openshift will be collected
         # Cloud platform and network plugins metadata