Refactor and formatting

Signed-off-by: Raul Sevilla <[email protected]>

Author: rsevilla87

fmatch/logrus.py

@@ -17,27 +17,27 @@ class SingletonLogger:
 
     def __new__(cls, debug: int, name: str):
         if (not cls.instance) or name not in cls.instance:
-            cls.instance[name] = cls._initialize_logger(debug,name)
+            cls.instance[name] = cls._initialize_logger(debug, name)
         return cls.instance[name]
 
     @staticmethod
     def _initialize_logger(debug: int, name: str) -> logging.Logger:
         level = debug  # if debug else logging.INFO
         logger = logging.getLogger(name)
-        logger.propagate=False
+        logger.propagate = False
         if not logger.hasHandlers():
             logger.setLevel(level)
             handler = logging.StreamHandler(sys.stdout)
             handler.setLevel(level)
             formatter = logging.Formatter(
-                "%(asctime)s - %(name)-10s - %(levelname)s - file: %(filename)s - line: %(lineno)d - %(message)s" # pylint: disable = line-too-long
+                "%(asctime)s - %(name)-10s - %(levelname)s - file: %(filename)s - line: %(lineno)d - %(message)s"  # pylint: disable = line-too-long
             )
             handler.setFormatter(formatter)
             logger.addHandler(handler)
         return logger
 
     @classmethod
-    def getLogger(cls, name:str) -> logging.Logger:
+    def getLogger(cls, name: str) -> logging.Logger:
         """Return logger in instance
 
         Args:

fmatch/matcher.py

@@ -28,87 +28,73 @@ def __init__(
         verify_certs: bool = True,
     ):
         self.index = index
-        self.es_url = ES_URL
         self.search_size = 10000
         self.logger = SingletonLogger(debug=level, name="Matcher")
-        self.es = Elasticsearch([self.es_url], timeout=30, verify_certs=verify_certs)
-        self.data = None
+        self.es = Elasticsearch([ES_URL], timeout=30, verify_certs=verify_certs)
 
-    def get_metadata_by_uuid(self, uuid: str, index: str = None) -> dict:
+    def get_metadata_by_uuid(self, uuid: str) -> dict:
         """Returns back metadata when uuid is given
 
         Args:
             uuid (str): uuid of the run
-            index (str, optional): index to be searched in. Defaults to None.
 
         Returns:
             _type_: _description_
         """
-        if index is None:
-            index = self.index
         query = Q("match", uuid=uuid)
         result = {}
-        s = Search(using=self.es, index=index).query(query)
-        res = self.query_index(index, s)
+        s = Search(using=self.es, index=self.index).query(query)
+        res = self.query_index(s)
         hits = res.hits.hits
         if hits:
             result = dict(hits[0].to_dict()["_source"])
         return result
 
-    def query_index(self, index: str, search: Search) -> Response:
+    def query_index(self, search: Search) -> Response:
         """generic query function
 
         Args:
-            index (str): _description_
             search (Search) : Search object with query
         """
-        self.logger.info("Executing query against index=%s", index)
+        self.logger.info("Executing query against index: %s", self.index)
         self.logger.debug("Executing query \r\n%s", search.to_dict())
         return search.execute()
 
     def get_uuid_by_metadata(
         self,
         meta: Dict[str, Any],
-        index: str = None,
         lookback_date: datetime = None,
         lookback_size: int = 10000,
     ) -> List[Dict[str, str]]:
         """gets uuid by metadata
 
         Args:
             meta (Dict[str, Any]): metadata of the runs
-            index (str, optional): Index to search. Defaults to None.
-            lookback_date (datetime, optional): 
+            lookback_date (datetime, optional):
             The cutoff date to get the uuids from. Defaults to None.
-            lookback_size (int, optional): 
+            lookback_size (int, optional):
             Maximum number of runs to get, gets the latest. Defaults to 10000.
 
