This is an automated email from the ASF dual-hosted git repository. ephraimbuddy pushed a commit to branch dag-processor-deployment in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 411b576078ee8467c23df665aa740f2e23232ad5 Author: Ephraim Anierobi <[email protected]> AuthorDate: Thu Oct 15 11:52:53 2026 +0100 Add API-mode Dag processor deployment support API-mode processors need health probes that work without database credentials. Each processor session must also remain exclusive while pods are replaced, and isolated deployments need explicit credential and network boundaries around the trusted API server. --- .../docs/howto/docker-compose/docker-compose.yaml | 9 +- .../security/dag-processor-api/helm-values.yaml | 74 ++++++++++++++ .../security/dag-processor-api/network-policy.yaml | 51 ++++++++++ .../docs/security/jwt_token_authentication.rst | 22 +++++ chart/templates/_helpers.yaml | 5 + .../dag-processor/dag-processor-deployment.yaml | 21 +++- .../helm_tests/airflow_core/test_dag_processor.py | 107 +++++++++++++++++++++ chart/values.schema.json | 4 +- chart/values.yaml | 7 +- .../test_dag_processor_api_deployment.py | 99 +++++++++++++++++++ 10 files changed, 393 insertions(+), 6 deletions(-) diff --git a/airflow-core/docs/howto/docker-compose/docker-compose.yaml b/airflow-core/docs/howto/docker-compose/docker-compose.yaml index b68f1afd3cc..54f4dfe603f 100644 --- a/airflow-core/docs/howto/docker-compose/docker-compose.yaml +++ b/airflow-core/docs/howto/docker-compose/docker-compose.yaml @@ -155,7 +155,14 @@ services: <<: *airflow-common command: dag-processor healthcheck: - test: ["CMD-SHELL", 'airflow jobs check --job-type DagProcessorJob --hostname "$${HOSTNAME}"'] + test: + - CMD-SHELL + - >- + if [ -n "$${AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE:-}" ]; then + airflow dag-processor --check-health; + else + airflow jobs check --job-type DagProcessorJob --hostname "$${HOSTNAME}"; + fi interval: 30s timeout: 10s retries: 5 diff --git a/airflow-core/docs/security/dag-processor-api/helm-values.yaml b/airflow-core/docs/security/dag-processor-api/helm-values.yaml new file mode 100644 index 00000000000..a4ecfcf877d --- /dev/null +++ b/airflow-core/docs/security/dag-processor-api/helm-values.yaml @@ -0,0 +1,74 @@ +# 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. +--- +# Install as a separate processor-only release, using an image with API mode. +airflowVersion: "3.5.0" +defaultAirflowTag: "3.5.0" +executor: LocalExecutor +scheduler: + enabled: false +apiServer: + enabled: false +triggerer: + enabled: false +createUserJob: + enabled: false +migrateDatabaseJob: + enabled: false +postgresql: + enabled: false +redis: + enabled: false +statsd: + enabled: false +rbac: + create: false +enableBuiltInSecretEnvVars: + AIRFLOW__CORE__FERNET_KEY: false + AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: false + AIRFLOW_CONN_AIRFLOW_DB: false + AIRFLOW__API__SECRET_KEY: false + AIRFLOW__API_AUTH__JWT_SECRET: false + AIRFLOW__CELERY__RESULT_BACKEND: false + AIRFLOW__CELERY__BROKER_URL: false +config: + core: + execution_api_server_url: http://airflow-api-server.airflow.svc.cluster.local:8080/execution/ + dag_processor: + execution_api_token_file: /run/dag-processor/processor.jwt +dagProcessor: + replicas: 1 + waitForMigrations: + enabled: false + serviceAccount: + automountServiceAccountToken: false + env: + - name: CONNECTION_CHECK_MAX_COUNT + value: "0" + extraVolumes: + - name: processor-token + secret: + secretName: dag-processor-session + defaultMode: 0440 + extraVolumeMounts: + - name: processor-token + mountPath: /run/dag-processor + readOnly: true +dags: + persistence: + enabled: true + existingClaim: processor-dags diff --git a/airflow-core/docs/security/dag-processor-api/network-policy.yaml b/airflow-core/docs/security/dag-processor-api/network-policy.yaml new file mode 100644 index 00000000000..ae021ce64da --- /dev/null +++ b/airflow-core/docs/security/dag-processor-api/network-policy.yaml @@ -0,0 +1,51 @@ +# 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. +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: dag-processor-api-only +spec: + podSelector: + matchLabels: + component: dag-processor + policyTypes: [Ingress, Egress] + ingress: [] + egress: + - to: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: airflow + podSelector: + matchLabels: + component: api-server + release: airflow + ports: + - protocol: TCP + port: 8080 + - to: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: kube-system + podSelector: + matchLabels: + k8s-app: kube-dns + ports: + - protocol: UDP + port: 53 + - protocol: TCP + port: 53 diff --git a/airflow-core/docs/security/jwt_token_authentication.rst b/airflow-core/docs/security/jwt_token_authentication.rst index e978e054bf2..c0c5f01e903 100644 --- a/airflow-core/docs/security/jwt_token_authentication.rst +++ b/airflow-core/docs/security/jwt_token_authentication.rst @@ -683,6 +683,14 @@ without querying the database. Give each processor its own writable health file. manager fails liveness even while the API remains healthy; a transient API outage affects readiness without immediately failing liveness. +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 +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 +database-denying network policies automatically; the isolated examples below do. Set ``CONNECTION_CHECK_MAX_COUNT=0`` in the processor container too: the standard image entrypoint otherwise checks the database before starting the command. Do not set ``_AIRFLOW_DB_MIGRATE`` or ``_AIRFLOW_WWW_USER_CREATE`` in that container. @@ -696,6 +704,20 @@ Synchronize the bundle catalog and run token provisioning in the trusted deploym described above. Keep the token provisioner running, and mount its directory rather than a single file so atomic token replacement remains visible. +For Kubernetes, use the :download:`processor-only Helm values <dag-processor-api/helm-values.yaml>` +in a **separate release and namespace**. These values disable other components and built-in +server credential injection. Do not apply them to the release running the scheduler/API server. +Create the ``dag-processor-session`` Secret containing ``processor.jwt`` and the +``processor-dags`` PVC in that namespace, then configure the API service address and image. +The processor service account has no mounted API token; a trusted provisioner updates the Secret. + +Apply the :download:`egress policy <dag-processor-api/network-policy.yaml>` in the processor +namespace before starting its pods. It permits only DNS and the trusted API server. Adjust +its namespace/pod selectors and ports to match the deployment. NetworkPolicy enforcement +requires a supporting cluster network plugin; other policies selecting these pods must not +grant broader egress. This example uses local Dag files. Remote bundles need additional, +explicit egress to their source service. + For Compose, use the :download:`standalone processor definition <dag-processor-api/docker-compose.yaml>` as its own project, rather than merging it with the quick-start file. It has no shared ``env_file``, server configuration mount, or dependency on a database/migration service. diff --git a/chart/templates/_helpers.yaml b/chart/templates/_helpers.yaml index 9a5238fdc8c..fb84e9c7d3b 100644 --- a/chart/templates/_helpers.yaml +++ b/chart/templates/_helpers.yaml @@ -838,8 +838,13 @@ server_tls_key_file = /etc/pgbouncer/server.key - sh - -c - | + {{- if (get (.Values.config.dag_processor | default dict) "execution_api_token_file") }} + CONNECTION_CHECK_MAX_COUNT=0 AIRFLOW__LOGGING__LOGGING_LEVEL=ERROR exec /entrypoint \ + airflow dag-processor --check-health + {{- else }} CONNECTION_CHECK_MAX_COUNT=0 AIRFLOW__LOGGING__LOGGING_LEVEL=ERROR exec /entrypoint \ airflow jobs check --local --job-type DagProcessorJob + {{- end }} {{- end }} {{- define "registry_docker_config" }} diff --git a/chart/templates/dag-processor/dag-processor-deployment.yaml b/chart/templates/dag-processor/dag-processor-deployment.yaml index ca8453c2bfe..28e3d45f213 100644 --- a/chart/templates/dag-processor/dag-processor-deployment.yaml +++ b/chart/templates/dag-processor/dag-processor-deployment.yaml @@ -21,6 +21,9 @@ ## Airflow Dag Processor Deployment ################################# {{- if .Values.dagProcessor.enabled }} +{{- 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 }} {{- $nodeSelector := or .Values.dagProcessor.nodeSelector .Values.nodeSelector }} {{- $affinity := or .Values.dagProcessor.affinity .Values.affinity }} {{- $tolerations := or .Values.dagProcessor.tolerations .Values.tolerations }} @@ -58,7 +61,11 @@ spec: tier: airflow component: dag-processor release: {{ .Release.Name }} - {{- if .Values.dagProcessor.strategy }} + {{- if (get (.Values.config.dag_processor | default dict) "execution_api_token_file") }} + {{- /* Pods share one session token, and a new pod can register its Job only after the old one stops. */}} + strategy: + type: Recreate + {{- else if .Values.dagProcessor.strategy }} strategy: {{- toYaml .Values.dagProcessor.strategy | nindent 4 }} {{- end }} template: @@ -189,6 +196,18 @@ spec: {{- include "custom_airflow_environment" . | indent 10 }} {{- include "standard_airflow_environment" . | indent 10 }} {{- include "container_extra_envs" (list . .Values.dagProcessor.env) | indent 10 }} + {{- if (get (.Values.config.dag_processor | default dict) "execution_api_token_file") }} + readinessProbe: + timeoutSeconds: {{ .Values.dagProcessor.livenessProbe.timeoutSeconds }} + periodSeconds: {{ .Values.dagProcessor.livenessProbe.periodSeconds }} + exec: + command: + - sh + - -c + - >- + CONNECTION_CHECK_MAX_COUNT=0 AIRFLOW__LOGGING__LOGGING_LEVEL=ERROR exec /entrypoint + airflow dag-processor --check-ready + {{- end }} livenessProbe: initialDelaySeconds: {{ .Values.dagProcessor.livenessProbe.initialDelaySeconds }} timeoutSeconds: {{ .Values.dagProcessor.livenessProbe.timeoutSeconds }} 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 ec6dcaab475..7fdff7fb35a 100644 --- a/chart/tests/helm_tests/airflow_core/test_dag_processor.py +++ b/chart/tests/helm_tests/airflow_core/test_dag_processor.py @@ -16,8 +16,13 @@ # under the License. from __future__ import annotations +from configparser import ConfigParser +from pathlib import Path +from subprocess import CalledProcessError + import jmespath import pytest +import yaml from chart_utils.helm_template_generator import render_chart from chart_utils.log_groomer import LogGroomerTestBase @@ -434,6 +439,108 @@ class TestDagProcessor: in jmespath.search("spec.template.spec.containers[0].livenessProbe.exec.command", docs[0])[-1] ) + def test_api_mode_uses_local_liveness_probe(self): + docs = render_chart( + values={"config": {"dag_processor": {"execution_api_token_file": "/run/processor.jwt"}}}, + show_only=["templates/dag-processor/dag-processor-deployment.yaml"], + ) + command = jmespath.search("spec.template.spec.containers[0].livenessProbe.exec.command", docs[0])[-1] + assert "airflow dag-processor --check-health" in command + readiness = docs[0]["spec"]["template"]["spec"]["containers"][0]["readinessProbe"]["exec"]["command"] + assert "airflow dag-processor --check-ready" in readiness[-1] + assert "airflow jobs check" not in command + + def test_api_mode_replaces_pods_with_recreate(self): + docs = render_chart( + values={ + "config": {"dag_processor": {"execution_api_token_file": "/run/processor.jwt"}}, + "dagProcessor": { + "strategy": {"rollingUpdate": {"maxSurge": "100%", "maxUnavailable": "50%"}} + }, + }, + show_only=["templates/dag-processor/dag-processor-deployment.yaml"], + ) + + assert jmespath.search("spec.strategy", docs[0]) == {"type": "Recreate"} + + @pytest.mark.parametrize(("replicas", "rejected"), [(0, False), (1, False), (2, True)]) + def test_api_mode_rejects_replicas_sharing_a_session_token(self, replicas, rejected): + values = { + "config": {"dag_processor": {"execution_api_token_file": "/run/processor.jwt"}}, + "dagProcessor": {"replicas": replicas}, + } + show_only = ["templates/dag-processor/dag-processor-deployment.yaml"] + if rejected: + with pytest.raises(CalledProcessError, match="must not exceed 1"): + render_chart(values=values, show_only=show_only) + else: + docs = render_chart(values=values, show_only=show_only) + assert jmespath.search("spec.replicas", docs[0]) == replicas + + 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()) + docs = render_chart(values=values, namespace="dag-processors") + workloads = [doc for doc in docs if doc["kind"] in {"Deployment", "StatefulSet", "Job", "CronJob"}] + assert len(workloads) == 1 + deployment = workloads[0] + assert deployment["metadata"]["labels"]["component"] == "dag-processor" + assert deployment["spec"]["strategy"] == {"type": "Recreate"} + pod = deployment["spec"]["template"]["spec"] + assert not pod.get("initContainers") + for container in pod["containers"]: + 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") + assert {env["name"]: env["value"] for env in processor["env"]} == { + "AIRFLOW_HOME": "/opt/airflow", + "CONNECTION_CHECK_MAX_COUNT": "0", + } + assert "--check-health" in processor["livenessProbe"]["exec"]["command"][-1] + assert "--check-ready" in processor["readinessProbe"]["exec"]["command"][-1] + assert {v["secret"]["secretName"] for v in pod["volumes"] if "secret" in v} == { + "dag-processor-session" + } + token_mount = next(m for m in processor["volumeMounts"] if m["name"] == "processor-token") + assert token_mount["readOnly"] + assert "subPath" not in token_mount + account = next( + doc + for doc in docs + if doc["kind"] == "ServiceAccount" and doc["metadata"]["name"] == pod["serviceAccountName"] + ) + assert account["automountServiceAccountToken"] is False + configmap = next( + doc for doc in docs if doc["kind"] == "ConfigMap" and "airflow.cfg" in doc.get("data", {}) + ) + config = ConfigParser(interpolation=None) + config.read_string(configmap["data"]["airflow.cfg"]) + for section, key in ( + ("database", "sql_alchemy_conn"), + ("core", "fernet_key"), + ("api_auth", "jwt_secret"), + ("api_auth", "jwt_private_key_path"), + ): + assert not config.has_option(section, key) + assert config["dag_processor"]["execution_api_token_file"] == "/run/dag-processor/processor.jwt" + policy = yaml.safe_load((examples / "network-policy.yaml").read_text())["spec"] + labels = deployment["spec"]["template"]["metadata"]["labels"] + assert policy["podSelector"]["matchLabels"].items() <= labels.items() + assert set(policy["policyTypes"]) == {"Ingress", "Egress"} + assert not policy["ingress"] + assert len(policy["egress"]) == 2 + assert policy["egress"][0] == { + "to": [ + { + "namespaceSelector": {"matchLabels": {"kubernetes.io/metadata.name": "airflow"}}, + "podSelector": {"matchLabels": {"component": "api-server", "release": "airflow"}}, + } + ], + "ports": [{"protocol": "TCP", "port": 8080}], + } + assert policy["egress"][1]["to"][0]["podSelector"]["matchLabels"] == {"k8s-app": "kube-dns"} + assert {port["port"] for port in policy["egress"][1]["ports"]} == {53} + @pytest.mark.parametrize( ("log_values", "expected_volume"), [ diff --git a/chart/values.schema.json b/chart/values.schema.json index 49251ffa3d2..6e13f38afe1 100644 --- a/chart/values.schema.json +++ b/chart/values.schema.json @@ -4639,7 +4639,7 @@ } }, "replicas": { - "description": "Number of dag processors to run.", + "description": "Number of dag processors to run. At most 1 when ``config.dag_processor.execution_api_token_file`` is set, since the pods would share one session token.", "type": "integer", "default": 1 }, @@ -4679,7 +4679,7 @@ ] }, "strategy": { - "description": "Specifies the strategy used to replace old Pods by new ones when deployed as a Deployment.", + "description": "Specifies the strategy used to replace old Pods by new ones when deployed as a Deployment. Ignored when ``config.dag_processor.execution_api_token_file`` is set; those pods share one session token and are replaced with the Recreate strategy.", "type": [ "null", "object" diff --git a/chart/values.yaml b/chart/values.yaml index 3edbbf55c58..e8976363680 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -2185,7 +2185,8 @@ dagProcessor: # classpath: "airflow.dag_processing.bundles.local.LocalDagBundle" # kwargs: {} - # Number of Airflow dag processors in the Deployment + # Number of Airflow dag processors in the Deployment. At most 1 when + # config.dag_processor.execution_api_token_file is set, since the pods would share one session token. replicas: 1 # Max number of old ReplicaSets to retain @@ -2197,7 +2198,9 @@ dagProcessor: # Args to use when running Airflow dag processor (templated). args: ["bash", "-c", "exec airflow dag-processor"] - # Update Strategy for dag processors + # Update Strategy for dag processors. Ignored when + # config.dag_processor.execution_api_token_file is set: those pods share one session + # token, so they are always replaced with the Recreate strategy. strategy: rollingUpdate: maxSurge: "100%" diff --git a/docker-tests/tests/docker_tests/test_dag_processor_api_deployment.py b/docker-tests/tests/docker_tests/test_dag_processor_api_deployment.py new file mode 100644 index 00000000000..b2e067da35c --- /dev/null +++ b/docker-tests/tests/docker_tests/test_dag_processor_api_deployment.py @@ -0,0 +1,99 @@ +# 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 subprocess + +import pytest +import yaml +from python_on_whales import DockerClient + +from docker_tests.constants import AIRFLOW_ROOT_PATH + + [email protected]( + ("token_file", "expected_args"), + [ + (None, ["jobs", "check", "--job-type", "DagProcessorJob", "--hostname", "test-processor"]), + ("/run/processor.jwt", ["dag-processor", "--check-health"]), + ], +) +def test_quick_start_processor_healthcheck_mode(tmp_path, token_file, expected_args): + example = AIRFLOW_ROOT_PATH / "airflow-core/docs/howto/docker-compose/docker-compose.yaml" + healthcheck = yaml.safe_load(example.read_text())["services"]["airflow-dag-processor"]["healthcheck"][ + "test" + ] + assert healthcheck[0] == "CMD-SHELL" + airflow = tmp_path / "airflow" + airflow.write_text("#!/bin/sh\nprintf '%s\\n' \"$@\"\n") + airflow.chmod(0o700) + env = {"PATH": str(tmp_path), "HOSTNAME": "test-processor"} + if token_file is not None: + env["AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE"] = token_file + # Compose unescapes doubled dollar signs before passing CMD-SHELL to the container. + result = subprocess.run( + ["/bin/sh", "-c", healthcheck[1].replace("$$", "$")], + env=env, + check=True, + capture_output=True, + text=True, + ) + assert result.stdout.splitlines() == expected_args + + [email protected]("uid", [50000, 54321]) +def test_isolated_processor_compose_example(tmp_path, monkeypatch, uid): + monkeypatch.setenv("AIRFLOW_IMAGE_NAME", "processor-api:test") + monkeypatch.setenv("DAG_PROCESSOR_API_NETWORK", "test-api-only") + monkeypatch.setenv("AIRFLOW_UID", str(uid)) + credentials = { + "AIRFLOW__DATABASE__SQL_ALCHEMY_CONN": "postgresql://must-not-reach-the-processor", + "AIRFLOW__CORE__FERNET_KEY": "must-not-reach-the-processor", + "AIRFLOW__API_AUTH__JWT_SECRET": "must-not-reach-the-processor", + "AIRFLOW__API_AUTH__JWT_PRIVATE_KEY_PATH": "/server/private.key", + } + for name, value in credentials.items(): + monkeypatch.setenv(name, value) + (tmp_path / ".env").write_text("\n".join(f"{name}={value}" for name, value in credentials.items())) + example = AIRFLOW_ROOT_PATH / "airflow-core/docs/security/dag-processor-api/docker-compose.yaml" + rendered = DockerClient(compose_files=[example], compose_project_directory=tmp_path).compose.config( + return_json=True + ) + assert set(rendered["services"]) == {"dag-processor"} + processor = rendered["services"]["dag-processor"] + assert processor["command"] == ["dag-processor"] + assert processor["user"] == f"{uid}:0" + assert processor["environment"]["USER"] == "airflow" + assert not processor.get("depends_on") + assert not processor.get("env_file") + assert not credentials.keys() & processor["environment"].keys() + assert processor["environment"]["CONNECTION_CHECK_MAX_COUNT"] == "0" + assert ( + processor["environment"]["AIRFLOW__DAG_PROCESSOR__EXECUTION_API_TOKEN_FILE"] + == "/run/dag-processor/processor.jwt" + ) + assert processor["environment"]["AIRFLOW_CONFIG"] == "/tmp/airflow.cfg" + assert processor["healthcheck"]["test"] == ["CMD", "airflow", "dag-processor", "--check-health"] + assert set(processor["networks"]) == {"processor-api"} + assert rendered["networks"]["processor-api"]["name"] == "test-api-only" + assert rendered["networks"]["processor-api"]["external"] + mounts = {mount["target"]: mount for mount in processor["volumes"]} + assert set(mounts) == {"/run/dag-processor", "/opt/airflow/dags", "/opt/airflow/logs"} + assert mounts["/run/dag-processor"]["read_only"] + assert mounts["/opt/airflow/dags"]["read_only"] + assert processor["read_only"] + assert processor["cap_drop"] == ["ALL"]
