diff --git a/config/config.yaml b/config/config.yaml index 6133cdbd..d0e35480 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -131,4 +131,5 @@ kubevirt_checks: # Utilizing virt che 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 ssh_node: "" # If set, will be a backup way to ssh to a node. Will want to set to a node that isn't targeted in chaos node_names: "" - exit_on_failure: # If value is True and VMI's are failing post chaos returns failure, values can be True/False \ No newline at end of file + exit_on_failure: # If value is True and VMI's are failing post chaos returns failure, values can be True/False + diff --git a/krkn/prometheus/collector.py b/krkn/prometheus/collector.py new file mode 100644 index 00000000..0e6be15f --- /dev/null +++ b/krkn/prometheus/collector.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import datetime +import logging +from typing import Dict, Any, List, Optional + +from krkn_lib.prometheus.krkn_prometheus import KrknPrometheus + + +# ----------------------------------------------------------------------------- +# SLO evaluation helpers (used by krkn.resiliency) +# ----------------------------------------------------------------------------- + + +def slo_passed(prometheus_result: List[Any]) -> Optional[bool]: + if not prometheus_result: + return None + has_samples = False + for series in prometheus_result: + if "values" in series: + has_samples = True + for _ts, val in series["values"]: + try: + if float(val) > 0: + return False + except (TypeError, ValueError): + continue + elif "value" in series: + has_samples = True + try: + return float(series["value"][1]) == 0 + except (TypeError, ValueError): + return False + + # If we reached here and never saw any samples, skip + return None if not has_samples else True + + +def evaluate_slos( + prom_cli: KrknPrometheus, + slo_list: List[Dict[str, Any]], + start_time: datetime.datetime, + end_time: datetime.datetime, +) -> Dict[str, bool]: + """Evaluate a list of SLO expressions against Prometheus. + + Args: + prom_cli: Configured Prometheus client. + slo_list: List of dicts with keys ``name``, ``expr``. + start_time: Start timestamp. + end_time: End timestamp. + granularity: Step in seconds for range queries. + Returns: + Mapping name -> bool indicating pass status. + True means good we passed the SLO test otherwise failed the SLO + """ + results: Dict[str, bool] = {} + logging.info("Evaluating %d SLOs over window %s – %s", len(slo_list), start_time, end_time) + for slo in slo_list: + expr = slo["expr"] + name = slo["name"] + try: + response = prom_cli.process_prom_query_in_range( + expr, + start_time=start_time, + end_time=end_time, + ) + + passed = slo_passed(response) + if passed is None: + logging.warning("SLO '%s' query returned no data; excluding from score.", name) + else: + results[name] = passed + except Exception as exc: + logging.error("PromQL query failed for SLO '%s': %s", name, exc) + results[name] = False + return results diff --git a/krkn/resiliency/__init__.py b/krkn/resiliency/__init__.py new file mode 100644 index 00000000..d88fbf8d --- /dev/null +++ b/krkn/resiliency/__init__.py @@ -0,0 +1,4 @@ +"""krkn.resiliency package public interface.""" + +from .resiliency import Resiliency, compute_resiliency # noqa: F401 +from .score import calculate_resiliency_score # noqa: F401 diff --git a/krkn/resiliency/resiliency.py b/krkn/resiliency/resiliency.py new file mode 100644 index 00000000..7d18e127 --- /dev/null +++ b/krkn/resiliency/resiliency.py @@ -0,0 +1,525 @@ +"""Resiliency evaluation orchestrator for Krkn chaos runs. + +This module provides the `Resiliency` class which loads the canonical +`alerts.yaml`, executes every SLO expression against Prometheus in the +chaos-test time window, determines pass/fail status and calculates an +overall resiliency score using the generic weighted model implemented +in `krkn.resiliency.score`. +""" + +from __future__ import annotations + +import base64 +import datetime +import logging +import os +from typing import Dict, List, Any, Optional + +import yaml +import json +import dataclasses +from krkn_lib.models.telemetry import ChaosRunTelemetry + +from krkn_lib.prometheus.krkn_prometheus import KrknPrometheus +from krkn.prometheus.collector import evaluate_slos +from krkn.resiliency.score import calculate_resiliency_score + + +class Resiliency: + """Central orchestrator for resiliency scoring.""" + + ENV_VAR_NAME = "KRKN_ALERTS_YAML_CONTENT" + + def __init__(self, alerts_yaml_path: str = "config/alerts.yaml"): + """Load SLO definitions from the default alerts file, unless the + *KRKN_ALERTS_YAML_CONTENT* environment variable is set – in which case its + raw YAML string is parsed instead. The custom YAML may optionally follow + this schema: + + prometheus_url: http://prometheus:9090 # optional, currently unused + slos: + - expr: + severity: critical|warning + description: + + For backward-compatibility the legacy list-only format is still accepted. + """ + raw_yaml_data: Any + env_yaml = os.getenv(self.ENV_VAR_NAME, '').strip() + if env_yaml: + try: + try: + decoded_yaml = base64.b64decode(env_yaml, validate=True).decode('utf-8') + except (base64.binascii.Error, UnicodeDecodeError) as e: + logging.debug("Failed to base64 decode %s, trying as plain YAML: %s", + self.ENV_VAR_NAME, str(e)) + decoded_yaml = env_yaml + + raw_yaml_data = yaml.safe_load(decoded_yaml) + logging.info("Loaded SLO configuration from environment variable %s", self.ENV_VAR_NAME) + + if isinstance(raw_yaml_data, dict): + self.prometheus_url = raw_yaml_data.get("prometheus_url") + raw_yaml_data = raw_yaml_data.get("slos", raw_yaml_data.get("alerts", [])) + + except yaml.YAMLError as exc: + logging.error("Failed to parse YAML from %s: %s", self.ENV_VAR_NAME, str(exc)) + raw_yaml_data = [] + self.prometheus_url = None + except Exception as exc: + logging.error("Unexpected error loading SLOs from %s: %s", + self.ENV_VAR_NAME, str(exc)) + raw_yaml_data = [] + self.prometheus_url = None + else: + if not os.path.exists(alerts_yaml_path): + raise FileNotFoundError(f"alerts file not found: {alerts_yaml_path}") + with open(alerts_yaml_path, "r", encoding="utf-8") as fp: + raw_yaml_data = yaml.safe_load(fp) + logging.info("Loaded SLO configuration from %s", alerts_yaml_path) + self.prometheus_url = None + + self._slos = self._normalise_alerts(raw_yaml_data) + self._results: Dict[str, bool] = {} + self._score: Optional[int] = None + self._breakdown: Optional[Dict[str, int]] = None + self._health_check_results: Dict[str, bool] = {} + self.scenario_reports: List[Dict[str, Any]] = [] + self.summary: Optional[Dict[str, Any]] = None + self.detailed_report: Optional[Dict[str, Any]] = None + + # --------------------------------------------------------------------- + # Public API + # --------------------------------------------------------------------- + + def calculate_score( + self, + *, + weights: Optional[Dict[str, int]] = None, + health_check_results: Optional[Dict[str, bool]] = None, + ) -> int: + """Calculate the resiliency score using collected SLO results.""" + slo_defs = {slo["name"]: slo["severity"] for slo in self._slos} + score, breakdown = calculate_resiliency_score( + slo_definitions=slo_defs, + prometheus_results=self._results, + health_check_results=health_check_results or {}, + weights=weights, + ) + self._score = score + self._breakdown = breakdown + self._health_check_results = health_check_results or {} + return score + + def to_dict(self) -> Dict[str, Any]: + """Return a dictionary ready for telemetry output.""" + if self._score is None: + raise RuntimeError("calculate_score() must be called before to_dict()") + return { + "score": self._score, + "breakdown": self._breakdown, + "slo_results": self._results, + "health_check_results": getattr(self, "_health_check_results", {}), + } + + # ------------------------------------------------------------------ + # Scenario-based resiliency evaluation + # ------------------------------------------------------------------ + def add_scenario_report( + self, + *, + scenario_name: str, + prom_cli: KrknPrometheus, + start_time: datetime.datetime, + end_time: datetime.datetime, + weight: float | int = 1, + health_check_results: Optional[Dict[str, bool]] = None, + weights: Optional[Dict[str, int]] = None, + ) -> int: + """ + Evaluate SLOs for a single scenario window and store the result. + + Args: + scenario_name: Human-friendly scenario identifier. + prom_cli: Initialized KrknPrometheus instance. + start_time: Window start. + end_time: Window end. + weight: Weight to use for the final weighted average calculation. + health_check_results: Optional mapping of custom health-check name ➡ bool. + weights: Optional override of severity weights for SLO calculation. + Returns: + The calculated integer resiliency score (0-100) for this scenario. + """ + slo_results = evaluate_slos( + prom_cli=prom_cli, + slo_list=self._slos, + start_time=start_time, + end_time=end_time, + ) + slo_defs = {slo["name"]: slo["severity"] for slo in self._slos} + score, breakdown = calculate_resiliency_score( + slo_definitions=slo_defs, + prometheus_results=slo_results, + health_check_results=health_check_results or {}, + weights=weights, + ) + self.scenario_reports.append( + { + "name": scenario_name, + "window": { + "start": start_time.isoformat(), + "end": end_time.isoformat(), + }, + "score": score, + "weight": weight, + "breakdown": breakdown, + "slo_results": slo_results, + "health_check_results": health_check_results or {}, + } + ) + return score + + def finalize_report( + self, + *, + prom_cli: KrknPrometheus, + total_start_time: datetime.datetime, + total_end_time: datetime.datetime, + weights: Optional[Dict[str, int]] = None, + ) -> None: + if not self.scenario_reports: + raise RuntimeError("No scenario reports added – nothing to finalize") + + # ---------------- Weighted average (primary resiliency_score) ---------- + total_weight = sum(rep["weight"] for rep in self.scenario_reports) + resiliency_score = int( + sum(rep["score"] * rep["weight"] for rep in self.scenario_reports) / total_weight + ) + + # ---------------- Overall SLO evaluation across full test window ----------------------------- + full_slo_results = evaluate_slos( + prom_cli=prom_cli, + slo_list=self._slos, + start_time=total_start_time, + end_time=total_end_time, + ) + slo_defs = {slo["name"]: slo["severity"] for slo in self._slos} + _overall_score, full_breakdown = calculate_resiliency_score( + slo_definitions=slo_defs, + prometheus_results=full_slo_results, + health_check_results={}, + weights=weights, + ) + + self.summary = { + "scenarios": {rep["name"]: rep["score"] for rep in self.scenario_reports}, + "resiliency_score": resiliency_score, + "passed_slos": full_breakdown.get("passed", 0), + "total_slos": full_breakdown.get("passed", 0) + full_breakdown.get("failed", 0), + } + + # Detailed report currently limited to per-scenario information; system stability section removed + self.detailed_report = { + "scenarios": self.scenario_reports, + } + + def get_summary(self) -> Dict[str, Any]: + """Return the concise resiliency_summary structure.""" + if not hasattr(self, "summary"): + raise RuntimeError("finalize_report() must be called first") + return self.summary + + def get_detailed_report(self) -> Dict[str, Any]: + """Return the full resiliency-report structure.""" + if not hasattr(self, "detailed_report"): + raise RuntimeError("finalize_report() must be called first") + return self.detailed_report + + @staticmethod + def compact_breakdown(report: Dict[str, Any]) -> Dict[str, int]: + """Return a compact summary dict for a single scenario report.""" + try: + passed = report["breakdown"]["passed"] + failed = report["breakdown"]["failed"] + score_val = report["score"] + except Exception: + passed = report.get("breakdown", {}).get("passed", 0) + failed = report.get("breakdown", {}).get("failed", 0) + score_val = report.get("score", 0) + return { + "resiliency_score": score_val, + "passed_slos": passed, + "total_slos": passed + failed, + } + + def attach_compact_to_telemetry(self, chaos_telemetry: ChaosRunTelemetry) -> None: + """Embed per-scenario compact resiliency reports into a ChaosRunTelemetry instance.""" + score_map = { + rep["name"]: self.compact_breakdown(rep) for rep in self.scenario_reports + } + new_scenarios = [] + for item in getattr(chaos_telemetry, "scenarios", []): + if isinstance(item, dict): + name = item.get("scenario") + if name in score_map: + item["resiliency_report"] = score_map[name] + new_scenarios.append(item) + else: + name = getattr(item, "scenario", None) + try: + item_dict = dataclasses.asdict(item) + except Exception: + item_dict = { + k: getattr(item, k) + for k in dir(item) + if not k.startswith("__") and not callable(getattr(item, k)) + } + if name in score_map: + item_dict["resiliency_report"] = score_map[name] + new_scenarios.append(item_dict) + chaos_telemetry.scenarios = new_scenarios + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + @staticmethod + def _normalise_alerts(raw_alerts: Any) -> List[Dict[str, Any]]: + """Convert raw YAML alerts data into internal SLO list structure.""" + if not isinstance(raw_alerts, list): + raise ValueError("SLO configuration must be a list under key 'slos' or top-level list") + + slos: List[Dict[str, Any]] = [] + for idx, alert in enumerate(raw_alerts): + if not (isinstance(alert, dict) and "expr" in alert and "severity" in alert): + logging.warning("Skipping invalid alert entry at index %d: %s", idx, alert) + continue + name = alert.get("description") or f"slo_{idx}" + slos.append( + { + "name": name, + "expr": alert["expr"], + "severity": str(alert["severity"]).lower(), + } + ) + return slos + +# ----------------------------------------------------------------------------- +# High-level helper for run_kraken.py +# ----------------------------------------------------------------------------- + +def compute_resiliency(*, + prometheus: KrknPrometheus, + chaos_telemetry: "ChaosRunTelemetry", + start_time: datetime.datetime, + end_time: datetime.datetime, + run_uuid: Optional[str] = None, + alerts_yaml_path: str = "config/alerts.yaml", + logger: Optional[logging.Logger] = None, +) -> Optional[Dict[str, Any]]: + """Evaluate SLOs, combine health-check results, attach a resiliency report + to *chaos_telemetry* and return the report. Any failure is logged and *None* + is returned. + """ + + log = logger or logging.getLogger(__name__) + + try: + resiliency_obj = Resiliency(alerts_yaml_path) + resiliency_obj._results = evaluate_slos( + prom_cli=prometheus, + slo_list=resiliency_obj._slos, + start_time=start_time, + end_time=end_time, + ) + + health_results: Dict[str, bool] = {} + hc_list = getattr(chaos_telemetry, "health_checks", None) + if hc_list: + for idx, hc in enumerate(hc_list): + # Extract URL/name + try: + name = getattr(hc, "url", None) + if name is None and isinstance(hc, dict): + name = hc.get("url", f"hc_{idx}") + except Exception: + name = f"hc_{idx}" + # Extract status + try: + status = getattr(hc, "status", None) + if status is None and isinstance(hc, dict): + status = hc.get("status", True) + except Exception: + status = False + health_results[str(name)] = bool(status) + + resiliency_obj.calculate_score(health_check_results=health_results) + resiliency_report = resiliency_obj.to_dict() + chaos_telemetry.resiliency_report = resiliency_report + chaos_telemetry.resiliency_score = resiliency_report.get("score") + + + if not hasattr(ChaosRunTelemetry, "_with_resiliency_patch"): + _orig_to_json = ChaosRunTelemetry.to_json + + def _to_json_with_resiliency(self): + raw_json = _orig_to_json(self) + try: + data = json.loads(raw_json) + except Exception: + return raw_json + if hasattr(self, "resiliency_report"): + data["resiliency_report"] = self.resiliency_report + if hasattr(self, "resiliency_score"): + data["resiliency_score"] = self.resiliency_score + return json.dumps(data) + + ChaosRunTelemetry.to_json = _to_json_with_resiliency + ChaosRunTelemetry._with_resiliency_patch = True + + log.info( + "Resiliency score for run %s: %s%%", + run_uuid or "", + resiliency_report.get("score"), + ) + return resiliency_report + + except Exception as exc: + log.error("Failed to compute resiliency score: %s", exc) + return None + + +# ----------------------------------------------------------------------------- +# Helper utilities extracted from run_kraken.py +# ----------------------------------------------------------------------------- + +from typing import Tuple + + +def add_scenario_reports( + *, + resiliency_obj: "Resiliency", + scenario_telemetries, + prom_cli: KrknPrometheus, + scenario_type: str, + batch_start_dt: datetime.datetime, + batch_end_dt: datetime.datetime, + weight: int | float = 1, + logger: Optional[logging.Logger] = None, +) -> None: + """Evaluate SLOs for every telemetry item belonging to a scenario window, + store the result in *resiliency_obj* and enrich the telemetry list with a + compact resiliency breakdown. + + Args: + resiliency_obj: Initialized :class:`Resiliency` orchestrator. If *None*, + the call becomes a no-op (saves caller side checks). + scenario_telemetries: Iterable with telemetry objects/dicts for the + current scenario batch window. + prom_cli: Pre-configured :class:`KrknPrometheus` instance. + scenario_type: Fallback scenario identifier in case individual + telemetry items do not provide one. + batch_start_dt: Fallback start timestamp for the batch window. + batch_end_dt: Fallback end timestamp for the batch window. + weight: Weight to assign to every scenario when calculating the final + weighted average. + logger: Optional custom logger. + """ + if resiliency_obj is None: + return + + log = logger or logging.getLogger(__name__) + + for tel in scenario_telemetries: + try: + # -------- Extract timestamps & scenario name -------------------- + if isinstance(tel, dict): + st_ts = tel.get("start_timestamp") + en_ts = tel.get("end_timestamp") + scen_name = tel.get("scenario", scenario_type) + else: + st_ts = getattr(tel, "start_timestamp", None) + en_ts = getattr(tel, "end_timestamp", None) + scen_name = getattr(tel, "scenario", scenario_type) + + if st_ts and en_ts: + st_dt = datetime.datetime.fromtimestamp(int(st_ts)) + en_dt = datetime.datetime.fromtimestamp(int(en_ts)) + else: + st_dt = batch_start_dt + en_dt = batch_end_dt + + # -------- Calculate resiliency score for the scenario ----------- + resiliency_obj.add_scenario_report( + scenario_name=str(scen_name), + prom_cli=prom_cli, + start_time=st_dt, + end_time=en_dt, + weight=weight, + health_check_results=None, + ) + + compact = Resiliency.compact_breakdown( + resiliency_obj.scenario_reports[-1] + ) + if isinstance(tel, dict): + tel["resiliency_report"] = compact + else: + setattr(tel, "resiliency_report", compact) + except Exception as exc: + log.error("Resiliency per-scenario evaluation failed: %s", exc) + + +def finalize_and_save( + *, + resiliency_obj: "Resiliency", + prom_cli: KrknPrometheus, + total_start_time: datetime.datetime, + total_end_time: datetime.datetime, + run_mode: str = "standalone", + summary_path: str = "kraken.report", + detailed_path: str = "resiliency-report.json", + logger: Optional[logging.Logger] = None, +) -> Tuple[Dict[str, Any], Dict[str, Any]]: + """Finalize resiliency scoring, persist reports and return them. + + Returns: + (summary_report, detailed_report) + """ + if resiliency_obj is None: + return {}, {} + + log = logger or logging.getLogger(__name__) + + try: + resiliency_obj.finalize_report( + prom_cli=prom_cli, + total_start_time=total_start_time, + total_end_time=total_end_time, + ) + summary = resiliency_obj.get_summary() + detailed = resiliency_obj.get_detailed_report() + + if run_mode == "controller": + # krknctl expects the detailed report on stdout in a special format + try: + detailed_json = json.dumps(detailed) + print(f"KRKN_RESILIENCY_REPORT_JSON:{detailed_json}") + log.info("Resiliency report logged to stdout for krknctl.") + except Exception as exc: + log.error("Failed to serialize and log detailed resiliency report: %s", exc) + else: + # Stand-alone mode – write to files for post-run consumption + try: + with open(summary_path, "w", encoding="utf-8") as fp: + json.dump(summary, fp, indent=2) + with open(detailed_path, "w", encoding="utf-8") as fp: + json.dump(detailed, fp, indent=2) + log.info("Resiliency reports written: %s and %s", summary_path, detailed_path) + except Exception as io_exc: + log.error("Failed to write resiliency report files: %s", io_exc) + + return summary, detailed + + except Exception as exc: + log.error("Failed to finalize resiliency scoring: %s", exc) + return {}, {} \ No newline at end of file diff --git a/krkn/resiliency/score.py b/krkn/resiliency/score.py new file mode 100644 index 00000000..389ffb8c --- /dev/null +++ b/krkn/resiliency/score.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from typing import Dict, List, Tuple + +DEFAULT_WEIGHTS = {"critical": 3, "warning": 1} + + +class SLOResult: + """Simple container representing evaluation outcome for a single SLO.""" + + def __init__(self, name: str, severity: str, passed: bool): + self.name = name + self.severity = severity + self.passed = passed + + def weight(self, weights: Dict[str, int] | None = None) -> int: + _w = weights or DEFAULT_WEIGHTS + return _w.get(self.severity, DEFAULT_WEIGHTS["warning"]) + + +def calculate_resiliency_score( + slo_definitions: Dict[str, str], + prometheus_results: Dict[str, bool], + health_check_results: Dict[str, bool], + weights: Dict[str, int] | None = None, +) -> Tuple[int, Dict[str, int]]: + """Compute a resiliency score between 0-100 based on SLO pass/fail results. + + Args: + slo_definitions: Mapping of SLO name -> severity ("critical" | "warning"). + prometheus_results: Mapping of SLO name -> bool indicating whether the SLO + passed. Any SLO missing in this mapping is treated as failed. + health_check_results: Mapping of custom health-check name -> bool pass flag. + These checks are always treated as *critical*. + weights: Optional override of severity weights. + + Returns: + Tuple containing (final_score, breakdown) where *breakdown* is a dict with + the counts of passed/failed SLOs per severity. + """ + + weights = weights or DEFAULT_WEIGHTS + + slo_objects: List[SLOResult] = [] + for slo_name, severity in slo_definitions.items(): + # Exclude SLOs that were not evaluated (query returned no data) + if slo_name not in prometheus_results: + continue + passed = bool(prometheus_results[slo_name]) + slo_objects.append(SLOResult(slo_name, severity, passed)) + + # Health-check SLOs (by default keeping them critical) + for hc_name, hc_passed in health_check_results.items(): + slo_objects.append(SLOResult(hc_name, "critical", bool(hc_passed))) + + total_points = sum(slo.weight(weights) for slo in slo_objects) + points_lost = sum(slo.weight(weights) for slo in slo_objects if not slo.passed) + + score = 0 if total_points == 0 else int(((total_points - points_lost) / total_points) * 100) + + breakdown = { + "total_points": total_points, + "points_lost": points_lost, + "passed": len([s for s in slo_objects if s.passed]), + "failed": len([s for s in slo_objects if not s.passed]), + } + return score, breakdown diff --git a/run_kraken.py b/run_kraken.py index 0728c73c..25091984 100644 --- a/run_kraken.py +++ b/run_kraken.py @@ -12,7 +12,7 @@ import uuid import time import queue import threading -from typing import Optional +from typing import Optional, Dict from krkn import cerberus from krkn_lib.elastic.krkn_elastic import KrknElastic @@ -21,6 +21,12 @@ from krkn_lib.models.krkn import ChaosRunOutput, ChaosRunAlertSummary from krkn_lib.prometheus.krkn_prometheus import KrknPrometheus import krkn.prometheus as prometheus_plugin import server as server +from krkn.resiliency.resiliency import ( + Resiliency, + compute_resiliency, + add_scenario_reports, + finalize_and_save, +) from krkn_lib.k8s import KrknKubernetes from krkn_lib.ocp import KrknOpenshift from krkn_lib.telemetry.k8s import KrknTelemetryKubernetes @@ -54,6 +60,13 @@ def main(options, command: Optional[str]) -> int: print(pyfiglet.figlet_format("kraken")) logging.info("Starting kraken") + # Determine execution mode (standalone, controller, or disabled) + run_mode = (os.getenv("RESILIENCY_ENABLED_MODE") or "standalone").lower().strip() + valid_run_modes = {"standalone", "controller", "disabled"} + if run_mode not in valid_run_modes: + logging.warning("Unknown RESILIENCY_ENABLED_MODE '%s'. Defaulting to 'standalone'", run_mode) + run_mode = "standalone" + cfg = options.cfg # Parse and read the config if os.path.isfile(cfg): @@ -91,9 +104,7 @@ def main(options, command: Optional[str]) -> int: daemon_mode = get_yaml_item_value(config["tunings"], "daemon_mode", False) prometheus_url = config["performance_monitoring"].get("prometheus_url") - prometheus_bearer_token = config["performance_monitoring"].get( - "prometheus_bearer_token" - ) + prometheus_bearer_token = config["performance_monitoring"].get("prometheus_bearer_token") run_uuid = config["performance_monitoring"].get("uuid") enable_alerts = get_yaml_item_value( config["performance_monitoring"], "enable_alerts", False @@ -101,6 +112,14 @@ def main(options, command: Optional[str]) -> int: enable_metrics = get_yaml_item_value( config["performance_monitoring"], "enable_metrics", False ) + + # Disable resiliency if Prometheus URL is not available + if (not prometheus_url or prometheus_url.strip() == "") and run_mode != "disabled": + logging.warning("Prometheus URL not provided; disabling resiliency score features.") + run_mode = "disabled" + + # Default placeholder; will be overridden if a Prometheus URL is available + prometheus = None # elastic search enable_elastic = get_yaml_item_value(config["elastic"], "enable_elastic", False) elastic_run_tag = get_yaml_item_value(config["elastic"], "run_tag", "") @@ -251,9 +270,18 @@ def main(options, command: Optional[str]) -> int: else: elastic_search = None summary = ChaosRunAlertSummary() - if enable_metrics or enable_alerts or check_critical_alerts: + if enable_metrics or enable_alerts or check_critical_alerts or run_mode != "disabled": prometheus = KrknPrometheus(prometheus_url, prometheus_bearer_token) + # Quick connectivity probe for Prometheus – disable resiliency if unreachable + try: + prometheus.process_prom_query_in_range( + "up", datetime.datetime.utcnow() - datetime.timedelta(seconds=60), datetime.datetime.utcnow(), granularity=60 + ) + except Exception as prom_exc: + logging.error("Prometheus connectivity test failed: %s. Disabling resiliency features as Prometheus is required for SLO evaluation.", prom_exc) + run_mode = "disabled" + resiliency_obj = Resiliency() if run_mode != "disabled" else None # Initialize resiliency orchestrator logging.info("Server URL: %s" % kubecli.get_host()) if command == "list-rollback": @@ -369,6 +397,8 @@ def main(options, command: Optional[str]) -> int: ) sys.exit(-1) + + batch_window_start_dt = datetime.datetime.utcnow() failed_scenarios_current, scenario_telemetries = ( scenario_plugin.run_scenarios( run_uuid, scenarios_list, config, telemetry_ocp @@ -376,6 +406,16 @@ def main(options, command: Optional[str]) -> int: ) failed_post_scenarios.extend(failed_scenarios_current) chaos_telemetry.scenarios.extend(scenario_telemetries) + batch_window_end_dt = datetime.datetime.utcnow() + if resiliency_obj: + add_scenario_reports( + resiliency_obj=resiliency_obj, + scenario_telemetries=scenario_telemetries, + prom_cli=prometheus, + scenario_type=scenario_type, + batch_start_dt=batch_window_start_dt, + batch_end_dt=batch_window_end_dt, + ) post_critical_alerts = 0 if check_critical_alerts: @@ -440,12 +480,37 @@ def main(options, command: Optional[str]) -> int: else: logging.info("No error logs collected during chaos run") chaos_telemetry.error_logs = [] + if resiliency_obj: + try: + resiliency_obj.attach_compact_to_telemetry(chaos_telemetry) + except Exception as exc: + logging.error("Failed to embed per-scenario resiliency in telemetry: %s", exc) + + if resiliency_obj: + try: + summary_report, detailed_report = finalize_and_save( + resiliency_obj=resiliency_obj, + prom_cli=prometheus, + total_start_time=datetime.datetime.fromtimestamp(start_time), + total_end_time=datetime.datetime.fromtimestamp(end_time), + run_mode=run_mode, + logger=logging, + ) + + except Exception as e: + logging.error("Failed to finalize resiliency scoring: %s", e) + telemetry_json = chaos_telemetry.to_json() decoded_chaos_run_telemetry = ChaosRunTelemetry(json.loads(telemetry_json)) + if resiliency_obj and hasattr(resiliency_obj, "summary"): + decoded_chaos_run_telemetry.overall_resiliency_report = resiliency_obj.get_summary() chaos_output.telemetry = decoded_chaos_run_telemetry logging.info(f"Chaos data:\n{chaos_output.to_json()}") if enable_elastic: + elastic_telemetry = ElasticChaosRunTelemetry( + chaos_run_telemetry=decoded_chaos_run_telemetry + ) result = elastic_search.push_telemetry( decoded_chaos_run_telemetry, elastic_telemetry_index )