canary-load: keep going on any exception, fail at the end if any collected

test/canary-load/mzcompose.py

@@ -13,15 +13,14 @@
 import os
 import threading
 import time
+import traceback
 import urllib.parse
 from collections.abc import Callable
 from textwrap import dedent
 
 import psycopg
 import requests
 from psycopg import Connection, Cursor
-from psycopg.errors import IdleInTransactionSessionTimeout, OperationalError
-from requests.exceptions import ConnectionError, ReadTimeout
 
 from materialize.cloudtest.util.jwt_key import fetch_jwt
 from materialize.mz_env_util import (
@@ -56,13 +55,32 @@
     "consuming input failed: SSL SYSCALL error: EOF detected",
     "unexpected eof while reading",
     "consuming input failed: SSL connection has been closed unexpectedly",
+    # The agent's resolver, not the environment: seen as
+    # `NameResolutionError(... [Errno -3] Temporary failure in name resolution)`
+    # wrapped in a requests.ConnectionError.
+    "Temporary failure in name resolution",
 ]
 
 
 def is_connection_error(msg: str) -> bool:
     return any(s in msg for s in CONNECTION_ERROR_STRINGS)
 
 
+def failure_messages(e: Exception) -> list[tuple[str, str | None]]:
+    """Every (message, details) pair an exception escaping a chunk stands for.
+
+    A FailedTestExecutionError carries one per testdrive error; anything else
+    is one entry, with the traceback as details for exception types the test
+    did not anticipate.
+    """
+    if isinstance(e, FailedTestExecutionError):
+        assert len(e.errors) > 0, "Exception contains no errors"
+        return [(error.message, error.details) for error in e.errors]
+    if isinstance(e, CommandFailureCausedUIError):
+        return [((e.stdout or "") + (e.stderr or ""), None)]
+    return [(f"{type(e).__name__}: {e}", traceback.format_exc())]
+
+
 class WebhookSender:
     """Sends one webhook per second with the current timestamp in a background thread."""
 
@@ -193,58 +211,32 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
                     )
                     close_connection_and_cursor(conn2, cursor_on_mv, "subscribe_mv")
 
-                except (
-                    OperationalError,
-                    ReadTimeout,
-                    ConnectionError,
-                    IdleInTransactionSessionTimeout,
-                ) as e:
-                    error_msg_str = str(e)
-                    if is_connection_error(error_msg_str):
+                except Exception as e:
+                    # Whatever escaped the chunk, the run keeps going: a
+                    # connection error restarts the chunk and stays out of the
+                    # verdict, anything else is collected and rethrown at the
+                    # end. Nothing aborts the run any more; an abort used to
+                    # skip the rethrow below and lose every failure collected
+                    # before it.
+                    for message, details in failure_messages(e):
+                        # TODO(def-): Remove when database-issues#6825 is fixed
+                        if "Non-positive multiplicity in DistinctBy" in message:
+                            continue
                         now = datetime.datetime.now(
                             tz=datetime.timezone.utc
                         ).isoformat()
-                        connection_failures.append((now, error_msg_str))
-                        print(f"Connection failure at {now}: {e}; retrying")
-                    else:
-                        raise
-                except FailedTestExecutionError as e:
-                    assert len(e.errors) > 0, "Exception contains no errors"
-                    for error in e.errors:
-                        # TODO(def-): Remove when database-issues#6825 is fixed
-                        if "Non-positive multiplicity in DistinctBy" in error.message:
-                            continue
-                        if is_connection_error(error.message):
-                            now = datetime.datetime.now(
-                                tz=datetime.timezone.utc
-                            ).isoformat()
-                            connection_failures.append((now, error.message))
+                        if is_connection_error(message):
+                            connection_failures.append((now, message))
                             print(
-                                f"Connection failure at {now}: {error.message}; continuing."
+                                f"Connection failure at {now}: {message}; restarting chunk."
                             )
                             continue
                         print(
-                            f"Test failure occurred ({error.message}), collecting it, and continuing."
+                            f"Test failure occurred ({message}), collecting it, and continuing."
+                        )
+                        failures.append(
+                            TestFailureDetails(message=message, details=details)
                         )
-                        # collect, continue, and rethrow at the end
-                        failures.append(error)
-                except CommandFailureCausedUIError as e:
-                    msg = (e.stdout or "") + (e.stderr or "")
-                    # TODO(def-): Remove when database-issues#6825 is fixed
-                    if "Non-positive multiplicity in DistinctBy" in msg:
-                        continue
-                    if is_connection_error(msg):
-                        now = datetime.datetime.now(
-                            tz=datetime.timezone.utc
-                        ).isoformat()
-                        connection_failures.append((now, msg))
-                        print(f"Connection failure at {now}: {msg}; continuing.")
-                        continue
-                    print(
-                        f"Test failure occurred ({msg}), collecting it, and continuing."
-                    )
-                    # collect, continue, and rethrow at the end
-                    failures.append(TestFailureDetails(message=msg, details=None))
         finally:
             webhook_sender.stop()
 
@@ -274,6 +266,20 @@ def fetch_token(user_name: str, password: str) -> str:
     )
 
 
+def http_status_failure(
+    status_code: int, text: str, query: str
+) -> FailedTestExecutionError:
+    return FailedTestExecutionError(
+        error_summary="HTTP SQL query failed",
+        errors=[
+            TestFailureDetails(
+                message=f"HTTP {status_code} from /api/sql: {text.strip()}",
+                details=f"Occurred at {datetime.datetime.now()}. Query: {query}",
+            )
+        ],
+    )
+
+
 def http_sql_query(
     host: str,
     query: str,
@@ -302,7 +308,15 @@ def http_sql_query(
         print("Got 401, refreshing token and retrying")
         new_token = refresh_token()
         return http_sql_query(host, query, new_token, retries, refresh_token=None)
-    assert r.status_code == 200, f"{r}\n{r.text}"
+    if r.status_code >= 500 and retries > 0:
+        # The balancer answers 502 "upstream server not available" while
+        # environmentd restarts, e.g. during a version cutover. Give it a
+        # moment; a persistent 5xx is a collected failure, not an abort.
+        print(f"Got {r.status_code}, retrying in 5s: {r.text.strip()}")
+        time.sleep(5)
+        return http_sql_query(host, query, token, retries - 1, refresh_token)
+    if r.status_code != 200:
+        raise http_status_failure(r.status_code, r.text, query)
     results = r.json()["results"]
     assert len(results) == 1, results