fix: set base_log_folder to display task logs in the UI

.gitignore

@@ -2,6 +2,7 @@ tests/_work/
 debug/
 target/
 **/*.rs.bk
+__pycache__/
 
 .idea/
 *.iws

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 ([#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
+[#834]: https://github.com/stackabletech/airflow-operator/pull/834
 
 ## [26.7.0] - 2026-07-21
 

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();
@@ -322,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
@@ -490,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,
-        }
-    }
 }

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 %}

tests/templates/kuttl/smoke/task-logs.py

@@ -0,0 +1,218 @@
+#!/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
+
+# 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(
+        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 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()
+    return response.json()
+
+
+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(
+            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."
+        )
+    sys.exit(f"Expected at least {MIN_LOG_LINES} log lines, got {len(lines)}")
+
+
+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)