ephraimbuddy commented on code in PR #74794:
URL: https://github.com/apache/airflow/pull/74794#discussion_r4425530587
##########
airflow-core/src/airflow/cli/cli_config.py:
##########
@@ -2304,6 +2304,21 @@ class GroupCommand(NamedTuple):
help="Start a dag processor instance",
func=lazy_load_command("airflow.cli.commands.dag_processor_command.dag_processor"),
args=(
+ Arg(
+ ("--check-health",),
+ action="store_true",
+ help="Check the API-mode processor's local loop heartbeat and
exit",
+ ),
+ Arg(
+ ("--check-ready",),
+ action="store_true",
+ help="Check the API-mode processor's loop and API heartbeat
and exit",
+ ),
+ Arg(
+ ("--sync-bundles-only",),
Review Comment:
Correction to my earlier reply: the separate command is implemented.
`airflow dag-processor-sync-bundles` synchronizes the bundle catalog from a
trusted host with database access and exits without starting a processor, even
when `execution_api_token_file` is set. The `--sync-bundles-only` flag is gone,
and in API mode `airflow dag-processor` only checks that its bundles are
provisioned.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -797,9 +828,47 @@ def claim_priority_files(self) -> list[DagFileInfo]:
Default implementation reads from the metadata DB; override to source
requests from an API.
Review Comment:
The requested-work delivery is now behind `RequestedWorkSource` in
`dag_processing/requested_work.py`, added in #74793.
`DatabaseRequestedWorkSource` wraps the existing database claims, and this PR's
`ExecutionAPIWorkSource` holds the API claim, attempt and acknowledgment state;
`DagProcessorJobRunner.run_with_api` installs it, and the manager calls
`attempt_finished`, `bundle_unusable`, `has_outstanding_work` and `flush`. What
remains coupled: the manager still picks `ExecutionAPIWorkSource` as its
default when it has an API client, and a few `requires_acknowledgment` branches
stay because the database source deletes work when it claims it. Does this
clear the blocker?
##########
airflow-core/tests/integration/dag_processing/test_api_isolation.py:
##########
@@ -0,0 +1,314 @@
+# 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.
+
+"""Run with Breeze and PostgreSQL; the processor joins an API-only internal
network."""
Review Comment:
The test does run in CI. The core integration jobs run all of
`airflow-core/tests/integration` without passing `--integration` to pytest, so
the marker check doesn't skip it, and the "Integration core otel" job ran and
passed both cases. I updated the docstring to say every core integration job
runs it and added the missing `__init__.py`; if you'd rather it ran in a single
job, I can add an integration marker.
##########
airflow-core/src/airflow/jobs/dag_processor_job_runner.py:
##########
@@ -49,12 +62,102 @@ def __init__(
):
super().__init__(job)
self.processor = processor
+ self._last_api_heartbeat = 0.0
+ self._next_api_heartbeat = 0.0
self.processor.heartbeat = lambda: perform_heartbeat(
job=self.job,
heartbeat_callback=self.heartbeat_callback,
only_if_necessary=True,
)
+ def run_with_api(self, client: DagProcessorAPIClient) -> int | None:
+ """Mirror ``run_job`` metrics and listener ordering while the API owns
Job persistence."""
+ self.processor.api_client = client
+ self.job.id = self._register_api_job(client)
+ self.job.state = JobState.RUNNING
+ self.job.start_date = self.job.latest_heartbeat = timezone.utcnow()
+ self._last_api_heartbeat = monotonic()
+ self._next_api_heartbeat = self._last_api_heartbeat +
self.job.heartrate
+ self.processor.heartbeat = lambda: self._heartbeat_api(client)
+ stats.incr("job_start", 1, 1)
+ failed = True
+ try:
+ result = execute_job(self.job, execute_callable=self._execute)
+ failed = False
+ return result
+ finally:
+ self.job.state = JobState.FAILED if failed else JobState.SUCCESS
+ try:
+ get_listener_manager().hook.before_stopping(component=self.job)
+ except Exception:
+ self.log.exception("Error calling Dag processor shutdown
listener")
+ self.job.end_date = timezone.utcnow()
+ try:
+ client.complete_job(TerminalJobState(self.job.state))
+ except Exception:
+ if not failed:
+ raise
+ self.log.exception("Unable to record the failed Dag processor
Job")
Review Comment:
The retirement messages now include the time since the last successful
heartbeat and `[dag_processor] job_heartbeat_timeout`. I kept the
`complete_job` call: re-provisioning the token file under a new session retires
the registration while the old Job is still open, and skipping completion left
that Job's claimed work unavailable until scheduler cleanup ended it. When an
ended or replaced Job refuses completion, that's logged as a warning without a
traceback.
##########
airflow-core/docs/security/security_model.rst:
##########
@@ -775,6 +786,12 @@ model — Airflow does not enforce these natively.
value itself. However, access to the file it points to should be restricted
to the Scheduler
(which generates tokens) and the API Server (which validates them).
+ A Dag File Processor running with ``[dag_processor]
execution_api_token_file`` needs none of the
+ metadata database variables, the Fernet key or JWT signing keys; it
authenticates with the token
+ file instead. It also no longer needs asset manager credentials for
publishing parsing results.
Review Comment:
I took the docs option. `security_model.rst` now says to configure `[core]
asset_manager_class` and `[core] asset_manager_kwargs` on the API server and
remove both from the processor, and explains that removing only the credentials
can stop the processor from starting. Resolution is still eager; I can make it
lazy if you'd prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
