post chaos (#939)

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

.github/workflows/tests.yml

@@ -47,6 +47,17 @@ jobs:
           kubectl --namespace default port-forward $es_pod_name 9200 &
           prom_name=$(kubectl get pods -n monitoring -l "app.kubernetes.io/name=prometheus" -o name)
           kubectl --namespace monitoring port-forward $prom_name 9090 &
+
+          # Wait for Elasticsearch to be ready
+          echo "Waiting for Elasticsearch to be ready..."
+          for i in {1..30}; do
+            if curl -k -s -u elastic:$ELASTIC_PASSWORD https://localhost:9200/_cluster/health > /dev/null 2>&1; then
+              echo "Elasticsearch is ready!"
+              break
+            fi
+            echo "Attempt $i: Elasticsearch not ready yet, waiting..."
+            sleep 2
+          done
           kubectl apply -f CI/templates/outage_pod.yaml
           kubectl wait --for=condition=ready pod -l scenario=outage --timeout=300s
           kubectl apply -f CI/templates/container_scenario_pod.yaml
@@ -73,7 +84,7 @@ jobs:
             yq -i '.kraken.performance_monitoring="localhost:9090"' CI/config/common_test_config.yaml
             yq -i '.elastic.elastic_port=9200' CI/config/common_test_config.yaml
             yq -i '.elastic.elastic_url="https://localhost"' CI/config/common_test_config.yaml
-            yq -i '.elastic.enable_elastic=False' CI/config/common_test_config.yaml
+            yq -i '.elastic.enable_elastic=True' CI/config/common_test_config.yaml
             yq -i '.elastic.password="${{env.ELASTIC_PASSWORD}}"' CI/config/common_test_config.yaml
             yq -i '.performance_monitoring.prometheus_url="http://localhost:9090"' CI/config/common_test_config.yaml
             echo "test_service_hijacking" > ./CI/tests/functional_tests
@@ -89,7 +100,6 @@ jobs:
             echo "test_io_hog" >> ./CI/tests/functional_tests
             echo "test_pod_network_filter" >> ./CI/tests/functional_tests
 
-
       # Push on main only steps + all other functional to collect coverage
       # for the badge
       - name: Configure AWS Credentials
@@ -135,38 +145,38 @@ jobs:
           cat ./CI/results.markdown >> $GITHUB_STEP_SUMMARY
           echo >> $GITHUB_STEP_SUMMARY
       - name: Upload CI logs
-        if: ${{ success() || failure() }}
+        if: ${{ always() }}
         uses: actions/upload-artifact@v4
         with:
           name: ci-logs
           path: CI/out
           if-no-files-found: error
       - name: Collect coverage report
-        if: ${{ success() || failure() }}
+        if: ${{ always() }}
         run: |
           python -m coverage html
           python -m coverage json
       - name: Publish coverage report to job summary
-        if: ${{ success() || failure() }}
+        if: ${{ always() }}
         run: |
           pip install html2text
           html2text --ignore-images --ignore-links -b 0 htmlcov/index.html >> $GITHUB_STEP_SUMMARY
       - name: Upload coverage data
-        if: ${{ success() || failure() }}
+        if: ${{ always() }}
         uses: actions/upload-artifact@v4
         with:
           name: coverage
           path: htmlcov
           if-no-files-found: error
       - name: Upload json coverage
-        if: ${{ success() || failure() }}
+        if: ${{ always() }}
         uses: actions/upload-artifact@v4
         with:
           name: coverage.json
           path: coverage.json
           if-no-files-found: error
       - name: Check CI results
-        if: ${{ success() || failure() }}
+        if: ${{ always() }}
         run: "! grep Fail CI/results.markdown"
 
   badge:

CI/tests/test_app_outages.sh

@@ -19,6 +19,7 @@ function functional_test_app_outage {
   kubectl get pods 
   envsubst < CI/config/common_test_config.yaml > CI/config/app_outage.yaml
   cat $scenario_file
+  cat CI/config/app_outage.yaml
   python3 -m coverage run -a run_kraken.py -c CI/config/app_outage.yaml
   echo "App outage scenario test: Success"
 }

CI/tests/test_pod.sh

@@ -10,7 +10,7 @@ function functional_test_pod_crash {
   export scenario_file="scenarios/kind/pod_etcd.yml"
   export post_config=""
   envsubst < CI/config/common_test_config.yaml > CI/config/pod_config.yaml
-
+  cat CI/config/pod_config.yaml
   python3 -m coverage run -a run_kraken.py -c CI/config/pod_config.yaml
   echo "Pod disruption scenario test: Success"
 }

krkn/utils/VirtChecker.py

@@ -1,6 +1,7 @@
 
 import time
 import logging
+import math
 import queue
 from datetime import datetime
 from krkn_lib.models.telemetry.models import VirtCheck
@@ -20,6 +21,8 @@ class VirtChecker:
         self.vm_list = []
         self.threads = []
         self.threads_limit = threads_limit
+        # setting to 0 in case no variables are set, so no threads later get made
+        self.batch_size = 0
         if self.namespace == "":
             logging.info("kube virt checks config is not defined, skipping them")
             return
@@ -29,19 +32,32 @@ class VirtChecker:
         self.only_failures =  get_yaml_item_value(kubevirt_check_config, "only_failures", False)
         self.interval = get_yaml_item_value(kubevirt_check_config, "interval", 2)
         self.ssh_node = get_yaml_item_value(kubevirt_check_config, "ssh_node", "")
+        self.node_names = get_yaml_item_value(kubevirt_check_config, "node_names", "")
         try:
             self.kube_vm_plugin = KubevirtVmOutageScenarioPlugin()
             self.kube_vm_plugin.init_clients(k8s_client=krkn_lib)
             vmis = self.kube_vm_plugin.get_vmis(vmi_name_match,self.namespace)
         except Exception as e:
             logging.error('Virt Check init exception: ' + str(e))
-            return 
-        
+            return
+        # See if multiple node names exist
+        node_name_list = self.node_names.split(',')
+
         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, "new_ip_address":""}))
