kaxil commented on code in PR #74923:
URL: https://github.com/apache/airflow/pull/74923#discussion_r4304536093
##########
providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py:
##########
@@ -330,14 +354,14 @@ def _read_single_logs_page(self, log_filter: str,
page_token: str | None = None)
def _task_instance_to_labels(ti) -> dict[str, str]:
"""Convert a task instance to Cloud Logging labels."""
- labels = {
- LABEL_TASK_ID: ti.task_id,
- LABEL_DAG_ID: ti.dag_id,
- LABEL_LOGICAL_DATE: str(ti.logical_date.isoformat())
- if AIRFLOW_V_3_0_PLUS
- else str(ti.execution_date.isoformat()),
- LABEL_TRY_NUMBER: str(ti.try_number),
- }
+ labels = {LABEL_TASK_ID: ti.task_id, LABEL_DAG_ID: ti.dag_id}
+ if AIRFLOW_V_3_0_PLUS:
Review Comment:
I don't think this helps 3.2.2 on the default config. There `stackdriver://`
still sets up `StackdriverTaskHandler` and no `REMOTE_TASK_LOG` (that moved to
`StackdriverRemoteLogIO` in 3.3.0), and with no `set_context` caller `emit`
sends no task labels at all. So I'd scope the "3.2.2 to 3.3.x" line in the
description to 3.3.x, plus 3.2.x with a custom `REMOTE_TASK_LOG`.
##########
providers/google/tests/unit/google/cloud/log/test_cloud_logging_task_handler.py:
##########
@@ -328,6 +380,7 @@ def test_processors_fallback_to_event_labels(self,
mock_transport_prop):
"run_id": "test_run_id",
"try_number": "2",
"map_index": "-1",
+ "log_path": "some/path.py",
Review Comment:
`relative_path_from_logger` returns a `Path`, but the mock above returns a
`str`, so dropping the `str(log_path)` in the processor wouldn't fail this
test. Returning `Path("some/path.py")` from the mock would make this assertion
catch it.
##########
providers/google/src/airflow/providers/google/cloud/log/cloud_logging_task_handler.py:
##########
@@ -276,14 +281,33 @@ 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
+ if match_ti_id:
# Entries written before the ``ti_id`` label existed are matched
by the other labels.
legacy_conditions = [*label_conditions.values(), f"NOT
labels.{LABEL_TI_ID}:*"]
- log_filters.append(f"({ti_id_condition} OR ({' AND
'.join(legacy_conditions)}))")
+ task_filters = [f"({ti_id_condition} OR ({' AND
'.join(legacy_conditions)}))"]
else:
- log_filters.extend(label_conditions.values())
+ task_filters = list(label_conditions.values())
+ if log_path:
+ # Workers before 3.4 ship a task's stdout and the supervisor's own
lines without task context.
+ # The path is matched only for such lines, so other tries that
wrote to the same path stay out.
+ path_conditions = [
+ f"labels.{LABEL_LOG_PATH}={escape_label_value(log_path)}",
Review Comment:
Before 3.2.2 the triggerer writes a deferred task's lines to `<task
path>.trigger.<job id>.log` and binds only `trigger_id`, so those entries get
that trigger path as `log_path` and no task labels. This exact match never
finds them, while the S3 reader does since it lists by prefix. It only matters
on 3.0 to 3.2.1 with a custom `REMOTE_TASK_LOG`, so either a regex like
`labels.log_path=~"^<escaped path>(\.trigger\.[0-9]+\.log)?$"` or a line in the
known limitations would be fine.
--
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]
