fix: add program abort on k8s_service_watcher fail

Signed-off-by: Alexis Asseman <alexis@semiotic.ai>

autoagora/k8s_service_watcher.py

@@ -8,6 +8,8 @@
 from kubernetes.client.api_client import ApiClient
 from kubernetes.client.rest import ApiException
 
+from autoagora.misc import async_exit_on_exception
+
 
 class K8SServiceEndpointsWatcher:
     def __init__(self, service_name: str) -> None:
@@ -47,6 +49,7 @@ def __init__(self, service_name: str) -> None:
         # Starts the async _loop immediately
         self._future = aio.ensure_future(self._watch_loop())
 
+    @async_exit_on_exception()
     async def _watch_loop(self) -> None:
         """Restarts the k8s watch on expiration."""
         while True:

autoagora/misc.py

@@ -0,0 +1,26 @@
+import functools
+import logging
+
+
+def async_exit_on_exception(exit_code: int = -1):
+    """Returns decorator that logs any exception and exits the program immediately.
+
+    The goal of this function is to easily trigger an immediate program abort from any
+    asynchronous function.
+
+    Args:
+        exit_code (int, optional): Self explanatory. Defaults to -1.
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs):
+            try:
+                return await func(*args, **kwargs)
+            except:
+                logging.exception("exit_on_exception triggered")
+                exit(exit_code)
+
+        return wrapper
+
+    return decorator