From 75e32145cafb6da765e1541aa42ca74723349655 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Wed, 5 Aug 2026 17:27:04 +0200 Subject: [PATCH] chore: demonstrate that task logs are not aggregated in kuttl test --- .../logging/55-task-log-aggregation.yaml.j2 | 50 +++++++++++ .../airflow-vector-aggregator-values.yaml.j2 | 12 +++ .../kuttl/logging/task-log-aggregation.py | 85 +++++++++++++++++++ 3 files changed, 147 insertions(+) create mode 100644 tests/templates/kuttl/logging/55-task-log-aggregation.yaml.j2 create mode 100644 tests/templates/kuttl/logging/task-log-aggregation.py diff --git a/tests/templates/kuttl/logging/55-task-log-aggregation.yaml.j2 b/tests/templates/kuttl/logging/55-task-log-aggregation.yaml.j2 new file mode 100644 index 00000000..d2b44a11 --- /dev/null +++ b/tests/templates/kuttl/logging/55-task-log-aggregation.yaml.j2 @@ -0,0 +1,50 @@ +{% if test_scenario['values']['executor'] == 'celery' %} +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +metadata: + name: task-log-aggregation +# The script polls the aggregator itself, so it should not be retried, thus TestStep. +timeout: 600 +commands: + # Show that the task logs are on disk, and which files the Vector agent looks at. Step 52 + # results in a succeeded task, thus logs have been written. + - script: | + set -eu + + FOUND="" + + for POD in airflow-worker-automatic-log-config-0 airflow-worker-custom-log-config-0; do + # Airflow >= 3.1 writes attempt logs below "[logging] base_log_folder" from airflow.cfg + # ($AIRFLOW_HOME/logs), earlier versions below the "task" handler's base_log_folder from + # the operator-generated log_config.py (/stackable/log/airflow). Both are searched so that + # this works for every supported Airflow version. + # Issues around base_log_path are documented in + # https://github.com/stackabletech/airflow-operator/pull/834. + TASK_LOGS=$( + kubectl exec -n "$NAMESPACE" "$POD" -c airflow -- sh -c \ + 'find "$AIRFLOW_HOME/logs" /stackable/log/airflow \ + \( -name "attempt=*.log" -o -name "[0-9]*.log" \) 2>/dev/null' || true + ) + if [ -n "$TASK_LOGS" ]; then + echo "Task-attempt logs written by $POD:" + echo "$TASK_LOGS" + FOUND=yes + fi + done + + if [ -z "$FOUND" ]; then + echo "No worker wrote a task-attempt log, so nothing can be said about their aggregation." + exit 1 + fi + + echo + echo "File sources of the Vector agent config generated for the worker:" + kubectl get configmap -n "$NAMESPACE" airflow-worker-automatic-log-config \ + -o jsonpath='{.data.vector\.yaml}' | grep -E '^ +- \$\{LOG_DIR\}/' + - script: kubectl cp -n $NAMESPACE ./task-log-aggregation.py test-airflow-python-0:/tmp + timeout: 240 + - script: >- + kubectl exec -n $NAMESPACE test-airflow-python-0 -- + python /tmp/task-log-aggregation.py +{% endif %} diff --git a/tests/templates/kuttl/logging/airflow-vector-aggregator-values.yaml.j2 b/tests/templates/kuttl/logging/airflow-vector-aggregator-values.yaml.j2 index 714e3476..b46de584 100644 --- a/tests/templates/kuttl/logging/airflow-vector-aggregator-values.yaml.j2 +++ b/tests/templates/kuttl/logging/airflow-vector-aggregator-values.yaml.j2 @@ -121,6 +121,18 @@ customConfig: condition: >- .pod == "airflow-worker-custom-log-config-0" && .container == "vector" + # Task-attempt logs of the DAG run triggered in step 52. Celery hands each task to whichever + # worker picks it up first, so both worker role groups are accepted. + # `.file` is the path below the container's log directory, as set by the Vector agent's + # `extended_logs_files` transform. Airflow 3 names the attempt files `attempt=.log`, + # Airflow 2 `.log`. + filteredWorkerTaskLogs: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "airflow-worker-") && + .container == "airflow" && + match(string(.file) ?? "", r'(attempt=\d+|/\d+)\.log$') {% elif test_scenario['values']['executor'] == 'kubernetes' %} filteredExampleTriggerTargetDagBashTaskBase: type: filter diff --git a/tests/templates/kuttl/logging/task-log-aggregation.py b/tests/templates/kuttl/logging/task-log-aggregation.py new file mode 100644 index 00000000..ebc969d5 --- /dev/null +++ b/tests/templates/kuttl/logging/task-log-aggregation.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python +"""Assert that Airflow task logs reach the Vector aggregator. + +The Vector agent config generated by the operator defines exactly three file sources, each of them +one directory level deep with a fixed suffix: + + ${LOG_DIR}/*/*.stdout.log + ${LOG_DIR}/*/*.stderr.log + ${LOG_DIR}/*/*.py.json + +Airflow writes a task-attempt log to + + /dag_id=/run_id=/task_id=/attempt=.log + +which is four levels deep and, from Airflow 3.1 on, below `$AIRFLOW_HOME/logs` instead of +`${LOG_DIR}` (= /stackable/log). Neither the depth nor the suffix can be matched by any of the +three globs, so task logs are never forwarded to the aggregator - no matter which Airflow version +is used or whether Vector aggregation is enabled. + +The task logs checked here belong to the `example_trigger_target_dag` run that step 52 triggers. +The calling test step has already asserted that the worker wrote the attempt logs, so a zero event +count here means the files are there but no Vector source picks them up. +""" + +import json +import subprocess +import sys +import time + +# Filters events whose file name looks like a task attempt log, see the transform of that name in +# airflow-vector-aggregator-values.yaml.j2. +TRANSFORM = "filteredWorkerTaskLogs" + +# Vector rescans its source globs periodically (`glob_minimum_cooldown_ms`, 60s by default), so the +# attempt log being on disk does not mean it has been read yet. +TIMEOUT = 180 +INTERVAL = 10 + + +def sent_events(component_id: str) -> int | None: + """Return the number of events the given component sent, or None if it does not exist.""" + response = subprocess.run( + [ + "grpcurl", + "-plaintext", + "-d", + '{"limit": 100}', + "airflow-vector-aggregator:8686", + "vector.observability.v1.ObservabilityService/GetComponents", + ], + capture_output=True, + text=True, + check=True, # Raise a CalledProcessError if non-zero return + timeout=20, # seconds + ) + for component in json.loads(response.stdout).get("components", []): + if component.get("componentId") == component_id: + return int(component["metrics"]["sentEventsTotal"] or 0) + return None + + +def check_task_logs_are_aggregated() -> None: + deadline = time.time() + TIMEOUT + while True: + events = sent_events(TRANSFORM) + if events is None: + sys.exit(f'The aggregator has no transform "{TRANSFORM}".') + if events > 0: + print(f"{events} task log events reached the aggregator.") + return + if time.time() >= deadline: + break + time.sleep(INTERVAL) + + sys.exit( + f"No task log event reached the aggregator within {TIMEOUT}s although the worker wrote " + "the attempt log (see the output of the previous command). The Vector agent config " + "generated by the operator has no file source matching the task log directory tree, so " + "task logs are silently left out of log aggregation." + ) + + +if __name__ == "__main__": + check_task_logs_are_aggregated() + print("Test successful!")