fix(scenarios): fix cerberus call signature, missing imports, prometheus return, HealthChecker bool flags, network_chaos_ng len check

Author: Nitesh

krkn/cerberus/setup.py

@@ -1,3 +1,19 @@
+#!/usr/bin/env python
+#
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import logging
 import requests
 import sys
@@ -19,14 +35,13 @@ def set_url(config):
         cerberus_url = get_yaml_item_value(config["cerberus"],"cerberus_url", "")
         global check_application_routes
         check_application_routes = \
-            get_yaml_item_value(config["cerberus"],"check_applicaton_routes","")
+            get_yaml_item_value(config["cerberus"],"check_application_routes","")
 
 def get_status(start_time, end_time):
     """
     Get cerberus status
     """
     cerberus_status = True
-    check_application_routes = False
     application_routes_status = True
     if cerberus_enabled:
         if not cerberus_url:
@@ -42,7 +57,6 @@ def get_status(start_time, end_time):
         # experience downtime during the chaos
         if check_application_routes:
             application_routes_status, unavailable_routes = application_status(
-                cerberus_url,
                 start_time,
                 end_time
             )
@@ -83,26 +97,12 @@ def publish_kraken_status( start_time, end_time):
     if not cerberus_status:
         if exit_on_failure:
             logging.info(
-                "Cerberus status is not healthy and post action scenarios "
-                "are still failing, exiting kraken run"
-            )
-            sys.exit(1)
-        else:
-            logging.info(
-                "Cerberus status is not healthy and post action scenarios "
-                "are still failing"
-            )
-    else:
-        if exit_on_failure:
-            logging.info(
-                "Cerberus status is healthy but post action scenarios "
-                "are still failing, exiting kraken run"
+                "Cerberus status is not healthy, exiting kraken run"
             )
             sys.exit(1)
         else:
             logging.info(
-                "Cerberus status is healthy but post action scenarios "
-                "are still failing"
+                "Cerberus status is not healthy"
             )
 
 

krkn/prometheus/client.py

@@ -1,3 +1,19 @@
+#!/usr/bin/env python
+#
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 from __future__ import annotations
 
 import datetime
@@ -165,8 +181,10 @@ def critical_alerts(
 
     if not firing_alerts:
         logging.info("No critical alerts are firing!!")
-    
-   
+
+    return firing_alerts
+
+
 def metrics(
     prom_cli: KrknPrometheus,
     elastic: KrknElastic,
@@ -243,37 +261,37 @@ def metrics(
                         pass
         telemetry_json = json.loads(telemetry_json)
         for scenario in telemetry_json['scenarios']:
-            for k,v in scenario["affected_pods"].items():
+            for pod_type, pod_list in scenario["affected_pods"].items():
                 metric_name = "affected_pods_recovery"
-                metric = {"metricName": metric_name, "type": k}
-                if type(v) is list:
-                    for pod in v:
-                        for k,v in pod.items():
-                            metric[k] = v
+                metric = {"metricName": metric_name, "type": pod_type}
+                if type(pod_list) is list:
+                    for pod in pod_list:
+                        for pod_k, pod_v in pod.items():
+                            metric[pod_k] = pod_v
                             metric['timestamp'] = str(datetime.datetime.now())
                         logging.debug("adding pod %s", metric)
                         metrics_list.append(metric.copy())
             for affected_node in scenario["affected_nodes"]:
                 metric_name = "affected_nodes_recovery"
                 metric = {"metricName": metric_name}
-                for k,v in affected_node.items():
-                    metric[k] = v
+                for node_k, node_v in affected_node.items():
+                    metric[node_k] = node_v
                     metric['timestamp'] = str(datetime.datetime.now())
                 metrics_list.append(metric.copy())
         if telemetry_json['health_checks']:
             for health_check in telemetry_json["health_checks"]:
                     metric_name = "health_check_recovery"
                     metric = {"metricName": metric_name}
-                    for k,v in health_check.items():
-                        metric[k] = v
+                    for hc_k, hc_v in health_check.items():
+                        metric[hc_k] = hc_v
                         metric['timestamp'] = str(datetime.datetime.now())
                     metrics_list.append(metric.copy())
         if telemetry_json['virt_checks']:
             for virt_check in telemetry_json["virt_checks"]:
                     metric_name = "virt_check_recovery"
                     metric = {"metricName": metric_name}
-                    for k,v in virt_check.items():
-                        metric[k] = v
+                    for vc_k, vc_v in virt_check.items():
+                        metric[vc_k] = vc_v
                         metric['timestamp'] = str(datetime.datetime.now())
                     metrics_list.append(metric.copy())
 

krkn/scenario_plugins/network_chaos_ng/network_chaos_ng_scenario_plugin.py

@@ -1,3 +1,19 @@
+#!/usr/bin/env python
+#
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import logging
 import queue
 import random
@@ -51,19 +67,15 @@ def run(
                             f"with target type {network_chaos_type}"
                         )
 
-                    if (
-                        network_chaos_config.instance_count != 0
-                        and network_chaos_config.instance_count > len(targets)
-                    ):
+                    if network_chaos_config.instance_count != 0 and len(targets) > network_chaos_config.instance_count:
                         targets = random.sample(
                             targets, network_chaos_config.instance_count
                         )
-
                     if network_chaos_config.execution == "parallel":
                         self.run_parallel(targets, network_chaos)
                     else:
                         self.run_serial(targets, network_chaos)
-                    if len(config) > 1:
+                    if len(scenario_config) > 1:
                         logging.info(
                             f"waiting {network_chaos_config.wait_duration} seconds before running the next "
                             f"Network Chaos NG Module"

krkn/scenario_plugins/zone_outage/zone_outage_scenario_plugin.py

@@ -1,7 +1,23 @@
-import base64
-import json
+#!/usr/bin/env python
+#
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import logging
 import time
+import base64
+import json
 
 import yaml
 

krkn/utils/HealthChecker.py

@@ -1,3 +1,19 @@
+#!/usr/bin/env python
+#
+# Copyright 2025 The Krkn Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import requests
 import time
 import logging
@@ -6,10 +22,10 @@
 from krkn_lib.models.telemetry.models import HealthCheck
 
 class HealthChecker:
-    current_iterations: int = 0
-    ret_value = 0
     def __init__(self, iterations):
         self.iterations = iterations
+        self.current_iterations: int = 0
+        self.ret_value = 0
 
     def make_request(self, url, auth=None, headers=None, verify=True):
         response_data = {}
@@ -50,8 +66,10 @@ def run_health_check(self, health_check_config, health_check_telemetry_queue: qu
                             "start_timestamp": start_timestamp
                         }
                         if response["status_code"] != 200: 
-                            if response_tracker[config["url"]] != False: response_tracker[config["url"]] = False
-                            if config["exit_on_failure"] and config["exit_on_failure"] == True and self.ret_value==0: self.ret_value = 2
+                            if response_tracker[config["url"]] is not False:
+                                response_tracker[config["url"]] = False
+                            if config["exit_on_failure"] is True and self.ret_value == 0:
+                                self.ret_value = 2
                     else:
                             if response["status_code"] != health_check_tracker[config["url"]]["status_code"]:
                                 end_timestamp = datetime.now()
@@ -86,4 +104,5 @@ def run_health_check(self, health_check_config, health_check_telemetry_queue: qu
 
             health_check_telemetry_queue.put(health_check_telemetry)
         else:
-            logging.info("health checks config is not defined, skipping them")
+            logging.info("health checks config is not defined, skipping them")
+        return self.ret_value