This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch ci-runs-on-fewer-test-slices in repository https://gitbox.apache.org/repos/asf/airflow.git
commit b5887139696920f015a92200267143ba3e265f38 Author: Jarek Potiuk <[email protected]> AuthorDate: Wed Oct 28 02:16:25 2026 +0100 Size CI test jobs on RunsOn runners to the parallelism they can use Jobs that run tests with --run-in-parallel start one serial pytest per test type, so they never use more CPUs than they have test types. On RunsOn, splitting core test types into two jobs only doubles the jobs: each half still lasts as long as its slowest test type, and the out-of-memory risk the split guards against on public runners does not apply to the larger RunsOn runners. The lowest-dependencies provider jobs left their 32 CPUs partly idle, and the integration-core and system t [...] Generated-by: Claude Opus 5 --- .github/workflows/integration-system-tests.yml | 2 +- .github/workflows/special-tests.yml | 2 +- dev/breeze/doc/ci/04_selective_checks.md | 14 ++++++- dev/breeze/src/airflow_breeze/global_constants.py | 7 ++++ .../src/airflow_breeze/utils/selective_checks.py | 27 +++++++------ dev/breeze/tests/test_selective_checks.py | 44 ++++++++++++++++++++++ 6 files changed, 81 insertions(+), 15 deletions(-) diff --git a/.github/workflows/integration-system-tests.yml b/.github/workflows/integration-system-tests.yml index 8b24223a55f..e61e4070e0c 100644 --- a/.github/workflows/integration-system-tests.yml +++ b/.github/workflows/integration-system-tests.yml @@ -75,7 +75,7 @@ jobs: timeout-minutes: 30 if: inputs.testable-core-integrations != '[]' name: "Integration core ${{ matrix.integration }}" - runs-on: ${{ format(fromJSON(inputs.runners)[0], '32+64', '128+256') }} + runs-on: ${{ format(fromJSON(inputs.runners)[0], '8+16', '32+64') }} strategy: fail-fast: false max-parallel: 20 diff --git a/.github/workflows/special-tests.yml b/.github/workflows/special-tests.yml index 09658246df9..5eca561ba59 100644 --- a/.github/workflows/special-tests.yml +++ b/.github/workflows/special-tests.yml @@ -352,7 +352,7 @@ jobs: contents: read packages: read with: - runners: ${{ format(inputs.runners, '32+64', '128+256') }} + runners: ${{ format(inputs.runners, '8+16', '32+64') }} platform: ${{ inputs.platform }} test-name: "SystemTest" test-scope: "System" diff --git a/dev/breeze/doc/ci/04_selective_checks.md b/dev/breeze/doc/ci/04_selective_checks.md index 4bf5fffa284..4646442c0d0 100644 --- a/dev/breeze/doc/ci/04_selective_checks.md +++ b/dev/breeze/doc/ci/04_selective_checks.md @@ -651,7 +651,7 @@ GitHub Actions to pass the list of parameters to a command to execute | all-versions | If set to true, then all python, k8s, DB versions are used for tests. | false | | | basic-checks-only | Whether to run all static checks ("false") or only basic set of static checks ("true") | false | | | ci-image-build | Whether CI image build is needed | true | | -| core-test-types-list-as-strings-in-json | Which test types should be run for unit tests for core | API Always Providers | * | +| core-test-types-list-as-strings-in-json | Which test types should be run for unit tests for core (2 slices; 1 on RunsOn) | API Always Providers | * | | debug-resources | Whether resources usage should be printed during parallel job execution ("true"/ "false") | false | | | default-branch | Which branch is default for the build ("main" for main branch, "v2-4-test" for 2.4 line etc.) | main | | | default-constraints-branch | Which branch is default for the build ("constraints-main" for main branch, "constraints-2-4" etc.) | constraints-main | | @@ -675,7 +675,7 @@ GitHub Actions to pass the list of parameters to a command to execute | helm-version | Which Helm version to use for tests | v3.15.3 | | | in-memory-runner-type | Like runner-type, but the RunsOn template puts docker in RAM (`extras=tmpfs`) unless disabled | \["ubuntu-22.04"\] | | | include-success-outputs | Whether to include outputs of successful parallel tests ("true"/"false") | false | | -| individual-providers-test-types-list-as-strings-in-json | Which test types should be run for unit tests for providers (individually listed) | Providers[\amazon\] Providers\[google\] | * | +| individual-providers-test-types-list-as-strings-in-json | Which test types should be run for unit tests for providers (individually listed; 5 slices, 3 on RunsOn)| Providers[\amazon\] Providers\[google\] | * | | is-committer-build | Whether the build is triggered by a committer | false | | | is-legacy-ui-api-labeled | Whether the PR is labeled as legacy UI/API | false | | | java-sdk-version | JDK version used to build the lang-SDK Java artifacts natively in CI | 17 | | @@ -767,6 +767,16 @@ picks its size class with `format(fromJSON(inputs.runners)[0], <cpu>, <ram>)` (` `'8+16', '32+64'` or `'32+64', '128+256'`), or, for a single-job reusable workflow, its caller passes `format(<runners>, <cpu>, <ram>)`. Each class is a range up to twice its size, so Spot can come from the larger instance types too; xdist follows the host's CPUs, so a larger runner runs more test workers. +Jobs using `--run-in-parallel` (the DB, lowest-dependencies and special tests) run one serial pytest per +test type instead, so they never use more CPUs than they have test types, and jobs running a single pytest +(integration and system tests) use one. The size classes follow that: the xdist jobs and the +lowest-dependencies provider tests take `'32+64', '128+256'`, the others `'8+16', '32+64'` or smaller. +The matrix slices follow it too. On public runners the core test types are split into +`NUMBER_OF_CORE_SLICES` (2) jobs, to keep the Non-DB xdist collection inside the runner's memory, and +the individually listed providers into `NUMBER_OF_LOW_DEP_SLICES` (5). On RunsOn the core test types +stay in one job (`NUMBER_OF_CORE_SLICES_ON_RUNS_ON`), because a core DB job lasts as long as its +slowest test type whichever slice it is in, and the providers are split into 3 jobs +(`NUMBER_OF_LOW_DEP_SLICES_ON_RUNS_ON`), so each job has about one provider per CPU of its runner. `spot=co` takes Spot from the pools with the most capacity regardless of price, which is cheaper than the on-demand launch and whole-job rerun a capacity miss or interruption costs; on a public label the same call returns the label unchanged. Steps that only make sense on RunsOn are diff --git a/dev/breeze/src/airflow_breeze/global_constants.py b/dev/breeze/src/airflow_breeze/global_constants.py index efa1c3425b0..f4436496fa1 100644 --- a/dev/breeze/src/airflow_breeze/global_constants.py +++ b/dev/breeze/src/airflow_breeze/global_constants.py @@ -968,6 +968,13 @@ NUMBER_OF_LOW_DEP_SLICES = 5 # Number of slices for core test types (splitting reduces memory per xdist collection) NUMBER_OF_CORE_SLICES = 2 +# Slice counts on the self-hosted RunsOn runners. Jobs using `--run-in-parallel` run one serial pytest per +# test type, so a job never uses more CPUs than it has test types. The core jobs are bound by their slowest +# test type and the Non-DB xdist collection gets enough memory on RunsOn, so one slice only removes jobs. +# Low dep providers run on 32-CPU runners: a third of the providers per job keeps those CPUs busy. +NUMBER_OF_CORE_SLICES_ON_RUNS_ON = 1 +NUMBER_OF_LOW_DEP_SLICES_ON_RUNS_ON = 3 + # Milestone Tag Assistant configuration # Labels indicating a bug fix PR that should have a milestone MILESTONE_BUG_LABELS: frozenset[str] = frozenset({"kind:bug", "type:bug-fix"}) diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py b/dev/breeze/src/airflow_breeze/utils/selective_checks.py index b5415d504d3..8ca28020748 100644 --- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py +++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py @@ -52,7 +52,9 @@ from airflow_breeze.global_constants import ( JAVA_SDK_VERSION, KIND_VERSION, NUMBER_OF_CORE_SLICES, + NUMBER_OF_CORE_SLICES_ON_RUNS_ON, NUMBER_OF_LOW_DEP_SLICES, + NUMBER_OF_LOW_DEP_SLICES_ON_RUNS_ON, PROVIDERS_COMPATIBILITY_TESTS_MATRIX, PUBLIC_AMD_RUNNERS, PUBLIC_ARM_RUNNERS, @@ -1877,9 +1879,10 @@ class SelectiveChecks: if not self.run_unit_tests: return None current_test_types = sorted(set(self._get_core_test_types_to_run())) - if len(current_test_types) <= NUMBER_OF_CORE_SLICES: + number_of_slices = NUMBER_OF_CORE_SLICES_ON_RUNS_ON if self._is_on_runs_on else NUMBER_OF_CORE_SLICES + if len(current_test_types) <= number_of_slices: return json.dumps(_get_test_list_as_json([current_test_types])) - list_of_list_of_types = _split_list(current_test_types, NUMBER_OF_CORE_SLICES) + list_of_list_of_types = _split_list(current_test_types, number_of_slices) return json.dumps(_get_test_list_as_json(list_of_list_of_types)) @cached_property @@ -1966,11 +1969,12 @@ class SelectiveChecks: current_test_types = sorted(self._get_individual_providers_list()) if not current_test_types: return None - # We are hard-coding the number of lists as reasonable starting point to split the - # list of test types - and we can modify it in the future - if len(current_test_types) < NUMBER_OF_LOW_DEP_SLICES: + number_of_slices = ( + NUMBER_OF_LOW_DEP_SLICES_ON_RUNS_ON if self._is_on_runs_on else NUMBER_OF_LOW_DEP_SLICES + ) + if len(current_test_types) < number_of_slices: return json.dumps(_get_test_list_as_json([current_test_types])) - list_of_list_of_types = _split_list(current_test_types, NUMBER_OF_LOW_DEP_SLICES) + list_of_list_of_types = _split_list(current_test_types, number_of_slices) return json.dumps(_get_test_list_as_json(list_of_list_of_types)) @cached_property @@ -2322,17 +2326,18 @@ class SelectiveChecks: ) return json.dumps([template.format(run_id=run_id)]) + @cached_property + def _is_on_runs_on(self) -> bool: + # Reuses the routing decision of runner_type, which may have asked the GitHub API. + return self.runner_type not in (PUBLIC_AMD_RUNNERS, PUBLIC_ARM_RUNNERS) + @cached_property def in_memory_runner_type(self) -> str: """Runner labels like ``runner_type``, with docker on a RAM disk when the run is on RunsOn. The repository variable ``USE_IN_MEMORY_DOCKER=false`` keeps docker on disk. """ - # Reuses the routing decision of runner_type, which may have asked the GitHub API. - if ( - self.runner_type in (PUBLIC_AMD_RUNNERS, PUBLIC_ARM_RUNNERS) - or os.environ.get("USE_IN_MEMORY_DOCKER", "").casefold() == "false" - ): + if not self._is_on_runs_on or os.environ.get("USE_IN_MEMORY_DOCKER", "").casefold() == "false": return self.runner_type template = ( RUNS_ON_ARM_IN_MEMORY_RUNNER_TEMPLATE diff --git a/dev/breeze/tests/test_selective_checks.py b/dev/breeze/tests/test_selective_checks.py index ddce03204ad..00b133b40dc 100644 --- a/dev/breeze/tests/test_selective_checks.py +++ b/dev/breeze/tests/test_selective_checks.py @@ -37,7 +37,9 @@ from airflow_breeze.global_constants import ( DEFAULT_PYTHON_MAJOR_MINOR_VERSION, JAVA_SDK_VERSION, NUMBER_OF_CORE_SLICES, + NUMBER_OF_CORE_SLICES_ON_RUNS_ON, NUMBER_OF_LOW_DEP_SLICES, + NUMBER_OF_LOW_DEP_SLICES_ON_RUNS_ON, PROVIDERS_COMPATIBILITY_TESTS_MATRIX, GithubEvents, get_airflow_version, @@ -2250,6 +2252,48 @@ def test_list_splitting(): ) [email protected]( + ("github_event", "github_context_dict", "expected_core_slices", "expected_low_dep_slices"), + [ + pytest.param( + GithubEvents.PULL_REQUEST, + {}, + NUMBER_OF_CORE_SLICES, + NUMBER_OF_LOW_DEP_SLICES, + id="public runners", + ), + pytest.param( + GithubEvents.PUSH, + {"run_id": 42}, + NUMBER_OF_CORE_SLICES_ON_RUNS_ON, + NUMBER_OF_LOW_DEP_SLICES_ON_RUNS_ON, + id="RunsOn runners", + ), + ], +) +def test_number_of_test_type_slices_follows_runners( + github_event: GithubEvents, + github_context_dict: dict[str, Any], + expected_core_slices: int, + expected_low_dep_slices: int, +): + selective_checks = SelectiveChecks( + pr_labels=("full tests needed",), + default_branch="main", + github_event=github_event, + github_repository="apache/airflow", + github_context_dict=github_context_dict, + ) + core_slices = json.loads(selective_checks.core_test_types_list_as_strings_in_json) + low_dep_slices = json.loads(selective_checks.individual_providers_test_types_list_as_strings_in_json) + assert len(core_slices) == expected_core_slices + assert len(low_dep_slices) == expected_low_dep_slices + assert sorted(" ".join(s["test_types"] for s in core_slices).split()) == sorted( + ALL_CI_SELECTIVE_TEST_TYPES.split() + ) + assert sum(len(s["test_types"].split()) for s in low_dep_slices) == len(LIST_OF_ALL_PROVIDER_TESTS) + + def test_excluded_providers(): stderr = SelectiveChecks( files=(),
