diff --git a/.github/workflows/test-integration-runtime.yml b/.github/workflows/test-integration-runtime.yml index a90e68fa9cb..d8b2a2c7a59 100644 --- a/.github/workflows/test-integration-runtime.yml +++ b/.github/workflows/test-integration-runtime.yml @@ -79,3 +79,15 @@ jobs: KAFKA_BOOTSTRAP_SERVERS: ${{ vars.CI_KAFKA_BOOTSTRAP }} SCHEMA_REGISTRY_URL: ${{ vars.CI_SCHEMA_REGISTRY }} RUN_ID: 1 # we use this to run a single variant of the Kafka tests in test_kafka_avro.py + + # Test teardown only runs while pytest is alive. A cancel sentinel firing + # for some other job, a runner eviction or a job timeout leaves tests + # running on the shared instance to burn compute. + # This step reclaims the pipelines this run named after itself + - name: Stop pipelines this run leaked + if: ${{ always() && vars.CI_DRY_RUN != 'true' }} + timeout-minutes: 10 + run: uv run --locked python -m tests.stop_ci_run_pipelines + working-directory: python + env: + PYTHONPATH: ${{ github.workspace }}/python diff --git a/python/feldera/testutils.py b/python/feldera/testutils.py index 010ffb373d1..b53a044a1ff 100644 --- a/python/feldera/testutils.py +++ b/python/feldera/testutils.py @@ -99,18 +99,19 @@ def _token_expiry(token: str) -> float: return time.time() -def _feldera_credential(): - """Bearer credential for the test client. - - Under GitHub Actions the credential is an ID token that expires well inside - a long test run, so hand the SDK the callable rather than a token: it - re-resolves per request and retries once on 401. A configured OIDC login - flow takes precedence over both. +def feldera_bearer_token() -> Optional[str]: + """The bearer token to send with a request issued now. + + A configured OIDC login flow wins, then a GitHub Actions ID token, then the + static API key. None when nothing is configured, which is how a local + instance without authentication runs. Callers that build their own requests + resolve this per request: under Actions the token expires well inside a test + run, and one read at import time starts returning 401 partway through. """ if os.environ.get("OIDC_TEST_ISSUER"): return _get_effective_api_key() if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL"): - return _github_oidc_token + return _github_oidc_token() return API_KEY @@ -155,9 +156,16 @@ def __init__(self): def _ensure(self): if self._client is None: + # Under Actions the token expires inside a run, so the SDK gets the + # resolver itself: it re-resolves per request and retries once on + # 401. Elsewhere the credential is fixed for the process. self._client = FelderaClient( connection_timeout=10, - api_key=_feldera_credential(), + api_key=( + feldera_bearer_token + if os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL") + else feldera_bearer_token() + ), ) return self._client @@ -287,6 +295,42 @@ def unique_pipeline_name(base_name: str) -> str: return name +# Teardown must not block on one wedged pipeline: the SDK otherwise polls for +# `Stopped` forever, and when the runner finally kills the job every pipeline +# the run started keeps consuming compute on the shared instance. +RECLAIM_TIMEOUT_SECONDS = 60.0 + + +def reclaim_pipeline(name: str, delete: bool = False) -> List[str]: + """Force-stop `name`, clear its storage, and optionally delete it. + + Every step runs even when the one before it raised: a pipeline that never + reports `Stopped` can still release its storage, and abandoning the rest on + the first error leaves them running on a shared instance. Returns one + message per failed step, empty when the pipeline is fully reclaimed, so the + caller decides whether a cleanup failure is worth failing a test over. + """ + failures = [] + + try: + TEST_CLIENT.stop_pipeline(name, force=True, timeout_s=RECLAIM_TIMEOUT_SECONDS) + except Exception as error: + failures.append(f"{name}: stop: {error}") + + try: + TEST_CLIENT.clear_storage(name, timeout_s=RECLAIM_TIMEOUT_SECONDS) + except Exception as error: + failures.append(f"{name}: clear storage: {error}") + + if delete: + try: + TEST_CLIENT.delete_pipeline(name) + except Exception as error: + failures.append(f"{name}: delete: {error}") + + return failures + + def enterprise_only(fn): fn._enterprise_only = True return unittest.skipUnless( diff --git a/python/tests/README.md b/python/tests/README.md index db9af0778ad..2b4c167c695 100644 --- a/python/tests/README.md +++ b/python/tests/README.md @@ -111,3 +111,21 @@ after the test completed. def test_some_property(pipeline_name): pass ``` + +## Cleaning up after a run + +Teardown only runs while the test process is alive. A cancelled workflow run, an +evicted runner or a job timeout kills pytest outright, and every pipeline that +was running at that moment keeps consuming compute on the shared CI instance. + +`tests/stop_ci_run_pipelines.py` sweeps up what such a run left behind. It +matches the prefix `unique_pipeline_name` stamps on every test pipeline. + +```bash +cd python +PYTHONPATH=`pwd` uv run python -m tests.stop_ci_run_pipelines --prefix _ +``` + +Whatever cleanup a test does itself should go through +`feldera.testutils.reclaim_pipeline`: it bounds every wait, and it clears +storage even when the stop failed instead of abandoning a running pipeline. diff --git a/python/tests/platform/TEST_AUTH.md b/python/tests/platform/TEST_AUTH.md index 49e73e0dc41..f17712351e5 100644 --- a/python/tests/platform/TEST_AUTH.md +++ b/python/tests/platform/TEST_AUTH.md @@ -136,8 +136,13 @@ uv run pytest tests/platform -v The platform tests automatically select authentication in this order: 1. **OIDC Token** (if all OIDC environment variables are set) -2. **API Key** (if `FELDERA_API_KEY` is set) -3. **No Authentication** (fallback for development) +2. **GitHub Actions ID token** (if `ACTIONS_ID_TOKEN_REQUEST_URL` is set) +3. **API Key** (if `FELDERA_API_KEY` is set) +4. **No Authentication** (fallback for development) + +`feldera.testutils.feldera_bearer_token()` resolves this order. Call it per +request rather than caching what it returns: an ID token expires well inside a +long suite, and the function re-mints one shortly before its expiry. ## CI/CD Configuration diff --git a/python/tests/platform/helper.py b/python/tests/platform/helper.py index 048ae378c68..d9013e916e9 100644 --- a/python/tests/platform/helper.py +++ b/python/tests/platform/helper.py @@ -13,6 +13,7 @@ from __future__ import annotations import json +import logging import os import unittest from http import HTTPStatus @@ -21,11 +22,14 @@ import pytest import requests -from feldera.testutils import FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS -from feldera.testutils_oidc import get_oidc_test_helper +from feldera.testutils import ( + FELDERA_TEST_NUM_HOSTS, + FELDERA_TEST_NUM_WORKERS, + feldera_bearer_token, + reclaim_pipeline, +) from tests import ( - API_KEY, BASE_URL, FELDERA_REQUESTS_VERIFY, TEST_CLIENT, @@ -35,19 +39,18 @@ API_PREFIX = "/v0" +logger = logging.getLogger(__name__) + def _base_headers() -> Dict[str, str]: headers = { "Accept": "application/json", } - # Try OIDC authentication first, then fall back to API_KEY - oidc_helper = get_oidc_test_helper() - if oidc_helper is not None: - token = oidc_helper.obtain_access_token() + # Resolved per request: a GitHub Actions ID token expires mid-suite. + token = feldera_bearer_token() + if token: headers["Authorization"] = f"Bearer {token}" - elif API_KEY: - headers["Authorization"] = f"Bearer {API_KEY}" return headers @@ -162,22 +165,25 @@ def pause_pipeline(name: str, wait: bool = True): TEST_CLIENT.pause_pipeline(name, wait=wait) -def stop_pipeline(name: str, force: bool = True, wait: bool = True): - TEST_CLIENT.stop_pipeline(name, force=force, wait=wait) +def stop_pipeline( + name: str, force: bool = True, wait: bool = True, timeout_s: float | None = None +): + TEST_CLIENT.stop_pipeline(name, force=force, wait=wait, timeout_s=timeout_s) -def clear_pipeline(name: str, wait: bool = True): +def clear_pipeline(name: str, wait: bool = True, timeout_s: float | None = None): r = post_no_body(f"{API_PREFIX}/pipelines/{name}/clear") if wait and r.status_code == HTTPStatus.ACCEPTED: - TEST_CLIENT.clear_storage(name) + TEST_CLIENT.clear_storage(name, timeout_s=timeout_s) return r def reset_pipeline(name: str): + """Force-stop `name` and clear its storage.""" if get_pipeline(name, "status").status_code != HTTPStatus.OK: return - stop_pipeline(name, force=True) - clear_pipeline(name) + for failure in reclaim_pipeline(name): + logger.warning("pipeline cleanup: %s", failure) def delete_pipeline(name: str): diff --git a/python/tests/runtime_aggtest/aggtst_base.py b/python/tests/runtime_aggtest/aggtst_base.py index df420cb5fce..abb5d0d4963 100644 --- a/python/tests/runtime_aggtest/aggtst_base.py +++ b/python/tests/runtime_aggtest/aggtst_base.py @@ -9,7 +9,11 @@ from feldera.enums import CompilationProfile from feldera.rest.errors import FelderaAPIError from feldera.runtime_config import Resources, RuntimeConfig -from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS +from feldera.testutils import ( + FELDERA_TEST_NUM_WORKERS, + FELDERA_TEST_NUM_HOSTS, + reclaim_pipeline, +) from tests import TEST_CLIENT, unique_pipeline_name JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None @@ -325,8 +329,8 @@ def run_pipeline( raise if pipeline is not None: - pipeline.stop(force=True) - pipeline.delete(True) + for failure in reclaim_pipeline(pipeline_name, delete=True): + print(f"WARNING: {failure}") def assert_expected_error(self, obj: SqlObject, actual_exception: Exception): """Validate the error produced by the failing pipeline with the expected error type""" diff --git a/python/tests/shared_test_pipeline.py b/python/tests/shared_test_pipeline.py index afcfdaaddec..3da91c98820 100644 --- a/python/tests/shared_test_pipeline.py +++ b/python/tests/shared_test_pipeline.py @@ -1,3 +1,4 @@ +import logging import unittest from feldera import Pipeline, PipelineBuilder @@ -5,10 +6,13 @@ from feldera.testutils import ( FELDERA_TEST_NUM_HOSTS, FELDERA_TEST_NUM_WORKERS, + reclaim_pipeline, unique_pipeline_name, ) from tests import TEST_CLIENT +logger = logging.getLogger(__name__) + def sql(text_or_iterable): """ @@ -59,30 +63,27 @@ def setUpClass(cls): ).create_or_replace() def setUp(self): - p = PipelineBuilder( - self.client, - unique_pipeline_name(self._testMethodName), - sql=self.ddl, - runtime_config=RuntimeConfig( - workers=FELDERA_TEST_NUM_WORKERS, - hosts=FELDERA_TEST_NUM_HOSTS, - logging="debug", - ), - ).create_or_replace() - self.p = p + self._owned_pipelines = [] + self.p = self._build_pipeline(self._testMethodName) def tearDown(self): - self.p.stop(force=True) - self.p.clear_storage() + """Force-stop and clear every pipeline the test owns.""" + for pipeline in self._owned_pipelines: + for failure in reclaim_pipeline(pipeline.name): + logger.warning("pipeline teardown: %s", failure) @property def pipeline(self) -> Pipeline: return self.p def new_pipeline_with_suffix(self, suffix: str) -> Pipeline: - return PipelineBuilder( + return self._build_pipeline(f"{self._testMethodName}_{suffix}") + + def _build_pipeline(self, base_name: str) -> Pipeline: + """Create a pipeline over the class DDL and hand it to `tearDown`.""" + pipeline = PipelineBuilder( self.client, - unique_pipeline_name(f"{self._testMethodName}_{suffix}"), + unique_pipeline_name(base_name), sql=self.ddl, runtime_config=RuntimeConfig( workers=FELDERA_TEST_NUM_WORKERS, @@ -90,3 +91,5 @@ def new_pipeline_with_suffix(self, suffix: str) -> Pipeline: logging="debug", ), ).create_or_replace() + self._owned_pipelines.append(pipeline) + return pipeline diff --git a/python/tests/stop_ci_run_pipelines.py b/python/tests/stop_ci_run_pipelines.py new file mode 100644 index 00000000000..f4eada043e9 --- /dev/null +++ b/python/tests/stop_ci_run_pipelines.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Stop every pipeline the current CI run left behind on a shared instance. + +Per-test teardown only runs while the test process is alive. A cancelled +workflow run, an evicted runner or a job timeout kills pytest outright, and +every pipeline that was running at that moment keeps burning compute on the +shared instance until the daily sweep finds it. Run this as an `if: always()` +step at the end of a job that talks to a shared instance. + +The sweep matches on the prefix `feldera.testutils.unique_pipeline_name` +stamps on every test pipeline, so it only ever touches this run's own +pipelines. It never fails the job: a red cleanup step on an otherwise green +run tells nobody anything the warnings do not, and the daily sweep is still +the backstop for a runner that dies before this step runs. + +It stops pipelines and clears their storage; it never deletes them. Stopped +and cleared, a pipeline consumes nothing, and its record is what you read to +work out why the run failed. Deleting is the daily sweep's job. + +Run it as a module, not as a path: `tests/` holds a `platform` package that +shadows the standard library one for anything imported by a script living +there. + +Usage: + PYTHONPATH=$PWD uv run python -m tests.stop_ci_run_pipelines [--prefix P] +""" + +from __future__ import annotations + +import argparse +import os +import sys +from concurrent.futures import ThreadPoolExecutor + +from feldera.testutils import ( + BASE_URL, + TEST_CLIENT, + reclaim_pipeline, + unique_pipeline_name, +) + +# A cancelled job gets a short grace window before the runner kills it, so the +# sweep works on several pipelines at once rather than serially. +MAX_CONCURRENT_RECLAIMS = 8 + + +def warn(message: str) -> None: + # `::warning::` reaches the run's annotation summary, which is where a + # cleanup that found something has to show up. Nothing reads it elsewhere. + if os.environ.get("GITHUB_ACTIONS"): + print(f"::warning::{message}", flush=True) + else: + print(f"WARNING: {message}", flush=True) + + +def is_reclaimed(deployment_status: str | None, storage_status: str | None) -> bool: + """True once a pipeline holds neither compute nor storage.""" + return deployment_status == "Stopped" and storage_status == "Cleared" + + +def main() -> int: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--prefix", + default=None, + help="Pipeline name prefix to sweep. Defaults to this run's own prefix, " + "the first five characters of GITHUB_SHA plus FELDERA_TEST_TAG_SUFFIX.", + ) + args = parser.parse_args() + + # `unique_pipeline_name` stamps this prefix on every pipeline of this run. + prefix = args.prefix or unique_pipeline_name("") + print(f"Sweeping pipelines named '{prefix}*' on {BASE_URL}", flush=True) + + try: + pipelines = TEST_CLIENT.pipelines() + except Exception as error: + warn(f"could not list pipelines on {BASE_URL}: {error}") + return 0 + + to_reclaim = [ + pipeline.name + for pipeline in pipelines + if pipeline.name.startswith(prefix) + and not is_reclaimed(pipeline.deployment_status, pipeline.storage_status) + ] + + if not to_reclaim: + print("No leaked pipelines.", flush=True) + return 0 + + warn( + f"{len(to_reclaim)} pipeline(s) outlived their test and are being reclaimed: " + + ", ".join(sorted(to_reclaim)) + ) + + def reclaim(name: str) -> tuple[str, list[str]]: + return name, reclaim_pipeline(name) + + with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_RECLAIMS) as pool: + for name, failures in pool.map(reclaim, to_reclaim): + if failures: + warn("; ".join(failures)) + else: + print(f"Reclaimed {name}", flush=True) + + return 0 + + +if __name__ == "__main__": + sys.exit(main())