kaxil commented on code in PR #74806:
URL: https://github.com/apache/airflow/pull/74806#discussion_r4289941598
##########
airflow-core/docs/authoring-and-scheduling/loops.rst:
##########
@@ -330,8 +351,9 @@ In the Grid, select the loop's task group in a Dag run to
open its Task Instance
tab. The **Iteration** filter narrows the table to one iteration, or shows
**All iterations**, and the **Iteration** column shows which iteration each
task
instance belongs to. The gate's state and logs explain why the loop continued,
-stopped, or failed. Each task's tries and logs remain accessible within its
-iteration.
+stopped, or failed. When ``until`` returned a ``LoopDecision`` with a reason,
the
+Iteration filter shows that reason next to the iteration. Each task's tries and
Review Comment:
I don't think the Iteration filter shows the reason. `LoopOptionLabel` isn't
touched here, and the only places that read `decision_reason` are the Outcome
stat on the loop's group page (final iteration, once the loop is done) and the
"Decision" stat in a task instance header. Should this sentence point at those
instead?
##########
task-sdk/src/airflow/sdk/execution_time/loop.py:
##########
@@ -76,29 +78,50 @@ def _result(self, *, previous_iteration: bool) -> Any:
def execute_loop_gate(gate: LoopGateOperator, context: Context) -> None:
- # task_runner imports LoopContextAccessor from this module.
- from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
-
loop = context["loop"]
+ ti = cast("RuntimeTaskInstance", context["ti"])
+ reason: str | None = None
if gate.until is None:
stop = loop.index + 1 >= loop.max_iterations
else:
- stop = gate.until(**determine_kwargs(gate.until, (), context))
- if not isinstance(stop, bool):
- raise AirflowFailException(f"until must return a bool, got
{type(stop).__name__}")
- if not stop and loop.index + 1 >= loop.max_iterations:
- raise LoopMaxIterationsExceeded(
- f"Loop condition remained false at
max_iterations={loop.max_iterations}"
+ match result := gate.until(**determine_kwargs(gate.until, (),
context)):
+ case bool():
+ stop = result
+ case LoopDecision(stop=bool(), reason=None | str()):
+ stop, reason = result.stop, result.reason or None
+ case LoopDecision(stop=bool()):
+ raise AirflowFailException(
+ f"LoopDecision reason must be a str or None, got
{type(result.reason).__name__}"
+ )
+ case LoopDecision():
+ raise AirflowFailException(
+ f"LoopDecision stop must be a bool, got
{type(result.stop).__name__}"
+ )
+ case _:
+ raise AirflowFailException(
+ f"until must return a bool or LoopDecision, got
{type(result).__name__}"
+ )
+ if reason is not None:
+ # Mask here, where mask_secret() registered the value: the API server
showing it later does not.
+ reason = cast("str", redact(reason))
+ try:
+ size = len(reason.encode())
Review Comment:
A NUL character in the reason passes this check and
`LoopDecisionPayload._check_reason_size`, but Postgres JSONB rejects `\u0000`
when the server writes the XCom. Since that write now runs in the same
transaction as the gate's state update, the gate's terminal state update fails
with it, not just the reason. Could we reject NUL here next to the size check,
and in the server validator too?
##########
airflow-core/pyproject.toml:
##########
@@ -135,7 +135,8 @@ dependencies = [
'pendulum>=3.1.0',
"pluggy>=1.5.0",
"psutil>=5.8.0",
- "pydantic>=2.11.0",
+ # 2.12 adds Field(exclude_if=...), which API responses use to omit absent
optional fields
+ "pydantic>=2.12.0",
Review Comment:
Is omitting `decision_reason` worth raising the core pydantic floor?
`exclude_if` has one user (`LoopIterationSummary.decision_reason`), the sibling
fields on that model serialize `null`, and `OmitsMissingRegion` in
`api_fastapi/common/region.py` already omits fields with a `model_serializer`
under the current floor. If we keep the bump, the comment's "API responses use"
overstates it a bit for now.
##########
airflow-core/src/airflow/example_dags/example_task_loops.py:
##########
@@ -38,7 +38,8 @@ def evaluate(estimate):
evaluate(improve())
def accurate_enough(*, loop):
- return loop.result["error"] < 0.000001
+ error = loop.result["error"]
+ return LoopDecision(stop=error < 0.000001, reason=f"error {error:.2g},
needs < 1e-06")
Review Comment:
Nit: this gets pulled into the docs via `exampleinclude`, so people will
copy it. A `threshold = 1e-6` used in both the comparison and the message
(`needs < {threshold:g}`) keeps the two from drifting.
--
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]
