humit0 commented on code in PR #54431:
URL: https://github.com/apache/airflow/pull/54431#discussion_r2281093038
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1597,12 +1597,7 @@ def process_log_messages_from_subprocess(
if ts := event.get("timestamp"):
# We use msgspec to decode the timestamp as it does it orders of
magnitude quicker than
# datetime.strptime cn
- #
- # We remove the timezone info here, as the json encoding has
`+00:00`, and since the log came
- # from a subprocess we know that the timezone of the log message
is the same, so having some
- # messages include tz (from subprocess) but others not (ones from
supervisor process) is
- # confusing.
- event["timestamp"] = msgspec.json.decode(f'"{ts}"',
type=datetime).replace(tzinfo=None)
+ event["timestamp"] = msgspec.json.decode(f'"{ts}"', type=datetime)
Review Comment:
@ashb
Sorry for late response. Here are the raw log file contents!
before
```jsonl
{"timestamp":"2025-08-13T01:38:27.459132","level":"info","event":"DAG
bundles loaded:
dags-folder","logger":"airflow.dag_processing.bundles.manager.DagBundlesManager"}
{"timestamp":"2025-08-13T01:38:27.459490","level":"info","event":"Filling up
the DagBag from /files/dags/dag.py","logger":"airflow.models.dagbag.DagBag"}
{"timestamp":"2025-08-13T01:38:27.910325","level":"info","event":"Tmp dir
root location:
/tmp","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T01:38:27.910602","level":"info","event":"Running
command: ['/usr/bin/bash', '-c', \"echo
'2022-12-31'\"]","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T01:38:27.913484","level":"info","event":"Output:","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T01:38:27.914388","level":"info","event":"2022-12-31","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T01:38:27.914490","level":"info","event":"Command
exited with return code
0","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T01:38:27.914823","level":"info","event":"Pushing
xcom","ti":"RuntimeTaskInstance(id=UUID('0198a114-1dbf-7bdb-a7d2-1bc32154fb72'),
task_id='test_task', dag_id='test_dag',
run_id='scheduled__2022-12-31T15:00:00+00:00', try_number=2,
dag_version_id=UUID('0198a10a-fdbd-775f-97de-a0b2c89a6eee'), map_index=-1,
hostname='0a3aaf1ba399', context_carrier=None, task=<Task(BashOperator):
test_task>, bundle_instance=LocalDagBundle(name=dags-folder), max_tries=2,
start_date=datetime.datetime(2025, 8, 13, 1, 38, 27, 81666,
tzinfo=datetime.timezone.utc), end_date=None, state=<TaskInstanceState.RUNNING:
'running'>, is_mapped=False, rendered_map_index=None,
log_url='http://localhost:8080/dags/test_dag/runs/scheduled__2022-12-31T15%3A00%3A00%2B00%3A00/tasks/test_task?try_number=2')","logger":"task"}
```
after
```jsonl
{"timestamp":"2025-08-13T02:05:21.148361Z","level":"info","event":"DAG
bundles loaded:
dags-folder","logger":"airflow.dag_processing.bundles.manager.DagBundlesManager"}
{"timestamp":"2025-08-13T02:05:21.148831Z","level":"info","event":"Filling
up the DagBag from /files/dags/dag.py","logger":"airflow.models.dagbag.DagBag"}
{"timestamp":"2025-08-13T02:05:21.687594Z","level":"info","event":"Tmp dir
root location:
/tmp","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T02:05:21.687884Z","level":"info","event":"Running
command: ['/usr/bin/bash', '-c', \"echo
'2022-12-31'\"]","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T02:05:21.690570Z","level":"info","event":"Output:","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T02:05:21.691518Z","level":"info","event":"2022-12-31","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T02:05:21.691581Z","level":"info","event":"Command
exited with return code
0","logger":"airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"}
{"timestamp":"2025-08-13T02:05:21.691911Z","level":"info","event":"Pushing
xcom","ti":"RuntimeTaskInstance(id=UUID('0198a12c-ba73-709e-8d48-14b87d48688d'),
task_id='test_task', dag_id='test_dag',
run_id='scheduled__2022-12-31T15:00:00+00:00', try_number=3,
dag_version_id=UUID('0198a10a-fdbd-775f-97de-a0b2c89a6eee'), map_index=-1,
hostname='d3d7af713f58', context_carrier=None, task=<Task(BashOperator):
test_task>, bundle_instance=LocalDagBundle(name=dags-folder), max_tries=3,
start_date=datetime.datetime(2025, 8, 13, 2, 5, 20, 744889,
tzinfo=datetime.timezone.utc), end_date=None, state=<TaskInstanceState.RUNNING:
'running'>, is_mapped=False, rendered_map_index=None,
log_url='http://localhost:8080/dags/test_dag/runs/scheduled__2022-12-31T15%3A00%3A00%2B00%3A00/tasks/test_task?try_number=3')","logger":"task"}
```
--
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]
