integration WIP

Signed-off-by: Tullio Sebastiani <[email protected]>

kraken/arcaflow_plugin/arcaflow_plugin.py

@@ -5,23 +5,30 @@
 import logging
 from pathlib import Path
 from typing import List
+
 from .context_auth import ContextAuth
 from krkn_lib.telemetry.k8s import KrknTelemetryKubernetes
 from krkn_lib.models.telemetry import ScenarioTelemetry
 
+from .. import utils
+
 
 def run(scenarios_list: List[str], kubeconfig_path: str, telemetry: KrknTelemetryKubernetes) -> (list[str], list[ScenarioTelemetry]):
     scenario_telemetries: list[ScenarioTelemetry] = []
     failed_post_scenarios = []
     for scenario in scenarios_list:
         scenario_telemetry = ScenarioTelemetry()
         scenario_telemetry.scenario = scenario
-        scenario_telemetry.start_timestamp = time.time()
-        telemetry.set_parameters_base64(scenario_telemetry,scenario)
+        start_time = time.time()
+        scenario_telemetry.start_timestamp = start_time
+        telemetry.set_parameters_base64(scenario_telemetry, scenario)
         engine_args = build_args(scenario)
         status_code = run_workflow(engine_args, kubeconfig_path)
-        scenario_telemetry.end_timestamp = time.time()
+        end_time = time.time()
+        scenario_telemetry.end_timestamp = end_time
         scenario_telemetry.exit_status = status_code
+
+        utils.populate_cluster_events(scenario_telemetry, scenario, telemetry.kubecli, int(start_time), int(end_time))
         scenario_telemetries.append(scenario_telemetry)
         if status_code != 0:
             failed_post_scenarios.append(scenario)

kraken/utils/__init__.py

@@ -1 +1,2 @@
 from .TeeLogHandler import TeeLogHandler
+from .functions import *

kraken/utils/functions.py

@@ -0,0 +1,26 @@
+import krkn_lib.utils
+from krkn_lib.k8s import KrknKubernetes
+from krkn_lib.models.telemetry import ScenarioTelemetry
+from tzlocal.unix import get_localzone
+
+
+def populate_cluster_events(scenario_telemetry: ScenarioTelemetry,
+                                  scenario_config: dict,
+                                  kubecli: KrknKubernetes,
+                                  start_timestamp: int,
+                                  end_timestamp: int
+                                  ):
+    namespaces = []
+    events = []
+    namespaces.extend(krkn_lib.utils.deep_search_attribute("namespace", scenario_config))
+    namespace_patterns = krkn_lib.utils.deep_search_attribute("namespace_pattern", scenario_config)
+    for pattern in namespace_patterns:
+        namespaces.extend(kubecli.list_namespaces_by_regex(pattern))
+
+    if len(namespaces) == 0:
+        events.extend(kubecli.collect_and_parse_cluster_events(start_timestamp, end_timestamp, str(get_localzone())))
+    else:
+        for namespace in namespaces:
+            events.extend(kubecli.collect_and_parse_cluster_events(start_timestamp, end_timestamp, str(get_localzone()), namespace=namespace))
+
+    scenario_telemetry.set_cluster_events(events)

requirements.txt

@@ -15,7 +15,7 @@ google-api-python-client==2.116.0
 ibm_cloud_sdk_core==3.18.0
 ibm_vpc==0.20.0
 jinja2==3.1.4
-krkn-lib==3.0.0
+krkn-lib@git+https://github.com/krkn-chaos/krkn-lib.git@namespaced_logs_alerts
 lxml==5.1.0
 kubernetes==28.1.0
 numpy==1.26.4

run_kraken.py

@@ -14,6 +14,8 @@
 from krkn_lib.models.elastic import ElasticChaosRunTelemetry
 from krkn_lib.models.krkn import ChaosRunOutput, ChaosRunAlertSummary
 from krkn_lib.prometheus.krkn_prometheus import KrknPrometheus
+from tzlocal.unix import get_localzone
+
 import kraken.time_actions.common_time_functions as time_actions
 import kraken.performance_dashboards.setup as performance_dashboards
 import kraken.pod_scenarios.setup as pod_scenarios
@@ -452,7 +454,8 @@ def main(cfg) -> int:
         else:
             telemetry_k8s.collect_cluster_metadata(chaos_telemetry)
 
-        decoded_chaos_run_telemetry = ChaosRunTelemetry(json.loads(chaos_telemetry.to_json()))
+        telemetry_json = chaos_telemetry.to_json()
+        decoded_chaos_run_telemetry = ChaosRunTelemetry(json.loads(telemetry_json))
         chaos_output.telemetry = decoded_chaos_run_telemetry
         logging.info(f"Chaos data:\n{chaos_output.to_json()}")
         if enable_elastic:
@@ -468,7 +471,8 @@ def main(cfg) -> int:
             logging.info(f"telemetry upload log: {safe_logger.log_file_name}")
             try:
                 telemetry_k8s.send_telemetry(config["telemetry"], telemetry_request_id, chaos_telemetry)
-                telemetry_k8s.put_cluster_events(telemetry_request_id, config["telemetry"], start_time, end_time)
+
+                telemetry_k8s.put_cluster_events(telemetry_request_id, config["telemetry"], events_file)
                 telemetry_k8s.put_critical_alerts(telemetry_request_id, config["telemetry"], summary)
                 # prometheus data collection is available only on Openshift
                 if config["telemetry"]["prometheus_backup"]: