litmus and cluster events deprecation (#126)

linting

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

Author: tsebastiani

src/krkn_lib/k8s/krkn_kubernetes.py

@@ -12,10 +12,8 @@
 from functools import partial
 from queue import Queue
 from typing import Any, Dict, List, Optional
-from krkn_lib.version import __version__
 
 import arcaflow_lib_kubernetes
-import deprecation
 import kubernetes
 import urllib3
 import yaml
@@ -31,10 +29,7 @@
     PVC,
     AffectedPod,
     ApiRequestException,
-    ChaosEngine,
-    ChaosResult,
     Container,
-    LitmusChaosObject,
     Pod,
     PodsMonitorThread,
     PodsStatus,
@@ -1575,82 +1570,6 @@ def get_pod_info(self, name: str, namespace: str = "default") -> Pod:
             )
             return None
 
-    @deprecation.deprecated(
-        deprecated_in="3.1.0",
-        removed_in="3.2.0",
-        current_version=__version__,
-        details="litmus support dropped from krkn",
-    )
-    def get_litmus_chaos_object(
-        self, kind: str, name: str, namespace: str = "default"
-    ) -> LitmusChaosObject:
-        """
-        Retrieves Litmus Chaos CRDs
-
-        :param kind: the custom resource type
-        :param name: the object name
-        :param namespace: the namespace (optional default `default`)
-        :return: data class object of a subclass of LitmusChaosObject
-        """
-
-        group = "litmuschaos.io"
-        version = "v1alpha1"
-
-        if kind.lower() == "chaosengine":
-            plural = "chaosengines"
-            response = self.custom_object_client.get_namespaced_custom_object(
-                group=group,
-                plural=plural,
-                version=version,
-                namespace=namespace,
-                name=name,
-            )
-            try:
-                engine_status = response["status"]["engineStatus"]
-                exp_status = response["status"]["experiments"][0]["status"]
-            except Exception:
-                engine_status = "Not Initialized"
-                exp_status = "Not Initialized"
-            custom_object = ChaosEngine(
-                kind="ChaosEngine",
-                group=group,
-                namespace=namespace,
-                name=name,
-                plural=plural,
-                version=version,
-                engineStatus=engine_status,
-                expStatus=exp_status,
-            )
-        elif kind.lower() == "chaosresult":
-            plural = "chaosresults"
-            response = self.custom_object_client.get_namespaced_custom_object(
-                group=group,
-                plural=plural,
-                version=version,
-                namespace=namespace,
-                name=name,
-            )
-            try:
-                verdict = response["status"]["experimentStatus"]["verdict"]
-                fail_step = response["status"]["experimentStatus"]["failStep"]
-            except Exception:
-                verdict = "N/A"
-                fail_step = "N/A"
-            custom_object = ChaosResult(
-                kind="ChaosResult",
-                group=group,
-                namespace=namespace,
-                name=name,
-                plural=plural,
-                version=version,
-                verdict=verdict,
-                failStep=fail_step,
-            )
-        else:
-            logging.error("Invalid litmus chaos custom resource name")
-            custom_object = None
-        return custom_object
-
     def check_if_namespace_exists(self, name: str) -> bool:
         """
         Check if a namespace exists by parsing through
@@ -2596,89 +2515,6 @@ def collect_and_parse_cluster_events(
 
         return events
 
-    @deprecation.deprecated(
-        deprecated_in="3.1.0",
-        removed_in="3.2.0",
-        current_version=__version__,
-        details="replaced by `collect_and_parse_cluster_events`",
-    )
-    def collect_cluster_events(
-        self,
-        start_timestamp: int,
-        end_timestamp: int,
-        local_timezone: str,
-        cluster_timezone: str = "UTC",
-        limit: int = 500,
-        namespace: str = None,
-    ) -> Optional[str]:
-        """
-        Collects cluster events querying `/api/v1/events`
-        filtered in a given time interval and writes them in
-        a temporary file in json format.
-
-        :param start_timestamp: timestamp of the minimum date
-            after that the event is relevant
-        :param end_timestamp: timestamp of the maximum date
-            before that the event is relevant
-        :param local_timezone: timezone of the local system
-        :param cluster_timezone: timezone of the remote cluster
-        :param limit: limit of the number of events to be fetched
-            from the cluster
-        :param namespace: Namespace from which the events must be
-            collected, if None all-namespaces will be selected
-
-        """
-
-        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"
-
-            (data) = self.api_client.call_api(
-                path,
-                "GET",
-                path_params,
-                query_params,
-                header_params,
-                response_type="str",
-                auth_settings=auth_settings,
-            )
-            events = []
-            json_obj = ast.literal_eval(data[0])
-            events_list = reversed(json_obj["items"])
-            for obj in events_list:
-                filtered_obj = filter_dictionary(
-                    obj,
-                    "firstTimestamp",
-                    start_timestamp,
-                    end_timestamp,
-                    cluster_timezone,
-                    local_timezone,
-                )
-                if filtered_obj:
-                    events.append(filtered_obj)
-
-            if len(events) > 0:
-                file_content = json.dumps(events, indent=2)
-                with tempfile.NamedTemporaryFile(
-                    delete=False, mode="w"
-                ) as file:
-                    file.write(file_content)
-                    file.flush()
-                    return file.name
-            return None
-        except Exception as e:
-            logging.error(str(e))
-            return None
-
     def parse_events_from_file(
         self, events_filename: str
     ) -> Optional[list[ClusterEvent]]:

src/krkn_lib/models/k8s/models.py

@@ -1,9 +1,6 @@
 from concurrent.futures import Future, ThreadPoolExecutor
 from dataclasses import dataclass
 from typing import Optional
-from krkn_lib.version import __version__
-
-import deprecation
 
 
 @dataclass(frozen=True, order=False)
@@ -119,79 +116,6 @@ class Pod:
     """
 
 
-class LitmusChaosObject:
-    """
-    Data class to hold information regarding
-    a custom object of litmus project
-    """
-
-    @deprecation.deprecated(
-        deprecated_in="3.1.0",
-        removed_in="3.2.0",
-        current_version=__version__,
-        details="litmus support removed from krkn",
-    )
-    def __init__(self):
-        pass
-
-    kind: str
-    """
-    Litmus Object Kind
-    """
-    group: str
-    """
-    Api Group
-    """
-    namespace: str
-    """
-    Namespace where the object is deployed
-    """
-    name: str
-    """
-    Object name
-    """
-    plural: str
-    """
-    CRD plural
-    """
-    version: str
-    """
-    Version
-    """
-
-
-class ChaosEngine(LitmusChaosObject):
-    """
-    Data class to hold information
-    regarding a ChaosEngine object
-    """
-
-    engineStatus: str
-    """
-    Litmus Chaos engine status
-    """
-    expStatus: str
-    """
-    Litmus Chaos Engine experiment status
-    """
-
-
-class ChaosResult(LitmusChaosObject):
-    """
-    Data class to hold information
-    regarding a ChaosResult object
-    """
-
-    verdict: str
-    """
-    Verdict of the chaos experiment
-    """
-    failStep: str
-    """
-    Flag to show the failure step of the ChaosExperiment
-    """
-
-
 class ApiRequestException(Exception):
     """
     Generic API Exception raised by k8s package