kaxil commented on code in PR #66608:
URL: https://github.com/apache/airflow/pull/66608#discussion_r3376647552
##########
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py:
##########
@@ -234,16 +247,38 @@ def _target():
_log.debug(
"Added bundle path to sys.path",
bundle_name=bundle_info.name, path=bundle_path
)
+ # DAG processor loads bundle files with a mangled module
name
+ # (unusual_prefix_{hash}_{stem}) to avoid collisions. The
callback path
+ # was serialized using that mangled name. Register the
module under that
+ # name so import_string can find it in the subprocess.
+ if callback_path and
callback_path.startswith("unusual_prefix_"):
+ _register_unusual_prefix_module(callback_path,
bundle.path, _log)
Review Comment:
This pre-execs the DAG module into `sys.modules` via
`load_mangled_dag_module` (which calls `exec_module`), but `execute_callback`
then builds a fresh spec from `abs_path` and `exec_module`s the same file again
(`callback_supervisor.py:144`) without consulting `sys.modules`. So a callback
defined inside a DAG file runs that file's module-level code twice in the
subprocess, doubling any import-time side effects. If the registration is only
there so the mangled name resolves, `execute_callback` could reuse the
already-registered module (`import_module(module_path)`) instead of re-exec'ing.
##########
airflow-core/src/airflow/triggers/callback.py:
##########
@@ -17,21 +17,61 @@
from __future__ import annotations
+import asyncio
import logging
+import sys
import traceback
from collections.abc import AsyncIterator
+from pathlib import Path
from typing import Any
-from airflow._shared.module_loading import accepts_context, import_string,
qualname
+from airflow._shared.module_loading import accepts_context, import_string,
load_mangled_dag_module, qualname
+from airflow.dag_processing.bundles.manager import DagBundlesManager
from airflow.models.callback import CallbackState
from airflow.triggers.base import BaseTrigger, TriggerEvent
+from airflow.utils.file import get_unique_dag_module_name
log = logging.getLogger(__name__)
PAYLOAD_STATUS_KEY = "state"
PAYLOAD_BODY_KEY = "body"
+def _ensure_bundle_module_registered(callback_path: str) -> None:
+ """
+ Register an unusual_prefix_{hash}_{stem} module so import_string can find
it.
+
+ The triggerer event loop doesn't run the DAG processor, so DAG files with
mangled
+ module names must be registered manually. Walks all configured bundles to
find the
+ file, then delegates the actual loading to _load_mangled_module in
callback_supervisor.
+ """
+ mod_name = callback_path.split(".")[0]
+ if mod_name in sys.modules:
+ return
+
+ # unusual_prefix_{hex40}_{stem} → {stem}.py
+ parts = mod_name.split("_", 3)
+ if len(parts) < 4:
+ return
+ stem = parts[3]
+
+ try:
+ for bundle in DagBundlesManager().get_all_dag_bundles():
Review Comment:
`get_all_dag_bundles()` returns bundles at their currently-configured
version, and the match key is `get_unique_dag_module_name`, a sha1 of the
file's absolute path. So an older versioned run's callback loads current bundle
code (there's no run-version pin, since `RunTrigger` carries no bundle
identity), and a triggerer host whose bundle path differs from the
DAG-processor's that minted the mangled name won't hash-match, so the callback
ends FAILED. The within-host wrong-file case is already handled by the `*.py`
walk + hash compare. Is cross-version / cross-host path skew in scope here, or
an accepted limitation worth a code comment?
##########
airflow-core/src/airflow/models/deadline.py:
##########
@@ -217,46 +218,35 @@ def prune_deadlines(cls, *, session: Session, conditions:
dict[Mapped, Any]) ->
def handle_miss(self, session: Session):
"""Handle a missed deadline by queueing the callback."""
-
- def get_simple_context():
- from airflow.api_fastapi.core_api.datamodels.dag_run import
DAGRunResponse
- from airflow.models import DagRun
-
- # TODO: Use the TaskAPI from within Triggerer to fetch full
context instead of sending this context
- # from the scheduler
-
- # Fetch the DagRun from the database again to avoid errors when
self.dagrun's relationship fields
- # are not in the current session.
- dagrun = session.get(DagRun, self.dagrun_id)
-
- return {
- "dag_run":
DAGRunResponse.model_validate(dagrun).model_dump(mode="json"),
- "deadline": {"id": self.id, "deadline_time":
self.deadline_time},
- }
+ # Routing identifiers: stored at the top level of callback.data so the
triggerer/executor
+ # can locate the DagRun and build execution context *before* invoking
the callback.
+ # These are NOT user payload — they are consumed by infrastructure
(_fetch_callback_dag_run_data
+ # in triggerer_job_runner and _fetch_and_build_context in
callback_supervisor) and stripped
+ # before the callback function is called.
+ # TODO: A dedicated `routing_data` field on Callback would make this
separation explicit
+ # and avoid mixing routing metadata with user kwargs in the same JSON
blob. Deferred because
+ # changing the Callback model would require a migration and
coordination with the executor path.
+ self.callback.data["dag_id"] = self.dagrun.dag_id
+ self.callback.data["run_id"] = self.dagrun.run_id
+
+ # Deadline-specific info goes into the routing data (not user kwargs).
+ # The context builder (_fetch_and_build_context /
_fetch_callback_dag_run_data) exposes
+ # these as context["deadline"] = {"id": ..., "deadline_time": ...} so
templates like
+ # {{ deadline.deadline_time }} continue to work.
+ self.callback.data["deadline_id"] = str(self.id)
+ self.callback.data["deadline_time"] = self.deadline_time.isoformat()
Review Comment:
These fields land at the top level of `callback.data`, and
`Callback.get_metric_info` builds metric tags as `{"result": result,
**self.data}` (`models/callback.py:160`), stripping only `prefix` and
`kwargs["context"]`. So `dag_id`, `run_id`, `deadline_id`, and `deadline_time`
all become metric tags. `deadline_time` (a per-miss timestamp) and
`deadline_id` are effectively unbounded, which is high-cardinality for
statsd/OTel. On 3.2.x these sat under `kwargs["context"]` (stripped), so this
is net-new tag cardinality for the deadline callback metric. Worth excluding
the routing fields from the tag set (overlaps the deferred `routing_data`
namespace TODO).
##########
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py:
##########
@@ -234,16 +247,38 @@ def _target():
_log.debug(
"Added bundle path to sys.path",
bundle_name=bundle_info.name, path=bundle_path
)
+ # DAG processor loads bundle files with a mangled module
name
+ # (unusual_prefix_{hash}_{stem}) to avoid collisions. The
callback path
+ # was serialized using that mangled name. Register the
module under that
+ # name so import_string can find it in the subprocess.
+ if callback_path and
callback_path.startswith("unusual_prefix_"):
Review Comment:
Minor: this uses the literal `"unusual_prefix_"` while
`UNUSUAL_MODULE_PREFIX` (same value) is already imported and used at line 132.
Same literal in `triggers/callback.py:98`. Worth the constant at both sites.
--
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]
