diff --git a/test/canary-load/mzcompose.py b/test/canary-load/mzcompose.py index 18cd5321514aa..1f1881ca8d21f 100644 --- a/test/canary-load/mzcompose.py +++ b/test/canary-load/mzcompose.py @@ -13,6 +13,7 @@ import os import threading import time +import traceback import urllib.parse from collections.abc import Callable from textwrap import dedent @@ -20,8 +21,6 @@ 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,6 +55,10 @@ "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", ] @@ -63,6 +66,21 @@ 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