chore: demonstrate that task logs are not aggregated in kuttl test

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

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=<n>.log`,
+    # Airflow 2 `<try_number>.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

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
+
+    <base_log_folder>/dag_id=<dag>/run_id=<run>/task_id=<task>/attempt=<n>.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!")