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 1/4] 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) From 78ab5cded1b5abd1e65e924a9492f69615aa0ce7 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 6 Aug 2026 16:41:03 +0200 Subject: [PATCH 2/4] serve logs from base log folder for the UI --- CHANGELOG.md | 5 ++ .../build/properties/product_logging/mod.rs | 76 ++++++++++++++++++- 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fd53da5d..eeb9fdbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,11 +20,16 @@ that direction the UI and the other destinations open up together and cannot be set apart ([#829]). +### Fixed + +- Task logs are served from `BASE_LOG_FOLDER` instead of the `task` handler's `base_log_folder` at the Vector agent log directory ([#XXX]). + [#814]: https://github.com/stackabletech/airflow-operator/pull/814 [#821]: https://github.com/stackabletech/airflow-operator/pull/821 [#827]: https://github.com/stackabletech/airflow-operator/pull/827 [#828]: https://github.com/stackabletech/airflow-operator/pull/828 [#829]: https://github.com/stackabletech/airflow-operator/pull/829 +[#XXX]: https://github.com/stackabletech/airflow-operator/pull/XXX ## [26.7.0] - 2026-07-21 diff --git a/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs b/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs index bde3d165..3768c4ef 100644 --- a/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs @@ -212,6 +212,7 @@ LOGGING_CONFIG['loggers']['{name}']['level'] = {level} import logging import os from airflow.config_templates import airflow_local_settings +from airflow.configuration import conf os.makedirs('{log_dir}', exist_ok=True) @@ -247,7 +248,9 @@ LOGGING_CONFIG = {{ 'class': 'airflow.utils.log.file_task_handler.FileTaskHandler', 'level': {task_log_level}, 'formatter': 'airflow', - 'base_log_folder': '{log_dir}', + # `serve_logs` on the workers serves task logs from this directory, so it must be + # the folder the Task SDK writes task logs to, not the Vector agent log directory. + 'base_log_folder': os.path.expanduser(conf.get('logging', 'BASE_LOG_FOLDER')), 'filters': ['mask_secrets_core'] }} }}, @@ -294,6 +297,77 @@ mod tests { use super::*; + fn resolved_image(product_version: &str) -> ResolvedProductImage { + ResolvedProductImage { + product_version: product_version.to_string(), + app_version_label_value: product_version.parse().expect("valid label value"), + image: format!("oci.example.org/sdp/airflow:{product_version}-stackable0.0.0-dev"), + image_pull_policy: "IfNotPresent".to_string(), + pull_secrets: None, + } + } + + /// The Vector agent tails `{log_dir}/airflow.py.json` (see the `files_py` source in + /// `vector.yaml`), so every generated log config must create the log directory and write + /// the rotating JSON log file there. + #[test] + fn test_vector_log_file() { + let log_config = AutomaticContainerLogConfig::default(); + + for content in [ + create_airflow_stdlib_config( + &log_config, + "/stackable/log/airflow", + &resolved_image("3.0.6"), + ), + create_airflow_structlog_config(&log_config, "/stackable/log/airflow"), + ] { + assert!(content.contains("os.makedirs('/stackable/log/airflow', exist_ok=True)")); + assert!(content.contains("'filename': '/stackable/log/airflow/airflow.py.json'")); + } + } + + /// Only the last version line before the stdlib/structlog switch gets the stdlib config; + /// all later (including future) versions must get the structlog one. + #[test] + fn test_logging_variant_selection() { + // The stdlib config copies Airflow's default logging config, the structlog one + // defines its own `mask_secrets_core` filter. + let log_config = + ValidatedContainerLogConfigChoice::Automatic(AutomaticContainerLogConfig::default()); + let stdlib_content = create_airflow_config( + &log_config, + "/stackable/log/airflow", + &resolved_image("3.0.6"), + ) + .expect("automatic log config produces content"); + let structlog_content = create_airflow_config( + &log_config, + "/stackable/log/airflow", + &resolved_image("3.1.6"), + ) + .expect("automatic log config produces content"); + assert!(stdlib_content.contains("deepcopy(airflow_local_settings.DEFAULT_LOGGING_CONFIG)")); + assert!(structlog_content.contains("mask_secrets_core")); + } + + #[test] + fn test_structlog_task_log_folder() { + let log_config = AutomaticContainerLogConfig::default(); + + let content = create_airflow_structlog_config(&log_config, "/stackable/log/airflow"); + + // `serve_logs` on the workers serves task logs from the `task` handler's + // `base_log_folder`, so it must point to the folder the Task SDK writes task logs to + // (`[logging] base_log_folder`), not to the Vector agent log directory. + assert!(content.contains( + "'base_log_folder': os.path.expanduser(conf.get('logging', 'BASE_LOG_FOLDER'))" + )); + assert!(!content.contains("'base_log_folder': '/stackable/log/airflow'")); + // The generated config must import `conf` itself. + assert!(content.contains("from airflow.configuration import conf")); + } + #[test] fn test_vector_config_file_content() { let content = vector_config_file_content(); From c976f6154b36a0ba56fe672edf0df1a834cdc9bc Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 7 Aug 2026 12:36:18 +0200 Subject: [PATCH 3/4] changelog --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eeb9fdbb..370db136 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,14 +22,14 @@ ### Fixed -- Task logs are served from `BASE_LOG_FOLDER` instead of the `task` handler's `base_log_folder` at the Vector agent log directory ([#XXX]). +- Task logs are served from `BASE_LOG_FOLDER` instead of the `task` handler's `base_log_folder` at the Vector agent log directory ([#834]). [#814]: https://github.com/stackabletech/airflow-operator/pull/814 [#821]: https://github.com/stackabletech/airflow-operator/pull/821 [#827]: https://github.com/stackabletech/airflow-operator/pull/827 [#828]: https://github.com/stackabletech/airflow-operator/pull/828 [#829]: https://github.com/stackabletech/airflow-operator/pull/829 -[#XXX]: https://github.com/stackabletech/airflow-operator/pull/XXX +[#834]: https://github.com/stackabletech/airflow-operator/pull/834 ## [26.7.0] - 2026-07-21 From 12d7154e2d58138fb05efc869d7804d5f8a5ff96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20M=C3=BCller?= Date: Fri, 7 Aug 2026 13:38:56 +0200 Subject: [PATCH 4/4] chore: cleanups --- .gitignore | 1 + .../build/properties/product_logging/mod.rs | 17 +------- tests/templates/kuttl/smoke/task-logs.py | 39 +++++++++++++++---- 3 files changed, 33 insertions(+), 24 deletions(-) diff --git a/.gitignore b/.gitignore index 696bc411..a2b5030c 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ tests/_work/ debug/ target/ **/*.rs.bk +__pycache__/ .idea/ *.iws diff --git a/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs b/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs index 3768c4ef..c99247d5 100644 --- a/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs @@ -396,12 +396,7 @@ mod tests { } fn stdlib_config(log_config: &AutomaticContainerLogConfig) -> String { - let resolved_product_image = ResolvedProductImage { - product_version: "2.10.0".to_string(), - ..resolved_product_image_stub() - }; - - create_airflow_stdlib_config(log_config, "/stackable/log", &resolved_product_image) + create_airflow_stdlib_config(log_config, "/stackable/log", &resolved_image("2.10.0")) } /// The requested level paired with the `task` handler level and the `airflow.task` logger @@ -564,14 +559,4 @@ mod tests { "logging.INFO" ); } - - fn resolved_product_image_stub() -> ResolvedProductImage { - ResolvedProductImage { - product_version: "0.0.0".to_string(), - app_version_label_value: "0.0.0".parse().unwrap(), - image: "oci.example.org/product:0.0.0".to_string(), - image_pull_policy: "Always".to_string(), - pull_secrets: None, - } - } } diff --git a/tests/templates/kuttl/smoke/task-logs.py b/tests/templates/kuttl/smoke/task-logs.py index a5092688..24b822a2 100644 --- a/tests/templates/kuttl/smoke/task-logs.py +++ b/tests/templates/kuttl/smoke/task-logs.py @@ -33,6 +33,11 @@ # A successful `runme_0` run produces far more than this; the failure mode produces none at all. MIN_LOG_LINES = 3 +# The log is fetched right after the task instance reports success, so the worker's log server +# may not have flushed the file yet. +LOG_FETCH_ATTEMPTS = 3 +LOG_FETCH_INTERVAL = 5 + def get_token() -> str: response = requests.post( @@ -137,20 +142,41 @@ def sources(payload) -> list[str]: return result -def assert_log_is_readable(headers, dag_run_id: str) -> None: +def fetch_log(headers, dag_run_id: str): 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) + return response.json() - # The api-server reports where it read the log from in a "Log message source details" group. + +def print_sources(payload) -> None: + """Print where the api-server read the log from, as far as it reports it.""" for source in sources(payload): print(f"Log source: {source}") + +def assert_log_is_readable(headers, dag_run_id: str) -> None: + for attempt in range(1, LOG_FETCH_ATTEMPTS + 1): + payload = fetch_log(headers, dag_run_id) + lines, text = log_lines(payload) + + if LOG_NOT_FOUND not in text and len(lines) >= MIN_LOG_LINES: + print_sources(payload) + print(f"Read back {len(lines)} log lines for {DAG_ID}.{TASK_ID}") + return + + if attempt < LOG_FETCH_ATTEMPTS: + print( + f"Log not readable yet (attempt {attempt}/{LOG_FETCH_ATTEMPTS}), retrying in " + f"{LOG_FETCH_INTERVAL}s" + ) + time.sleep(LOG_FETCH_INTERVAL) + + print_sources(payload) + if LOG_NOT_FOUND in text: print(f"Log response: {text}") sys.exit( @@ -159,10 +185,7 @@ def assert_log_is_readable(headers, dag_run_id: str) -> None: "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}") + sys.exit(f"Expected at least {MIN_LOG_LINES} log lines, got {len(lines)}") def main(airflow_version: str) -> None: