Fix the isolated Dag processor deployment examples and chart checks The token provisioner example signed with the key in the chart's generated <release>-jwt-secret. Every helm upgrade of the trusted release writes a new key there and rolls the API server onto it, but the provisioner keeps the old key in its environment until it restarts, so the API server rejected every token it issued after an upgrade. The example now reads a pre-created Secret that the trusted release also uses as jwtSecretName. The processor-only Helm values set LocalExecutor just to pass the chart's validation, which also made the processor parse Dags with a different [core] executor from the trusted deployment. Tasks naming another executor then failed to import. The example now sets the processor's executor to match, and the guide says it must. The chart chooses probes, update strategy and the replica limit only from config.dag_processor.execution_api_token_file. Setting the token file through AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE instead kept a liveness probe that fails without database credentials, so the chart now rejects that variable. The chart tests now also cover rendering without a token file, including a null config.dag_processor. The guide overstated how close to token expiry a Kubernetes API outage can end without stopping the processor, because it left out the retry backoff and the Secret projection delay. Its install command also lacked a chart version, and a few smaller statements were wrong. No CI job ran the Compose deployment example tests, so the quick-start Compose job now runs them.
diff --git a/.github/workflows/additional-prod-image-tests.yml b/.github/workflows/additional-prod-image-tests.yml index 5df6fd2..dc24b01 100644 --- a/.github/workflows/additional-prod-image-tests.yml +++ b/.github/workflows/additional-prod-image-tests.yml
@@ -230,6 +230,10 @@ use-uv: ${{ inputs.use-uv }} make-mnt-writeable-and-cleanup: true id: breeze + - name: "Test Dag processor API deployment examples" + run: > + cd ./docker-tests && + uv run pytest tests/docker_tests/test_dag_processor_api_deployment.py --color=yes - name: "Test docker-compose quick start" run: breeze testing docker-compose-tests
diff --git a/airflow-core/docs/security/dag-processor-api/helm-values.yaml b/airflow-core/docs/security/dag-processor-api/helm-values.yaml index 1013e3e..1b06b3e 100644 --- a/airflow-core/docs/security/dag-processor-api/helm-values.yaml +++ b/airflow-core/docs/security/dag-processor-api/helm-values.yaml
@@ -19,6 +19,7 @@ # Apply kubernetes-token-provisioner.yaml first, using an image with API mode. airflowVersion: "3.5.0" defaultAirflowTag: "3.5.0" +# Only satisfies the chart's executor validation; the processor's executor is set in dagProcessor.env. executor: LocalExecutor scheduler: enabled: false @@ -60,6 +61,9 @@ env: - name: CONNECTION_CHECK_MAX_COUNT value: "0" + # Match the trusted deployment's [core] executor: parsing fails for tasks naming any other executor. + - name: AIRFLOW__CORE__EXECUTOR + value: CeleryExecutor extraVolumes: - name: processor-token secret:
diff --git a/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml b/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml index 3f664a9..b264f11 100644 --- a/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml +++ b/airflow-core/docs/security/dag-processor-api/kubernetes-token-provisioner.yaml
@@ -116,8 +116,10 @@ - name: AIRFLOW__API_AUTH__JWT_SECRET valueFrom: secretKeyRef: - # Match the trusted API server's jwtSecretName (default: <release>-jwt-secret). - name: airflow-jwt-secret + # A pre-created Secret that the trusted release uses as its jwtSecretName. Do not use + # the chart-generated <release>-jwt-secret: every helm upgrade replaces its key, and + # this variable keeps the old key until the provisioner restarts. + name: airflow-jwt-signing-key key: jwt-secret securityContext: allowPrivilegeEscalation: false
diff --git a/airflow-core/docs/security/jwt_token_authentication.rst b/airflow-core/docs/security/jwt_token_authentication.rst index 8acab3d..5656109 100644 --- a/airflow-core/docs/security/jwt_token_authentication.rst +++ b/airflow-core/docs/security/jwt_token_authentication.rst
@@ -705,7 +705,11 @@ For Helm, set the token-file option under ``config.dag_processor`` to select the local probe. The chart then replaces processor pods with the ``Recreate`` strategy, because pods that share a session token cannot run their Jobs side by side, and rejects more than one -replica for the same reason. Disable the processor's migration-wait init container, provision the catalog +replica for the same reason. It does not select those settings for the +``AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE`` environment variable, and rejects that +variable in ``env`` and ``dagProcessor.env``. Do not set it through ``secret``, ``extraEnv``, +or ``extraEnvFrom`` either; the chart does not check those. +Disable the processor's migration-wait init container, provision the catalog separately, and remove server credentials from the pod. The Compose example selects the local probe when ``AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE`` is set. Neither the main chart release nor the quick-start Compose file removes shared credentials or adds @@ -728,6 +732,10 @@ Token provisioning itself needs only the API's JWT signing configuration and the configured bundle names; it does not access the metadata database. +Configure the processor's ``[core] executor``, and ``[core] multi_team`` with its team +executors, to match the trusted deployment. Parsing records an import error for any Dag with +a task whose ``executor`` is not configured on the processor. + Before starting the isolated processor, stop the trusted deployment's processor for the bundles being moved. If that processor must continue serving other bundles, restrict it with ``--bundle-name`` to those other bundles. This prevents conflicting inventory reconciliation @@ -753,15 +761,27 @@ * Set both images to an image containing this feature. The manifest assumes the trusted release and namespace are named ``airflow``; update namespace references, the API service address, and NetworkPolicy selectors together if yours differ. -* Point the provisioner's ``AIRFLOW__API_AUTH__JWT_SECRET`` at the API server's signing - Secret. For the Helm chart's default configuration this is ``<release>-jwt-secret``, - key ``jwt-secret``; ``jwtSecretName`` overrides the Secret name. For asymmetric signing, - replace this environment variable with the matching private-key mount and +* Pin the trusted release's signing key, and point the provisioner's + ``AIRFLOW__API_AUTH__JWT_SECRET`` at the same Secret. Create a Secret with the key + ``jwt-secret``, and set the trusted release's ``jwtSecretName`` to its name, as the + :doc:`chart production guide <helm-chart:production-guide>` recommends. The manifest + expects it as ``airflow-jwt-signing-key``. Do not use the chart-generated + ``<release>-jwt-secret`` unless ``jwtSecret`` holds a fixed value: otherwise every + ``helm upgrade`` of the trusted release generates a new key. The API server switches to + the new key, but the provisioner keeps signing with the old one until it restarts, so the + API rejects the processor's requests. For asymmetric signing, replace this + environment variable with the matching private-key mount and ``AIRFLOW__API_AUTH__JWT_PRIVATE_KEY_PATH``. * Copy any non-default signing algorithm, key ID, issuer, and execution API audience settings from the API server. Do not copy database credentials or the entire server configuration into the provisioner. Restart the provisioner when its signing - configuration changes, including a signing Secret supplied through environment variables. + configuration changes, including a signing Secret supplied through environment variables + (``kubectl -n airflow rollout restart deployment/dag-processor-token-provisioner``). +* Set ``AIRFLOW__CORE__EXECUTOR`` in the values file's ``dagProcessor.env`` to the trusted + deployment's ``[core] executor``; the example assumes the chart's default + ``CeleryExecutor``. The values file's top-level ``executor`` only satisfies the chart's + validation for a release without a broker, and the chart rejects a ``config.core.executor`` + that differs from it. * Match ``--bundle-name`` and ``AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST`` to the catalog and processor configuration. Repeat ``--bundle-name`` for additional bundles. Provisioning reads their names without loading bundle classes or fetching Dag files. @@ -792,13 +812,16 @@ the last-applied configuration and cause the provisioner to create a new session. After preparing the files for a fresh deployment or adoption, start the provisioner -and then the processor:: +and then the processor. Replace ``<chart-version>`` with a chart release that supports +``config.dag_processor.execution_api_token_file``. Earlier releases keep the database-backed +liveness probe, which fails without database credentials, and rolling updates, in which the +new pod cannot register while the old pod holds the session:: kubectl apply -f kubernetes-token-provisioner.yaml kubectl -n airflow rollout status deployment/dag-processor-token-provisioner kubectl -n airflow-dag-processor wait --for=jsonpath='{.data.processor\.jwt}' secret/dag-processor-session --timeout=120s kubectl -n airflow-dag-processor apply -f network-policy.yaml - helm upgrade --install dag-processor apache-airflow/airflow \ + helm upgrade --install dag-processor apache-airflow/airflow --version <chart-version> \ --namespace airflow-dag-processor --values helm-values.yaml For Secrets created by this example, the manifest deliberately omits ``data`` and @@ -827,10 +850,16 @@ and reloads that credential on each request. In rotation mode, Kubernetes read and update failures are retried with backoff, including during initial issuance. Initial signing or configuration errors exit the command and Kubernetes restarts it. Later rotation failures -also retry; the previous token remains usable until it expires. An outage lasting beyond expiry -prevents the processor from renewing its API credentials. Monitor provisioner failures -and processor readiness, and restore Kubernetes API access or signing configuration -before the token expires. Monitor the token's expiry without logging its value. +also retry; the previous token remains usable until it expires. The delay between retries +doubles from 2 seconds up to the smaller of 1024 seconds and a quarter of the token +lifetime: 150 seconds with ``--valid-for 600``. The processor keeps its API credentials +only if a retry succeeds and Kubernetes projects the new token before the previous one +expires. Restore Kubernetes API access or signing configuration at least that maximum +delay plus your cluster's projection delay before the token expires. Otherwise the +processor stops when the API rejects its +expired credentials, and recovers only after Kubernetes restarts it with a newly projected +token. Monitor provisioner failures and processor readiness, and monitor the token's expiry +without logging its value. Compose token provisioner """""""""""""""""""""""""
diff --git a/chart/templates/dag-processor/dag-processor-deployment.yaml b/chart/templates/dag-processor/dag-processor-deployment.yaml index 28e3d45..f892bac 100644 --- a/chart/templates/dag-processor/dag-processor-deployment.yaml +++ b/chart/templates/dag-processor/dag-processor-deployment.yaml
@@ -24,6 +24,11 @@ {{- if and (get (.Values.config.dag_processor | default dict) "execution_api_token_file") (gt (int .Values.dagProcessor.replicas) 1) }} {{- fail "`dagProcessor.replicas` must not exceed 1 when `config.dag_processor.execution_api_token_file` is set, because the pods would share one session token. Deploy a separate release, with its own token, for each additional processor." }} {{- end }} +{{- range concat (.Values.env | default list) (.Values.dagProcessor.env | default list) }} +{{- if eq .name "AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE" }} +{{- fail "Set the Dag processor token file with `config.dag_processor.execution_api_token_file`, not the `AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE` environment variable: the chart selects probes, update strategy and replica limits that work without database credentials only from the config value." }} +{{- end }} +{{- end }} {{- $nodeSelector := or .Values.dagProcessor.nodeSelector .Values.nodeSelector }} {{- $affinity := or .Values.dagProcessor.affinity .Values.affinity }} {{- $tolerations := or .Values.dagProcessor.tolerations .Values.tolerations }}
diff --git a/chart/tests/helm_tests/airflow_core/test_dag_processor.py b/chart/tests/helm_tests/airflow_core/test_dag_processor.py index 3825747..01f89c4 100644 --- a/chart/tests/helm_tests/airflow_core/test_dag_processor.py +++ b/chart/tests/helm_tests/airflow_core/test_dag_processor.py
@@ -478,6 +478,37 @@ docs = render_chart(values=values, show_only=show_only) assert jmespath.search("spec.replicas", docs[0]) == replicas + @pytest.mark.parametrize( + "values", + [ + {"env": [{"name": "AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE", "value": "/run/t"}]}, + { + "dagProcessor": { + "env": [{"name": "AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE", "value": "/run/t"}] + } + }, + ], + ids=["env", "dagProcessor.env"], + ) + def test_rejects_the_token_file_environment_variable(self, values): + with pytest.raises(CalledProcessError, match="config.dag_processor.execution_api_token_file"): + render_chart(values=values, show_only=["templates/dag-processor/dag-processor-deployment.yaml"]) + + @pytest.mark.parametrize("config", [{}, {"dag_processor": None}]) + def test_database_mode_keeps_database_probe_and_replicas(self, config): + docs = render_chart( + values={"config": config, "dagProcessor": {"replicas": 2}}, + show_only=["templates/dag-processor/dag-processor-deployment.yaml"], + ) + + assert jmespath.search("spec.replicas", docs[0]) == 2 + container = jmespath.search("spec.template.spec.containers[0]", docs[0]) + assert "readinessProbe" not in container + assert ( + "airflow jobs check --local --job-type DagProcessorJob" + in container["livenessProbe"]["exec"]["command"][-1] + ) + def test_isolated_api_deployment_example(self): examples = Path(__file__).resolve().parents[4] / "airflow-core/docs/security/dag-processor-api" values = yaml.safe_load((examples / "helm-values.yaml").read_text()) @@ -493,9 +524,13 @@ assert not container.get("envFrom") assert all("valueFrom" not in env for env in container.get("env", [])) processor = next(container for container in pod["containers"] if container["name"] == "dag-processor") + trusted_executor = yaml.safe_load((Path(__file__).resolve().parents[3] / "values.yaml").read_text())[ + "executor" + ] assert {env["name"]: env["value"] for env in processor["env"]} == { "AIRFLOW_HOME": "/opt/airflow", "CONNECTION_CHECK_MAX_COUNT": "0", + "AIRFLOW__CORE__EXECUTOR": trusted_executor, } assert "--check-health" in processor["livenessProbe"]["exec"]["command"][-1] assert "--check-ready" in processor["readinessProbe"]["exec"]["command"][-1] @@ -621,7 +656,7 @@ assert not provisioner.get("envFrom") env = {entry["name"]: entry for entry in provisioner["env"]} signing_secret = env.pop("AIRFLOW__API_AUTH__JWT_SECRET")["valueFrom"]["secretKeyRef"] - assert signing_secret == {"name": "airflow-jwt-secret", "key": "jwt-secret"} + assert signing_secret == {"name": "airflow-jwt-signing-key", "key": "jwt-secret"} bundles = json.loads(env.pop("AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST")["value"]) assert {bundle["name"] for bundle in bundles} == {"dags-folder"} assert {name: entry["value"] for name, entry in env.items()} == { @@ -638,6 +673,26 @@ assert token_volume["secret"]["secretName"] != signing_secret["name"] assert provisioner["image"] == f"apache/airflow:{values['defaultAirflowTag']}" + def test_kubernetes_token_provisioner_example_uses_a_pinned_signing_key(self): + examples = Path(__file__).resolve().parents[4] / "airflow-core/docs/security/dag-processor-api" + docs = yaml.safe_load_all((examples / "kubernetes-token-provisioner.yaml").read_text()) + deployment = next(doc for doc in docs if doc["kind"] == "Deployment") + env = deployment["spec"]["template"]["spec"]["containers"][0]["env"] + signing_secret = next(entry for entry in env if entry["name"] == "AIRFLOW__API_AUTH__JWT_SECRET") + policy = yaml.safe_load((examples / "network-policy.yaml").read_text())["spec"] + trusted_release = policy["egress"][0]["to"][0]["podSelector"]["matchLabels"]["release"] + + generated = render_chart(name=trusted_release, show_only=["templates/secrets/jwt-secret.yaml"]) + generated_names = {doc["metadata"]["name"] for doc in generated} + assert generated_names == {f"{trusted_release}-jwt-secret"} + assert signing_secret["valueFrom"]["secretKeyRef"]["name"] not in generated_names + pinned = render_chart( + name=trusted_release, + values={"jwtSecretName": signing_secret["valueFrom"]["secretKeyRef"]["name"]}, + show_only=["templates/api-server/api-server-deployment.yaml"], + ) + assert signing_secret in jmespath.search("spec.template.spec.containers[0].env", pinned[0]) + @pytest.mark.parametrize( ("log_values", "expected_volume"), [
diff --git a/contributing-docs/testing/docker_compose_tests.rst b/contributing-docs/testing/docker_compose_tests.rst index a225264..2b3aa0e 100644 --- a/contributing-docs/testing/docker_compose_tests.rst +++ b/contributing-docs/testing/docker_compose_tests.rst
@@ -64,6 +64,13 @@ to see the output of the test as it happens (it can be also set via ``WAIT_FOR_CONTAINERS_TIMEOUT`` environment variable) +The same CI job also checks the Dag processor API deployment examples. That test needs only +``docker compose config``, not a running deployment or the PROD image: + +.. code-block:: bash + + cd docker-tests && uv run pytest tests/docker_tests/test_dag_processor_api_deployment.py + The test can be also run manually with ``pytest docker_tests/test_docker_compose_quick_start.py`` command, provided that you have a local Airflow venv with ``dev`` extra set and the ``DOCKER_IMAGE`` environment variable is set to the image you want to test. The variable defaults