kaxil commented on code in PR #74794: URL: https://github.com/apache/airflow/pull/74794#discussion_r4517220828
########## airflow-core/src/airflow/dag_processing/processor.py: ########## @@ -183,6 +193,21 @@ class DagFileParsingResult(BaseModel): ] [email protected]() +class _DagParseCommsDecoder(CommsDecoder[ToDagProcessor, ToManager]): + api_request_failed: bool = attrs.field(default=False, init=False) + """Whether the Dag processor could not answer a request of this process through the Execution API.""" + + def _from_frame(self, frame) -> ToDagProcessor | None: Review Comment: `api_request_failed` is only set once a response is decoded, so an import that times out while its request is still waiting never sets it. During an outage that wait can be long: the manager only reads child requests in `_service_processor_sockets`, so a request sits while the loop's own heartbeat, claims, publication and ack each wait up to `execution_api_timeout`, and other children's handlers go first. Once `dagbag_import_timeout` fires, `python_importer.py` catches the `AirflowTaskTimeout` as an import error, the child exits 0, and `_finish_callback_attempt` acknowledges the callback without running it. Could a request still waiting for its response count as failed too, e.g. a flag set before the write in `send`/`asend` and cleared in `_from_frame`? ########## airflow-core/src/airflow/dag_processing/manager.py: ########## @@ -2716,10 +2853,15 @@ def terminate(self): processor.kill(signal.SIGTERM, escalation_delay=5.0) def end(self): - """Kill all child processes on exit since we don't want to leave them as orphaned.""" + """Kill child processes so none are orphaned, then acknowledge work that already finished.""" pids_to_kill = [p.pid for p in self._processors.values()] if pids_to_kill: kill_child_processes_by_pids(pids_to_kill) + # Claims the server releases after this Job ends are delivered again, repeating their callbacks. + try: + self.work_source.flush(deadline=math.inf) + except Exception: + self.log.exception("Unable to acknowledge requested work before exiting") Review Comment: `flush()` only catches `httpx2.HTTPError`, but a 403 `job_closed` comes out of `request()` as `DagProcessorRegistrationRetired`. So when the registration retires with acks still queued, which is likely after an outage, this logs a full traceback for the case `run_with_api` already logs as a warning when `client.restart_required` is set. Could this do the same? ########## airflow-core/tests/unit/dag_processing/test_manager.py: ########## @@ -8013,18 +8995,147 @@ def test_result_over_the_publication_limit_is_reported_without_sending_it(): "a.py": "The parsing result was not published: it exceeds the 1 MiB publication limit." } assert pending.body.result_unavailable + assert get_api_work_source(manager)._pending_work_acks == [("priority", work)] client.publish_parse_result.assert_not_called() [email protected]("method", ["claim_priority_files", "fetch_callbacks"]) -def test_internal_api_manager_leaves_requested_work_for_activation(method, mocker): +def test_acknowledgments_are_sent_within_the_loop_budget(): Review Comment: This only calls `work_source.flush`, with no manager code on the path, and `test_flush_preserves_pending_work_after_deadline` in `test_api_requested_work.py` already covers it. The loop also never passes a deadline, so the name points at a manager budget that doesn't exist. Could it go the way of the refusal test? ########## airflow-core/docs/security/jwt_token_authentication.rst: ########## @@ -557,14 +587,174 @@ parsing subprocess can without this client. ``Variable.set`` upserts on the key multi-team mode a write can overwrite another team's Variable with the same key, as it can with task tokens. -The client attempts renewal at 80% of token lifetime. Transient early-renewal failures leave -the existing token usable until expiry. A retired registration signals that its owner must -restart. Completion retains its original identity and outcome across retries. - -File and attempt identity are carried in the token but are not currently used for access -checks or request attribution. Archives retain one file identity under the current processor -model. These credentials do not isolate hostile code from other credentials accessible on -the processor host; the process and deployment limitations above still apply. +For each ``dag_parse`` token, the manager chooses the bundle, bundle-relative file location, +and parsing-attempt UUID. + +The manager publishes completed imports through ``POST /jobs/{job_id}/parse-results`` +using its Job credential. Parsing tokens cannot publish. A request contains one file or +container's serialized Dags, diagnostics, source text (or an explicit unavailable marker), +and the bundle version captured at dispatch. The API never reads the processor's source +paths. It checks bundle access, existing Dag ownership and team-scoped plugin classes +before deserializing results and rechecks ownership on the rows used for persistence. +A Dag can move between files within its bundle. Recovery from a removed bundle is +limited to stale legacy rows with neither a relative file location nor a Dag version. + +Each request carries an attempt UUID and a dispatch sequence within its Job. A publication +receipt and the metadata writes commit together. Replaying the latest identical request returns +the saved receipt; conflicting payloads and superseded sequences are rejected. Publication +receipts retain one entry per Job and source and are removed with Job cleanup. There is no total +ordering between different Jobs publishing a source without versioning; an already-accepted +request's retry does not overwrite another Job's result. Listener side effects are not +transactional and may repeat after a failed transaction. + +Publication and inventory request bodies are limited to 16 MiB, and the API refuses a +larger body before reading the rest of it. A rejected publication does not fall back +to direct database writes, and does not prevent other files from being processed. Transport +failures and server errors are retried; exhausting those retries preserves the last accepted +result. When the API refuses a result for its content, or the result is over the limit, the +processor publishes an import error for the file instead, so the problem is visible. +Dags whose definition path matches an import error are marked stale until a later parse Review Comment: This doesn't hold for the import error the processor publishes in place of a refused or oversized result. `_build_rejection_report` sends it with `result_unavailable=True`, so `routes/dag_parsing.py` leaves those Dags to `stale_dag_threshold` and they keep scheduling until then. Could this sentence name that exception? ########## airflow-core/tests/unit/dag_processing/test_manager.py: ########## @@ -129,6 +133,7 @@ play_runtime, write_native_file, ) +from unit.dag_processing.requested_work_helpers import get_api_work_source, make_status_error Review Comment: Only `make_status_error` comes in here, so about 20 `ProcessorWorkItem(...)` in this module still build the item by hand, as do `test_priority_inventory_sequence.py:69`, the job-runner tests and the new `test_work_claims_and_acknowledgments_wait_for_a_slow_response`. Could they use `make_work` too? ########## airflow-core/docs/security/jwt_token_authentication.rst: ########## @@ -557,14 +587,174 @@ parsing subprocess can without this client. ``Variable.set`` upserts on the key multi-team mode a write can overwrite another team's Variable with the same key, as it can with task tokens. -The client attempts renewal at 80% of token lifetime. Transient early-renewal failures leave -the existing token usable until expiry. A retired registration signals that its owner must -restart. Completion retains its original identity and outcome across retries. - -File and attempt identity are carried in the token but are not currently used for access -checks or request attribution. Archives retain one file identity under the current processor -model. These credentials do not isolate hostile code from other credentials accessible on -the processor host; the process and deployment limitations above still apply. +For each ``dag_parse`` token, the manager chooses the bundle, bundle-relative file location, +and parsing-attempt UUID. + +The manager publishes completed imports through ``POST /jobs/{job_id}/parse-results`` +using its Job credential. Parsing tokens cannot publish. A request contains one file or +container's serialized Dags, diagnostics, source text (or an explicit unavailable marker), +and the bundle version captured at dispatch. The API never reads the processor's source +paths. It checks bundle access, existing Dag ownership and team-scoped plugin classes +before deserializing results and rechecks ownership on the rows used for persistence. +A Dag can move between files within its bundle. Recovery from a removed bundle is +limited to stale legacy rows with neither a relative file location nor a Dag version. + +Each request carries an attempt UUID and a dispatch sequence within its Job. A publication +receipt and the metadata writes commit together. Replaying the latest identical request returns +the saved receipt; conflicting payloads and superseded sequences are rejected. Publication +receipts retain one entry per Job and source and are removed with Job cleanup. There is no total +ordering between different Jobs publishing a source without versioning; an already-accepted +request's retry does not overwrite another Job's result. Listener side effects are not +transactional and may repeat after a failed transaction. + +Publication and inventory request bodies are limited to 16 MiB, and the API refuses a +larger body before reading the rest of it. A rejected publication does not fall back +to direct database writes, and does not prevent other files from being processed. Transport +failures and server errors are retried; exhausting those retries preserves the last accepted +result. When the API refuses a result for its content, or the result is over the limit, the +processor publishes an import error for the file instead, so the problem is visible. +Dags whose definition path matches an import error are marked stale until a later parse +publishes them again. Dags missing from an accepted parse are marked stale at once: the +publication is atomic, so there is no gap between file and Dag parse times to wait out. For a +container-level error, missing member Dags are marked stale only when their last successful +parse is older than ``[dag_processor] stale_dag_threshold``, using the server's clock. Invalid ``access_control`` produces an import error with both +current and supported older FAB releases. Permission backend failures remain retryable +server errors. Empty imports still publish a receipt and clear the file's earlier import +errors. A bundle refresh, including one forced by a parse request, waits for that bundle's +active imports instead of discarding them; only initializing a bundle again invalidates and +requeues older in-flight results from that manager. + +Inventories report discovered definitions and sources whose discovery failed separately. +Failed sources preserve their existing definitions and diagnostics, including archive members; +other sources still undergo deletion reconciliation. Each inventory carries the previously +observed server revision; conflicting refreshes must +rediscover before retrying. Imports carry the accepted revision, so an older snapshot cannot +publish after a changed inventory. Only accepted publications establish missing Dags within +a file; rejected requests and incomplete discovery do not establish absence. Every +``[scheduler] parsing_cleanup_interval``, the scheduler marks the Dags of inactive bundles stale, +removes stale warnings, and ends expired processor Jobs, in bounded batches and in +both modes. + +Callback and priority requests remain stored while claimed. A lost claim response or +acknowledgment can be retried. Claims use increasing sequences per Job and work type; the +client waits for a claim response before sending the next sequence for that type. +Retrying an empty claim remains empty, even if work arrives +afterward, and retrying an older claim cannot consume newly requested work. Requests from a +retired Job become available to another processor. A callback acknowledgment confirms +delivery, not successful execution of user code. As with other at-least-once delivery, a crash +after a callback's side effect and before acknowledgment can repeat that side effect. Before the +processor's Job ends, a graceful shutdown acknowledges callbacks whose processes have already exited +cleanly; callbacks that are still running or that the shutdown signal stopped are delivered again. +When a callback's bundle +version cannot be prepared, the processor retries with backoff for up to +``[dag_processor] job_heartbeat_timeout`` +and then acknowledges the request without running the callback. The same retry applies after an +Execution API request fails, but only when none of the callbacks processed together for a file +found its Dag or task, so none of them ran. When one of them did, a callback whose Dag or task did +not load is acknowledged without running. Priority requests are assigned +to an import started after their requested refresh. Only that import's publication or terminal +failure acknowledges the request; an older in-flight import cannot consume it. Discovery after +the requested refresh can also acknowledge a request if it confirms the definition is absent. +A new request for a file already being imported remains pending for a subsequent import, +even if the earlier import has not yet been acknowledged. + +The manager retains a bounded queue of pending publications. Each loop it sends queued +publications and acknowledgments for up to a second, stopping at the first failure that may +be transient. Connecting is limited to a second, but a publication, inventory, work claim or +acknowledgment waits up to ``[workers] execution_api_timeout`` for its response. The API server +commits each of them before it responds, so a request that gave up sooner would be sent again +and take just as long. Retries keep the same payload and identity while +heartbeats and subprocess supervision continue between attempts. Pending results +count against parsing capacity. Failed publication attempts throttle reparsing but +do not advance the timestamps used to detect stale Dags. + +The returned ``dag_parse`` token binds these identifiers to the processor's session and +Job. It authorizes parse-time operations within that bundle and has no Job-management or +token-exchange access. A bundle header cannot override its signed bundle. The token expires +no later than the management credential; ending or replacing the Job invalidates it. +The supervisor caches it for the attempt and attempts renewal at 80% of its lifetime. +Transient early-renewal failures leave the existing token usable until expiry. + +Parse-time requests use ``[workers] execution_api_timeout`` (five seconds by default), +unless the request specifies a timeout. Required parsing-token exchanges and expired +Job-token renewals use the same timeout. Each makes one attempt without retry sleeps in +the manager loop; optional early token renewal caps each HTTP timeout phase at one second. + +Heartbeats use ``[scheduler] job_heartbeat_sec`` and are also sent between bundle +operations and before each callback's bundle is prepared. Transient API failures are +retried on the next heartbeat interval. Once a Job has gone +``[dag_processor] job_heartbeat_timeout`` without a recorded heartbeat, the scheduler's next cleanup (every +``[scheduler] parsing_cleanup_interval``) ends it and releases its claimed work, and a processor +that could not reach the API for that long stops. Set the timeout above the longest single +bundle initialization, refresh, or listing, and set it for the scheduler as well as the +processor: the scheduler's value decides when the Job ends. The processor does not update its +local health file during one of those operations either, so ``health_check_threshold`` and the +liveness probe's failure budget must also cover it. Likewise, the API server reads +``[dag_processor] health_check_threshold`` to decide whether a session's previous Job is still +alive, so keep it the same for the API server and the processor. A restart request, replaced Job, or retired +registration also stops parsing and terminates its children. The deployment's process +supervisor must restart the command, which creates a new registration. After a crash, +registration retries ``job_running`` with backoff for up to the health-check threshold +plus one heartbeat interval, retaining the same registration ID while waiting for the old +Job to expire. +Failed runs attempt to record failure without masking the original error; completion +failures after a successful run are reported to the caller. + +Use ``airflow dag-processor --check-health`` for local liveness and ``--check-ready`` for +local liveness plus a recent API heartbeat. Both read ``[dag_processor] health_check_file`` +without querying the database. Give each processor its own writable health file. A paused +manager fails liveness even while the API remains healthy; a transient API outage affects +readiness without immediately failing liveness. While the API is slow, each request can wait up +to ``[workers] execution_api_timeout``. While the manager answers its parsing processes' requests, +it refreshes the health file once a second has passed, so a slow request is always followed by a +refresh. One loop also makes up to five requests of its own (a heartbeat, two work claims, a +publication and an acknowledgment), so keep +``health_check_threshold`` above five times that timeout. Review Comment: `_publish_pending_results` and `flush` keep sending until their 1s budget runs out, so each can take a second plus the timeout, and the worst gap is five timeouts plus a couple of seconds. The defaults still fit, but could this say so, so someone raising `execution_api_timeout` to exactly a fifth of the threshold doesn't get a flapping probe? -- 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]
