This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ce75cee3b44 Use task instance end date as OpenLineage terminal event
time (#75328)
ce75cee3b44 is described below
commit ce75cee3b44192c498c76c7c7554649392a2de1d
Author: Kacper Muda <[email protected]>
AuthorDate: Wed Oct 28 10:21:47 2026 +0100
Use task instance end date as OpenLineage terminal event time (#75328)
* Use the task instance end date as OpenLineage terminal event time
The Airflow 3 task runner sets the end date and reports it to the API
server before it notifies listeners, but the listener stamped complete,
fail and skipped events with the time of its own notification. That is
later than the stored end date by however long the Airflow callbacks took.
The task instance facet also never carried a duration on Airflow 3: the
runtime task instance has none, as the API server computes it after the
listener ran. It is now derived from the same end date as the event time.
Co-Authored-By: Claude <[email protected]>
* Keep the stored task instance duration in the OpenLineage facet
Airflow 2 stores the duration before the listener is notified, while the
end date passed to the facet there is the notification time, taken after
the success callback ran. Computing the duration from it made the facet
disagree with the duration shown in the UI.
Co-Authored-By: Claude <[email protected]>
* Note that Airflow 3.4 keeps the end date of skipped task instances
Since apache/airflow#75685 the task runner sets the end date on the task
instance in the skip and existing Dag run cases too, so the fallback to the
notification time there only applies to older Airflow versions.
Co-Authored-By: Claude <[email protected]>
* Include start and end dates in the OpenLineage task instance facet
Consumers had only the event time and a duration to work out when a task
instance ran, and the event time of a terminal event isn't always its end
date. The dates the task instance holds are the ones the UI shows.
Co-Authored-By: Claude <[email protected]>
* Check task instance dates in the OpenLineage system tests
The facet now carries the start and end dates; locking them in on the
success, skip, failure and retry paths keeps a regression there from
going unnoticed.
Co-Authored-By: Claude <[email protected]>
* Require a number for the duration in the OpenLineage system tests
A numeric string would pass the previous check, while the duration must be
emitted as a JSON number, as in the Airflow 2 expectations.
Co-Authored-By: Claude <[email protected]>
* Keep OpenLineage task instance dates consistent with what Airflow stores
Address review: on Airflow 2 the event time and the facet's end date came
from different values, a retry reported the previous try's duration, and
before Airflow 3.2.2 a task resumed from deferral reported its resume time
as start date and a duration that left out the time before it deferred.
Co-Authored-By: Claude <[email protected]>
* Check task instance dates per Airflow version in the OpenLineage system
tests
A task resumed from deferral reports no start date and duration before
Airflow 3.2.2, and Airflow 2 events now carry the dates too, so the
expectations need to say which versions get what. That boundary is a
patch release, which the expected-event files couldn't express.
Co-Authored-By: Claude <[email protected]>
* Expect no duration from a running Airflow 2 task instance in the
OpenLineage test
Its stored duration is the previous try's, which the facet no longer
reports.
Co-Authored-By: Claude <[email protected]>
---------
Co-authored-by: Claude <[email protected]>
---
.../openlineage/facets/AirflowRunFacet.json | 10 ++
.../providers/openlineage/plugins/listener.py | 28 ++-
.../airflow/providers/openlineage/utils/utils.py | 63 ++++++-
.../providers/openlineage/version_compat.py | 9 +-
.../example_openlineage_defer_simple_dag.py | 2 +
.../system/openlineage/expected_events/__init__.py | 18 +-
.../openlineage_base_complex_dag__af2.json | 63 +++++--
.../openlineage_base_complex_dag__af3.json | 70 ++++++--
.../openlineage_custom_operator_failure_dag.json | 5 +-
.../openlineage_defer_simple_dag.json | 10 +-
...json => openlineage_defer_simple_dag__af3.json} | 10 +-
...penlineage_defer_simple_dag__af3_2_2_plus.json} | 10 +-
.../expected_events/openlineage_retry_dag.json | 12 ++
.../openlineage_retry_dag__af2.json | 20 ++-
.../openlineage_retry_dag__af3_0.json | 12 ++
.../openlineage_sensor_reschedule_dag.json | 10 +-
.../openlineage_sensor_reschedule_dag__af2.json | 10 +-
.../openlineage_sensor_reschedule_dag__af3_0.json | 10 +-
.../openlineage_task_states_dag.json | 12 ++
.../openlineage_task_states_dag__af3_0.json | 6 +
.../unit/openlineage/plugins/test_listener.py | 110 +++++++++++-
.../tests/unit/openlineage/utils/test_utils.py | 193 ++++++++++++++++++++-
22 files changed, 631 insertions(+), 62 deletions(-)
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/facets/AirflowRunFacet.json
b/providers/openlineage/src/airflow/providers/openlineage/facets/AirflowRunFacet.json
index 1e18bcfbbac..8422e27c367 100644
---
a/providers/openlineage/src/airflow/providers/openlineage/facets/AirflowRunFacet.json
+++
b/providers/openlineage/src/airflow/providers/openlineage/facets/AirflowRunFacet.json
@@ -308,6 +308,16 @@
"type": "string",
"format": "date-time"
},
+ "start_date": {
+ "description": "Absent from the worker's events of a task resumed
from deferral before Airflow 3.2.2, whose start date there is the resume time.",
+ "type": "string",
+ "format": "date-time"
+ },
+ "end_date": {
+ "description": "Set once the task instance finished: on the COMPLETE
and FAIL events.",
+ "type": "string",
+ "format": "date-time"
+ },
"log_url": {
"type": "string",
"format": "uri"
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
index dda70e50fec..51a42b3c48e 100644
---
a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
+++
b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
@@ -163,6 +163,25 @@ 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.
+
+ Airflow 2 and the Airflow 3 task runner set ``end_date`` before they
notify listeners, and that value is 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 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.
+ """
+ end_date = getattr(task_instance, "end_date", None)
+ if isinstance(end_date, datetime):
+ return end_date
+ return timezone.utcnow()
+
+
class OpenLineageListener:
"""OpenLineage listener sends events on task instance and dag run starts,
completes and failures."""
@@ -395,7 +414,7 @@ class OpenLineageListener:
self._on_task_instance_success(task_instance, dag, dagrun, task)
def _on_task_instance_success(self, task_instance: RuntimeTaskInstance,
dag, dagrun, task):
- end_date = timezone.utcnow()
+ end_date = _get_task_end_date(task_instance)
controls = resolve_task_emission_policy(
operator=task,
@@ -493,6 +512,7 @@ class OpenLineageListener:
task_uuid,
include_full_task_info=controls.include_full_task_info,
ti_position=position,
+ end_date=end_date,
),
**get_airflow_debug_facet(),
},
@@ -567,7 +587,7 @@ class OpenLineageListener:
task,
error: None | str | BaseException = None,
) -> None:
- end_date = timezone.utcnow()
+ end_date = _get_task_end_date(task_instance)
controls = resolve_task_emission_policy(
operator=task,
@@ -665,6 +685,7 @@ class OpenLineageListener:
task_uuid,
include_full_task_info=controls.include_full_task_info,
ti_position=position,
+ end_date=end_date,
),
**get_airflow_debug_facet(),
},
@@ -715,7 +736,7 @@ class OpenLineageListener:
dagrun,
task,
) -> None:
- end_date = timezone.utcnow()
+ end_date = _get_task_end_date(task_instance)
controls = resolve_task_emission_policy(
operator=task,
@@ -812,6 +833,7 @@ class OpenLineageListener:
task_uuid,
include_full_task_info=controls.include_full_task_info,
ti_position=position,
+ end_date=end_date,
),
**get_airflow_debug_facet(),
},
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
b/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
index 34c235abc24..320a324da8d 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
@@ -73,6 +73,7 @@ from airflow.providers.openlineage.utils.selective_enable
import (
)
from airflow.providers.openlineage.version_compat import (
AIRFLOW_V_3_0_PLUS,
+ AIRFLOW_V_3_2_2_PLUS,
AIRFLOW_V_3_2_PLUS,
AIRFLOW_V_3_3_PLUS,
AIRFLOW_V_3_4_PLUS,
@@ -1188,7 +1189,7 @@ class DagRunInfo(InfoJsonEncodable):
class TaskInstanceInfo(InfoJsonEncodable):
"""Defines encoding TaskInstance object to JSON."""
- includes = ["duration", "log_url", "max_tries", "pool", "queued_dttm",
"state", "try_number"]
+ includes = ["log_url", "max_tries", "pool", "queued_dttm", "state",
"try_number"]
casts = {
"log_url": lambda ti: getattr(ti, "log_url", None),
"dag_version_id": lambda ti: (
@@ -1209,7 +1210,12 @@ class TaskInstanceInfo(InfoJsonEncodable):
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.
@@ -1217,9 +1223,57 @@ class TaskInstanceInfo(InfoJsonEncodable):
# 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_start_date(self) -> datetime.datetime | None:
+ if not isinstance(start_date := getattr(self.obj, "start_date", None),
datetime.datetime):
+ return None
+ # Before Airflow 3.2.2 a runtime task instance resumed from deferral
starts at the resume time, while the
+ # stored start date stays the original one. Leaving it out keeps a
wrong duration out of the facet too.
+ if not AIRFLOW_V_3_2_2_PLUS and self._is_resumed_from_deferral():
+ return None
+ return start_date
+
+ def _is_resumed_from_deferral(self) -> bool:
+ ti_context = getattr(self.obj, "_ti_context_from_server", None)
+ return bool(getattr(ti_context, "next_method", None))
+
+ 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. Airflow 2 keeps the
previous try's duration when it clears
+ # end_date for the next try, so the stored duration only counts
alongside an end_date.
+ if (duration := getattr(self.obj, "duration", None)) is not None and
isinstance(
+ getattr(self.obj, "end_date", None), datetime.datetime
+ ):
+ return duration
+ if start_date is None or end_date is None:
+ return None
+ try:
+ return (end_date - start_date).total_seconds()
+ except TypeError: # naive and aware datetimes can't be subtracted
+ return None
+
def _extend_fields(self) -> None:
+ start_date = self._get_start_date()
+ end_date = self._get_end_date()
+ for field, field_value in (
+ ("start_date", start_date),
+ ("end_date", end_date),
+ ("duration", self._get_duration(start_date, end_date)),
+ ):
+ if field_value is not None:
+ setattr(self, field, field_value)
+ self._fields.append(field)
position = self._position
if position is None:
# A stored task instance whose task isn't attached can't be placed
without its region, and a facet
@@ -1552,13 +1606,16 @@ def get_airflow_run_facet(
task_uuid: str,
include_full_task_info: bool = False,
ti_position: TaskInstancePosition | None = None,
+ end_date: datetime.datetime | None = None,
) -> dict[str, AirflowRunFacet]:
runtime_assets = get_runtime_outlet_assets(task_instance)
return {
"airflow": AirflowRunFacet(
dag=DagInfo(dag),
dagRun=DagRunInfo(dag_run),
- taskInstance=TaskInstanceInfo(task_instance, position=ti_position,
dag=dag, dag_run=dag_run),
+ taskInstance=TaskInstanceInfo(
+ task_instance, position=ti_position, dag=dag, dag_run=dag_run,
end_date=end_date
+ ),
task=(
TaskInfoComplete(task, runtime_assets=runtime_assets)
if include_full_task_info
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/version_compat.py
b/providers/openlineage/src/airflow/providers/openlineage/version_compat.py
index 548ae8df9fc..5a42154445a 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/version_compat.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/version_compat.py
@@ -34,8 +34,15 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
AIRFLOW_V_3_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 0)
+AIRFLOW_V_3_2_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 2)
AIRFLOW_V_3_3_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
AIRFLOW_V_3_4_PLUS = get_base_airflow_version_tuple() >= (3, 4, 0)
-__all__ = ["AIRFLOW_V_3_0_PLUS", "AIRFLOW_V_3_2_PLUS", "AIRFLOW_V_3_3_PLUS",
"AIRFLOW_V_3_4_PLUS"]
+__all__ = [
+ "AIRFLOW_V_3_0_PLUS",
+ "AIRFLOW_V_3_2_PLUS",
+ "AIRFLOW_V_3_2_2_PLUS",
+ "AIRFLOW_V_3_3_PLUS",
+ "AIRFLOW_V_3_4_PLUS",
+]
diff --git
a/providers/openlineage/tests/system/openlineage/example_openlineage_defer_simple_dag.py
b/providers/openlineage/tests/system/openlineage/example_openlineage_defer_simple_dag.py
index d0937e0964a..f7dcf68762d 100644
---
a/providers/openlineage/tests/system/openlineage/example_openlineage_defer_simple_dag.py
+++
b/providers/openlineage/tests/system/openlineage/example_openlineage_defer_simple_dag.py
@@ -20,6 +20,8 @@ Simple DAG with deferrable operator.
It checks:
- that at least two task START events (before and after deferral) are
emitted and
the try_num remains at 1
+ - taskInstance's `start_date` and `duration` after the resume: absent on
Airflow 3.0 to 3.2.1, where the
+ worker's start date is the resume time, and present otherwise
"""
from __future__ import annotations
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/__init__.py
b/providers/openlineage/tests/system/openlineage/expected_events/__init__.py
index 3bd0dafd402..0f081502a98 100644
--- a/providers/openlineage/tests/system/openlineage/expected_events/__init__.py
+++ b/providers/openlineage/tests/system/openlineage/expected_events/__init__.py
@@ -47,8 +47,10 @@ def get_expected_event_file_path(dag_id: str) -> str:
1. A file named `{dag_id}__af{major_version}_{minor_version}.json`
(e.g., `example_dag__af2_10.json` for Airflow 2.10.x)
2. A file named `{dag_id}__af{major_version}_{N}_plus.json` where N <=
current minor version
- (e.g., `example_dag__af3_3_plus.json` matches Airflow 3.3, 3.4, 3.5, …)
- When multiple such files exist, the one with the highest N wins.
+ (e.g., `example_dag__af3_3_plus.json` matches Airflow 3.3, 3.4, 3.5,
…), or
+ `{dag_id}__af{major_version}_{N}_{P}_plus.json` for a patch release
+ (e.g., `example_dag__af3_2_2_plus.json` matches Airflow 3.2.2, 3.2.3,
3.3.0, …)
+ When multiple such files exist, the one with the highest version wins.
3. A file named `{dag_id}__af{major_version}.json`
(e.g., `example_dag__af3.json` for any Airflow 3.x version)
4. A generic file named `{dag_id}.json` without version suffix
@@ -76,17 +78,17 @@ def get_expected_event_file_path(dag_id: str) -> str:
if os.path.exists(exact_minor):
return exact_minor
- # 2. Minimum-version files: {dag_id}__af{major}_{N}_plus.json where N <=
current minor
+ # 2. Minimum-version files: {dag_id}__af{major}_{N}[_{P}]_plus.json where
N[.P] <= current minor[.micro]
plus_candidates = []
for f in
base_path.glob(f"{dag_id}__af{AIRFLOW_VERSION.major}_*_plus.json"):
m = re.fullmatch(
-
rf"{re.escape(dag_id)}__af{AIRFLOW_VERSION.major}_(\d+)_plus\.json",
+
rf"{re.escape(dag_id)}__af{AIRFLOW_VERSION.major}_(\d+)(?:_(\d+))?_plus\.json",
f.name,
)
if m:
- file_minor = int(m.group(1))
- if file_minor <= AIRFLOW_VERSION.minor:
- plus_candidates.append((file_minor, str(f)))
+ file_version = (int(m.group(1)), int(m.group(2) or 0))
+ if file_version <= (AIRFLOW_VERSION.minor, AIRFLOW_VERSION.micro):
+ plus_candidates.append((file_version, str(f)))
if plus_candidates:
plus_candidates.sort(key=lambda x: x[0], reverse=True)
return plus_candidates[0][1]
@@ -103,6 +105,6 @@ def get_expected_event_file_path(dag_id: str) -> str:
f"Could not locate expected event files for dag_id {dag_id} "
f"(Airflow {AIRFLOW_VERSION.major}.{AIRFLOW_VERSION.minor}). "
f"Checked:
{dag_id}__af{AIRFLOW_VERSION.major}_{AIRFLOW_VERSION.minor}.json, "
- f"{dag_id}__af{AIRFLOW_VERSION.major}_N_plus.json
(N<={AIRFLOW_VERSION.minor}), "
+ f"{dag_id}__af{AIRFLOW_VERSION.major}_N[_P]_plus.json
(N[.P]<={AIRFLOW_VERSION.minor}.{AIRFLOW_VERSION.micro}), "
f"{dag_id}__af{AIRFLOW_VERSION.major}.json, {dag_id}.json"
)
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
index 362d5c1bcc9..11035532d85 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
@@ -332,7 +332,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_1.id.with.dots.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_1.id.with.dots.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -512,7 +515,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_1.id.with.dots.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"depends_on_past": false,
@@ -691,7 +696,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_2.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_2.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -885,7 +893,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_2.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"depends_on_past": false,
@@ -1072,7 +1082,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_3.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_3.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -1255,7 +1268,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_3.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"depends_on_past": false,
@@ -1431,7 +1446,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_4.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_4.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"deferrable": "{{ result == true }}",
@@ -1597,7 +1615,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=task_4.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"deferrable": "{{ result == true }}",
@@ -1762,7 +1782,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.task_5.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.task_5.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -1932,7 +1955,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.task_5.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"depends_on_past": false,
@@ -2101,7 +2126,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.section_2.task_6.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.section_2.task_6.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -2283,7 +2311,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.section_2.task_6.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"depends_on_past": false,
@@ -2464,7 +2494,10 @@
"try_number": "{{ result is number }}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.section_2.section_3.task_7.*\")
}}"
+ "log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.section_2.section_3.task_7.*\")
}}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -2644,7 +2677,9 @@
"log_url": "{{ regex_match(result,
\"^http[s]?:\\/\\/.*\\/dags\\/openlineage_base_complex_dag\\/grid\\?dag_run_id=.*?\\&task_id=section_1.section_2.section_3.task_7.*\")
}}",
"pool": "{{ result is string }}",
"queued_dttm": "{{ is_datetime(result) }}",
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}"
},
"task": {
"depends_on_past": false,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
index 05fb49bfd16..7a3b9f1d654 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
@@ -333,7 +333,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -508,7 +511,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
@@ -683,7 +689,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -873,7 +882,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
@@ -1057,7 +1069,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -1235,7 +1250,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
@@ -1407,7 +1425,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"deferrable": "{{ result == true }}",
@@ -1568,7 +1589,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"deferrable": "{{ result == true }}",
@@ -1729,7 +1753,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -1894,7 +1921,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
@@ -2059,7 +2089,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -2235,7 +2268,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
@@ -2411,7 +2447,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"depends_on_past": false,
@@ -2586,7 +2625,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
index da4b7f60472..98d9590ebe9 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
@@ -80,7 +80,10 @@
"start_date": "{{ is_datetime(result) }}"
},
"taskInstance": {
- "try_number": "{{ result is number }}"
+ "try_number": "{{ result is number }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"depends_on_past": false,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
index d3604f916d0..0dbf5ab0af9 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
@@ -5,7 +5,10 @@
"facets": {
"airflow": {
"taskInstance": {
- "try_number": 1
+ "try_number": 1,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
}
}
}
@@ -20,7 +23,10 @@
"facets": {
"airflow": {
"taskInstance": {
- "try_number": 1
+ "try_number": 1,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
}
}
}
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag__af3.json
similarity index 62%
copy from
providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
copy to
providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag__af3.json
index d3604f916d0..cea3487bae6 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag__af3.json
@@ -5,7 +5,10 @@
"facets": {
"airflow": {
"taskInstance": {
- "try_number": 1
+ "try_number": 1,
+ "start_date": null,
+ "end_date": null,
+ "duration": null
}
}
}
@@ -20,7 +23,10 @@
"facets": {
"airflow": {
"taskInstance": {
- "try_number": 1
+ "try_number": 1,
+ "start_date": null,
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": null
}
}
}
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag__af3_2_2_plus.json
similarity index 58%
copy from
providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
copy to
providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag__af3_2_2_plus.json
index d3604f916d0..0dbf5ab0af9 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_defer_simple_dag__af3_2_2_plus.json
@@ -5,7 +5,10 @@
"facets": {
"airflow": {
"taskInstance": {
- "try_number": 1
+ "try_number": 1,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
}
}
}
@@ -20,7 +23,10 @@
"facets": {
"airflow": {
"taskInstance": {
- "try_number": 1
+ "try_number": 1,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
}
}
}
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag.json
index 2b3a8a24cf4..c0079cf927d 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag.json
@@ -29,6 +29,9 @@
"taskInstance": {
"try_number": 1,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -80,6 +83,9 @@
"taskInstance": {
"try_number": 1,
"state": "up_for_retry",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -131,6 +137,9 @@
"taskInstance": {
"try_number": 2,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -182,6 +191,9 @@
"taskInstance": {
"try_number": 2,
"state": "success",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af2.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af2.json
index c4c95d87aed..76b72411964 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af2.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af2.json
@@ -39,7 +39,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": null,
- "dag_bundle_version": null
+ "dag_bundle_version": null,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"retries": 1,
@@ -90,7 +93,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": null,
- "dag_bundle_version": null
+ "dag_bundle_version": null,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"retries": 1,
@@ -141,7 +147,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": null,
- "dag_bundle_version": null
+ "dag_bundle_version": null,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"task": {
"retries": 1,
@@ -192,7 +201,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": null,
- "dag_bundle_version": null
+ "dag_bundle_version": null,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"task": {
"retries": 1,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af3_0.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af3_0.json
index ede391a026c..ad654eb9eb6 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af3_0.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_retry_dag__af3_0.json
@@ -29,6 +29,9 @@
"taskInstance": {
"try_number": 1,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -80,6 +83,9 @@
"taskInstance": {
"try_number": 1,
"state": "up_for_retry",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -131,6 +137,9 @@
"taskInstance": {
"try_number": 2,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -182,6 +191,9 @@
"taskInstance": {
"try_number": 2,
"state": "success",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"max_tries": 1,
"id": "{{ is_uuid(result) }}",
"map_index": null,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag.json
index e46cec58913..c2b70810dc4 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag.json
@@ -37,7 +37,10 @@
"region_index": null,
"dag_version_id": "{{ is_uuid(result) }}",
"dag_bundle_name": "{{ result is string }}",
- "dag_bundle_version": "{{ result is string }}"
+ "dag_bundle_version": "{{ result is string }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"dag": {
"dag_id": "openlineage_sensor_reschedule_dag"
@@ -86,7 +89,10 @@
"region_index": null,
"dag_version_id": "{{ is_uuid(result) }}",
"dag_bundle_name": "{{ result is string }}",
- "dag_bundle_version": "{{ result is string }}"
+ "dag_bundle_version": "{{ result is string }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"dag": {
"dag_id": "openlineage_sensor_reschedule_dag"
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af2.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af2.json
index 51bd0716c15..5c0c6652b8a 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af2.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af2.json
@@ -37,7 +37,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": null,
- "dag_bundle_version": null
+ "dag_bundle_version": null,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"dag": {
"dag_id": "openlineage_sensor_reschedule_dag"
@@ -86,7 +89,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": null,
- "dag_bundle_version": null
+ "dag_bundle_version": null,
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"dag": {
"dag_id": "openlineage_sensor_reschedule_dag"
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af3_0.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af3_0.json
index d942dbbda86..3094b7e20d3 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af3_0.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_sensor_reschedule_dag__af3_0.json
@@ -37,7 +37,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": "{{ result is string }}",
- "dag_bundle_version": "{{ result is string }}"
+ "dag_bundle_version": "{{ result is string }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null
},
"dag": {
"dag_id": "openlineage_sensor_reschedule_dag"
@@ -86,7 +89,10 @@
"region_index": null,
"dag_version_id": null,
"dag_bundle_name": "{{ result is string }}",
- "dag_bundle_version": "{{ result is string }}"
+ "dag_bundle_version": "{{ result is string }}",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}"
},
"dag": {
"dag_id": "openlineage_sensor_reschedule_dag"
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag.json
index c0b16b8cb57..d948a58fc44 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag.json
@@ -41,6 +41,9 @@
"try_number": 1,
"max_tries": 0,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"dag_version_id": "{{ is_uuid(result) }}",
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -92,6 +95,9 @@
"try_number": 1,
"max_tries": 0,
"state": "skipped",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"dag_version_id": "{{ is_uuid(result) }}",
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -150,6 +156,9 @@
"try_number": 1,
"max_tries": 2,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"dag_version_id": "{{ is_uuid(result) }}",
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -201,6 +210,9 @@
"try_number": 1,
"max_tries": 2,
"state": "failed",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"dag_version_id": "{{ is_uuid(result) }}",
"id": "{{ is_uuid(result) }}",
"map_index": null,
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag__af3_0.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag__af3_0.json
index ac87ea35b21..52cee13d52b 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag__af3_0.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_task_states_dag__af3_0.json
@@ -55,6 +55,9 @@
"try_number": 1,
"max_tries": 2,
"state": "running",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": null,
+ "duration": null,
"dag_version_id": null,
"id": "{{ is_uuid(result) }}",
"map_index": null,
@@ -106,6 +109,9 @@
"try_number": 1,
"max_tries": 2,
"state": "failed",
+ "start_date": "{{ is_datetime(result) }}",
+ "end_date": "{{ is_datetime(result) }}",
+ "duration": "{{ result is number }}",
"dag_version_id": null,
"id": "{{ is_uuid(result) }}",
"map_index": null,
diff --git
a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
index 32a6b7570db..35b7e28987e 100644
--- a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
+++ b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
@@ -45,7 +45,7 @@ from airflow.providers.common.compat.sdk import (
)
from airflow.providers.openlineage.extractors.base import OperatorLineage
from airflow.providers.openlineage.plugins.adapter import OpenLineageAdapter
-from airflow.providers.openlineage.plugins.listener import OpenLineageListener
+from airflow.providers.openlineage.plugins.listener import
OpenLineageListener, _get_task_end_date
from airflow.providers.openlineage.plugins.macros import lineage_run_id
from airflow.providers.openlineage.utils.emission_policy import EmissionPolicy
from airflow.providers.openlineage.utils.selective_enable import
disable_lineage, enable_lineage
@@ -887,6 +887,49 @@ class TestOpenLineageListenerAirflow2:
assert
listener.adapter.complete_task.call_args.kwargs["job_description"] == "Test DAG
Description"
assert
listener.adapter.complete_task.call_args.kwargs["job_description_type"] ==
"text/plain"
+ @pytest.mark.parametrize(
+ ("hook_name", "hook_kwargs", "adapter_method"),
+ [
+ pytest.param("on_task_instance_success", {}, "complete_task",
id="success"),
+ pytest.param("on_task_instance_failed", {"error":
ValueError("boom")}, "fail_task", id="failed"),
+ ],
+ )
+ @mock.patch("airflow.providers.openlineage.conf.debug_mode",
return_value=True)
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_debug_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_task_parent_run_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_run_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_user_provided_run_facets")
+ @mock.patch(
+
"airflow.providers.openlineage.plugins.listener.OpenLineageListener._execute",
new=regular_call
+ )
+ def test_terminal_event_time_comes_from_task_instance_end_date(
+ self,
+ mock_get_user_provided_run_facets,
+ mock_get_airflow_run_facet,
+ mock_get_task_parent_run_facet,
+ mock_debug_facet,
+ mock_debug_mode,
+ time_machine,
+ hook_name,
+ hook_kwargs,
+ adapter_method,
+ ):
+ time_machine.move_to(timezone.datetime(2023, 1, 3, 13, 1, 1),
tick=False)
+ listener, task_instance = self._create_listener_and_task_instance()
+ task_instance.end_date = timezone.datetime(2023, 1, 2, 3, 4, 5)
+ mock_get_user_provided_run_facets.return_value = {}
+ mock_get_airflow_run_facet.return_value = {}
+ mock_get_task_parent_run_facet.return_value = {}
+ mock_debug_facet.return_value = {}
+
+ getattr(listener, hook_name)(
+ previous_state=None, task_instance=task_instance, session=None,
**hook_kwargs
+ )
+
+ adapter_call = getattr(listener.adapter, adapter_method)
+ assert adapter_call.call_args.kwargs["end_time"] ==
"2023-01-02T03:04:05+00:00"
+ assert mock_get_airflow_run_facet.call_args.kwargs["end_date"] ==
task_instance.end_date
+
@mock.patch(
"airflow.providers.openlineage.plugins.listener.OpenLineageListener._execute",
new=regular_call
)
@@ -2624,6 +2667,71 @@ class TestOpenLineageListenerAirflow3:
map_index=-1,
)
+ @pytest.mark.parametrize(
+ ("hook_name", "hook_kwargs", "adapter_method"),
+ [
+ pytest.param("on_task_instance_success", {}, "complete_task",
id="success"),
+ pytest.param("on_task_instance_failed", {"error":
ValueError("boom")}, "fail_task", id="failed"),
+ pytest.param("on_task_instance_skipped", {}, "complete_task",
id="skipped"),
+ ],
+ )
+ @pytest.mark.parametrize(
+ ("ti_end_date", "expected_end_time"),
+ [
+ pytest.param(
+ timezone.datetime(2023, 1, 2, 3, 4, 5),
+ "2023-01-02T03:04:05+00:00",
+ id="end_date_set_by_task_runner",
+ ),
+ pytest.param(None, "2023-01-03T13:01:01+00:00",
id="end_date_not_set_falls_back_to_now"),
+ ],
+ )
+ @mock.patch("airflow.providers.openlineage.conf.debug_mode",
return_value=True)
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_debug_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_task_parent_run_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_run_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_user_provided_run_facets")
+ @mock.patch(
+
"airflow.providers.openlineage.plugins.listener.OpenLineageListener._execute",
new=regular_call
+ )
+ def test_terminal_event_time_comes_from_runtime_task_instance_end_date(
+ self,
+ mock_get_user_provided_run_facets,
+ mock_get_airflow_run_facet,
+ mock_get_task_parent_run_facet,
+ mock_debug_facet,
+ mock_debug_mode,
+ time_machine,
+ hook_name,
+ hook_kwargs,
+ adapter_method,
+ ti_end_date,
+ expected_end_time,
+ ):
+ time_machine.move_to(timezone.datetime(2023, 1, 3, 13, 1, 1),
tick=False)
+ listener, task_instance = self._create_listener_and_task_instance()
+ task_instance.end_date = ti_end_date
+ mock_get_user_provided_run_facets.return_value = {}
+ mock_get_airflow_run_facet.return_value = {}
+ mock_get_task_parent_run_facet.return_value = {}
+ mock_debug_facet.return_value = {}
+
+ getattr(listener, hook_name)(previous_state=None,
task_instance=task_instance, **hook_kwargs)
+
+ adapter_call = getattr(listener.adapter, adapter_method)
+ adapter_call.assert_called_once()
+ assert adapter_call.call_args.kwargs["end_time"] == expected_end_time
+ # The facet must describe the same end date as the event, or its
duration would disagree with it
+ assert mock_get_airflow_run_facet.call_args.kwargs["end_date"] ==
timezone.parse(expected_end_time)
+
+ @pytest.mark.parametrize("end_date", [None, "2023-01-02T03:04:05+00:00",
MagicMock()])
+ @mock.patch("airflow.providers.openlineage.plugins.listener.timezone")
+ def test_get_task_end_date_ignores_values_that_are_not_datetimes(self,
mock_timezone, end_date):
+ mock_timezone.utcnow.return_value = timezone.datetime(2023, 1, 3, 13,
1, 1)
+ task_instance = MagicMock(end_date=end_date)
+
+ assert _get_task_end_date(task_instance) == timezone.datetime(2023, 1,
3, 13, 1, 1)
+
@mock.patch("airflow.providers.openlineage.plugins.listener.resolve_task_emission_policy")
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_run_facet")
@mock.patch("airflow.providers.openlineage.plugins.listener.get_user_provided_run_facets")
diff --git a/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
b/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
index 8db37c2c355..9c9518b97bc 100644
--- a/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
+++ b/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
@@ -3600,6 +3600,9 @@ def test_taskinstance_info_af3():
runtime_ti.state = TaskInstanceState.UP_FOR_RETRY
assert dict(TaskInstanceInfo(runtime_ti)) == {
+ "duration": 12.345,
+ "start_date": "2025-01-01T00:00:00+00:00",
+ "end_date": "2025-01-01T00:00:12.345000+00:00",
"id": str(runtime_ti.id),
"log_url": runtime_ti.log_url,
"map_index": 2,
@@ -3645,11 +3648,11 @@ def test_taskinstance_info_af2():
ti = TaskInstance(
task=task_obj, run_id="task_instance_run_id",
state=TaskInstanceState.RUNNING, map_index=2
)
+ # Left from the previous try, as end_date is cleared when the next one
starts, so it isn't reported
ti.duration = 12.345
ti.queued_dttm = some_date
assert dict(TaskInstanceInfo(ti)) == {
- "duration": 12.345,
"map_index": 2,
"rendered_map_index": None,
"pool": "default_pool",
@@ -6114,6 +6117,194 @@ def
test_convert_to_json_safe_list_returns_none_on_failure(mock_log):
assert str(mock_log.warning.call_args.args[2]) == "Circular reference
detected"
+_DURATION_START = datetime.datetime(2024, 6, 1, 12, 0, 0, tzinfo=datetime.UTC)
+
+
[email protected](
+ ("ti_attrs", "end_date", "expected_duration"),
+ [
+ pytest.param({"duration": 3.5, "end_date": _DURATION_START}, None,
3.5, id="stored_duration"),
+ pytest.param(
+ {"duration": 3.5, "start_date": _DURATION_START, "end_date":
_DURATION_START},
+ _DURATION_START + datetime.timedelta(seconds=10),
+ 3.5,
+ id="stored_duration_wins_over_passed_end_date",
+ ),
+ pytest.param(
+ {"duration": 3.5, "start_date": _DURATION_START, "end_date": None},
+ None,
+ None,
+ id="stored_duration_of_previous_try_without_end_date",
+ ),
+ pytest.param(
+ {"duration": 3.5, "start_date": _DURATION_START, "end_date": None},
+ _DURATION_START + datetime.timedelta(seconds=10),
+ 10.0,
+ id="passed_end_date_over_stored_duration_without_end_date",
+ ),
+ pytest.param(
+ {"duration": None, "start_date": _DURATION_START},
+ _DURATION_START + datetime.timedelta(seconds=10, milliseconds=500),
+ 10.5,
+ id="passed_end_date_when_stored_duration_is_none",
+ ),
+ pytest.param(
+ {"start_date": _DURATION_START},
+ _DURATION_START + datetime.timedelta(minutes=2),
+ 120.0,
+ id="passed_end_date_when_task_instance_has_none_yet",
+ ),
+ pytest.param(
+ {"start_date": _DURATION_START, "end_date": _DURATION_START +
datetime.timedelta(seconds=7)},
+ None,
+ 7.0,
+ id="end_date_of_task_instance",
+ ),
+ pytest.param(
+ {
+ "duration": None,
+ "start_date": _DURATION_START,
+ "end_date": _DURATION_START + datetime.timedelta(seconds=7),
+ },
+ None,
+ 7.0,
+ id="end_date_of_task_instance_when_stored_duration_is_none",
+ ),
+ pytest.param(
+ {"start_date": _DURATION_START, "end_date": _DURATION_START +
datetime.timedelta(seconds=7)},
+ _DURATION_START + datetime.timedelta(seconds=9),
+ 7.0,
+ id="end_date_of_task_instance_wins_over_passed_end_date",
+ ),
+ pytest.param({"start_date": _DURATION_START}, None, None,
id="running_task_instance"),
+ pytest.param({}, None, None, id="no_dates"),
+ pytest.param(
+ {},
+ _DURATION_START + datetime.timedelta(seconds=10),
+ None,
+ id="passed_end_date_without_start_date",
+ ),
+ pytest.param(
+ {"start_date": None, "end_date": _DURATION_START},
+ None,
+ None,
+ id="start_date_is_none",
+ ),
+ pytest.param(
+ {"start_date": "2024-06-01T12:00:00+00:00", "end_date":
"2024-06-01T12:00:07+00:00"},
+ None,
+ None,
+ id="dates_are_not_datetimes",
+ ),
+ pytest.param(
+ {"start_date": datetime.datetime(2024, 6, 1, 12, 0, 0)},
+ _DURATION_START + datetime.timedelta(seconds=10),
+ None,
+ id="naive_start_date_and_aware_end_date",
+ ),
+ ],
+)
+def test_task_instance_info_duration(ti_attrs, end_date, expected_duration):
+ ti = SimpleNamespace(id=uuid7(), map_index=-1, **ti_attrs)
+
+ info = dict(TaskInstanceInfo(ti, end_date=end_date))
+
+ if expected_duration is None:
+ assert "duration" not in info
+ else:
+ assert info["duration"] == expected_duration
+
+
[email protected](
+ ("ti_attrs", "end_date", "expected_dates"),
+ [
+ pytest.param(
+ {"start_date": _DURATION_START, "end_date": _DURATION_START +
datetime.timedelta(seconds=7)},
+ _DURATION_START + datetime.timedelta(seconds=9),
+ {"start_date": "2024-06-01T12:00:00+00:00", "end_date":
"2024-06-01T12:00:07+00:00"},
+ id="end_date_of_task_instance_wins_over_passed_end_date",
+ ),
+ pytest.param(
+ {"start_date": _DURATION_START, "end_date": None},
+ _DURATION_START + datetime.timedelta(seconds=9),
+ {"start_date": "2024-06-01T12:00:00+00:00", "end_date":
"2024-06-01T12:00:09+00:00"},
+ id="passed_end_date_when_task_instance_has_none",
+ ),
+ pytest.param(
+ {"start_date": _DURATION_START},
+ None,
+ {"start_date": "2024-06-01T12:00:00+00:00"},
+ id="running_task_instance",
+ ),
+ pytest.param(
+ {"start_date": "2024-06-01T12:00:00+00:00", "end_date":
"2024-06-01T12:00:07+00:00"},
+ None,
+ {},
+ id="dates_are_not_datetimes",
+ ),
+ ],
+)
+def test_task_instance_info_dates(ti_attrs, end_date, expected_dates):
+ ti = SimpleNamespace(id=uuid7(), map_index=-1, **ti_attrs)
+
+ info = dict(TaskInstanceInfo(ti, end_date=end_date))
+
+ assert {key: info[key] for key in ("start_date", "end_date") if key in
info} == expected_dates
+
+
[email protected](
+ ("airflow_3_2_2_plus", "next_method", "expected_fields"),
+ [
+ pytest.param(False, "execute_complete", {}, id="resumed_before_3_2_2"),
+ pytest.param(
+ True,
+ "execute_complete",
+ {"start_date": "2024-06-01T12:00:00+00:00", "duration": 9.0},
+ id="resumed_since_3_2_2",
+ ),
+ pytest.param(
+ False,
+ None,
+ {"start_date": "2024-06-01T12:00:00+00:00", "duration": 9.0},
+ id="not_resumed_before_3_2_2",
+ ),
+ ],
+)
+def test_task_instance_info_start_date_of_task_resumed_from_deferral(
+ airflow_3_2_2_plus, next_method, expected_fields
+):
+ ti = SimpleNamespace(
+ id=uuid7(),
+ map_index=-1,
+ start_date=_DURATION_START,
+ end_date=None,
+ _ti_context_from_server=SimpleNamespace(next_method=next_method),
+ )
+
+ with
patch("airflow.providers.openlineage.utils.utils.AIRFLOW_V_3_2_2_PLUS",
airflow_3_2_2_plus):
+ info = dict(TaskInstanceInfo(ti, end_date=_DURATION_START +
datetime.timedelta(seconds=9)))
+
+ assert {key: info[key] for key in ("start_date", "duration") if key in
info} == expected_fields
+ assert info["end_date"] == "2024-06-01T12:00:09+00:00"
+
+
+@patch("airflow.providers.openlineage.utils.utils.get_runtime_outlet_assets",
return_value=[])
+def
test_get_airflow_run_facet_passes_end_date_to_task_instance_facet(mock_get_runtime):
+ task = BashOperator(task_id="t", bash_command="echo 1")
+ ti = SimpleNamespace(id=uuid7(), map_index=-1, start_date=_DURATION_START)
+
+ facet = get_airflow_run_facet(
+ MagicMock(),
+ MagicMock(),
+ ti,
+ task,
+ "task_uuid",
+ end_date=_DURATION_START + datetime.timedelta(seconds=4),
+ )["airflow"]
+
+ assert facet.taskInstance["duration"] == 4.0
+
+
@patch("airflow.providers.openlineage.utils.utils.get_runtime_outlet_assets")
@patch("airflow.providers.openlineage.conf.include_full_task_info")
def test_get_airflow_run_facet_plumbs_runtime_assets(mock_include_full,
mock_get_runtime):
