Ssh Test and Debugging Fixups

.github/workflows/devcontainer.yml

@@ -208,13 +208,6 @@ jobs:
         # Now actually run the tests.
         docker exec --user vscode --env USER=vscode mlos-devcontainer make CONDA_INFO_LEVEL=-v test
 
-    - name: Upload coverage report as build artifact
-      if: always() # ensures it runs even if tests fail
-      uses: actions/upload-artifact@v4
-      with:
-        name: coverage-report
-        path: coverage.xml
-
     - name: Generate and test binary distribution files
       timeout-minutes: 10
       run: |
@@ -298,6 +291,21 @@ jobs:
         docker tag mlos-devcontainer:latest ${{ secrets.ACR_LOGINURL }}/mlos-devcontainer:$image_tag
         docker push ${{ secrets.ACR_LOGINURL }}/mlos-devcontainer:$image_tag
 
+    # ensure these run even if tests fail, but only after everything else is done or skipped
+    - name: Upload coverage report as build artifact
+      if: always()
+      uses: actions/upload-artifact@v5
+      with:
+        name: coverage-report
+        path: coverage.xml
+
+    - name: Upload pytest logs
+      if: always()
+      uses: actions/upload-artifact@v5
+      with:
+        name: devcontainer-pytest-logs
+        path: logs/pytest*.log
+        retention-days: 7
 
   PublishDocs:
     name: Publish Documentation

.github/workflows/linux.yml

@@ -181,3 +181,11 @@ jobs:
 
     - name: Generate and test binary distribution files
       run: make CONDA_ENV_NAME=$CONDA_ENV_NAME CONDA_INFO_LEVEL=-v dist dist-test
+
+    - name: Upload pytest logs
+      if: always()
+      uses: actions/upload-artifact@v5
+      with:
+        name: linux-${{ matrix.python_version }}-pytest-logs
+        path: logs/pytest*.log
+        retention-days: 7

.github/workflows/macos.yml

@@ -163,6 +163,13 @@ jobs:
     - name: Generate and test binary distribution files
       run: make CONDA_ENV_NAME=$CONDA_ENV_NAME CONDA_INFO_LEVEL=-v dist dist-test
 
+    - name: Upload pytest logs
+      if: always()
+      uses: actions/upload-artifact@v5
+      with:
+        name: macos-pytest-logs
+        path: logs/pytest*.log
+        retention-days: 7
 
   MacOSDevContainerBuildTest:
     name: MacOS DevContainer Build/Test

.github/workflows/windows.yml

@@ -142,6 +142,13 @@ jobs:
       run: |
         .github/workflows/build-dist-test.ps1
 
+    - name: Upload pytest logs
+      if: always()
+      uses: actions/upload-artifact@v5
+      with:
+        name: windows-pytest-logs
+        path: logs/pytest*.log
+        retention-days: 7
 
   WindowsDevContainerBuildTest:
     name: Windows DevContainer Build/Test

.gitignore