+            # If node_name_list exists, only add if node name is in list
+
+            if len(node_name_list) > 0 and node_name in node_name_list:
+                self.vm_list.append(VirtCheck({'vm_name':vmi_name, 'ip_address': ip_address, 'namespace':self.namespace, 'node_name':node_name, "new_ip_address":""}))
+            elif len(node_name_list) == 0:
+
+                # If node_name_list is blank, add all vms
+                self.vm_list.append(VirtCheck({'vm_name':vmi_name, 'ip_address': ip_address, 'namespace':self.namespace, 'node_name':node_name, "new_ip_address":""}))
+
+        self.batch_size = math.ceil(len(self.vm_list)/self.threads_limit)
+        
 
     def check_disconnected_access(self, ip_address: str, worker_name:str = '', vmi_name: str = ''):
         
@@ -50,7 +66,6 @@ class VirtChecker:
         all_out = invoke_no_exit(virtctl_vm_cmd)
         logging.debug(f"Checking disconnected access for {ip_address} on {worker_name} output: {all_out}")
         virtctl_vm_cmd = f"ssh core@{worker_name} 'ssh -o BatchMode=yes -o ConnectTimeout=5 -o StrictHostKeyChecking=no root@{ip_address} 2>&1 | grep Permission' && echo 'True' || echo 'False'"
-        logging.debug(f"Checking disconnected access for {ip_address} on {worker_name} with command: {virtctl_vm_cmd}")
         output = invoke_no_exit(virtctl_vm_cmd)
         if 'True' in output:
             logging.debug(f"Disconnected access for {ip_address} on {worker_name} is successful: {output}")
@@ -63,7 +78,6 @@ class VirtChecker:
             # if vm gets deleted, it'll start up with a new ip address
             if new_ip_address != ip_address:
                 virtctl_vm_cmd = f"ssh core@{worker_name} 'ssh -o BatchMode=yes -o ConnectTimeout=5 -o StrictHostKeyChecking=no root@{new_ip_address} 2>&1 | grep Permission' && echo 'True' || echo 'False'"
-                logging.debug(f"Checking disconnected access for {new_ip_address} on {worker_name} with command: {virtctl_vm_cmd}")
                 new_output = invoke_no_exit(virtctl_vm_cmd)
                 logging.debug(f"Disconnected access for {ip_address} on {worker_name}: {new_output}")
                 if 'True' in new_output:
@@ -71,7 +85,6 @@ class VirtChecker:
             # if node gets stopped, vmis will start up with a new node (and with new ip)
             if new_node_name != worker_name:
                 virtctl_vm_cmd = f"ssh core@{new_node_name} 'ssh -o BatchMode=yes -o ConnectTimeout=5 -o StrictHostKeyChecking=no root@{new_ip_address} 2>&1 | grep Permission' && echo 'True' || echo 'False'"