-            lookback_size and lookback_date get the data on the 
+            lookback_size and lookback_date get the data on the
             precedency of whichever cutoff comes first.
             Similar to a car manufacturer's warranty limits.
 
         Returns:
             List[Dict[str, str]]: _description_
         """
+        must_clause = []
         must_not_clause = []
-        if index is None:
-            index = self.index
 
         version = meta["ocpVersion"][:4]
 
-        must_clause = [
-            (
-                Q("match", **{field: str(value)})
-                if isinstance(value, str)
-                else Q("match", **{field: value})
-            )
-            for field, value in meta.items()
-            if field not in "ocpVersion"
-            if field not in "ocpMajorVersion"
-        ]
-
-        for field, value in meta.get("not", {}).items():
-            must_not_clause.append(Q("match", **{field: str(value)}))
+        for field, value in meta.items():
+            if field in ["ocpVersion", "ocpMajorVersion"]:
+                continue
+            if field != "not":
+                must_clause.append(Q("match", **{field: str(value)}))
+            else:
+                for not_field, not_value in meta["not"].items():
+                    must_not_clause.append(Q("match", **{not_field: str(not_value)}))
 
         if "ocpMajorVersion" in meta:
             version = meta["ocpMajorVersion"]
@@ -130,26 +116,32 @@ def get_uuid_by_metadata(
             filter=filter_clause,
         )
         s = (
-            Search(using=self.es, index=index)
+            Search(using=self.es, index=self.index)
             .query(query)
             .sort({"timestamp": {"order": "desc"}})
             .extra(size=lookback_size)
         )
-        result = self.query_index(index, s)
+        result = self.query_index(s)
         hits = result.hits.hits
         uuids_docs = []
         for hit in hits:
             if "buildUrl" in hit["_source"]:
-                uuids_docs.append({
+                uuids_docs.append(
+                    {
                         "uuid": hit.to_dict()["_source"]["uuid"],
-                        "buildUrl": hit.to_dict()["_source"]["buildUrl"]})
+                        "buildUrl": hit.to_dict()["_source"]["buildUrl"],
+                    }
+                )
             else:
-                uuids_docs.append({
+                uuids_docs.append(
+                    {
                         "uuid": hit.to_dict()["_source"]["uuid"],
-                        "buildUrl": "http://bogus-url"})
+                        "buildUrl": "http://bogus-url",
+                    }
+                )
         return uuids_docs
 
-    def match_kube_burner(self, uuids: List[str], index: str) -> List[Dict[str, Any]]:
+    def match_kube_burner(self, uuids: List[str]) -> List[Dict[str, Any]]:
         """match kube burner runs
         Args:
             uuids (list): list of uuids
@@ -165,9 +157,11 @@ def match_kube_burner(self, uuids: List[str], index: str) -> List[Dict[str, Any]
             ],
         )
         search = (
-            Search(using=self.es, index=index).query(query).extra(size=self.search_size)
+            Search(using=self.es, index=self.index)
+            .query(query)
+            .extra(size=self.search_size)
         )
-        result = self.query_index(index, search)
+        result = self.query_index(search)
         runs = [item.to_dict()["_source"] for item in result.hits.hits]
         return runs
 
@@ -188,16 +182,15 @@ def filter_runs(self, pdata: Dict[Any, Any], data: Dict[Any, Any]) -> List[str]:
         ids_df = ndf.loc[df["jobConfig.jobIterations"] == iterations]
         return ids_df["uuid"].to_list()
 
