kaxil commented on code in PR #71793:
URL: https://github.com/apache/airflow/pull/71793#discussion_r4468320217
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -7563,3 +7564,113 @@ def
test_stats_tags_with_standalone_and_key_value_tags(create_runtime_ti):
"task_id": "t",
"run_type": "manual",
}
+
+
+def _make_startup_details(make_ti_context, tags) -> tuple[StartupDetails,
BaseOperator]:
+ """Build a StartupDetails/task pair for a Dag carrying the given tags."""
+ with DAG("logging_tagged_dag", tags=tags):
+ task = BaseOperator(task_id="t")
+
+ ti = TaskInstance(
+ id=uuid7(),
+ task_id="t",
+ dag_id="logging_tagged_dag",
+ run_id="c",
+ try_number=1,
+ dag_version_id=uuid7(),
+ queue="default",
+ )
+ what = StartupDetails(
+ ti=ti,
+ dag_rel_path="",
+ bundle_info=FAKE_BUNDLE,
+ ti_context=make_ti_context(),
+ start_date=timezone.utcnow(),
+ sentry_integration="",
+ )
+ return what, task
+
+
+def _startup_with_dag_tags(
+ mocked_parse, mock_supervisor_comms, make_ti_context, tags
+) -> tuple[dict, TaskInstance]:
+ """
+ Run startup() for a Dag carrying the given tags.
+
+ Returns the bound structlog contextvars alongside the TaskInstance used,
so tests can assert
+ against ground-truth identifier values. Contextvars are cleared before
returning so a leftover
+ binding can't leak into unrelated tests that run afterwards.
+ """
+ what, task = _make_startup_details(make_ti_context, tags)
+ mock_supervisor_comms._get_response.return_value = what
+ mocked_parse(what, "logging_tagged_dag", task)
+
+ clear_contextvars()
Review Comment:
The autouse `_clear_structlog_contextvars` fixture at the top of this module
already clears contextvars after every test, so the `clear_contextvars()` calls
here and in the ordering test (and the new import) can go. The ordering test
could also reuse this helper instead of repeating its setup.
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -1298,6 +1298,13 @@ def startup(msg: StartupDetails) ->
tuple[RuntimeTaskInstance, Context, Logger]:
with _airflow_parsing_context_manager(dag_id=msg.ti.dag_id,
task_id=msg.ti.task_id):
ti = parse(msg, log)
+
+ if conf.getboolean("logging", "dag_tags_in_logs", fallback=False) and
ti.task.dag.tags:
+ # Nested under one key so free-form tags can't overwrite standard log
fields.
+ bind_contextvars(
Review Comment:
The UI's `tiContextFields` in `renderStructuredLog.tsx` hides the other
per-process fields bound just above, but not this one, so with the option on
every line in the log view gets `dag_tags={...}` appended. Should `dag_tags` go
in that list too?
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1008,6 +1008,17 @@ logging:
type: string
default: "filename,lineno"
example: ~
+ dag_tags_in_logs:
+ description: |
+ Set to ``True`` to include Dag tags in a ``dag_tags`` structured field
on task logs emitted
+ once the Dag has been parsed (log lines emitted during parsing itself
cannot carry them).
+ Tags that contain a colon (e.g. ``env:prod``) are split into a
key/value pair. Plain tags
+ (e.g. ``production``) map to an empty value. For example, tags
``env:prod`` and ``production``
+ are logged as ``dag_tags={"env": "prod", "production": ""}``.
+ version_added: 3.4.0
Review Comment:
main is 3.5.0 now and 3.4.0 is at RC, so this should be `3.5.0`.
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -7563,3 +7564,113 @@ def
test_stats_tags_with_standalone_and_key_value_tags(create_runtime_ti):
"task_id": "t",
"run_type": "manual",
}
+
+
+def _make_startup_details(make_ti_context, tags) -> tuple[StartupDetails,
BaseOperator]:
+ """Build a StartupDetails/task pair for a Dag carrying the given tags."""
+ with DAG("logging_tagged_dag", tags=tags):
+ task = BaseOperator(task_id="t")
+
+ ti = TaskInstance(
+ id=uuid7(),
+ task_id="t",
+ dag_id="logging_tagged_dag",
+ run_id="c",
+ try_number=1,
+ dag_version_id=uuid7(),
+ queue="default",
+ )
+ what = StartupDetails(
+ ti=ti,
+ dag_rel_path="",
+ bundle_info=FAKE_BUNDLE,
+ ti_context=make_ti_context(),
+ start_date=timezone.utcnow(),
+ sentry_integration="",
+ )
+ return what, task
+
+
+def _startup_with_dag_tags(
+ mocked_parse, mock_supervisor_comms, make_ti_context, tags
+) -> tuple[dict, TaskInstance]:
+ """
+ Run startup() for a Dag carrying the given tags.
+
+ Returns the bound structlog contextvars alongside the TaskInstance used,
so tests can assert
+ against ground-truth identifier values. Contextvars are cleared before
returning so a leftover
+ binding can't leak into unrelated tests that run afterwards.
+ """
+ what, task = _make_startup_details(make_ti_context, tags)
+ mock_supervisor_comms._get_response.return_value = what
+ mocked_parse(what, "logging_tagged_dag", task)
+
+ clear_contextvars()
+ try:
+ startup(get_startup_details())
+ return get_contextvars(), what.ti
+ finally:
+ clear_contextvars()
+
+
+def test_dag_tags_in_logs_bound_before_dag_file_parsed_log(
+ mocked_parse, mock_supervisor_comms, make_ti_context
+):
+ """Dag tags must be bound before the 'Dag file parsed' debug log, so that
event can carry them too."""
+ what, task = _make_startup_details(make_ti_context, ["env:prod"])
+ mock_supervisor_comms._get_response.return_value = what
+ mocked_parse(what, "logging_tagged_dag", task)
+
+ snapshot = {}
+
+ def _record_context_at_parse_complete(event, *args, **kwargs):
+ if event == "Dag file parsed":
+ snapshot.update(get_contextvars())
+
+ clear_contextvars()
+ try:
+ with (
+ conf_vars({("logging", "dag_tags_in_logs"): "True"}),
+ patch(
+ "airflow.sdk.execution_time.task_runner.log.debug",
+ side_effect=_record_context_at_parse_complete,
+ ),
+ ):
+ startup(get_startup_details())
+ finally:
+ clear_contextvars()
+
+ assert snapshot.get("dag_tags") == {"env": "prod"}
+
+
+def test_dag_tags_in_logs_disabled_by_default(mocked_parse,
mock_supervisor_comms, make_ti_context):
+ context, _ = _startup_with_dag_tags(
+ mocked_parse, mock_supervisor_comms, make_ti_context, ["env:prod",
"validation"]
+ )
+ assert "dag_tags" not in context
+
+
+@conf_vars({("logging", "dag_tags_in_logs"): "True"})
+def test_dag_tags_in_logs_without_dag_tags(mocked_parse,
mock_supervisor_comms, make_ti_context):
+ context, _ = _startup_with_dag_tags(mocked_parse, mock_supervisor_comms,
make_ti_context, [])
+ assert "dag_tags" not in context
+
+
+@conf_vars({("logging", "dag_tags_in_logs"): "True"})
+def test_dag_tags_in_logs_with_standalone_and_key_value_tags(
Review Comment:
These four could be one `@pytest.mark.parametrize` test over (enabled, tags,
expected). Worth adding a tag with more than one colon too, e.g.
`url:http://host:8080`; nothing here pins "split on the first colon", so
`rpartition` would still pass.
##########
airflow-core/newsfragments/71793.feature.rst:
##########
@@ -0,0 +1 @@
+Added ``[logging] dag_tags_in_logs`` to optionally include Dag tags in a
``dag_tags`` structured log field on task logs.
Review Comment:
We only take `significant` newsfragments now (see the newsfragments note in
`AGENTS.md`), and an opt-in option defaulting to `False` doesn't need one. Can
you drop this file?
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1008,6 +1008,17 @@ logging:
type: string
default: "filename,lineno"
example: ~
+ dag_tags_in_logs:
+ description: |
+ Set to ``True`` to include Dag tags in a ``dag_tags`` structured field
on task logs emitted
Review Comment:
This only binds in the task process. The supervisor writes into the same
task log without it: stdout/stderr via `forward_to_log`, and its own
`process_log` lines like the OOM/SIGKILL, SIGSEGV and "Process exited
abnormally" messages. Those explain why a task died, which is the failed-task
routing case in #76104. Can the description say it covers lines from the task
process only, or is the plan to get the tags to the supervisor too?
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -1298,6 +1298,13 @@ def startup(msg: StartupDetails) ->
tuple[RuntimeTaskInstance, Context, Logger]:
with _airflow_parsing_context_manager(dag_id=msg.ti.dag_id,
task_id=msg.ti.task_id):
ti = parse(msg, log)
+
+ if conf.getboolean("logging", "dag_tags_in_logs", fallback=False) and
ti.task.dag.tags:
+ # Nested under one key so free-form tags can't overwrite standard log
fields.
+ bind_contextvars(
+ dag_tags={key: value for key, _, value in (tag.partition(":") for
tag in ti.task.dag.tags)}
Review Comment:
`dag.tags` is a set, so when two tags share a key (`team:a` and `team:b`)
the value that wins here depends on the interpreter's hash seed. Tries of the
same task on different workers or pods can then carry different teams, and a
plain `owner` tag next to `owner:x` blanks the value to `""`. For log routing
that's the worst case, since nothing errors. Could this keep every value (a
sorted list per key, or the sorted raw tags), with a test for two tags sharing
a key?
--
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]
