kaxil commented on code in PR #66608:
URL: https://github.com/apache/airflow/pull/66608#discussion_r3337752291
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/variables.py:
##########
@@ -131,7 +140,10 @@ def get_variable(
@router.put(
"/{variable_key:path}",
- dependencies=[Depends(has_variable_access)],
+ dependencies=[
+ Security(require_auth, scopes=["token:execution", "token:workload"]),
Review Comment:
This widens `token:workload` onto the variable PUT and DELETE routes (and
the same pattern lands on xcoms POST and DELETE at `xcoms.py:342` and `:446`),
but the callback subprocess only ever issues reads: `CallbackToSupervisor` is
`GetConnection | GetDagRun | GetVariable | GetVariableKeys | MaskSecret`
(callback_supervisor.py:80) with no write or delete message in the set. Since
`has_variable_access`/`has_xcom_access` still return True unconditionally
(multi-tenancy isn't enforced yet), a workload token, or a compromised
callback, can now write or delete any Variable and any XCom cluster-wide. Could
you keep `token:workload` on the read routes only (GET variable/keys, GET/HEAD
xcom, GET connection, GET dag_run) and leave PUT/POST/DELETE as
`token:execution` only?
##########
airflow-core/src/airflow/models/deadline.py:
##########
@@ -215,46 +216,32 @@ 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 kwargs so it's passed to the
callback function.
+ self.callback.data["kwargs"]["deadline_id"] = str(self.id)
Review Comment:
Putting `deadline_id`/`deadline_time` into `data["kwargs"]` makes them flat
kwargs handed to the user callable. `execute_callback` instantiates a class
callback as `callback_callable(**kwargs_without_context)`, so the documented
`SlackWebhookNotifier(text=...)` becomes `SlackWebhookNotifier(text=...,
deadline_id=..., deadline_time=...)` and raises `TypeError: unexpected keyword
argument 'deadline_id'`. On main these weren't kwargs at all:
`get_simple_context()` exposed them as `context["deadline"] = {"id": ...,
"deadline_time": ...}` and `deadline_id` was a top-level `data` field. The
deadline-alerts docs depend on that, templating `{{ deadline.deadline_time }}`,
and `build_context_from_dag_run` no longer adds a `deadline` key, so that
template now renders empty as well. The model test mocks `queue()`, so it
exercises neither path. Could these go back under the context dict (a
`deadline` entry) rather than user kwargs?
##########
airflow-core/src/airflow/triggers/callback.py:
##########
@@ -32,6 +32,46 @@
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.
+ """
+ import sys
+ from pathlib import Path
+
+ mod_name = callback_path.split(".")[0]
Review Comment:
Two things on this lookup. The mangled name is
`unusual_prefix_{sha1(full_file_path)}_{stem}` (`get_unique_dag_module_name`,
utils/file.py:184), but this searches bundles for `{stem}.py` and registers the
first match under `mod_name` without checking that the hash recomputed from the
found path matches the one already baked into `mod_name`. Two bundles that both
contain e.g. `report.py` would register the wrong bundle's file, so the
callback imports the wrong function. Comparing
`get_unique_dag_module_name(str(file_path))` against `mod_name` and only
loading on an exact match would close that. Separately, this runs inside
`run()` on the triggerer's async event loop and does a synchronous
`bundle.initialize()` (a git/S3 fetch for remote bundles) plus a file load,
which blocks every other trigger on that triggerer until it returns. Worth
moving off the loop (e.g. `asyncio.to_thread`) or resolving the path before
`run()`.
##########
task-sdk/src/airflow/sdk/execution_time/callback_supervisor.py:
##########
@@ -209,14 +214,28 @@ 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)
except Exception:
_log.warning(
"Failed to initialize DAG bundle for callback",
bundle_name=bundle_info.name,
exc_info=True,
)
- success, error_msg = execute_callback(callback_path,
callback_kwargs, _log)
+ # When DagRun identifiers are provided, fetch the DagRun via
SUPERVISOR_COMMS
+ # and build a context dict to pass to the callback function.
+ effective_kwargs = dict(callback_kwargs)
+ if dag_id and run_id:
+ context =
_fetch_and_build_context(task_runner.SUPERVISOR_COMMS, dag_id, run_id, _log)
+ if context is not None:
Review Comment:
When `_fetch_and_build_context` fails, either an unexpected response or an
exception, it returns None (lines 414 and 424), and this guard then omits
`context` and runs the callback anyway. The triggerer path does the same:
`_fetch_callback_dag_run_data` returns None on a missing DagRun
(triggerer_job_runner.py:838) and the trigger runs with `dag_run_data=None`. So
a transient API/DB blip yields a callback that reports success but never
received the context it expects: a deadline notification templating `{{ dag_run
}}` renders empty or KeyErrors, while the deadline is marked handled and never
retried. Failing the callback on a context-fetch failure so it retries seems
safer than running degraded.
##########
airflow-core/src/airflow/triggers/callback.py:
##########
@@ -32,6 +32,46 @@
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.
+ """
+ import sys
Review Comment:
`import sys` and `from pathlib import Path` here are stdlib with no
circular-import reason to be inline, and the same goes for the
`importlib.machinery`/`importlib.util`/`sys`/`Path` block in
`load_mangled_dag_module` in the shared module_loading change. Those can hoist
to module top.
For the two `from airflow...` imports below (lines 57-58) I'd rather ask
than assume: `load_mangled_dag_module` comes from
`airflow._shared.module_loading`, which is already imported at the top of this
file on line 25, so it looks like it could just join that line. And
`DagBundlesManager` from `airflow.dag_processing.bundles.manager` is imported
at module top by `models/errors.py` and `models/dag_version.py`, so a top-level
import here probably isn't circular either. Is there a deliberate reason these
two are kept inline, e.g. avoiding pulling `dag_processing` in for every
importer of `triggers.callback` (including the serialization path), or can they
move up too?
--
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]