-    def getResults(
-        self, uuid: str, uuids: List[str], index_str: str, metrics: Dict[str, Any]
+    def get_results(
+        self, uuid: str, uuids: List[str], metrics: Dict[str, Any]
     ) -> Dict[Any, Any]:
         """
         Get results of elasticsearch data query based on uuid(s) and defined metrics
 
         Args:
             uuid (str): _description_
             uuids (list): _description_
-            index_str (str): _description_
             metrics (dict): _description_
 
         Returns:
@@ -224,22 +217,19 @@ def getResults(
             ],
         )
         search = (
-            Search(using=self.es, index=index_str)
+            Search(using=self.es, index=self.index)
             .query(query)
             .extra(size=self.search_size)
         )
-        result = self.query_index(index_str, search)
+        result = self.query_index(search)
         runs = [item.to_dict()["_source"] for item in result.hits.hits]
         return runs
 
-    def get_agg_metric_query(
-        self, uuids: List[str], index: str, metrics: Dict[str, Any]
-    ):
+    def get_agg_metric_query(self, uuids: List[str], metrics: Dict[str, Any]):
         """burner_metric_query will query for specific metrics data.
 
         Args:
             uuids (list): List of uuids
-            index (str): ES/OS Index to query from
             metrics (dict): metrics defined in es index metrics
         """
         metric_queries = []
@@ -261,7 +251,9 @@ def get_agg_metric_query(
             ],
         )
         search = (
-            Search(using=self.es, index=index).query(query).extra(size=self.search_size)
+            Search(using=self.es, index=self.index)
+            .query(query)
+            .extra(size=self.search_size)
         )
         agg_value = metrics["agg"]["value"]
         agg_type = metrics["agg"]["agg_type"]
@@ -271,7 +263,7 @@ def get_agg_metric_query(
         search.aggs.bucket(
             "uuid", "terms", field="uuid.keyword", size=self.search_size
         ).metric(agg_value, agg_type, field=metrics["metric_of_interest"])
-        result = self.query_index(index, search)
+        result = self.query_index(search)
         data = self.parse_agg_results(result, agg_value, agg_type)
         return data
 

fmatch/test_fmatch.py

@@ -5,6 +5,7 @@
 from datetime import datetime
 import sys
 import warnings
+
 # pylint: disable=import-error
 import pandas as pd
 
@@ -17,7 +18,7 @@
 )
 
 match = Matcher(index="perf_scale_ci*", verify_certs=False)
-res=match.get_metadata_by_uuid("b4afc724-f175-44d1-81ff-a8255fea034f",'perf_scale_ci*')
+res = match.get_metadata_by_uuid("b4afc724-f175-44d1-81ff-a8255fea034f")
 
 meta = {}
 meta["masterNodesType"] = "m6a.xlarge"
@@ -34,15 +35,16 @@
 # meta['fips'] = "false"
 
 uuids = match.get_uuid_by_metadata(meta)
-print("All uuids",len(uuids))
-date= datetime.strptime("2024-07-01T13:46:24Z","%Y-%m-%dT%H:%M:%SZ")
-uuids2= match.get_uuid_by_metadata(meta,lookback_date=date)
-print("lookback uuids",len(uuids2))
+print("All uuids", len(uuids))
+date = datetime.strptime("2024-07-01T13:46:24Z", "%Y-%m-%dT%H:%M:%SZ")
+uuids2 = match.get_uuid_by_metadata(meta, lookback_date=date)
+print("lookback uuids", len(uuids2))
 uuids2 = match.get_uuid_by_metadata(meta)
 if len(uuids) == 0:
     print("No UUID present for given metadata")
     sys.exit()
-runs = match.match_kube_burner(uuids,"ripsaw-kube-burner*")
+match = Matcher(index="ripsaw-kube-burner*", verify_certs=False)
+runs = match.match_kube_burner(uuids)
 
 ids = match.filter_runs(runs, runs)
 podl_metrics = {
@@ -52,25 +54,25 @@
     "metric_of_interest": "P99",
     "not": {"jobConfig.name": "garbage-collection"},
 }
-podl = match.getResults("", ids, "ripsaw-kube-burner*",metrics=podl_metrics)
+podl = match.get_results("", ids, metrics=podl_metrics)
 kapi_metrics = {
     "name": "apiserverCPU",
     "metricName": "containerCPU",
     "labels.namespace.keyword": "openshift-kube-apiserver",
     "metric_of_interest": "value",
     "agg": {"value": "cpu", "agg_type": "avg"},
 }
-kapi_cpu = match.get_agg_metric_query(ids, "ripsaw-kube-burner*", metrics=kapi_metrics)
+kapi_cpu = match.get_agg_metric_query(ids, metrics=kapi_metrics)
 podl_df = match.convert_to_df(
-    podl, columns=['uuid', 'timestamp', 'quantileName', 'P99'])
+    podl, columns=["uuid", "timestamp", "quantileName", "P99"]
+)
 kapi_cpu_df = match.convert_to_df(kapi_cpu)
 merge_df = pd.merge(kapi_cpu_df, podl_df, on="uuid")
-match.save_results(merge_df, "merged.csv", [
-                   "uuid", "timestamp_x", "cpu_avg", "P99"])
+match.save_results(merge_df, "merged.csv", ["uuid", "timestamp_x", "cpu_avg", "P99"])
 
 df = pd.read_csv("merged.csv")
 ls = df["uuid"].to_list()
 # Check merged csv data - Debug
 for i in ls:
     # Debug - Ensure they are all using the same networkType
-    print(match.get_metadata_by_uuid(i)['networkType'])
+    print(match.get_metadata_by_uuid(i)["networkType"])