From 057443365685678de3341c0883a81faacc3a417f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Tue, 4 Aug 2026 16:12:34 +0200 Subject: [PATCH] chore: demonstrate missing logs in kuttl test --- .../kuttl/smoke/85-task-logs.yaml.j2 | 23 +++ tests/templates/kuttl/smoke/task-logs.py | 195 ++++++++++++++++++ 2 files changed, 218 insertions(+) create mode 100644 tests/templates/kuttl/smoke/85-task-logs.yaml.j2 create mode 100644 tests/templates/kuttl/smoke/task-logs.py diff --git a/tests/templates/kuttl/smoke/85-task-logs.yaml.j2 b/tests/templates/kuttl/smoke/85-task-logs.yaml.j2 new file mode 100644 index 00000000..3afd8343 --- /dev/null +++ b/tests/templates/kuttl/smoke/85-task-logs.yaml.j2 @@ -0,0 +1,23 @@ +{% if test_scenario['values']['airflow'].find(",") > 0 %} +{% set airflow_version = test_scenario['values']['airflow'].split(',')[0] %} +{% else %} +{% set airflow_version = test_scenario['values']['airflow'] %} +{% endif %} +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +metadata: + name: task-logs +# The script waits for the task instance itself, so its verdict is final and must not be retried. +# Hence a TestStep (commands run once) rather than a TestAssert (commands are polled until the +# timeout, which would re-trigger a DAG run per attempt). +timeout: 480 +commands: +{% if test_scenario['values']['executor'] == 'celery' %} + # Extends the log-endpoint check above: the endpoint is not only reachable, it also serves the + # log of a task that ran. KubernetesExecutor task Pods are gone by the time the task finished, + # so their log can only be read back with remote logging (see the remote-logging test). + - script: kubectl cp -n $NAMESPACE task-logs.py test-airflow-python-0:/tmp + timeout: 240 + - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/task-logs.py --airflow-version "{{ airflow_version }}" +{% endif %} diff --git a/tests/templates/kuttl/smoke/task-logs.py b/tests/templates/kuttl/smoke/task-logs.py new file mode 100644 index 00000000..a5092688 --- /dev/null +++ b/tests/templates/kuttl/smoke/task-logs.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python +"""Assert that a task instance's log can be read back through the api-server. + +This guards against writer/reader disagreeing about where task logs live: from Airflow 3.1 on +the Task SDK writes them to `[logging] base_log_folder` (from `airflow.cfg`), while the +api-server and the worker's log server resolve them through +`LOGGING_CONFIG['handlers']['task']['base_log_folder']` of the custom logging config. If the two +point at different directories, every log page in the UI reports the log as missing even though +the task ran and wrote its log. + +The calling test step only runs this for the CeleryExecutor: KubernetesExecutor task Pods are +deleted once the task finishes, so their log server is gone and the api-server cannot read the log +back regardless of where it was written (that case needs remote logging, see the `remote-logging` +test). +""" + +import argparse +import sys +import time + +import requests + +DAG_ID = "example_bash_operator" +TASK_ID = "runme_0" + +REST_URL = "http://airflow-webserver:8080/api/v2" +TOKEN_URL = "http://airflow-webserver:8080/auth/token" + +# What the api-server returns instead of the log when it cannot find the file. The wording is +# checked as a substring because it is followed by the worker's host name. +LOG_NOT_FOUND = "Log file not found" + +# A successful `runme_0` run produces far more than this; the failure mode produces none at all. +MIN_LOG_LINES = 3 + + +def get_token() -> str: + response = requests.post( + TOKEN_URL, + headers={"Content-Type": "application/json"}, + json={"username": "airflow", "password": "airflow"}, + ) + response.raise_for_status() + return response.json()["access_token"] + + +def wait_for_dag(headers, timeout: int = 120) -> None: + """Wait until the DAG processor has registered the example DAGs. + + This is run once (see the calling TestStep), so it cannot rely on being retried. + """ + deadline = time.time() + timeout + while time.time() < deadline: + if ( + requests.get(f"{REST_URL}/dags/{DAG_ID}", headers=headers).status_code + == 200 + ): + return + time.sleep(5) + sys.exit(f"{DAG_ID} was not registered within {timeout}s") + + +def trigger_dag(headers) -> str: + requests.patch( + f"{REST_URL}/dags/{DAG_ID}", headers=headers, json={"is_paused": False} + ).raise_for_status() + + # An empty body is rejected with 422; `logical_date: null` triggers a run "now". + response = requests.post( + f"{REST_URL}/dags/{DAG_ID}/dagRuns", + headers=headers, + json={"logical_date": None}, + ) + response.raise_for_status() + return response.json()["dag_run_id"] + + +def wait_for_task_instance(headers, dag_run_id: str, timeout: int = 300) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + response = requests.get( + f"{REST_URL}/dags/{DAG_ID}/dagRuns/{dag_run_id}/taskInstances/{TASK_ID}", + headers=headers, + ) + if response.status_code == 200: + task_instance = response.json() + print( + f"{TASK_ID}: state={task_instance['state']} host={task_instance['hostname']}" + ) + if task_instance["state"] == "success": + return + if task_instance["state"] in ("failed", "upstream_failed", "skipped"): + sys.exit(f"{TASK_ID} ended up in state {task_instance['state']}") + time.sleep(5) + sys.exit(f"{TASK_ID} did not succeed within {timeout}s") + + +def log_lines(payload) -> tuple[list, str]: + """Return the structured log lines and the whole payload rendered as text. + + The shape of `content` changed over the 3.x line (plain text, list of strings, list of + structured messages), so both are derived defensively. Only actual log lines carry a + timestamp; the "Log message source details" group and error messages do not. + """ + content = payload["content"] + if isinstance(content, str): + return [], content + + lines = [] + text = [] + for entry in content: + if isinstance(entry, dict): + text.extend(str(value) for value in entry.values()) + if entry.get("timestamp"): + lines.append(entry) + else: + text.append(str(entry)) + return lines, "\n".join(text) + + +def sources(payload) -> list[str]: + """Return the log locations the api-server reports for this attempt (if any).""" + content = payload["content"] + if isinstance(content, str): + return [] + + result = [] + in_group = False + for entry in content: + event = entry.get("event", "") if isinstance(entry, dict) else str(entry) + if event == "::group::Log message source details": + in_group = True + elif event == "::endgroup::": + in_group = False + elif in_group: + result.append(event) + return result + + +def assert_log_is_readable(headers, dag_run_id: str) -> None: + response = requests.get( + f"{REST_URL}/dags/{DAG_ID}/dagRuns/{dag_run_id}/taskInstances/{TASK_ID}/logs/1", + headers=headers, + params={"full_content": "true"}, + ) + response.raise_for_status() + payload = response.json() + lines, text = log_lines(payload) + + # The api-server reports where it read the log from in a "Log message source details" group. + for source in sources(payload): + print(f"Log source: {source}") + + if LOG_NOT_FOUND in text: + print(f"Log response: {text}") + sys.exit( + f"The api-server cannot read back the log of {DAG_ID}.{TASK_ID}. The task ran and " + "wrote its log, so writer and reader disagree about the log directory: check that " + "the task handler's 'base_log_folder' in log_config.py matches " + "'[logging] base_log_folder' from airflow.cfg." + ) + if len(lines) < MIN_LOG_LINES: + sys.exit(f"Expected at least {MIN_LOG_LINES} log lines, got {len(lines)}") + + print(f"Read back {len(lines)} log lines for {DAG_ID}.{TASK_ID}") + + +def main(airflow_version: str) -> None: + if airflow_version.startswith("2."): + # Airflow 2 serves task logs through a different API; not covered here. + print(f"Skipping: not applicable to Airflow {airflow_version}") + return + + headers = { + "Authorization": f"Bearer {get_token()}", + "Content-Type": "application/json", + } + + wait_for_dag(headers) + + dag_run_id = trigger_dag(headers) + print(f"Triggered {DAG_ID}: {dag_run_id}") + + wait_for_task_instance(headers, dag_run_id) + assert_log_is_readable(headers, dag_run_id) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Airflow task log retrieval test") + parser.add_argument( + "--airflow-version", type=str, required=True, help="Airflow version" + ) + opts = parser.parse_args() + + main(opts.airflow_version)