mobuchowski commented on code in PR #75678:
URL: https://github.com/apache/airflow/pull/75678#discussion_r4402681529
##########
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:
On Airflow 3.0 a task that raises `AirflowSkipException` still sends START:
the 3.0.6 task runner calls `on_task_instance_running` before `execute()`, and
only skips the finish listener. This 3.0 file checks COMPLETE/FAIL as absent
for `t_skipped` and `t_skipped_full_info`, but has no START entry for either,
while the 3.1+ files do.
So on 3.0 nothing checks the two things the Dag comments promise for these
tasks: that `dag_excluded` is dropped from `t_skipped`'s START, and that
`bash_command` is in `t_skipped_full_info`'s START. For example, if exclusion
broke for skipped tasks on 3.0, this file would still pass.
Could you add the two START entries (same shape as in `__af3_1_plus.json`,
assuming 3.0 produces the same datasets/facets)? Also consider rewording the
reason to "Airflow 3.0 calls no finish listener when a task skips itself",
since the running listener does fire. If START is left out on 3.0 on purpose, a
short note saying why would help.
##########
airflow-e2e-tests/docker/openlineage.yml:
##########
@@ -35,10 +35,86 @@ x-ol-common-env: &ol-common-env
# fully-qualified class path; the OpenLineage client imports it directly.
AIRFLOW__OPENLINEAGE__TRANSPORT: '{"type":
"system.openlineage.transport.variable.VariableTransport"}'
AIRFLOW__OPENLINEAGE__DISABLED: 'false'
- # Deployment-wide policy required by the two conf-driven policy DAGs. Each
rule is scoped by
- # dag_id, so both rules coexist and do not affect any other DAG.
- AIRFLOW__OPENLINEAGE__EMISSION_POLICY: '[{"scope": {"dag_id":
"openlineage_policy_conf_locked_dag"}, "locked": true, "controls":
{"include_source_code": false}}, {"scope": {"dag_id":
"openlineage_policy_conf_source_code_dag"}, "controls": {"include_source_code":
false}}]'
+ # Deployment-wide policy required by the conf-driven policy DAGs. Rules are
scoped by dag_id or by
+ # names unique to a DAG (hook/operator classes, dataset patterns), so they
coexist and do not affect any other DAG.
+ AIRFLOW__OPENLINEAGE__EMISSION_POLICY: >-
+ [
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_dag"}, "locked":
true, "controls": {"include_source_code": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_source_code_dag"},
"controls": {"include_source_code": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_extract_off"}, "controls": {"extract_operator_metadata": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_hook_off"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_full_info"}, "controls": {"include_full_task_info": true}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_emit_off"}, "controls": {"emit": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_task_events_off"}, "controls": {"emit_task_events": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_task_events_beat_emit"}, "controls": {"emit": false, "emit_task_events":
true}},
+ {"scope": {"operator": ".*\\.PolicyConfControlsOperator"}, "match_mode":
"regex", "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_operator_task"}, "controls": {"hook_lineage": true}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_extract_overridden"}, "controls": {"extract_operator_metadata": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_hook_overridden"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_emit_overridden"}, "controls": {"emit": false}},
+ {"scope": {"operator": ".*\\.PolicyConfEmitOffOperator"}, "match_mode":
"regex", "controls": {"emit": false}},
+ {"scope": {"operator":
"airflow.providers.standard.operators.smooth.SmoothOperator"}, "controls":
{"emit": false}},
Review Comment:
The comment on line 39 says each rule is scoped by "names unique to a DAG
(hook/operator classes, dataset patterns)", so none affects any other Dag. That
holds for the test-defined classes, but this rule matches the built-in
`SmoothOperator`, deployment-wide. Nothing else uses it today. Still, anyone
who later adds an OpenLineage system Dag with a `SmoothOperator` will get no
task events, and the resulting "events missing" failure won't point here.
The PR description explains this, but the next person will read the yml, not
the PR. Could you add a short comment above this rule, e.g. `# Applies to every
Dag: do not use SmoothOperator in other OpenLineage system tests.`?
##########
providers/openlineage/tests/system/openlineage/example_openlineage_policy_conf_controls_dag.py:
##########
@@ -0,0 +1,200 @@
+# 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 every emission policy control in Dag + task and operator scoped
conf rules.
Review Comment:
Two small wording points (optional):
- The repo's naming rule (AGENTS.md / CLAUDE.md) asks for "Dag" in prose.
The new docstrings ("DAG exercising …" in all 8 new files) and the yml comments
on lines 39 and 115 ("any other DAG", "no other DAG") still use "DAG". The
existing policy Dags have the same header, so fixing those too would keep the
folder consistent.
- "every emission policy control" isn't quite true for this Dag: it has no
`include_source_code` or `exclude_datasets` rule (those are covered by the
locked-controls and `exclude_datasets_*` Dags). Something like "every
task-level boolean control" would avoid sending readers looking for them here.
##########
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:
Each of the 10 `TriggerDagRunOperator`s waits for its child Dag by polling
inside the worker process, so for as long as the child runs it holds one of the
Celery worker's slots (16 by default in the compose file the e2e uses). The
harness triggers every OpenLineage Dag in the same batch with a shared 360s
budget. So this Dag can take 10 slots while ~75 other Dags, plus its own child
`run` tasks, compete for what's left. Nothing deadlocks (the children are
quick), but it makes the batch slower and more likely to time out as the suite
grows.
The trigger tasks' own events aren't checked here, so `deferrable=True` (as
`example_openlineage_trigger_dag_deferrable` uses) looks like a safe way to
free the slots while waiting. Optional, but cheap.
##########
airflow-e2e-tests/docker/openlineage.yml:
##########
@@ -35,10 +35,86 @@ x-ol-common-env: &ol-common-env
# fully-qualified class path; the OpenLineage client imports it directly.
AIRFLOW__OPENLINEAGE__TRANSPORT: '{"type":
"system.openlineage.transport.variable.VariableTransport"}'
AIRFLOW__OPENLINEAGE__DISABLED: 'false'
- # Deployment-wide policy required by the two conf-driven policy DAGs. Each
rule is scoped by
- # dag_id, so both rules coexist and do not affect any other DAG.
- AIRFLOW__OPENLINEAGE__EMISSION_POLICY: '[{"scope": {"dag_id":
"openlineage_policy_conf_locked_dag"}, "locked": true, "controls":
{"include_source_code": false}}, {"scope": {"dag_id":
"openlineage_policy_conf_source_code_dag"}, "controls": {"include_source_code":
false}}]'
+ # Deployment-wide policy required by the conf-driven policy DAGs. Rules are
scoped by dag_id or by
+ # names unique to a DAG (hook/operator classes, dataset patterns), so they
coexist and do not affect any other DAG.
+ AIRFLOW__OPENLINEAGE__EMISSION_POLICY: >-
+ [
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_dag"}, "locked":
true, "controls": {"include_source_code": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_source_code_dag"},
"controls": {"include_source_code": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_extract_off"}, "controls": {"extract_operator_metadata": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_hook_off"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_full_info"}, "controls": {"include_full_task_info": true}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_emit_off"}, "controls": {"emit": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_task_events_off"}, "controls": {"emit_task_events": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_task_events_beat_emit"}, "controls": {"emit": false, "emit_task_events":
true}},
+ {"scope": {"operator": ".*\\.PolicyConfControlsOperator"}, "match_mode":
"regex", "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_operator_task"}, "controls": {"hook_lineage": true}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_extract_overridden"}, "controls": {"extract_operator_metadata": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_hook_overridden"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_emit_overridden"}, "controls": {"emit": false}},
+ {"scope": {"operator": ".*\\.PolicyConfEmitOffOperator"}, "match_mode":
"regex", "controls": {"emit": false}},
+ {"scope": {"operator":
"airflow.providers.standard.operators.smooth.SmoothOperator"}, "controls":
{"emit": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"grp.t_grouped"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_mapped"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_contr.*", "task_id":
"t_regex_.*"}, "match_mode": "regex", "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_conflict"}, "controls": {"hook_lineage": true}},
+ {"scope": {"dag_id": "openlineage_policy_conf_controls_dag", "task_id":
"t_conflict"}, "controls": {"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag",
"task_id": "t_locked_emit"}, "locked": true, "controls": {"emit": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag",
"task_id": "t_locked_task_events"}, "locked": true, "controls":
{"emit_task_events": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag",
"task_id": "t_locked_extract"}, "locked": true, "controls":
{"extract_operator_metadata": false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag",
"task_id": "t_locked_hook"}, "locked": true, "controls": {"hook_lineage":
false}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag",
"task_id": "t_locked_full"}, "locked": true, "controls":
{"include_full_task_info": true}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag"},
"locked": true, "controls": {"include_source_code": false, "exclude_datasets":
[".*/dag_locked_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_conf_locked_controls_dag",
"task_id": "t_floor"}, "controls": {"include_source_code": true}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_off_ok_child__notrigger"}, "controls":
{"emit_dag_events": false}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_off_fail_child__notrigger"}, "controls":
{"emit_dag_events": false}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_on_emit_off_ok_child__notrigger"}, "controls":
{"emit": false, "emit_dag_events": true}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_on_emit_off_fail_child__notrigger"}, "controls":
{"emit": false, "emit_dag_events": true}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_emit_off_child__notrigger"}, "controls":
{"emit": false}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_task_events_on_emit_off_child__notrigger"},
"controls": {"emit": false, "emit_task_events": true}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_task_events_off_child__notrigger"}, "controls":
{"emit_task_events": false}},
+ {"scope": {"dag_id":
"openlineage_policy_dag_events_locked_off_child__notrigger"}, "locked": true,
"controls": {"emit_dag_events": false}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag"},
"controls": {"exclude_datasets": [".*/conf_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag",
"hook": ".*\\.AuthHook"}, "match_mode": "regex", "controls":
{"exclude_datasets": [".*/conf_hook_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag",
"task_id": "t_locked"}, "locked": true, "controls": {"exclude_datasets":
[".*/locked_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag",
"task_id": "t_locked", "hook": ".*\\.AuthHook"}, "match_mode": "regex",
"controls": {"exclude_datasets": [".*/specific_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag",
"task_id": "t_locked_hook", "hook": ".*\\.LockedHook"}, "match_mode": "regex",
"locked": true, "controls": {"exclude_datasets":
[".*/locked_hook_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag",
"hook": ".*\\.AuthDroppedHook"}, "match_mode": "regex", "controls":
{"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_authoring_dag",
"task_id": "t_locked_dropped", "hook": ".*\\.AuthLockedDroppedHook"},
"match_mode": "regex", "locked": true, "controls": {"hook_lineage": false}},
+ {"scope": {}, "controls": {"exclude_datasets":
[".*/exclude_datasets_global_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag"},
"controls": {"exclude_datasets": [".*/dag_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"hook": ".*\\.ConfHook"}, "match_mode": "regex", "controls":
{"exclude_datasets": [".*/dag_hook_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"task_id": "t_task"}, "controls": {"exclude_datasets":
[".*/task_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"task_id": "t_task_hook"}, "controls": {"exclude_datasets":
[".*/task_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"task_id": "t_task_hook", "hook": ".*\\.ConfHook"}, "match_mode": "regex",
"controls": {"exclude_datasets": [".*/task_hook_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"task_id": "t_task_clear"}, "controls": {"exclude_datasets": []}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"hook": ".*\\.DroppedHook"}, "match_mode": "regex", "controls":
{"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"hook": ".*\\.KeptHook"}, "match_mode": "regex", "controls": {"hook_lineage":
true}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"task_id": "t_task_dropped", "hook": ".*\\.TaskDroppedHook"}, "match_mode":
"regex", "controls": {"hook_lineage": false}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsConfOperator"}, "match_mode":
"regex", "controls": {"exclude_datasets": [".*/operator_excluded_.*"]}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsConfOperator", "hook":
".*\\.ConfHook"}, "match_mode": "regex", "controls": {"exclude_datasets":
[".*/operator_hook_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_conf_dag",
"hook": "airflow.providers.standard.hooks.filesystem.FSHook"}, "controls":
{"exclude_datasets": [".*/fs_hook_excluded_.*"]}},
+ {"scope": {"hook": ".*\\.ExcludeDatasetsGlobalHook"}, "match_mode":
"regex", "controls": {"exclude_datasets": [".*/hook_excluded_.*"]}},
+ {"scope": {"hook": ".*\\.ExcludeDatasetsGlobalOtherHook"}, "match_mode":
"regex", "controls": {"exclude_datasets": [".*/other_hook_excluded_.*"]}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsGlobalOperator"},
"match_mode": "regex", "controls": {"exclude_datasets":
[".*/operator_excluded_.*"]}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsGlobalOperator", "hook":
".*\\.ExcludeDatasetsGlobalHook"}, "match_mode": "regex", "controls":
{"exclude_datasets": [".*/operator_hook_excluded_.*"]}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsGlobalClearOperator"},
"match_mode": "regex", "controls": {"exclude_datasets": []}},
+ {"scope": {"hook": ".*\\.ExcludeDatasetsGlobalDropHook"}, "match_mode":
"regex", "controls": {"hook_lineage": false}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsGlobalOperator", "hook":
".*\\.ExcludeDatasetsGlobalOpDropHook"}, "match_mode": "regex", "controls":
{"hook_lineage": false}},
+ {"scope": {"operator": ".*\\.ExcludeDatasetsGlobalLockedOperator"},
"match_mode": "regex", "locked": true, "controls": {"exclude_datasets":
[".*/operator_locked_.*"]}},
+ {"scope": {"hook": ".*\\.ExcludeDatasetsGlobalLockedHook"}, "match_mode":
"regex", "locked": true, "controls": {"exclude_datasets":
[".*/hook_locked_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_sources_dag"},
"controls": {"exclude_datasets": [".*/dag_excluded_.*", ".*sql_excluded.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_sources_dag",
"hook": ".*\\.SqlRuledHook"}, "match_mode": "regex", "controls":
{"exclude_datasets": [".*sql_hook_excluded.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_sources_dag",
"hook": ".*\\.SqlDroppedHook"}, "match_mode": "regex", "controls":
{"hook_lineage": false}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_sources_dag",
"hook": ".*\\.SourceHook"}, "match_mode": "regex", "controls":
{"exclude_datasets": [".*/hook_scoped_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_versioned_dag"},
"controls": {"exclude_datasets": [".*/dag_excluded_.*"]}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_versioned_dag",
"task_id": "t_skipped_emit_off"}, "controls": {"emit": false}},
+ {"scope": {"dag_id": "openlineage_policy_exclude_datasets_versioned_dag",
"task_id": "t_skipped_full_info"}, "controls": {"include_full_task_info": true}}
+ ]
AIRFLOW__OPENLINEAGE__EXECUTION_TIMEOUT: '60'
+ # openlineage_policy_exclude_datasets_versioned_dag checks that exclusion
runs after this cap; no other DAG
Review Comment:
The comment says the lower cap is safe because "no other DAG reports more
than a few assets per hook". But `max_assets_per_collector` limits the task's
whole hook-lineage collector: all hooks together, with inputs and outputs
counted separately (`task-sdk/src/airflow/sdk/lineage.py`,
`MAX_COLLECTED_ASSETS`). A task with many hooks that each report a few assets
could still hit 67 and lose lineage silently. Suggested wording:
```yaml
# ... no other Dag registers more than a few hook lineage inputs or outputs
per task, so 67
# (instead of the default 100) does not affect them.
```
--
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]
