mobuchowski commented on code in PR #75328:
URL: https://github.com/apache/airflow/pull/75328#discussion_r4389887348
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -163,6 +163,27 @@ def _emit_manual_state_change_event(adapter_method_name:
str, stats_key: str, **
Stats.gauge(stats_key, len(Serde.to_json(event).encode("utf-8")))
+def _get_task_end_date(task_instance: RuntimeTaskInstance | TaskInstance) ->
datetime:
+ """
+ Return the end date for a terminal event of a task instance that finished
in the current process.
+
+ The Airflow 3 task runner sets ``end_date`` before it notifies listeners
and reports that same value as
+ the stored one, so the event time matches it. In the cases below the time
of the notification is used
+ instead, so the event time is later than the stored ``end_date`` by
however long the task's state change
+ callbacks took:
+
+ - Airflow 2, where the event time is kept as it was before ``end_date``
was used.
+ - Airflow 3.0.0 to 3.0.4, whose task runner never sets ``end_date`` on the
task instance.
+ - Before Airflow 3.4.0, a task that raised ``AirflowSkipException``, or a
``TriggerDagRunOperator`` task
+ skipped or failed because the Dag run it triggers already exists.
+ """
+ if AIRFLOW_V_3_0_PLUS:
Review Comment:
On Airflow 2, `_get_task_end_date` still times the event with `utcnow()`,
but `TaskInstanceInfo._get_end_date` takes the task instance's own `end_date`
before the value passed in. So one event now carries two different end times.
On 2.10, `_run_raw_task` sets `ti.end_date` and calls `set_duration()`, then
runs `on_success_callback`, then `save_to_db`, and only then calls
`on_task_instance_success`. If a success callback takes 30s, the COMPLETE
event's `eventTime` is about 30s after `taskInstance.end_date`, and `start_date
+ duration` no longer reaches the event time. The new test says "The facet must
describe the same end date as the event, or its duration would disagree with
it". It checks the `end_date` passed to `get_airflow_run_facet`, but the facet
ignores that value whenever the task instance has its own, as it does on AF2.
`handle_failure` on AF2 also sets `end_date` before it calls
`on_task_instance_failed`. So I think the version gate can go, and
`ti.end_date` can be used whenever it's a `datetime`. Then the event and the
facet both come from one value. Is there an AF2 case I'm missing that made you
keep `now()` there?
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1209,17 +1209,56 @@ def get_rendered_map_index(ti: TaskInstance |
RuntimeTaskInstance) -> str | None
return getattr(ti, "rendered_map_index", None)
def __init__(
- self, obj, position: TaskInstancePosition | None = None, dag: Any =
None, dag_run: Any = None
+ self,
+ obj,
+ position: TaskInstancePosition | None = None,
+ dag: Any = None,
+ dag_run: Any = None,
+ end_date: datetime.datetime | None = None,
):
# Pass the position when the caller has already placed the task
instance, e.g. from its stored region or
# with get_task_instance_position; it is then used as is.
self._position = position
# A stored task instance's bundle comes from these, as the worker's
does, without a query
self._dag = dag
self._dag_run = dag_run
+ # The end date a terminal event is timed with, used when the task
instance carries none, as a skipped
+ # Airflow 3 runtime task instance doesn't before Airflow 3.4.
+ self._end_date = end_date
super().__init__(obj)
+ def _get_end_date(self) -> datetime.datetime | None:
+ if isinstance(end_date := getattr(self.obj, "end_date", None),
datetime.datetime):
+ return end_date
+ return self._end_date
+
+ def _get_duration(
+ self, start_date: datetime.datetime | None, end_date:
datetime.datetime | None
+ ) -> float | None:
+ # The stored duration is what the UI shows, so it wins. A runtime task
instance has none -- the API
+ # server computes it after the listener ran.
+ if (duration := getattr(self.obj, "duration", None)) is not None:
+ return duration
+ if start_date is None or end_date is None:
+ return None
+ try:
+ return (end_date - start_date).total_seconds()
Review Comment:
Before this PR, worker-side events sent `duration: null`. Now on Airflow
3.0–3.2, a task resumed from deferral reports a computed duration that's
clearly wrong. Take a deferrable sensor that runs 1s, defers for 2h, then
resumes and finishes in 1s. The COMPLETE facet says `duration ≈ 1` and gives
the resume time as `start_date`. The API server stores the original start, so
the UI shows about 2h.
The description calls this out, but consumers can't tell this value from a
correct one. I'd rather leave `duration` out than emit a number we know is
wrong. One option is to skip the computed duration before `AIRFLOW_V_3_3_PLUS`
when the task resumed from deferral. Another is to skip it entirely before 3.3,
if the runtime task instance can't tell you it resumed. What do you think?
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1209,17 +1209,56 @@ def get_rendered_map_index(ti: TaskInstance |
RuntimeTaskInstance) -> str | None
return getattr(ti, "rendered_map_index", None)
def __init__(
- self, obj, position: TaskInstancePosition | None = None, dag: Any =
None, dag_run: Any = None
+ self,
+ obj,
+ position: TaskInstancePosition | None = None,
+ dag: Any = None,
+ dag_run: Any = None,
+ end_date: datetime.datetime | None = None,
):
# Pass the position when the caller has already placed the task
instance, e.g. from its stored region or
# with get_task_instance_position; it is then used as is.
self._position = position
# A stored task instance's bundle comes from these, as the worker's
does, without a query
self._dag = dag
self._dag_run = dag_run
+ # The end date a terminal event is timed with, used when the task
instance carries none, as a skipped
+ # Airflow 3 runtime task instance doesn't before Airflow 3.4.
+ self._end_date = end_date
super().__init__(obj)
+ def _get_end_date(self) -> datetime.datetime | None:
+ if isinstance(end_date := getattr(self.obj, "end_date", None),
datetime.datetime):
+ return end_date
+ return self._end_date
+
+ def _get_duration(
+ self, start_date: datetime.datetime | None, end_date:
datetime.datetime | None
+ ) -> float | None:
+ # The stored duration is what the UI shows, so it wins. A runtime task
instance has none -- the API
+ # server computes it after the listener ran.
+ if (duration := getattr(self.obj, "duration", None)) is not None:
Review Comment:
Nothing new breaks here, but the "stored duration wins" rule produces a
misleading facet on Airflow 2 retries. When try 1 fails, `handle_failure` sets
`end_date` and `duration`. When try 2 starts,
`_check_and_change_state_before_execution` resets `ti.end_date = None` but
leaves `ti.duration` alone. The try-2 START event's facet then has try 2's
`start_date`, no `end_date`, and try 1's `duration`.
Could the stored duration be used only when the task instance also has an
`end_date`?
```python
if (duration := getattr(self.obj, "duration", None)) is not None and
getattr(self.obj, "end_date", None):
return duration
```
--
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]
