From 0de1b669839e6451fea9de7451844a24c4160e86 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 11 Sep 2026 14:29:29 +0000 Subject: [PATCH] Keep failed resource samples from aborting soak evidence collection --- scripts/perf/server_soak.py | 26 +++++++++++---- tests/Unit/Support/server_soak_test.py | 45 ++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 7 deletions(-) diff --git a/scripts/perf/server_soak.py b/scripts/perf/server_soak.py index e62b088f..ecdec05a 100755 --- a/scripts/perf/server_soak.py +++ b/scripts/perf/server_soak.py @@ -740,7 +740,7 @@ def parse_int_field(value: str) -> tuple[int, bool]: return 0, False -def mysql_counts(project: str) -> dict[str, int]: +def mysql_counts(project: str) -> dict[str, int | str]: query = ( "SELECT " "(SELECT COUNT(*) FROM workflow_namespaces) AS namespaces, " @@ -764,23 +764,35 @@ def mysql_counts(project: str) -> dict[str, int]: ) ) parts = result.stdout.strip().split() - if len(parts) >= 4: + if result.returncode == 0 and len(parts) == 4 and all(re.fullmatch(r"[0-9]+", part) for part in parts): return { - "mysql_sample_ok": 1 if result.returncode == 0 else 0, + "mysql_sample_ok": 1, "mysql_namespaces": int(parts[0]), "mysql_worker_registrations": int(parts[1]), "mysql_workflow_runs": int(parts[2]), "mysql_ready_tasks": int(parts[3]), } - return {"mysql_sample_ok": 0} + return { + "mysql_sample_ok": 0, + "mysql_sample_error": "command_failed" if result.returncode else "malformed_counts", + } def sample(project: str, include_sdk: bool = True) -> dict[str, Any]: row: dict[str, Any] = {"timestamp": time.time()} if project: - row.update(docker_stats(project, include_sdk)) - row.update(redis_info(project)) - row.update(mysql_counts(project)) + for collect, arguments, health_field in ( + (docker_stats, (project, include_sdk), "docker_stats_ok"), + (redis_info, (project,), "redis_sample_ok"), + (mysql_counts, (project,), "mysql_sample_ok"), + ): + try: + row.update(collect(*arguments)) + except (OSError, subprocess.TimeoutExpired) as error: + # Preserve the failed observation without losing the remaining + # load window and final report. It still disqualifies the run. + row[health_field] = 0 + row[health_field + "_error"] = type(error).__name__ return row diff --git a/tests/Unit/Support/server_soak_test.py b/tests/Unit/Support/server_soak_test.py index 23f10889..7a0538df 100644 --- a/tests/Unit/Support/server_soak_test.py +++ b/tests/Unit/Support/server_soak_test.py @@ -3,6 +3,7 @@ import importlib.util import json import os +import subprocess from pathlib import Path import unittest from unittest.mock import patch @@ -73,6 +74,50 @@ def test_request_error_fails_even_when_completion_floor_is_met(self) -> None: class RuntimeEvidenceConfigurationTest(unittest.TestCase): + def test_mysql_sampling_rejects_failed_or_malformed_output_without_crashing(self): + for code, output in ( + (1, "Unavailable: backend could not be reached"), + (0, "Unavailable: backend could not be reached"), + (1, "1 2 3 4"), (0, "1 2 3"), (0, "1 2 3 4 extra"), + (0, "1 2 -3 4"), (0, "1 2 3.0 4"), (0, ""), + ): + with self.subTest(code=code, output=output): + result = subprocess.CompletedProcess([], code, stdout=output) + with patch.object(server_soak, "run_command", return_value=result): + observation = server_soak.mysql_counts("fixture") + self.assertEqual(0, observation["mysql_sample_ok"]) + self.assertNotIn("mysql_ready_tasks", observation) + + def test_mysql_sampling_preserves_zero_and_valid_counts(self): + result = subprocess.CompletedProcess([], 0, stdout="10\t25\t49\t0\n") + with patch.object(server_soak, "run_command", return_value=result): + self.assertEqual({ + "mysql_sample_ok": 1, "mysql_namespaces": 10, + "mysql_worker_registrations": 25, "mysql_workflow_runs": 49, + "mysql_ready_tasks": 0, + }, server_soak.mysql_counts("fixture")) + + def test_failed_sampler_does_not_prevent_other_samples_or_later_recovery(self): + samplers = {"docker_stats": "docker_stats_ok", "redis_info": "redis_sample_ok", "mysql_counts": "mysql_sample_ok"} + for failing, field in samplers.items(): + for error in (OSError("fixture"), subprocess.TimeoutExpired(["fixture"], 30)): + with self.subTest(sampler=failing, error=type(error).__name__): + with patch.object(server_soak, "docker_stats", return_value={"docker_stats_ok": 1, "server_container_healthy": 1}), \ + patch.object(server_soak, "redis_info", return_value={"redis_sample_ok": 1}), \ + patch.object(server_soak, "mysql_counts", return_value={"mysql_sample_ok": 1}): + with patch.object(server_soak, failing, side_effect=error): + failed = server_soak.sample("fixture") + recovered = server_soak.sample("fixture") + self.assertEqual(0, failed[field]) + for health in samplers.values(): + self.assertEqual(1, recovered[health]) + if health != field: + self.assertEqual(1, failed[health]) + health = server_soak.sample_health([failed, recovered], "fixture") + self.assertEqual(1, health["unhealthy_samples"]) + self.assertEqual(1, health["unhealthy_field_counts"][field]) + self.assertFalse(health["unhealthy_final_sample"]) + def test_redis_sampling_targets_the_configured_cache_database(self) -> None: with patch.dict(os.environ, {"DW_PERF_REDIS_CACHE_DB": "3"}): self.assertEqual(3, server_soak.redis_cache_database())