shahar1 commented on code in PR #74923:
URL: https://github.com/apache/airflow/pull/74923#discussion_r4344654924
##########
providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py:
##########
@@ -254,11 +259,12 @@ def read(self, relative_path: str, ti: RuntimeTI | None)
-> LogResponse:
# Entries are found by task instance labels, so logs without a
task instance can't be read.
return [], None
ti_labels = _task_instance_to_labels(ti)
- log_filter = self.prepare_log_filter(ti_labels)
+ # The writer's path went through ``pathlib``, which drops empty and
``.`` segments.
+ log_filter = self.prepare_log_filter(ti_labels,
log_path=PurePosixPath(relative_path).as_posix())
messages, end_of_log, _ = self.read_logs(log_filter,
next_page_token=None, all_pages=True)
return [f"Reading remote log from Cloud Logging for {relative_path}"],
[messages] if messages else []
- def prepare_log_filter(self, ti_labels: dict[str, str]) -> str:
+ def prepare_log_filter(self, ti_labels: dict[str, str], log_path: str |
None = None) -> str:
Review Comment:
Added a warning just below the Changelog header in
`providers/google/docs/changelog.rst` (016e59b687): a subclass that overrides
`prepare_log_filter` must accept `log_path`, or every read fails with
`TypeError`.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py:
##########
@@ -254,11 +259,12 @@ def read(self, relative_path: str, ti: RuntimeTI | None)
-> LogResponse:
# Entries are found by task instance labels, so logs without a
task instance can't be read.
return [], None
ti_labels = _task_instance_to_labels(ti)
- log_filter = self.prepare_log_filter(ti_labels)
+ # The writer's path went through ``pathlib``, which drops empty and
``.`` segments.
+ log_filter = self.prepare_log_filter(ti_labels,
log_path=PurePosixPath(relative_path).as_posix())
Review Comment:
Kept it out of scope, and the description now says so under known
limitations: the `log_path` match covers `CloudLoggingRemoteLogIO` only, so
`StackdriverTaskHandler` as the task handler with a custom `REMOTE_TASK_LOG`
still reads, and links to, only the lines with task labels.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py:
##########
@@ -279,14 +285,36 @@ def escape_label_value(value: str) -> str:
key: f"labels.{escape_label_key(key)}={escape_label_value(value)}"
for key, value in ti_labels.items()
}
+ if ti_labels.get(LABEL_MAP_INDEX) == "0":
+ # Earlier provider releases wrote no ``map_index`` label for map
index 0.
+ label_conditions[LABEL_MAP_INDEX] = (
+ f"({label_conditions[LABEL_MAP_INDEX]} OR NOT
labels.{LABEL_MAP_INDEX}:*)"
+ )
ti_id_condition = label_conditions.pop(LABEL_TI_ID, None)
# Before 3.4 a cleared task instance gets a new id, which can differ
from the id its logs were written under.
- if ti_id_condition and AIRFLOW_V_3_4_PLUS:
+ match_ti_id = bool(ti_id_condition) and AIRFLOW_V_3_4_PLUS
Review Comment:
Confirmed on Cloud Logging: with entries labelled `run_id=Run_A` and
`run_id=run_a`, `labels.run_id="run_a"` returns both, while
`labels.run_id=~"^run_a$"` returns only `run_a`. In 016e59b687 `run_id` is
matched with an anchored regex, escaped the same way as the `log_path` regex.
`dag_id` and `task_id` still use `=`, as they did before this PR.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py:
##########
@@ -176,11 +226,17 @@ def test_read_logs(self, mock_client,
mock_get_creds_and_project_id):
else:
ti.execution_date = timezone.datetime(2016, 1, 1)
- messages, logs =
self.io.read("dag_id=test_dag/run_id=run1/task_id=test_task/attempt=1.log", ti)
+ messages, logs =
self.io.read("dag_id=test_dag/run_id=run1//task_id=test_task/./attempt=1.log",
ti)
assert len(messages) == 1
assert "Cloud Logging" in messages[0]
assert logs == ["MSG1\nMSG2"]
+ # The reader normalises the path the way the writer's pathlib path
does.
+ log_filter =
mock_client.return_value.list_log_entries.call_args.kwargs["request"].filter
+ assert (
Review Comment:
The TI in `test_read_logs` is now specced with `run_id` and `map_index` set,
and the test asserts the whole filter (016e59b687).
---
Drafted-by: Claude Code (Opus 5.5) (no human review 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]
