persist error functionality

Signed-off-by: adarsh0728 <gooneriitk@gmail.com>

Author: adarsh0728

pynumaflow/_constants.py

@@ -7,9 +7,11 @@
 SIDE_INPUT_DIR_PATH = "/var/numaflow/side-inputs"
 ENV_UD_CONTAINER_TYPE = "NUMAFLOW_UD_CONTAINER_TYPE"
 
-# Get container type from env var, default to unknown-container
+# Error Constants
+RUNTIME_APPLICATION_ERRORS_PATH = "/var/numaflow/runtime/application-errors"
+CURRENT_FILE = "current-udf.json"
+INTERNAL_ERROR = "Internal error"
 CONTAINER_TYPE = os.getenv(ENV_UD_CONTAINER_TYPE, "unknown-container")
-# UDF exception error string with container type
 ERR_UDF_EXCEPTION_STRING = f"UDF_EXECUTION_ERROR({CONTAINER_TYPE})"
 
 # Socket configs

pynumaflow/errors/__init__.py

@@ -0,0 +1,3 @@
+from pynumaflow.errors.errors import persist_critical_error
+
+__all__ = ["persist_critical_error"]

pynumaflow/errors/_dtypes.py

@@ -0,0 +1,18 @@
+class _RuntimeErrorEntry:
+    """Represents a runtime error entry to be persisted."""
+
+    def __init__(self, container: str, timestamp: int, code: str, message: str, details: str):
+        self.container = container
+        self.timestamp = timestamp
+        self.code = code
+        self.message = message
+        self.details = details
+
+    def to_dict(self) -> dict:
+        return {
+            "container": self.container,
+            "timestamp": self.timestamp,
+            "code": self.code,
+            "message": self.message,
+            "details": self.details,
+        }

pynumaflow/errors/errors.py

@@ -0,0 +1,82 @@
+import os
+import json
+import threading
+import time
+from pathlib import Path
+from pynumaflow._constants import (
+    CONTAINER_TYPE,
+    RUNTIME_APPLICATION_ERRORS_PATH,
+    CURRENT_FILE,
+    INTERNAL_ERROR,
+)
+from pynumaflow.errors._dtypes import _RuntimeErrorEntry
+
+
+class _PersistErrorOnce:
+    """Ensures that the persist_critical_error function is executed only once."""
+
+    def __init__(self):
+        self.done = False
+        self.lock = threading.Lock()
+
+    def execute(self, func, *args, **kwargs):
+        with self.lock:
+            if self.done:
+                raise RuntimeError("Persist critical error function has already been executed.")
+            self.done = True
+            return func(*args, **kwargs)
+
+
+_persist_error_once = _PersistErrorOnce()
+
+
+def persist_critical_error(
+    error_code: str, error_message: str, error_details: str
+) -> RuntimeError | None:
+    """
+    Persists a critical error to a file. This function will only execute once.
+    Logs the error if persisting to the file fails.
+    Returns None if successful, or raises RuntimeError if already executed.
+    """
+    try:
+        _persist_error_once.execute(
+            _persist_critical_error_to_file,
+            error_code,
+            error_message,
+            error_details,
+            RUNTIME_APPLICATION_ERRORS_PATH,
+        )
+    except RuntimeError as e:
+        return e
+    except Exception as e:
+        print(f"Error in persisting critical error: {e}")
+    return None
+
+
+def _persist_critical_error_to_file(
+    error_code: str, error_message: str, error_details: str, dir_path: str
+):
+    """Internal function to persist a critical error to a file."""
+
+    os.makedirs(dir_path, mode=0o777, exist_ok=True)
+    container_dir = os.path.join(dir_path, CONTAINER_TYPE)
+    os.makedirs(container_dir, mode=0o777, exist_ok=True)
+
+    current_file_path = os.path.join(container_dir, CURRENT_FILE)
+    error_code = error_code or INTERNAL_ERROR
+    current_timestamp = int(time.time())
+
+    runtime_error_entry = _RuntimeErrorEntry(
+        container=CONTAINER_TYPE,
+        timestamp=current_timestamp,
+        code=error_code,
+        message=error_message,
+        details=error_details,
+    )
+
+    with open(current_file_path, "w") as f:
+        json.dump(runtime_error_entry.to_dict(), f)
+
+    final_file_name = f"{current_timestamp}-udf.json"
+    final_file_path = os.path.join(container_dir, final_file_name)
+    os.rename(current_file_path, final_file_path)