-                logging.debug(f"Checking disconnected access for {new_ip_address} on {new_node_name} with command: {virtctl_vm_cmd}")
                 new_output = invoke_no_exit(virtctl_vm_cmd)
                 logging.debug(f"Disconnected access for {ip_address} on {new_node_name}: {new_output}")
                 if 'True' in new_output:
@@ -80,7 +93,6 @@ class VirtChecker:
             if self.ssh_node:
                 # using new_ip_address here since if it hasn't changed it'll match ip_address
                 virtctl_vm_cmd = f"ssh core@{self.ssh_node} 'ssh -o BatchMode=yes -o ConnectTimeout=5 -o StrictHostKeyChecking=no root@{new_ip_address} 2>&1 | grep Permission' && echo 'True' || echo 'False'"
-                logging.debug(f"Checking disconnected access for {new_ip_address} on {self.ssh_node} with command: {virtctl_vm_cmd}")
                 new_output = invoke_no_exit(virtctl_vm_cmd)
                 logging.debug(f"Disconnected access for {new_ip_address} on {self.ssh_node}: {new_output}")
                 if 'True' in new_output:
@@ -108,14 +120,19 @@ class VirtChecker:
         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 batch_list(self, queue: queue.Queue = None):
+        logging.info("batch size" + str(self.batch_size))
+        if self.batch_size > 0:
+            # Provided prints to easily visualize how the threads are processed.    
+            for i in range (0, len(self.vm_list),self.batch_size):
+                if i+self.batch_size > len(self.vm_list):
+                    sub_list = self.vm_list[i:  len(self.vm_list)-1]
+                else:
+                    sub_list = self.vm_list[i: i+self.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):
@@ -182,3 +199,60 @@ class VirtChecker:
             else:
                 virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm]))
         virt_check_telemetry_queue.put(virt_check_telemetry)
+    
+    def run_post_virt_check(self, vm_list_batch, virt_check_telemetry, post_virt_check_queue: queue.Queue):
+        
+        virt_check_telemetry = []
+        virt_check_tracker = {}
+        start_timestamp = datetime.now()
+        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, new_ip_address, new_node_name = self.check_disconnected_access(vm.ip_address, vm.node_name, vm.vm_name)
+                    if new_ip_address and vm.ip_address != new_ip_address:
+                        vm.new_ip_address = new_ip_address
+                    if new_node_name and vm.node_name != new_node_name:
+                        vm.node_name = new_node_name
+            except Exception:
+                vm_status = False
+            
+            if not vm_status:
+
+                virt_check_tracker= {
+                    "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.isoformat(),
+                    "new_ip_address": vm.new_ip_address,
+                    "duration": 0,
+                    "end_timestamp": start_timestamp.isoformat()
+                }
+                
+                virt_check_telemetry.append(VirtCheck(virt_check_tracker))
+        post_virt_check_queue.put(virt_check_telemetry)
+    
+
+    def gather_post_virt_checks(self, kubevirt_check_telem):
+
+        post_kubevirt_check_queue = queue.Queue()
+        post_threads = []
+
+        if self.batch_size > 0:
+            for i in range (0, len(self.vm_list),self.batch_size):
+                sub_list = self.vm_list[i: i+self.batch_size]
+                index = i
+                t = threading.Thread(target=self.run_post_virt_check,name=str(index), args=(sub_list,kubevirt_check_telem, post_kubevirt_check_queue))
+                post_threads.append(t)
+                t.start()
+
+            kubevirt_check_telem = []
+            for thread in post_threads:
+                thread.join()
+                if not post_kubevirt_check_queue.empty():
+                    kubevirt_check_telem.extend(post_kubevirt_check_queue.get_nowait())
+        return kubevirt_check_telem

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.11
+krkn-lib==5.1.12
 lxml==5.1.0
 kubernetes==34.1.0
 numpy==1.26.4

run_kraken.py

@@ -416,7 +416,8 @@ def main(options, command: Optional[str]) -> int:
                 break
             i+= 1
         chaos_telemetry.virt_checks = kubevirt_check_telem
-
+        post_kubevirt_check = kubevirt_checker.gather_post_virt_checks(kubevirt_check_telem)
+        chaos_telemetry.post_virt_checks = post_kubevirt_check
         # if platform is openshift will be collected
         # Cloud platform and network plugins metadata
         # through OCP specific APIs