diff --git a/config/config.yaml b/config/config.yaml index cfb5205b..236d55e7 100644 --- a/config/config.yaml +++ b/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 \ No newline at end of file diff --git a/containers/Dockerfile.template b/containers/Dockerfile.template index a86d4870..fd7a0530 100644 --- a/containers/Dockerfile.template +++ b/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 diff --git a/containers/krknctl-input.json b/containers/krknctl-input.json index 71a85036..ad7006e9 100644 --- a/containers/krknctl-input.json +++ b/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", diff --git a/krkn/invoke/command.py b/krkn/invoke/command.py index 0feb7cd5..ed233fbf 100644 --- a/krkn/invoke/command.py +++ b/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 diff --git a/krkn/scenario_plugins/abstract_scenario_plugin.py b/krkn/scenario_plugins/abstract_scenario_plugin.py index eed363a3..75b8f6f2 100644 --- a/krkn/scenario_plugins/abstract_scenario_plugin.py +++ b/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 - \ No newline at end of file + diff --git a/krkn/scenario_plugins/kubevirt_vm_outage/kubevirt_vm_outage_scenario_plugin.py b/krkn/scenario_plugins/kubevirt_vm_outage/kubevirt_vm_outage_scenario_plugin.py index 28790cf4..c087508c 100644 --- a/krkn/scenario_plugins/kubevirt_vm_outage/kubevirt_vm_outage_scenario_plugin.py +++ b/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] diff --git a/krkn/utils/VirtChecker.py b/krkn/utils/VirtChecker.py new file mode 100644 index 00000000..6d0c9702 --- /dev/null +++ b/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) diff --git a/requirements.txt b/requirements.txt index 8a4d1139..06e1883a 100644 --- a/requirements.txt +++ b/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 diff --git a/run_kraken.py b/run_kraken.py index 341e77d8..62730cf6 100644 --- a/run_kraken.py +++ b/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