adding core v1 event api

Signed-off-by: Paige Patton <[email protected]>

src/krkn_lib/k8s/krkn_kubernetes.py

@@ -37,7 +37,7 @@
     Volume,
     VolumeMount,
 )
-from krkn_lib.models.telemetry import NodeInfo, Taint, ClusterEvent
+from krkn_lib.models.telemetry import ClusterEvent, NodeInfo, Taint
 from krkn_lib.utils import filter_dictionary, get_random_string
 from krkn_lib.utils.safe_logger import SafeLogger
 
@@ -150,6 +150,7 @@ def __initialize_clients(self, kubeconfig_path: str = None):
                 config_file=kubeconfig_path
             )
             self.cli = client.CoreV1Api(self.k8s_client)
+
             self.version_client = client.VersionApi(self.api_client)
             self.apps_api = client.AppsV1Api(self.api_client)
             self.batch_cli = client.BatchV1Api(self.k8s_client)
@@ -1712,9 +1713,7 @@ def find_kraken_node(self) -> str:
                 raise e
         return node_name
 
-    def watch_node_status(
-        self, node: str, status: str, timeout: int
-    ):
+    def watch_node_status(self, node: str, status: str, timeout: int):
         """
         Watch for a specific node status
 
@@ -2473,41 +2472,22 @@ def collect_and_parse_cluster_events(
         """
         events = []
         try:
-            path_params: Dict[str, str] = {}
-            query_params = {"limit": limit}
-            header_params: Dict[str, str] = {}
-            auth_settings = ["BearerToken"]
-            header_params["Accept"] = self.api_client.select_header_accept(
-                ["application/json"]
-            )
-
-            path = "/api/v1/events"
             if namespace:
-                path = f"/api/v1/namespaces/{namespace}/events"
+                events_list = self.cli.list_namespaced_event(namespace)
 
-            (data) = self.api_client.call_api(
-                path,
-                "GET",
-                path_params,
-                query_params,
-                header_params,
-                response_type="str",
-                auth_settings=auth_settings,
-            )
-
-            json_obj = ast.literal_eval(data[0])
-            events_list = reversed(json_obj["items"])
+            else:
+                events_list = self.cli.list_event_for_all_namespaces()
+            events_list = events_list.items
             for obj in events_list:
