hanxdatadog commented on code in PR #71793:
URL: https://github.com/apache/airflow/pull/71793#discussion_r4484678121
##########
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:
Fixed: `dag_tags` is now the sorted raw tag list (`["owner", "owner:x",
"team:a", "team:b"]`), so nothing is dropped and the order doesn't depend on
the hash seed. Added a test with tags sharing a key.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @hanxdatadog before posting
##########
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:
Fixed: added `dag_tags` to `tiContextFields`, so it shows once in the
preamble.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @hanxdatadog before posting
##########
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:
Fixed: the description now says only task-process lines carry the field, and
names the supervisor's forwarded stdout/stderr and kill/abnormal-exit messages
as not covered. Getting the tags to the supervisor would need them sent over
comms, so I'd keep that out of this PR.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @hanxdatadog before posting
##########
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:
Fixed: `version_added: 3.5.0`.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @hanxdatadog before posting
##########
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:
Fixed: dropped the newsfragment.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @hanxdatadog before posting
##########
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:
Fixed: removed the `clear_contextvars()` calls and import, and the ordering
test reuses the helper.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @hanxdatadog before posting
--
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]