@@ -171,3 +171,6 @@ build/*.build-stamp
 *.duckdb
 *.db.wal
 *.duckdb.wal
+
+# pytest logs
+logs/pytest*.log

conftest.py

@@ -7,6 +7,7 @@
 # Note: This file is named conftest.py so that pytest picks it up automatically
 # without the need to adjust PYTHONPATH or sys.path as much.
 
+import logging
 import os
 import shutil
 from tempfile import mkdtemp
@@ -15,6 +16,14 @@
 import pytest
 from xdist.workermanage import WorkerController
 
+# See Also: setup.cfg and launcher.py
+LOG_FMT = (
+    "%(asctime)s.%(msecs)03d [%(process)d][%(threadName)s] "
+    "[%(filename)s:%(lineno)d %(funcName)s] "
+    "%(levelname)s: %(message)s"
+)
+DATE_FMT = "%Y-%m-%d %H:%M:%S"
+
 
 def is_master(config: pytest.Config) -> bool:
     """True if the code running the given pytest.config object is running in a xdist
@@ -56,6 +65,20 @@ def pytest_configure(config: pytest.Config) -> None:
         # Add it to the config so that it can passed to the worker nodes.
         setattr(config, "shared_temp_dir", mkdtemp())
 
+    # Configure per-worker log file.
+    worker_id = getattr(config, "workerinput", {}).get("workerid", "master")
+    os.makedirs("logs", exist_ok=True)
+    log_file_path = os.path.join("logs", f"pytest-{worker_id}.log")
+    file_handler = logging.FileHandler(log_file_path, mode="w")
+    if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
+        file_handler.setLevel(logging.DEBUG)
+    else:
+        file_handler.setLevel(logging.INFO)
+    # logging.basicConfig(level=file_handler.level, format=LOG_FMT, datefmt=DATE_FMT)
+    log_formatter = logging.Formatter(fmt=LOG_FMT, datefmt=DATE_FMT)
+    file_handler.setFormatter(log_formatter)
+    logging.getLogger().addHandler(file_handler)
+
 
 def pytest_configure_node(node: WorkerController) -> None:
     """Xdist hook used to inform workers of the location of the shared temp dir."""

mlos_bench/mlos_bench/launcher.py

@@ -37,6 +37,14 @@
 _LOG_FORMAT = "%(asctime)s %(filename)s:%(lineno)d %(funcName)s %(levelname)s %(message)s"
 logging.basicConfig(level=_LOG_LEVEL, format=_LOG_FORMAT)
 
+# TODO: Future PR: See Also: /conftest.py, setup.cfg
+# _LOG_FORMAT = (
+#    "%(asctime)s.%(msecs)03d [%(process)d][%(threadName)s] "
+#    "[%(filename)s:%(lineno)d %(funcName)s] "
+#    "%(levelname)s %(message)s"
+# )
+# logging.basicConfig(level=_LOG_LEVEL, format=_LOG_FORMAT, datefmt="%Y-%m-%d %H:%M:%S")
+
 _LOG = logging.getLogger(__name__)
 
 

mlos_bench/mlos_bench/services/remote/ssh/ssh_service.py

@@ -9,6 +9,7 @@
 from abc import ABCMeta
 from asyncio import Event as CoroEvent
 from asyncio import Lock as CoroLock
+from asyncio import sleep as async_sleep
 from collections.abc import Callable, Coroutine
 from threading import current_thread
 from types import TracebackType
@@ -172,34 +173,60 @@ async def get_client_connection(
             A tuple of (SSHClientConnection, SshClient).
         """
         _LOG.debug("%s: get_client_connection: %s", current_thread().name, connect_params)
-        async with self._cache_lock:
-            connection_id = SshClient.id_from_params(connect_params)
-            client: None | SshClient | asyncssh.SSHClient
-            _, client = self._cache.get(connection_id, (None, None))
-            if client:
-                _LOG.debug("%s: Checking cached client %s", current_thread().name, connection_id)
-                connection = await client.connection()
-                if not connection:
-                    _LOG.debug(
-                        "%s: Removing stale client connection %s from cache.",
-                        current_thread().name,
-                        connection_id,
-                    )
-                    self._cache.pop(connection_id)
-                    # Try to reconnect next.
-                else:
-                    _LOG.debug("%s: Using cached client %s", current_thread().name, connection_id)
-            if connection_id not in self._cache:
-                _LOG.debug(
-                    "%s: Establishing client connection to %s",
+        connection_id = SshClient.id_from_params(connect_params)
+        for i in range(3):  # TODO: make the retry count configurable
+            try:
+                async with self._cache_lock:
+                    client: None | SshClient | asyncssh.SSHClient
+                    _, client = self._cache.get(connection_id, (None, None))
+                    if client:
+                        _LOG.debug(
+                            "%s: Checking cached client %s", current_thread().name, connection_id
+                        )
+                        connection = await client.connection()
+                        if not connection:
+                            _LOG.debug(
+                                "%s: Removing stale client connection %s from cache.",
+                                current_thread().name,
+                                connection_id,
+                            )
+                            self._cache.pop(connection_id)
+                            # Try to reconnect next.
+                        else:
+                            _LOG.debug(
+                                "%s: Using cached client %s", current_thread().name, connection_id
+                            )
+                    if connection_id not in self._cache:
+                        _LOG.debug(
+                            "%s: Establishing client connection to %s",
+                            current_thread().name,
+                            connection_id,
+                        )
+                        connection, client = await asyncssh.create_connection(
+                            SshClient, **connect_params
+                        )
+                        assert isinstance(client, SshClient)
+                        self._cache[connection_id] = (connection, client)
+                        _LOG.debug(
+                            "%s: Created connection to %s.", current_thread().name, connection_id
+                        )
+                    return self._cache[connection_id]
+            except ConnectionRefusedError as ex:  # TODO: Add other error handling here too
+                _LOG.warning(
+                    "%s: Attempt %d: Failed to connect to %s: %s",
                     current_thread().name,
+                    i + 1,
                     connection_id,
+                    ex,
                 )
-                connection, client = await asyncssh.create_connection(SshClient, **connect_params)
-                assert isinstance(client, SshClient)
-                self._cache[connection_id] = (connection, client)
-                _LOG.debug("%s: Created connection to %s.", current_thread().name, connection_id)
-            return self._cache[connection_id]
+                if i < 2:  # TODO: adjust to match max range
+                    await async_sleep(1.0)  # TODO: Make this configurable
+                if i == 2:  # TODO: adjust to match max range
+                    _LOG.error(
+                        "%s: Giving up connecting to %s", current_thread().name, connection_id
+                    )
+                    raise
+        raise RuntimeError("Unreachable code in get_client_connection")
 
     def cleanup(self) -> None:
         """Closes all cached connections."""

mlos_bench/mlos_bench/tests/__init__.py

@@ -9,6 +9,7 @@
 """
 import filecmp
 import json
+import logging
 import os
 import shutil
 import socket
@@ -24,6 +25,8 @@
 
 from mlos_bench.util import get_class_from_name, nullable
 
+_LOG = logging.getLogger(__name__)
+
 ZONE_NAMES = [
     # Explicit time zones.
     "UTC",
@@ -171,11 +174,24 @@ def wait_docker_service_healthy(
 
 def wait_docker_service_socket(docker_services: DockerServices, hostname: str, port: int) -> None:
     """Wait until a docker service is ready."""
-    docker_services.wait_until_responsive(
-        check=lambda: check_socket(hostname, port),
-        timeout=60.0,
-        pause=0.5,
-    )
+    _LOG.info("Waiting for %s:%d to become responsive", hostname, port)
+
+    def check_with_logging() -> bool:
+        result = check_socket(hostname, port)
+        if not result:
+            _LOG.debug("Socket check failed for %s:%d", hostname, port)
+        return result
+
+    try:
+        docker_services.wait_until_responsive(
+            check=check_with_logging,
+            timeout=60.0,
+            pause=0.5,
+        )
+        _LOG.info("Socket %s:%d is now responsive", hostname, port)
+    except Exception as e:
+        _LOG.error("Failed waiting for %s:%d: %s", hostname, port, e)
+        raise
 
 
 def check_socket(host: str, port: int, timeout: float = 1.0) -> bool:
@@ -192,10 +208,27 @@ def check_socket(host: str, port: int, timeout: float = 1.0) -> bool:
     -------
     bool
     """
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-        sock.settimeout(timeout)  # seconds
-        result = sock.connect_ex((host, port))
-        return result == 0
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.settimeout(timeout)  # seconds
+            result = sock.connect_ex((host, port))
+            success = result == 0
+            if not success:
+                _LOG.debug(
+                    "Socket connection to %s:%d failed with code %d",
+                    host,
+                    port,
+                    result,
+                )
+            return success
+    except (OSError, TimeoutError) as e:
+        _LOG.debug(
+            "Socket check exception for %s:%d: %s",
+            host,
+            port,
+            e,
+        )
+        return False
 
 
 def resolve_host_name(host: str) -> str | None: