Merge branch 'main' into count-lines-for-regex

.flake8

@@ -1,3 +1,4 @@
 [flake8]
 ignore = E501,W503,E203
-exclude = .git,__pycache__,venv
+disable = W391
+exclude = .git,__pycache__,venv,.venv

README.md

@@ -53,7 +53,7 @@ Install with:
 
 If you have cloned the git and want to develop locally, replace last step with:
 
-    python -m pip install --editable .
+    python -m pip install --editable .[dev]
 
 Running unit tests
 ------------------

core/opl/args.py

@@ -90,7 +90,12 @@ def add_kafka_opts(parser):
     parser.add_argument(
         "--kafka-host",
         default=os.getenv("KAFKA_HOST", "localhost"),
-        help="Kafka host (also use env variable KAFKA_HOST)",
+        help="Kafka host (also use env variable KAFKA_HOST). Can get overriden by --kafka-hosts arg or KAFKA_HOSTS envvar.",
+    )
+    parser.add_argument(
+        "--kafka-hosts",
+        default=os.getenv("KAFKA_HOSTS", ""),
+        help="Comma-separated list of hosts, including their ports (also use env variable KAFKA_HOSTS). Takes precedence over --kafka-host and --kafka-port or their envvar variants.",
     )
     parser.add_argument(
         "--kafka-port",
@@ -131,7 +136,7 @@ def add_kafka_opts(parser):
         help="The client is going to wait this much time for the server to respond to a request (also use env variable KAFKA_REQUEST_TIMEOUT_MS)",
     )
     parser.add_argument(
-        "--kafka-max_block_ms",
+        "--kafka-max-block-ms",
         type=int,
         default=int(os.getenv("KAFKA_MAX_BLOCK_MS", 60000)),
         help="Max time to block send e.g. because buffer is full (also use env variable KAFKA_MAX_BLOCK_MS)",

core/opl/cluster_read.py

@@ -11,10 +11,12 @@
 import jinja2.exceptions
 import boto3
 import urllib3
+import tempfile
 
 from . import data
 from . import date
-from tenacity import *  # noqa: F403
+from . import status_data
+from . import retry
 
 
 def execute(command):
@@ -179,8 +181,7 @@ def _sanitize_target(self, target):
         target = target.replace("$Cloud", self.args.grafana_prefix)
         return target
 
-    # pylint: disable-next=undefined-variable
-    @retry(stop=(stop_after_delay(10) | stop_after_attempt(10)))  # noqa: F405
+    @retry.retry_on_traceback(max_attempts=10, wait_seconds=1)
     def measure(self, ri, name, grafana_target):
         assert (
             ri.start is not None and ri.end is not None
@@ -378,10 +379,10 @@ def measure(self, ri, name, copy_from):
         """
         Just return value from previously answered item
         """
-        try:
-            return name, dict(ri._responses)[copy_from]
-        except KeyError:
+        if ri.sd is None:
             return name, None
+        else:
+            return name, ri.sd.get(copy_from)
 
 
 class TestFailMePlugin(BasePlugin):
@@ -455,7 +456,9 @@ def get_source(self, environment, path):
 
 
 class RequestedInfo:
-    def __init__(self, config, start=None, end=None, args=argparse.Namespace()):
+    def __init__(
+        self, config, start=None, end=None, args=argparse.Namespace(), sd=None
+    ):
         """
         "config" is input for config_stuff function
         "start" and "end" are datetimes needed if config file contains some
@@ -466,9 +469,9 @@ def __init__(self, config, start=None, end=None, args=argparse.Namespace()):
         self.start = start
         self.end = end
         self.args = args
+        self.sd = sd
 
         self._index = 0  # which config item are we processing?
-        self._responses = []  # what responses we have gave so far
         self._token = None  # OCP token - we will take it from `oc whoami -t` if needed
         self.measurement_plugins = (
             {}
@@ -512,7 +515,6 @@ def __next__(self):
                         f"Failed to measure {self.config[i]['name']}: {e}"
                     )
                     output = (None, None)
-                self._responses.append(output)
                 return output
             else:
                 raise Exception(f"Unknown config '{self.config[i]}'")
@@ -530,11 +532,14 @@ def doit(args):
     else:
         config = args.requested_info_config
 
+    sd = status_data.StatusData(tempfile.NamedTemporaryFile().name)
+
     requested_info = RequestedInfo(
         config,
         args.monitoring_start,
         args.monitoring_end,
         args=args,
+        sd=sd,
     )
 
     if args.render_config:

core/opl/data.py

@@ -217,13 +217,13 @@ def data_stats(data):
             "max": max(data),
             "sum": sum(data),
             "mean": statistics.mean(data),
-            "non_zero_mean": statistics.mean(non_zero_data)
-            if len(non_zero_data) > 0
-            else 0.0,
+            "non_zero_mean": (
+                statistics.mean(non_zero_data) if len(non_zero_data) > 0 else 0.0
+            ),
             "median": statistics.median(data),
-            "non_zero_median": statistics.median(non_zero_data)
-            if len(non_zero_data) > 0
-            else 0.0,
+            "non_zero_median": (
+                statistics.median(non_zero_data) if len(non_zero_data) > 0 else 0.0
+            ),
             "stdev": statistics.stdev(data) if len(data) > 1 else 0.0,
             "range": max(data) - min(data),
             "percentile25": q25,

core/opl/date.py

@@ -30,3 +30,17 @@ def my_fromisoformat(string):
         out = datetime.datetime.strptime(string, "%Y-%m-%dT%H:%M:%S")
     out = out.replace(tzinfo=string_tz)
     return out
+
+
+def get_now_str() -> str:
+    """
+    return current datetime in UTC string format
+    """
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def get_now() -> datetime.datetime:
+    """
+    return current datetime in UTC datetime format
+    """
+    return datetime.datetime.now(datetime.timezone.utc)

core/opl/retry.py

@@ -0,0 +1,43 @@
+import logging
+import time
+from functools import wraps
+
+
+def retry_on_traceback(max_attempts=10, wait_seconds=1):
+    """
+    Retries a function until it succeeds or the maximum number of attempts
+    or wait time is reached.
+
+    This is to mimic `@retry` decorator from Tenacity so we do not depend
+    on it.
+
+    Args:
+        max_attempts: The maximum number of attempts to retry the function.
+        wait_seconds: The number of seconds to wait between retries.
+
+    Returns:
+        A decorator that retries the wrapped function.
+    """
+    assert max_attempts >= 0, "It does not make sense to have less than 0 retries"
+    assert wait_seconds >= 0, "It does not make sense to wait les than 0 seconds"
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            attempt = 0
+            while True:
+                try:
+                    return func(*args, **kwargs)
+                except Exception as e:
+                    if attempt >= max_attempts:
+                        raise  # Reraise the exception after all retries are exhausted
+
+                    attempt += 1
+                    logging.debug(
+                        f"Retrying in {wait_seconds} seconds. Attempt {attempt}/{max_attempts} failed with: {e}"
+                    )
+                    time.sleep(wait_seconds)
+
+        return wrapper
+
+    return decorator

core/opl/status_data.py

@@ -231,7 +231,7 @@ def clear(self):
         """
         self._data = {
             "name": None,
-            "started": get_now_str(),
+            "started": date.get_now_str(),
             "ended": None,
             "owner": None,
             "result": None,
@@ -280,12 +280,6 @@ def _save(self, filename):
         logging.debug(f"Saved status data to {filename}")
 
 
-def get_now_str():
-    now = datetime.datetime.utcnow()
-    now = now.replace(tzinfo=datetime.timezone.utc)
-    return now.isoformat()
-
-
 def doit_set(status_data, set_this):
     for item in set_this:
         if item == "":
@@ -298,7 +292,7 @@ def doit_set(status_data, set_this):
             value = value[1:-1]
 
         if value == "%NOW%":
-            value = get_now_str()
+            value = date.get_now_str()
         else:
             try:
                 value = int(value)
@@ -347,6 +341,7 @@ def doit_additional(status_data, additional, monitoring_start, monitoring_end, a
         start=monitoring_start,
         end=monitoring_end,
         args=args,
+        sd=status_data,
     )
 
     counter_ok = 0

core/opl/status_data_updater.py

@@ -13,6 +13,7 @@
 import opl.status_data
 
 import requests
+import requests.adapters
 
 import tabulate
 
@@ -36,6 +37,16 @@
 }
 
 
+def get_session():
+    session = requests.Session()
+    retry_adapter = requests.adapters.HTTPAdapter(
+        max_retries=urllib3.Retry(total=None, connect=10, backoff_factor=1)
+    )
+    session.mount("https://", retry_adapter)
+    session.mount("http://", retry_adapter)
+    return session
+
+
 def _es_get_test(session, args, key, val, size=1, sort_by="started"):
     url = f"{args.es_server}/{args.es_index}/_search"
     headers = {
@@ -65,12 +76,23 @@ def _es_get_test(session, args, key, val, size=1, sort_by="started"):
         )
 
     if session is None:
-        session = requests.Session()
+        session = get_session()
 
     logging.info(
         f"Querying ES with url={url}, headers={headers} and json={json.dumps(data)}"
     )
-    response = session.get(url, headers=headers, json=data)
+    attempt = 0
+    attempt_max = 10
+    while True:
+        try:
+            response = session.get(url, headers=headers, json=data)
+        except requests.exceptions.ConnectionError:
+            if attempt >= attempt_max:
+                raise
+            attempt += 1
+            time.sleep(attempt)
+        else:
+            break
     response.raise_for_status()
     logging.debug(
         f"Got back this: {json.dumps(response.json(), sort_keys=True, indent=4)}"
@@ -358,7 +380,7 @@ def doit_rp_to_es(args):
         urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
     # Start a session
-    session = requests.Session()
+    session = get_session()
 
     # Get 10 newest launches
     launches = _get_rp_launches(session, args)
@@ -455,7 +477,7 @@ def doit_rp_to_dashboard_new(args):
         urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
     # Start a session
-    session = requests.Session()
+    session = get_session()
 
     run_id = args.dashboard_run_id
     result = args.dashboard_result
@@ -522,9 +544,20 @@ def _update_es_dashboard_result(session, args, es_id, result_string):
     if args.dry_run:
         logging.debug("Skipped because of dry-run")
     else:
-        response = session.post(
-            url, json=data, headers=headers, verify=not args.rp_noverify
-        )
+        attempt = 0
+        attempt_max = 10
+        while True:
+            try:
+                response = session.post(
+                    url, json=data, headers=headers, verify=not args.rp_noverify
+                )
+            except requests.exceptions.ConnectionError:
+                if attempt >= attempt_max:
+                    raise
+                attempt += 1
+                time.sleep(attempt)
+            else:
+                break
         response.raise_for_status()
         logging.debug(
             f"Got back this: {json.dumps(response.json(), sort_keys=True, indent=4)}"

core/setup.py

@@ -31,7 +31,6 @@
         "requests",
         "tabulate",
         "deepdiff",
-        "tenacity",
     ],
     package_data={
         "opl": [