-                filtered_obj = filter_dictionary(
-                    obj,
-                    "firstTimestamp",
+                in_filtered_time = filter_dictionary(
+                    obj.first_timestamp,
                     start_timestamp,
                     end_timestamp,
                     cluster_timezone,
                     local_timezone,
                 )
-                if filtered_obj:
-                    events.append(ClusterEvent(k8s_json_dict=obj))
+                if in_filtered_time:
+                    events.append(ClusterEvent(k8s_obj=obj))
 
         except Exception as e:
             logging.error(str(e))

src/krkn_lib/models/telemetry/models.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
+
 import base64
 import json
-import yaml
 from dataclasses import dataclass
 from datetime import datetime, timezone
+
+import yaml
+
 from krkn_lib.models.k8s import PodsStatus
 
 relevant_event_reasons: frozenset[str] = frozenset(
@@ -292,7 +295,12 @@ class ClusterEvent:
     type: str
     """ Event severity"""
 
-    def __init__(self, k8s_json_dict: any = None, json_dict: any = None):
+    def __init__(
+        self,
+        k8s_json_dict: any = None,
+        json_dict: any = None,
+        k8s_obj: any = None,
+    ):
         self.name = ""
         self.creation = datetime.now(timezone.utc).strftime(
             "%Y-%m-%dT%H:%M:%SZ"
@@ -306,6 +314,20 @@ def __init__(self, k8s_json_dict: any = None, json_dict: any = None):
         self.involved_object_namespace = ""
         self.type = ""
 
+        if k8s_obj:
+            # This parses CoreV1Event
+            # (https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Event.md)
+            self.name = k8s_obj.metadata.name
+            self.creation = k8s_obj.metadata.creation_timestamp
+            self.reason = k8s_obj.reason
+            self.message = k8s_obj.message
+            self.namespace = k8s_obj.metadata.namespace
+            self.source_component = k8s_obj.source.component
+            self.involved_object_kind = k8s_obj.involved_object.kind
+            self.involved_object_name = k8s_obj.involved_object.name
+            self.involved_object_namespace = k8s_obj.involved_object.namespace
+            self.type = k8s_obj.type
+
         if k8s_json_dict:
             self.name = k8s_json_dict["metadata"]["name"]
             self.creation = k8s_json_dict["metadata"]["creationTimestamp"]

src/krkn_lib/tests/test_krkn_telemetry_models.py

@@ -1,10 +1,17 @@
 import json
 import unittest
 
+from kubernetes import client
+from kubernetes.client.models import (
+    V1EventSource,
+    V1ObjectMeta,
+    V1ObjectReference,
+)
+
 from krkn_lib.models.telemetry import (
     ChaosRunTelemetry,
-    ScenarioTelemetry,
     ClusterEvent,
+    ScenarioTelemetry,
 )
 
 
@@ -350,6 +357,48 @@ def test_cluster_event(self):
         self.assertEqual(event.involved_object_namespace, "default")
         self.assertEqual(event.type, "Normal")
 
+        json_dict = json.loads(krkn_json)
+
+        v1_object_reference = V1ObjectReference(
+            api_version="v1",
+            kind=json_dict["involved_object_kind"],
+            name=json_dict["involved_object_name"],
+            namespace=json_dict["involved_object_namespace"],
+            uid="2222666",  # Optional
+        )
+
+        event_source = V1EventSource(json_dict["source_component"])
+        metadata = V1ObjectMeta(
+            name=json_dict["involved_object_name"],
+            namespace=json_dict["involved_object_namespace"],
+            creation_timestamp=json_dict["creation"],
+        )
+        core_event = client.CoreV1Event(
+            message=json_dict["message"],
+            reason=json_dict["reason"],
+            type=json_dict["type"],
+            first_timestamp=json_dict["creation"],
+            involved_object=v1_object_reference,
+            source=event_source,
+            metadata=metadata,
+        )  # CoreV1Event
+
+        event = ClusterEvent(k8s_obj=core_event)
+
+        self.assertEqual(event.name, "test")
+        self.assertEqual(event.creation, "2024-09-02T14:00:53Z")
+        self.assertEqual(event.reason, "Failed")
+        self.assertEqual(
+            event.message,
+            "message",
+        )
+        self.assertEqual(event.namespace, "default")
+        self.assertEqual(event.source_component, "kubelet")
+        self.assertEqual(event.involved_object_kind, "Pod")
+        self.assertEqual(event.involved_object_name, "test")
+        self.assertEqual(event.involved_object_namespace, "default")
+        self.assertEqual(event.type, "Normal")
+
 
 if __name__ == "__main__":
     unittest.main()

src/krkn_lib/tests/test_utils.py

@@ -1,10 +1,11 @@
 import base64
-import json
+import datetime
 import os
 import re
 import tempfile
 
 import yaml
+from dateutil.tz import tzutc
 
 import krkn_lib.utils as utils
 from krkn_lib.tests import BaseTest
@@ -14,10 +15,10 @@
     filter_dictionary,
     filter_log_line,
     find_executable_in_path,
+    get_junit_test_case,
     get_random_string,
     get_yaml_item_value,
     is_host_reachable,
-    get_junit_test_case,
 )
 
 
@@ -254,72 +255,68 @@ def test_filter_file_log(self):
             )
 
     def test_filter_dictionary(self):
-        event = """
-                {
+        event = {
             "apiVersion": "v1",
             "count": 1,
-            "eventTime": null,
-            "firstTimestamp": "2023-10-04T09:51:06Z",
+            "eventTime": "null",
+            "firstTimestamp": datetime.datetime(
+                2023, 10, 4, 9, 51, 6, tzinfo=tzutc()
+            ),
             "involvedObject": {
                 "kind": "Node",
                 "name": "ip-10-0-143-127.us-west-2.compute.internal",
-                "uid": "ip-10-0-143-127.us-west-2.compute.internal"
+                "uid": "ip-10-0-143-127.us-west-2.compute.internal",
             },
             "kind": "Event",
             "lastTimestamp": "2023-10-04T09:51:06Z",
             "message": "Starting kubelet.",
             "metadata": {
                 "creationTimestamp": "2023-10-04T09:51:07Z",
-                "name": "ip-10-0-143-127.us-west-2.compute.internal.178adeb2335c61fe",
+                "name": "ip-10-0-143-127.us-west-2.compute.internal",
                 "namespace": "default",
                 "resourceVersion": "13109",
-                "uid": "e119423c-a5af-4bdc-b8ea-042d01390387"
+                "uid": "e119423c-a5af-4bdc-b8ea-042d01390387",
             },
             "reason": "Starting",
             "reportingComponent": "",
             "reportingInstance": "",
             "source": {
                 "component": "kubelet",
-                "host": "ip-10-0-143-127.us-west-2.compute.internal"
+                "host": "ip-10-0-143-127.us-west-2.compute.internal",
             },
-            "type": "Normal"
+            "type": "Normal",
         }
-        """  # NOQA
+
         ten_minutes_ago = 1696412513  # Wednesday, October 4, 2023 9:41:53 AM
         in_ten_minutes = 1696413713  # Wednesday, October 4, 2023 10:01:53 AM
 
-        event_dict = json.loads(event)
-
         result = filter_dictionary(
-            event_dict,
-            "firstTimestamp",
+            event["firstTimestamp"],
             ten_minutes_ago,
             in_ten_minutes,
             "UTC",
             "UTC",
         )
-        self.assertEqual(result, event_dict)
+        self.assertTrue(result)
 
         result = filter_dictionary(
-            event_dict,
-            "firstTimestamp",
+            event["firstTimestamp"],
             in_ten_minutes,
             None,
             "UTC",
             "UTC",
         )
 
-        self.assertIsNone(result)
+        self.assertFalse(result)
 
         result = filter_dictionary(
-            event_dict,
-            "apiVersion",
+            event["apiVersion"],
             in_ten_minutes,
             None,
             "UTC",
             "UTC",
         )
-        self.assertIsNone(result)
+        self.assertFalse(result)
 
     def test_get_yaml_item_value(self):
         cont = {"n_int": 1, "n_str": "value", "d_int": None, "d_str": None}

src/krkn_lib/utils/functions.py

@@ -3,17 +3,18 @@
 import os
 import random
 import re
+import socket
 import string
 import sys
-import socket
+import xml.etree.cElementTree as ET
 from queue import Queue
 from typing import Optional
 
 import pytz
 from base64io import Base64IO
 from dateutil import parser
 from dateutil.parser import ParserError
-import xml.etree.cElementTree as ET
+from dateutil.tz import tzutc # NOQA
 
 
 def decode_base64_file(source_filename: str, destination_filename: str):
@@ -241,7 +242,6 @@ def filter_log_line(
 
 
 def filter_dictionary(
-    dictionary: dict[str, any],
     datetime_key: str,
     start_timestamp: Optional[int],
     end_timestamp: Optional[int],
@@ -263,27 +263,27 @@ def filter_dictionary(
     :param interval_timezone: timezone of the interval within
         the dictionary will be checked
     """
-    date_time = dictionary.get(datetime_key)
-    if not date_time:
+    if not datetime_key:
         return None
 
     try:
-        log_date = parser.parse(date_time)
         if check_date_in_localized_interval(
             start_timestamp,
             end_timestamp,
-            int(log_date.timestamp()),
+            int(datetime_key.timestamp()),
             dictionary_timezone,
             interval_timezone,
         ):
-            return dictionary
-        return None
-    except ParserError:
-        logging.error(f"impossible to parse date: {str(date_time)}")
-        return None
+            return True
+        return False
     except TypeError:
-        logging.error(f"{str(date_time)} does not represent a valid datetime")
-        return None
+        logging.error(
+            f"{str(datetime_key)} does not represent a valid datetime"
+        )
+        return False
+    except Exception as e:
+        logging.error(f"{str(datetime_key)} " + str(e))
+        return False
 
 
 def filter_log_file_worker(