kacpermuda commented on code in PR #75678:
URL: https://github.com/apache/airflow/pull/75678#discussion_r4403408311
##########
providers/openlineage/tests/system/openlineage/expected_events/openlineage_policy_exclude_datasets_versioned_dag.json:
##########
@@ -0,0 +1,133 @@
+[
+ {
+ "eventType": "START",
+ "job": {
+ "name": "openlineage_policy_exclude_datasets_versioned_dag",
+ "facets": {
+ "airflow": {
+ "tasks": {
+ "t_skipped_emit_off": {
+ "emits_ol_events": "{{ result == false }}"
+ },
+ "t_skipped_full_info": {
+ "emits_ol_events": "{{ result == true }}"
+ },
+ "t_runtime_outlets": {
+ "emits_ol_events": "{{ result == true }}"
+ },
+ "t_collector_cap": {
+ "emits_ol_events": "{{ result == true }}"
+ },
+ "t_skipped": {
+ "emits_ol_events": "{{ result == true }}"
+ },
+ "check_events": {
+ "emits_ol_events": "{{ result == true }}"
+ }
+ }
+ }
+ }
+ }
+ },
+ {
+ "eventType": "START",
+ "inputs": [
+ {
+ "namespace": "file://host1",
+ "name": "/keep_declared_in.txt"
+ }
+ ],
+ "outputs": [
+ {
+ "namespace": "file://host1",
+ "name": "/keep_declared_out.txt"
+ }
+ ],
+ "job": {
+ "name":
"openlineage_policy_exclude_datasets_versioned_dag.t_runtime_outlets"
+ }
+ },
+ {
+ "eventType": "COMPLETE",
+ "inputs": [
+ {
+ "namespace": "file://host1",
+ "name": "/keep_declared_in.txt"
+ }
+ ],
+ "outputs": [
+ {
+ "namespace": "file://host1",
+ "name": "/keep_declared_out.txt"
+ }
+ ],
+ "job": {
+ "name":
"openlineage_policy_exclude_datasets_versioned_dag.t_runtime_outlets"
+ }
+ },
+ {
+ "eventType": "START",
+ "inputs": [],
+ "outputs": [],
+ "job": {
+ "name":
"openlineage_policy_exclude_datasets_versioned_dag.t_collector_cap"
+ }
+ },
+ {
+ "eventType": "COMPLETE",
+ "inputs": "{{ result | length == 65 and result[0].name ==
'/cap_keep_06_in.txt' and (result | last).name == '/cap_keep_70_in.txt' }}",
+ "outputs": "{{ result | length == 65 and result[0].name ==
'/cap_keep_06_out.txt' and (result | last).name == '/cap_keep_70_out.txt' }}",
+ "job": {
+ "name":
"openlineage_policy_exclude_datasets_versioned_dag.t_collector_cap"
+ }
+ },
+ {
+ "eventType": "COMPLETE",
+ "job": {
+ "name":
"openlineage_policy_exclude_datasets_versioned_dag.t_skipped"
+ },
+ "$absent": "Airflow 3.0 calls no listener when a task skips itself"
Review Comment:
Good catch, the START entries were left out on purpose but for the wrong
reason: the runId check rejects a task with START and no COMPLETE/FAIL. I added
both STARTs (same as 3.1+) and reworded the reason to “finish listener”.
check_run_ids is now turned off on 3.0 for this Dag only, with a comment saying
why.
##########
providers/openlineage/tests/system/openlineage/example_openlineage_policy_dag_events_trigger_dag.py:
##########
@@ -0,0 +1,121 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+DAG exercising the emission policy for Dag rules, including the Dag run
COMPLETE and FAIL events.
+
+A Dag run's own COMPLETE and FAIL events come after its checking task, so the
Dag triggers child Dags and waits for
+them, and its checking task reads their events. The children end in
``__notrigger`` so the harness does not
+trigger them itself, and the failing ones fail their trigger task, which the
checking task does not depend on.
+
+The comment above each task says which rules and authoring calls apply to it,
and what its events must show.
+
+The rules are in ``AIRFLOW__OPENLINEAGE__EMISSION_POLICY`` of
``airflow-e2e-tests/docker/openlineage.yml``.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow import DAG
+from airflow.providers.openlineage.api.emission_policy import
extend_global_openlineage_emission_policy
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.providers.standard.operators.trigger_dagrun import
TriggerDagRunOperator
+from airflow.utils.trigger_rule import TriggerRule
+
+from system.openlineage.constants import DEFAULT_DAGRUN_TIMEOUT
+from system.openlineage.expected_events import get_expected_event_file_path
+from system.openlineage.operator import OpenLineageTestOperator
+
+
+def _build_child_dag(dag_id: str, bash_command: str) -> DAG:
+ with DAG(
+ dagrun_timeout=DEFAULT_DAGRUN_TIMEOUT,
+ dag_id=dag_id,
+ start_date=datetime(2021, 1, 1),
+ schedule=None,
+ catchup=False,
+ default_args={"retries": 0},
+ ) as child_dag:
+ BashOperator(task_id="run", bash_command=bash_command)
+ return child_dag
+
+
+CHILD_COMMANDS = {
+ # No rule: Dag run START and COMPLETE, and the task's START and COMPLETE.
+ "ok": "echo ok",
+ # No rule: Dag run START and FAIL, and the task's START and FAIL.
+ "fail": "exit 1;",
+ # Conf Dag rule emit_dag_events: false, for a run that succeeds and one
that fails: no Dag run events, the task
+ # events are emitted.
+ "off_ok": "echo ok",
+ "off_fail": "exit 1;",
+ # Conf Dag rule emit: false with emit_dag_events: true, for a run that
succeeds and one that fails:
+ # emit_dag_events wins for the Dag run events, which are emitted, and the
task events are not.
+ "on_emit_off_ok": "echo ok",
+ "on_emit_off_fail": "exit 1;",
+ # Conf Dag rule emit: false: no Dag run events and no task events.
+ "emit_off": "echo ok",
+ # Conf Dag rule emit: false with emit_task_events: true: no Dag run
events, the task events are emitted.
+ "task_events_on_emit_off": "echo ok",
+ # Conf Dag rule emit_task_events: false: the Dag run events are emitted,
the task events are not.
+ "task_events_off": "echo ok",
+ # Conf locked Dag rule emit_dag_events: false, and the authored
emit_dag_events=True below is ignored: no Dag
+ # run events, the task events are emitted.
+ "locked_off": "echo ok",
+}
+CHILDREN = {
+ name:
_build_child_dag(f"openlineage_policy_dag_events_{name}_child__notrigger",
command)
+ for name, command in CHILD_COMMANDS.items()
+}
+extend_global_openlineage_emission_policy(CHILDREN["locked_off"],
emit_dag_events=True)
+
+
+DAG_ID = "openlineage_policy_dag_events_trigger_dag"
+
+with DAG(
+ dagrun_timeout=DEFAULT_DAGRUN_TIMEOUT,
+ dag_id=DAG_ID,
+ start_date=datetime(2021, 1, 1),
+ schedule=None,
+ catchup=False,
+ default_args={"retries": 0},
+) as dag:
+ # Not checks: each one triggers a child Dag and waits for it, so
check_events can read the child's events.
+ triggers = [
+ TriggerDagRunOperator(
Review Comment:
Done, the triggers are deferrable=True now (and poke_interval=10). Their
events aren’t checked, so the extra START on resume doesn’t matter.
--
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]
