diff --git a/scripts/prune_leaked_test_task_defs.py b/scripts/prune_leaked_test_task_defs.py new file mode 100644 index 00000000..58d0f0f2 --- /dev/null +++ b/scripts/prune_leaked_test_task_defs.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Delete task definitions left behind by integration-test runs. + +The integration suites register per-run task defs (``sync_basic_``, +``async_lease_heartbeat_task_``, ...) and, before the tearDownClass +cleanup was added, never removed them. On a shared server those accumulate +until the account hits its Task Definitions cap, at which point every +registration answers:: + + 402 System has reached the maximum allowed Task Definitions limit of 1000. + +and the integration jobs fail on unrelated branches. + +The suites now prune stale leftovers themselves at session start (see +tests/integration/leaked_task_defs.py), so this script is for pruning by hand +— including the defs too recent for the automatic pass to touch. + +Dry run by default: it prints what it would delete and exits. Pass --delete to +actually remove them. Reads the usual CONDUCTOR_SERVER_URL / +CONDUCTOR_AUTH_KEY / CONDUCTOR_AUTH_SECRET environment. + + python scripts/prune_leaked_test_task_defs.py # list matches + python scripts/prune_leaked_test_task_defs.py --delete # remove them +""" + +import argparse +import sys + +from conductor.client.configuration.configuration import Configuration +from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient +from tests.integration.leaked_task_defs import ( + STALE_AFTER_SECONDS, + is_leaked_task_def, + stale_leaked_task_defs, +) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--delete", + action="store_true", + help="actually unregister the matches (default: dry run)", + ) + parser.add_argument( + "--include-recent", + action="store_true", + help=( + "also prune defs newer than " + f"{STALE_AFTER_SECONDS // 3600}h — only safe when no integration " + "run is in flight, since a concurrent run's defs are fair game" + ), + ) + args = parser.parse_args() + + config = Configuration() + client = OrkesMetadataClient(config) + + all_defs = client.get_all_task_defs() + if args.include_recent: + leaked = sorted(d.name for d in all_defs if is_leaked_task_def(d.name)) + else: + leaked = stale_leaked_task_defs(all_defs) + + print(f"server: {config.host}") + print(f"task defs total: {len(all_defs)}") + print(f"test leftovers: {len(leaked)}") + + if not leaked: + return 0 + + if not args.delete: + for name in leaked: + print(f" would delete {name}") + print("\nDry run — re-run with --delete to remove these.") + return 0 + + failed = 0 + for name in leaked: + try: + client.unregister_task_def(name) + print(f" deleted {name}") + except Exception as e: + failed += 1 + print(f" FAILED {name}: {e}", file=sys.stderr) + + print(f"\ndeleted {len(leaked) - failed} of {len(leaked)}") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py index 2811d492..173a1cce 100644 --- a/tests/integration/client/orkes/test_orkes_clients.py +++ b/tests/integration/client/orkes/test_orkes_clients.py @@ -579,7 +579,10 @@ def __test_workflow_execution_lifecycle(self): workflow = self.workflow_client.get_workflow(workflow_uuid, False) except ApiException as e: assert e.code == 404 - assert str(e.message).lower() == "workflow with id: {} not found.".format(workflow_uuid) + # The server's wording for this 404 changes across releases + # ("Workflow with Id: X not found." -> "No execution found for id: X"), + # so assert on the durable part: it is a 404 naming this execution. + assert workflow_uuid in str(e.message) def __test_task_execution_lifecycle(self): diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index aede140e..3f7de2a7 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -52,6 +52,36 @@ def _check_server_connectivity(): return _server_available, _skip_reason +_reclaimed = False + + +def _reclaim_once(): + """Reclaim task-def quota leaked by earlier runs, before any test registers. + + Without this a filled-up account stays filled: registration answers 402 for + every branch, so no run gets far enough to clean up after itself. + + Called from both entry points below — pytest_sessionstart covers the CI + invocations (tests/integration/conftest.py is an initial conftest for every + bucket), skip_if_server_unavailable covers a session that reaches the + unittest suites some other way. The flag makes the second call a no-op. + """ + global _reclaimed + if _reclaimed: + return + _reclaimed = True + + available, _ = _check_server_connectivity() + if not available: + return + from tests.integration.leaked_task_defs import reclaim_task_def_quota + reclaim_task_def_quota(Configuration()) + + +def pytest_sessionstart(session): + _reclaim_once() + + def skip_if_server_unavailable(): """ Call from unittest.TestCase.setUpClass to skip the entire test class @@ -70,6 +100,61 @@ def setUpClass(cls): raise unittest.SkipTest(reason) +# --------------------------------------------------------------------------- +# Metadata cleanup +# --------------------------------------------------------------------------- + +def cleanup_metadata(config, task_defs=(), workflow_defs=()): + """Best-effort delete of metadata a test class registered. + + Suites that register per-run task defs (``worker_task(..., + register_task_def=True)`` with a RUN_ID-suffixed name) used to leave every + one behind, so each CI run added a handful to the shared server until it + hit its Task Definitions cap and answered 402 to all further + registrations. Call this from tearDownClass. + + Failures are logged and swallowed: cleanup must never turn a passing suite + red, and a def that was never registered is nothing to report. + + ``workflow_defs`` items are names (version 1 assumed) or (name, version). + """ + from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient + + client = OrkesMetadataClient(config) + + def _log(what, e): + # "no such definition" is the normal case for anything a deselected or + # skipped test never registered, so it logs at debug — only a cleanup + # that failed for some other reason is worth a warning. + level = logging.DEBUG if _is_missing(e) else logging.WARNING + logger.log(level, "cleanup: could not unregister %s: %s", what, e) + + for name in task_defs: + try: + client.unregister_task_def(name) + except Exception as e: + _log(f"task def {name}", e) + + for entry in workflow_defs: + name, version = entry if isinstance(entry, tuple) else (entry, 1) + try: + client.unregister_workflow_def(name, version) + except Exception as e: + _log(f"workflow def {name} v{version}", e) + + +def _is_missing(exc): + """True when a delete failed only because the definition was not there. + + The server reports this inconsistently — 404, or a 500 whose body reads + "No such task definition" / "No such workflow definition" — so match on + both the status and the text. + """ + if getattr(exc, "code", None) == 404: + return True + return "no such" in str(getattr(exc, "body", "") or exc).lower() + + # --------------------------------------------------------------------------- # Pytest session-scoped fixtures # --------------------------------------------------------------------------- diff --git a/tests/integration/leaked_task_defs.py b/tests/integration/leaked_task_defs.py new file mode 100644 index 00000000..098ca415 --- /dev/null +++ b/tests/integration/leaked_task_defs.py @@ -0,0 +1,112 @@ +"""Recognise and reclaim task definitions left behind by integration runs. + +The integration suites register per-run task defs +(``sync_basic_``, ``async_lease_heartbeat_task_``, ...). Until +the ``tearDownClass`` cleanup landed, every run left its own behind, and on the +shared server they accumulated until the account hit its cap:: + + 402 System has reached the maximum allowed Task Definitions limit of 1000. + +At that point registration fails for every branch, so unrelated PRs go red. +Cleanup on teardown stops the leak; this module reclaims what earlier runs +already leaked, and is also what ``scripts/prune_leaked_test_task_defs.py`` +matches on. +""" + +import logging +import os +import re +import time + +logger = logging.getLogger(__name__) + +# Prefixes owned by the integration suites, each followed by a per-run id. +# A name is only a candidate if it matches one of these AND ends in a run id, +# so a hand-registered or production task def is never touched. +TEST_PREFIXES = ( + # tests/integration/test_comprehensive_e2e.py + "sync_basic_", + "async_basic_", + "complex_schema_", + "task_in_progress_", + "failing_task_", + # tests/integration/test_lease_extension.py + "lease_heartbeat_task_", + "lease_no_heartbeat_task_", + # tests/integration/test_async_lease_extension.py + "async_lease_heartbeat_task_", + "async_lease_no_heartbeat_task_", + "async_lease_fast_with_hb_", + "async_lease_fast_no_hb_", + # tests/integration/client/orkes/test_orkes_clients.py (shortuuid suffix) + "IntegrationTestOrkesClientsTask_", +) + +# uuid4().hex[:8] for most suites; shortuuid for test_orkes_clients. +_RUN_ID = re.compile(r"^(?:[0-9a-f]{8}|[0-9A-Za-z]{20,25})$") + +# Four integration jobs run in parallel, and other PRs run at the same time. +# Only defs older than this are reclaimed, so a concurrent run never has the +# task def it is mid-test on deleted from under it. +STALE_AFTER_SECONDS = 2 * 60 * 60 + + +def is_leaked_task_def(name): + """True if ``name`` is a per-run task def owned by the integration suites.""" + for prefix in TEST_PREFIXES: + if name.startswith(prefix) and _RUN_ID.match(name[len(prefix):]): + return True + return False + + +def stale_leaked_task_defs(task_defs, now=None): + """Names in ``task_defs`` that are leaked AND older than STALE_AFTER_SECONDS. + + A def with no ``create_time`` is left alone: without an age there is no way + to tell it from one a concurrent run just registered. + """ + cutoff_ms = ((now if now is not None else time.time()) - STALE_AFTER_SECONDS) * 1000 + return sorted( + d.name + for d in task_defs + if is_leaked_task_def(d.name) and (d.create_time or 0) and d.create_time < cutoff_ms + ) + + +def reclaim_task_def_quota(config): + """Delete stale leaked task defs so registration has room again. + + Best-effort and never raises: this runs before the suites do, and a server + that will not answer is the tests' problem to report, not this helper's. + Returns the number deleted. Set CONDUCTOR_SKIP_TASK_DEF_RECLAIM=1 to skip. + """ + if os.environ.get("CONDUCTOR_SKIP_TASK_DEF_RECLAIM"): + return 0 + + from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient + + client = OrkesMetadataClient(config) + try: + all_defs = client.get_all_task_defs() + except Exception as e: + logger.warning("reclaim: could not list task defs: %s", e) + return 0 + + stale = stale_leaked_task_defs(all_defs) + if not stale: + return 0 + + deleted = 0 + for name in stale: + try: + client.unregister_task_def(name) + deleted += 1 + except Exception as e: + logger.warning("reclaim: could not unregister %s: %s", name, e) + + # print(), not logger: logging is not configured yet at session start. + print( + f"reclaimed {deleted} of {len(stale)} stale test task defs " + f"({len(all_defs)} defs on {config.host} before pruning)" + ) + return deleted diff --git a/tests/integration/test_async_lease_extension.py b/tests/integration/test_async_lease_extension.py index d384ac4d..b6829fbe 100644 --- a/tests/integration/test_async_lease_extension.py +++ b/tests/integration/test_async_lease_extension.py @@ -232,6 +232,17 @@ def tearDownClass(cls): handler = getattr(cls, '_task_handler', None) if handler is not None: handler.stop_processes() + from tests.integration.conftest import cleanup_metadata + cleanup_metadata( + cls.config, + task_defs=(HEARTBEAT_TASK, NO_HEARTBEAT_TASK, FAST_HB_TASK, FAST_NO_HB_TASK), + workflow_defs=( + f'test_async_lease_heartbeat_{RUN_ID}', + f'test_async_lease_no_heartbeat_{RUN_ID}', + f'test_async_perf_with_hb_{RUN_ID}', + f'test_async_perf_no_hb_{RUN_ID}', + ), + ) def _register_workflow(self, wf_name, task_names): """Register a workflow with one or more tasks in sequence.""" diff --git a/tests/integration/test_comprehensive_e2e.py b/tests/integration/test_comprehensive_e2e.py index 4fcfac8e..178c09f6 100644 --- a/tests/integration/test_comprehensive_e2e.py +++ b/tests/integration/test_comprehensive_e2e.py @@ -586,6 +586,12 @@ def tearDownClass(cls): if os.path.exists(cls.metrics_dir): import shutil shutil.rmtree(cls.metrics_dir) + from tests.integration.conftest import cleanup_metadata + cleanup_metadata( + cls.config, + task_defs=cls.EXPECTED_WORKERS, + workflow_defs=(WF_NAME,), + ) print("\n✓ Cleanup complete") diff --git a/tests/integration/test_lease_extension.py b/tests/integration/test_lease_extension.py index 3b986409..82f1aca4 100644 --- a/tests/integration/test_lease_extension.py +++ b/tests/integration/test_lease_extension.py @@ -146,6 +146,18 @@ def setUpClass(cls): cls.metadata_client = OrkesMetadataClient(cls.config) cls.workflow_client = OrkesWorkflowClient(cls.config) + @classmethod + def tearDownClass(cls): + from tests.integration.conftest import cleanup_metadata + cleanup_metadata( + cls.config, + task_defs=(HEARTBEAT_TASK, NO_HEARTBEAT_TASK), + workflow_defs=( + f'test_lease_heartbeat_{RUN_ID}', + f'test_lease_no_heartbeat_{RUN_ID}', + ), + ) + def _register_workflow(self, wf_name, task_name): """Register a single-task workflow.""" workflow = WorkflowDef(name=wf_name, version=1)