kaxil commented on code in PR #74923:
URL: https://github.com/apache/airflow/pull/74923#discussion_r4341794718
##########
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:
Small one: `read()` now passes `log_path=` here, so a subclass that
overrides `prepare_log_filter` with the old `(self, ti_labels)` signature gets
a `TypeError` on every log read. Probably rare, but a line in the provider
changelog would warn anyone who does.
##########
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:
Not from this PR, just worth knowing: the query language docs say string
comparisons with `=` ignore letter case, so when `match_ti_id` is false, two
runs of a Dag whose `run_id`s differ only in case would read each other's
lines. `dag_id` and `task_id` already work that way, so nothing to change here.
##########
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:
Only this reader passes `log_path`. `CloudLoggingTaskHandler.read` and
`get_external_log_url` still call `prepare_log_filter(ti_labels)`, so on
Airflow 3 a setup that keeps `StackdriverTaskHandler` as the task handler with
a custom `REMOTE_TASK_LOG` won't pick up the unlabelled lines, and the Cloud
Logging link won't either. Fine to keep out of scope, but the description could
say the fix covers the remote log reader.
##########
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:
Nit: `ti` above is a bare `MagicMock`, so `run_id` and `map_index` go into
the filter as `<MagicMock ...>` strings and only the `log_path` part is
asserted. A specced TI with those set, plus an assert on the label part, would
cover the whole filter.
--
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]
