test: fault tolerance tests

deploy/sdk/src/dynamo/sdk/cli/circus.py

@@ -86,6 +86,23 @@ def create_circus_watcher(
     use_sockets: bool = True,
     **kwargs: Any,
 ) -> Watcher:
+    log_dir = os.environ.get("DYN_CIRCUS_LOG_DIR", None)
+    if log_dir is not None:
+        prefix = f"{log_dir}/{name}"
+        os.makedirs(prefix, exist_ok=True)
+        stdout_stream = {
+            "class": "FileStream",
+            "filename": f"{prefix}/output.log",
+            "backup_count": 10,
+        }
+        stderr_stream = {
+            "class": "FileStream",
+            "filename": f"{prefix}/error.log",
+            "backup_count": 10,
+        }
+    else:
+        stdout_stream = None
+        stderr_stream = None
     return Watcher(
         name=name,
         cmd=shlex.quote(cmd) if psutil.POSIX else cmd,
@@ -94,7 +111,10 @@ def create_circus_watcher(
         stop_children=True,
         use_sockets=use_sockets,
         graceful_timeout=86400,
-        respawn=False,  # TODO
+        respawn=os.environ.get("DYN_CIRCUS_RESPAWN", "false").lower()
+        in ("true", "1", "yes"),
+        stdout_stream=stdout_stream,
+        stderr_stream=stderr_stream,
         **kwargs,
     )
 

pyproject.toml

@@ -162,6 +162,8 @@ markers = [
     "weekly: marks tests to run weekly",
     "gpu_1: marks tests to run on GPU",
     "gpu_2: marks tests to run on 2GPUs",
+    "gpu_4: marks tests to run on 4GPUs",
+    "gpu_8: marks tests to run on 8GPUs",
     "e2e: marks tests as end-to-end tests",
     "integration: marks tests as integration tests",
     "unit: marks tests as unit tests",

tests/conftest.py

@@ -15,6 +15,7 @@
 
 import logging
 import os
+import shutil
 import tempfile
 
 import pytest
@@ -23,15 +24,31 @@
 
 # Custom format inspired by your example
 LOG_FORMAT = "[TEST] %(asctime)s %(levelname)s %(name)s: %(message)s"
+DATE_FORMAT = "%Y-%m-%dT%H:%M:%S"
 
 # Configure logging
 logging.basicConfig(
     level=logging.INFO,
     format=LOG_FORMAT,
-    datefmt="%Y-%m-%dT%H:%M:%S",  # ISO 8601 UTC format
+    datefmt=DATE_FORMAT,  # ISO 8601 UTC format
 )
 
 
+@pytest.fixture(autouse=True)
+def logger(request):
+    log_path = os.path.join(request.node.name, "test.log.txt")
+    logger = logging.getLogger()
+    shutil.rmtree(request.node.name, ignore_errors=True)
+    os.makedirs(request.node.name, exist_ok=True)
+    handler = logging.FileHandler(log_path, mode="w")
+    formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT)
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    yield
+    handler.close()
+    logger.removeHandler(handler)
+
+
 def pytest_collection_modifyitems(config, items):
     """
     This function is called to modify the list of tests to run.
@@ -69,7 +86,7 @@ def __init__(self, request, port=2379, timeout=300):
             timeout=timeout,
             display_output=False,
             health_check_ports=[port],
-            data_dir=tempfile.mkdtemp(prefix="etcd_"),
+            data_dir=data_dir,
             log_dir=request.node.name,
         )