From e93230bbe6f1cbc0956ffaf1a746c9974a1fa23d Mon Sep 17 00:00:00 2001 From: PoAn Yang Date: Sat, 5 Sep 2026 14:44:49 +0900 Subject: [PATCH] Keep provider_config_fallback_defaults.cfg in sync with provider.yaml Signed-off-by: PoAn Yang --- .pre-commit-config.yaml | 12 + .../docs/cli-and-env-variables-ref.rst | 1 - airflow-core/docs/howto/set-config.rst | 1 - .../provider_config_fallback_defaults.cfg | 11 +- .../src/tests_common/test_utils/config.py | 2 +- ...check_provider_config_fallback_defaults.py | 200 +++++++++++++++ ...check_provider_config_fallback_defaults.py | 227 ++++++++++++++++++ 7 files changed, 441 insertions(+), 13 deletions(-) create mode 100755 scripts/ci/prek/check_provider_config_fallback_defaults.py create mode 100644 scripts/tests/ci/prek/test_check_provider_config_fallback_defaults.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index e71e7fedd09d0..d08be10e44cd7 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1148,6 +1148,18 @@ repos: pass_filenames: false require_serial: true additional_dependencies: ['packaging>=25', 'pyyaml', 'tomli>=2.0.1', 'rich>=13.6.0'] + - id: check-provider-config-fallback-defaults + name: Check provider_config_fallback_defaults.cfg is in sync with provider.yaml files + language: python + entry: ./scripts/ci/prek/check_provider_config_fallback_defaults.py + files: > + (?x) + ^airflow-core/src/airflow/config_templates/provider_config_fallback_defaults\.cfg$| + ^providers/.*/provider\.yaml$| + ^scripts/ci/prek/check_provider_config_fallback_defaults\.py$ + pass_filenames: false + require_serial: true + additional_dependencies: ['pyyaml', 'rich>=13.6.0'] - id: check-dependency-lower-bounds name: Check that dependencies in pyproject.toml have lower bounds language: python diff --git a/airflow-core/docs/cli-and-env-variables-ref.rst b/airflow-core/docs/cli-and-env-variables-ref.rst index 309d17e43561c..abf53aee77d8b 100644 --- a/airflow-core/docs/cli-and-env-variables-ref.rst +++ b/airflow-core/docs/cli-and-env-variables-ref.rst @@ -80,7 +80,6 @@ Environment Variables * ``broker_url`` in ``[celery]`` section * ``flower_basic_auth`` in ``[celery]`` section * ``result_backend`` in ``[celery]`` section -* ``password`` in ``[atlas]`` section * ``smtp_password`` in ``[smtp]`` section * ``secret_key`` in ``[api]`` section diff --git a/airflow-core/docs/howto/set-config.rst b/airflow-core/docs/howto/set-config.rst index 31c6b5362518b..7a98258459318 100644 --- a/airflow-core/docs/howto/set-config.rst +++ b/airflow-core/docs/howto/set-config.rst @@ -103,7 +103,6 @@ The following config options support this ``_cmd`` and ``_secret`` version: * ``broker_url`` in ``[celery]`` section * ``flower_basic_auth`` in ``[celery]`` section * ``result_backend`` in ``[celery]`` section -* ``password`` in ``[atlas]`` section * ``smtp_password`` in ``[smtp]`` section * ``secret_key`` in ``[api]`` section * ``jwt_secret`` in ``[api_auth]`` section diff --git a/airflow-core/src/airflow/config_templates/provider_config_fallback_defaults.cfg b/airflow-core/src/airflow/config_templates/provider_config_fallback_defaults.cfg index 691fe6064e8f0..354a486221a3d 100644 --- a/airflow-core/src/airflow/config_templates/provider_config_fallback_defaults.cfg +++ b/airflow-core/src/airflow/config_templates/provider_config_fallback_defaults.cfg @@ -32,13 +32,6 @@ # You've been warned! # -[atlas] -sasl_enabled = False -host = -port = 21000 -username = -password = - [hive] default_hive_mapred_queue = @@ -69,7 +62,6 @@ pool = prefork operation_timeout = 1.0 task_track_started = True task_publish_max_retries = 3 -worker_precheck = False [elasticsearch] host = @@ -86,7 +78,7 @@ write_to_es = False target_index = airflow-logs [elasticsearch_configs] -use_ssl = False +http_compress = False verify_certs = True [opensearch] @@ -132,5 +124,4 @@ tcp_keep_idle = 120 tcp_keep_intvl = 30 tcp_keep_cnt = 6 verify_ssl = True -worker_pods_queued_check_interval = 60 ssl_ca_cert = diff --git a/devel-common/src/tests_common/test_utils/config.py b/devel-common/src/tests_common/test_utils/config.py index 9a278346d3068..9893da665a4a7 100644 --- a/devel-common/src/tests_common/test_utils/config.py +++ b/devel-common/src/tests_common/test_utils/config.py @@ -45,7 +45,7 @@ CFG_FALLBACK_CONFIG_OPTIONS: list[tuple[str, str, str]] = [ ("celery", "flower_host", "0.0.0.0"), ("celery", "pool", "prefork"), - ("celery", "worker_precheck", "False"), + ("celery", "worker_prefetch_multiplier", "1"), ("kubernetes_executor", "in_cluster", "True"), ("kubernetes_executor", "verify_ssl", "True"), ("elasticsearch", "end_of_log_mark", "end_of_log"), diff --git a/scripts/ci/prek/check_provider_config_fallback_defaults.py b/scripts/ci/prek/check_provider_config_fallback_defaults.py new file mode 100755 index 0000000000000..3daaeb8e54145 --- /dev/null +++ b/scripts/ci/prek/check_provider_config_fallback_defaults.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# /// script +# requires-python = ">=3.10,<3.11" +# dependencies = [ +# "pyyaml", +# "rich>=13.6.0", +# ] +# /// +""" +Check that ``provider_config_fallback_defaults.cfg`` only carries options providers still declare. + +``airflow-core/src/airflow/config_templates/provider_config_fallback_defaults.cfg`` is a hand-maintained +subset of provider configuration: the defaults core may need while provider modules are being imported, +before the providers' own configuration has been loaded. Everything in it is merged into ``conf`` as if +the provider had declared it (``has_option``, ``as_dict``, ``airflow config list``), so an entry that no +``provider.yaml`` declares any more keeps a removed option alive indefinitely. + +This check enforces the "cfg is a subset of provider.yaml" direction only: every ``(section, option)`` +in the cfg must be declared in the ``config`` section of some ``providers/**/provider.yaml`` with the +same default. +""" + +from __future__ import annotations + +import sys +from configparser import ConfigParser +from pathlib import Path +from typing import TYPE_CHECKING, NamedTuple + +import yaml +from common_prek_utils import ( + AIRFLOW_CORE_SOURCES_PATH, + AIRFLOW_ROOT_PATH, + console, + get_all_provider_yaml_files, +) +from rich.markup import escape + +if TYPE_CHECKING: + from collections.abc import Iterable + +FALLBACK_CFG_PATH = ( + AIRFLOW_CORE_SOURCES_PATH / "airflow" / "config_templates" / "provider_config_fallback_defaults.cfg" +) + +# Options whose cfg default intentionally differs from the provider.yaml default. +# (section, option, provider.yaml default, cfg default) +# +# Keep in sync with PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK in +# devel-common/src/tests_common/test_utils/config.py - the unit test in +# scripts/tests/ci/prek/test_check_provider_config_fallback_defaults.py fails when the two lists differ. +PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK: list[tuple[str, str, str, str]] = [ + ( + "celery", + "celery_app_name", + "airflow.providers.celery.executors.celery_executor", + "airflow.executors.celery_executor", + ), +] + + +class ProviderOption(NamedTuple): + """Default declared for a config option in a provider.yaml (``None`` when declared as ``~``).""" + + default: str | None + provider_yaml: Path + + +def load_fallback_cfg(cfg_path: Path) -> dict[str, dict[str, str]]: + """Return ``{section: {option: value}}`` for the fallback cfg, lower-cased and without interpolation.""" + parser = ConfigParser(interpolation=None) + parser.read_string(cfg_path.read_text(), source=str(cfg_path)) + return { + section.lower(): {option.lower(): value for option, value in parser.items(section)} + for section in parser.sections() + } + + +def load_provider_config_options( + provider_yaml_files: Iterable[Path], +) -> dict[tuple[str, str], ProviderOption]: + """Return ``{(section, option): ProviderOption}`` for every config option the given provider.yaml files declare.""" + loader = getattr(yaml, "CSafeLoader", yaml.SafeLoader) + options: dict[tuple[str, str], ProviderOption] = {} + for provider_yaml in sorted(provider_yaml_files): + provider_info = yaml.load(provider_yaml.read_text(), Loader=loader) + for section, section_content in (provider_info.get("config") or {}).items(): + for option, option_content in ((section_content or {}).get("options") or {}).items(): + default = (option_content or {}).get("default") + options[(section.lower(), option.lower())] = ProviderOption( + default=None if default is None else str(default), + provider_yaml=provider_yaml, + ) + return options + + +def _display_path(path: Path) -> str: + try: + return path.relative_to(AIRFLOW_ROOT_PATH).as_posix() + except ValueError: + return path.as_posix() + + +def find_drift( + cfg: dict[str, dict[str, str]], + provider_options: dict[tuple[str, str], ProviderOption], + overrides: Iterable[tuple[str, str, str, str]] = PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK, +) -> list[str]: + """ + Return one message per cfg entry that no provider.yaml declares, or declares with a different default. + + Sections that no provider.yaml declares at all are reported once, not once per option. + """ + provider_sections = {section for section, _ in provider_options} + intentional = { + (section, option): (metadata_default, cfg_default) + for section, option, metadata_default, cfg_default in overrides + } + errors: list[str] = [] + for section, options in cfg.items(): + if section not in provider_sections: + errors.append( + f"[{section}]: no provider.yaml declares this section any more - remove the whole section" + ) + continue + for option, cfg_value in options.items(): + declared = provider_options.get((section, option)) + if declared is None: + errors.append( + f"[{section}] {option}: no provider.yaml declares this option any more - remove it" + ) + continue + expected = intentional.get((section, option)) + if expected is not None: + metadata_default, cfg_default = expected + if declared.default != metadata_default or cfg_value != cfg_default: + errors.append( + f"[{section}] {option}: intentional override is out of date - " + f"{_display_path(declared.provider_yaml)} declares {declared.default!r}, " + f"the cfg has {cfg_value!r}, but PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK " + f"expects ({metadata_default!r}, {cfg_default!r})" + ) + continue + if declared.default is None: + errors.append( + f"[{section}] {option} = {cfg_value!r}: " + f"{_display_path(declared.provider_yaml)} declares no default for it" + ) + elif declared.default != cfg_value: + errors.append( + f"[{section}] {option} = {cfg_value!r}: " + f"{_display_path(declared.provider_yaml)} declares default {declared.default!r}" + ) + return errors + + +def main() -> int: + cfg = load_fallback_cfg(FALLBACK_CFG_PATH) + provider_options = load_provider_config_options(get_all_provider_yaml_files()) + errors = find_drift(cfg, provider_options) + if errors: + console.print( + f"\n[red]{_display_path(FALLBACK_CFG_PATH)} is out of sync with provider.yaml files![/]\n" + ) + for error in errors: + # Section names look like Rich markup tags, so escape them before printing. + console.print(f" - {escape(error)}") + console.print( + "\n[yellow]Every option in the fallback cfg must be declared in some providers/**/provider.yaml " + "with the same default. If a provider dropped or renamed an option, drop it from the cfg too. " + "If a default has to differ on purpose, add it to PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK both " + "in this script and in devel-common/src/tests_common/test_utils/config.py.[/]\n" + ) + return 1 + checked = sum(len(options) for options in cfg.values()) + console.print( + f"[green]All {checked} fallback defaults in {len(cfg)} sections are declared " + "in provider.yaml files with matching defaults.[/]" + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/ci/prek/test_check_provider_config_fallback_defaults.py b/scripts/tests/ci/prek/test_check_provider_config_fallback_defaults.py new file mode 100644 index 0000000000000..fd219acdfe29c --- /dev/null +++ b/scripts/tests/ci/prek/test_check_provider_config_fallback_defaults.py @@ -0,0 +1,227 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import re +import textwrap +from pathlib import Path + +import check_provider_config_fallback_defaults as hook +import pytest +import yaml +from check_provider_config_fallback_defaults import ( + FALLBACK_CFG_PATH, + PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK, + ProviderOption, + find_drift, + load_fallback_cfg, + load_provider_config_options, +) +from common_prek_utils import get_all_provider_yaml_files + +CELERY_YAML = Path("providers/celery/provider.yaml") +CNCF_YAML = Path("providers/cncf/kubernetes/provider.yaml") + + +def _write_cfg(tmp_path: Path, content: str) -> Path: + path = tmp_path / "provider_config_fallback_defaults.cfg" + path.write_text(textwrap.dedent(content)) + return path + + +def _write_provider_yaml(tmp_path: Path, provider_id: str, config: dict) -> Path: + path = tmp_path / "providers" / provider_id / "provider.yaml" + path.parent.mkdir(parents=True) + path.write_text( + yaml.dump( + {"package-name": f"apache-airflow-providers-{provider_id}", "state": "ready", "config": config} + ) + ) + return path + + +def _declared(provider_yaml: Path, **defaults: str | None) -> dict[tuple[str, str], ProviderOption]: + """Build ``{("celery", option): ProviderOption}`` for the given ``option=default`` pairs.""" + return { + ("celery", option): ProviderOption(default, provider_yaml) for option, default in defaults.items() + } + + +class TestLoadFallbackCfg: + def test_lower_cases_sections_and_options_and_keeps_empty_values(self, tmp_path): + cfg_path = _write_cfg( + tmp_path, + """ + [Celery] + Pool = prefork + flower_url_prefix = + + [elasticsearch] + log_id_template = {dag_id}-{task_id}-{run_id}-{map_index}-{try_number} + """, + ) + assert load_fallback_cfg(cfg_path) == { + "celery": {"pool": "prefork", "flower_url_prefix": ""}, + "elasticsearch": {"log_id_template": "{dag_id}-{task_id}-{run_id}-{map_index}-{try_number}"}, + } + + def test_percent_signs_are_not_interpolated(self, tmp_path): + cfg_path = _write_cfg(tmp_path, "[logging]\nfmt = %(asctime)s\n") + assert load_fallback_cfg(cfg_path) == {"logging": {"fmt": "%(asctime)s"}} + + +class TestLoadProviderConfigOptions: + def test_collects_defaults_across_files(self, tmp_path): + celery_yaml = _write_provider_yaml( + tmp_path, + "celery", + { + "celery": {"options": {"pool": {"default": "prefork"}, "result_backend": {"default": None}}}, + "celery_kubernetes_executor": {"options": {"kubernetes_queue": {"default": "kubernetes"}}}, + }, + ) + cncf_yaml = _write_provider_yaml( + tmp_path, + "cncf/kubernetes", + {"kubernetes_executor": {"options": {"tcp_keep_idle": {"default": 120}}}}, + ) + assert load_provider_config_options([celery_yaml, cncf_yaml]) == { + ("celery", "pool"): ProviderOption("prefork", celery_yaml), + ("celery", "result_backend"): ProviderOption(None, celery_yaml), + ("celery_kubernetes_executor", "kubernetes_queue"): ProviderOption("kubernetes", celery_yaml), + ("kubernetes_executor", "tcp_keep_idle"): ProviderOption("120", cncf_yaml), + } + + def test_provider_without_config_is_ignored(self, tmp_path): + path = tmp_path / "provider.yaml" + path.write_text(yaml.dump({"package-name": "apache-airflow-providers-sqlite"})) + assert load_provider_config_options([path]) == {} + + +class TestFindDrift: + def test_in_sync_cfg_has_no_errors(self): + cfg = {"celery": {"pool": "prefork", "flower_url_prefix": ""}} + assert find_drift(cfg, _declared(CELERY_YAML, pool="prefork", flower_url_prefix="")) == [] + + def test_section_nobody_declares_is_reported_once(self): + cfg = {"atlas": {"host": "", "port": "21000"}, "celery": {"pool": "prefork"}} + errors = find_drift(cfg, _declared(CELERY_YAML, pool="prefork")) + assert len(errors) == 1 + assert errors[0].startswith("[atlas]:") + assert "remove the whole section" in errors[0] + + def test_option_nobody_declares_is_reported(self): + cfg = {"celery": {"pool": "prefork", "worker_precheck": "False"}} + errors = find_drift(cfg, _declared(CELERY_YAML, pool="prefork")) + assert errors == [ + "[celery] worker_precheck: no provider.yaml declares this option any more - remove it" + ] + + def test_different_default_is_reported_with_both_values(self): + cfg = {"celery": {"worker_concurrency": "16"}} + errors = find_drift(cfg, _declared(CELERY_YAML, worker_concurrency="32")) + assert errors == [ + "[celery] worker_concurrency = '16': providers/celery/provider.yaml declares default '32'" + ] + + def test_provider_declaring_no_default_is_reported(self): + cfg = {"celery": {"result_backend": ""}} + errors = find_drift(cfg, _declared(CELERY_YAML, result_backend=None)) + assert errors == [ + "[celery] result_backend = '': providers/celery/provider.yaml declares no default for it" + ] + + def test_intentional_override_is_accepted(self): + cfg = {"celery": {"celery_app_name": "old.module"}} + declared = _declared(CELERY_YAML, celery_app_name="new.module") + overrides = [("celery", "celery_app_name", "new.module", "old.module")] + assert find_drift(cfg, declared, overrides) == [] + + @pytest.mark.parametrize( + ("provider_default", "cfg_value"), + [ + pytest.param("renamed.module", "old.module", id="provider-default-changed"), + pytest.param("new.module", "edited.module", id="cfg-value-changed"), + ], + ) + def test_stale_intentional_override_is_reported(self, provider_default, cfg_value): + cfg = {"celery": {"celery_app_name": cfg_value}} + declared = _declared(CELERY_YAML, celery_app_name=provider_default) + overrides = [("celery", "celery_app_name", "new.module", "old.module")] + errors = find_drift(cfg, declared, overrides) + assert len(errors) == 1 + assert "intentional override is out of date" in errors[0] + assert "PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK" in errors[0] + + def test_intentional_override_does_not_resurrect_removed_option(self): + cfg = {"celery": {"celery_app_name": "old.module"}} + overrides = [("celery", "celery_app_name", "new.module", "old.module")] + errors = find_drift(cfg, _declared(CELERY_YAML, pool="prefork"), overrides) + assert errors == [ + "[celery] celery_app_name: no provider.yaml declares this option any more - remove it" + ] + + def test_all_problems_are_reported_together(self): + cfg = { + "atlas": {"host": ""}, + "celery": {"pool": "prefork", "worker_precheck": "False", "worker_concurrency": "16"}, + } + declared = _declared(CELERY_YAML, pool="prefork", worker_concurrency="32") + errors = find_drift(cfg, declared) + assert [error.split(":")[0] for error in errors] == [ + "[atlas]", + "[celery] worker_precheck", + "[celery] worker_concurrency = '16'", + ] + + +class TestMain: + def _run_main(self, monkeypatch, tmp_path, cfg_content: str, config: dict) -> int: + cfg_path = _write_cfg(tmp_path, cfg_content) + provider_yaml = _write_provider_yaml(tmp_path, "celery", config) + monkeypatch.setattr(hook, "FALLBACK_CFG_PATH", cfg_path) + monkeypatch.setattr(hook, "get_all_provider_yaml_files", lambda: [provider_yaml]) + return hook.main() + + def test_returns_zero_when_in_sync(self, monkeypatch, tmp_path): + config = {"celery": {"options": {"pool": {"default": "prefork"}}}} + assert self._run_main(monkeypatch, tmp_path, "[celery]\npool = prefork\n", config) == 0 + + def test_returns_one_on_drift(self, monkeypatch, tmp_path, capsys): + config = {"celery": {"options": {"pool": {"default": "prefork"}}}} + assert ( + self._run_main( + monkeypatch, tmp_path, "[celery]\npool = prefork\nworker_precheck = False\n", config + ) + == 1 + ) + plain_output = re.sub(r"\x1b\[[0-9;]*m", "", capsys.readouterr().out) + assert "[celery] worker_precheck: no provider.yaml declares this option any more" in plain_output + + +class TestRepositoryState: + def test_fallback_cfg_is_in_sync_with_provider_yaml_files(self): + cfg = load_fallback_cfg(FALLBACK_CFG_PATH) + provider_options = load_provider_config_options(get_all_provider_yaml_files()) + assert find_drift(cfg, provider_options) == [] + + def test_intentional_overrides_match_tests_common(self): + from tests_common.test_utils.config import ( + PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK as TESTS_COMMON_OVERRIDES, + ) + + assert sorted(PROVIDER_METADATA_OVERRIDES_CFG_FALLBACK) == sorted(TESTS_COMMON_OVERRIDES)