fix: Remote logging in Airflow 3.x

rust/operator-binary/src/airflow_controller.rs

@@ -806,6 +806,7 @@ fn build_rolegroup_config_map(
         container,
         &Container::Vector,
         &mut cm_builder,
+        resolved_product_image,
     )
     .context(InvalidLoggingConfigSnafu {
         cm_name: rolegroup.object_name(),

rust/operator-binary/src/product_logging.rs

@@ -3,6 +3,7 @@ use std::fmt::{Display, Write};
 use snafu::Snafu;
 use stackable_operator::{
     builder::configmap::ConfigMapBuilder,
+    commons::product_image_selection::ResolvedProductImage,
     kube::Resource,
     product_logging::{
         self,
@@ -43,6 +44,7 @@ pub fn extend_config_map_with_log_config<C, K>(
     main_container: &C,
     vector_container: &C,
     cm_builder: &mut ConfigMapBuilder,
+    resolved_product_image: &ResolvedProductImage,
 ) -> Result<()>
 where
     C: Clone + Ord + Display,
@@ -53,7 +55,10 @@ where
     }) = logging.containers.get(main_container)
     {
         let log_dir = format!("{STACKABLE_LOG_DIR}/{main_container}");
-        cm_builder.add_data(LOG_CONFIG_FILE, create_airflow_config(log_config, &log_dir));
+        cm_builder.add_data(
+            LOG_CONFIG_FILE,
+            create_airflow_config(log_config, &log_dir, resolved_product_image),
+        );
     }
 
     let vector_log_config = if let Some(ContainerLogConfig {
@@ -75,7 +80,11 @@ where
     Ok(())
 }
 
-fn create_airflow_config(log_config: &AutomaticContainerLogConfig, log_dir: &str) -> String {
+fn create_airflow_config(
+    log_config: &AutomaticContainerLogConfig,
+    log_dir: &str,
+    resolved_product_image: &ResolvedProductImage,
+) -> String {
     let loggers_config = log_config
         .loggers
         .iter()
@@ -92,18 +101,28 @@ LOGGING_CONFIG['loggers']['{name}']['level'] = {level}
             output
         });
 
+    let remote_task_log = if resolved_product_image.product_version.starts_with("2.") {
+        ""
+    } else {
+        "
+# This will cause the relevant RemoteLogIO handler to be initialized
+REMOTE_TASK_LOG = airflow_local_settings.REMOTE_TASK_LOG
+log = logging.getLogger(__name__)
+log.info('Custom logging remote task log %s', REMOTE_TASK_LOG)
+"
+    };
+
     format!(
         "\
 import logging
 import os
 from copy import deepcopy
-from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
+from airflow.config_templates import airflow_local_settings
 
 os.makedirs('{log_dir}', exist_ok=True)
 
-LOGGING_CONFIG = deepcopy(DEFAULT_LOGGING_CONFIG)
-
-REMOTE_TASK_LOG = None
+LOGGING_CONFIG = deepcopy(airflow_local_settings.DEFAULT_LOGGING_CONFIG)
+{remote_task_log}
 
 LOGGING_CONFIG.setdefault('loggers', {{}})
 for logger_name, logger_config in LOGGING_CONFIG['loggers'].items():

tests/templates/kuttl/remote-logging/00-patch-ns.yaml.j2

@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}

tests/templates/kuttl/remote-logging/00-range-limit.yaml.j2

@@ -0,0 +1,13 @@
+{% if test_scenario['values']['executor'] == 'celery' %}
+---
+apiVersion: v1
+kind: LimitRange
+metadata:
+  name: limit-request-ratio
+spec:
+  limits:
+  - type: "Container"
+    maxLimitRequestRatio:
+      cpu: 5
+      memory: 1
+{% endif %}

tests/templates/kuttl/remote-logging/02-s3-secret.yaml

@@ -0,0 +1,25 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: minio-credentials
+  labels:
+    secrets.stackable.tech/class: spark-pi-private-s3-credentials-class
+timeout: 240
+stringData:
+  accessKey: minioAccessKey
+  secretKey: minioSecretKey
+  # The following two entries are used by the Bitnami chart for MinIO to
+  # set up credentials for accessing buckets managed by the MinIO tenant.
+  root-user: minioAccessKey
+  root-password: minioSecretKey
+---
+apiVersion: secrets.stackable.tech/v1alpha1
+kind: SecretClass
+metadata:
+  name: spark-pi-private-s3-credentials-class
+spec:
+  backend:
+    k8sSearch:
+      searchNamespace:
+        pod: {}

tests/templates/kuttl/remote-logging/03-assert.yaml

@@ -0,0 +1,20 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 900
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: test-minio
+status:
+  readyReplicas: 1
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: minio-client
+  labels:
+    app: minio-client
+status:
+  phase: Running

tests/templates/kuttl/remote-logging/03-setup-minio.yaml

@@ -0,0 +1,55 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: >-
+      helm install test-minio
+      --namespace $NAMESPACE
+      --version 14.6.16
+      -f helm-bitnami-minio-values.yaml
+      --repo https://charts.bitnami.com/bitnami minio
+    timeout: 240
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: minio-client
+  labels:
+    app: minio-client
+spec:
+  selector:
+    matchLabels:
+      app: minio-client
+  restartPolicy: Never
+  containers:
+    - name: minio-client
+      image: docker.io/bitnamilegacy/minio-client:2024-debian-12
+      command: ["bash", "-c", "sleep infinity"]
+      stdin: true
+      tty: true
+      resources:
+        requests:
+          memory: 1Gi
+          cpu: "1"
+        limits:
+          memory: 1Gi
+          cpu: "2"
+      env:
+        - name: MINIO_SERVER_ACCESS_KEY
+          valueFrom:
+            secretKeyRef:
+              name: minio-credentials
+              key: root-user
+              optional: false
+        - name: MINIO_SERVER_SECRET_KEY
+          valueFrom:
+            secretKeyRef:
+              name: minio-credentials
+              key: root-password
+              optional: false
+        - name: MINIO_SERVER_HOST
+          value: test-minio
+        - name: MINIO_SERVER_PORT_NUMBER
+          value: "9000"
+        - name: MINIO_SERVER_SCHEME
+          value: http

tests/templates/kuttl/remote-logging/04-prepare-bucket.yaml.j2

@@ -0,0 +1,8 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  # give minio enough time to start
+  - command: sleep 10
+  - command: kubectl exec -n $NAMESPACE minio-client -- sh -c 'mc alias set test-minio http://test-minio:9000 $$MINIO_SERVER_ACCESS_KEY $$MINIO_SERVER_SECRET_KEY'
+  - command: kubectl exec -n $NAMESPACE minio-client -- mc mb test-minio/my-bucket

tests/templates/kuttl/remote-logging/10-assert.yaml

@@ -0,0 +1,14 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: test-airflow-postgresql
+timeout: 480
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-postgresql
+status:
+  readyReplicas: 1
+  replicas: 1

tests/templates/kuttl/remote-logging/10-install-postgresql.yaml

@@ -0,0 +1,12 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: >-
+      helm install airflow-postgresql
+      --namespace $NAMESPACE
+      --version 16.4.2
+      -f helm-bitnami-postgresql-values.yaml
+      oci://registry-1.docker.io/bitnamicharts/postgresql
+      --wait
+    timeout: 600

tests/templates/kuttl/remote-logging/20-assert.yaml.j2

@@ -0,0 +1,24 @@
+{% if test_scenario['values']['executor'] == 'celery' %}
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: test-airflow-redis
+timeout: 360
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-redis-master
+status:
+  readyReplicas: 1
+  replicas: 1
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-redis-replicas
+status:
+  readyReplicas: 1
+  replicas: 1
+{% endif %}

tests/templates/kuttl/remote-logging/20-install-redis.yaml.j2

@@ -0,0 +1,14 @@
+{% if test_scenario['values']['executor'] == 'celery' %}
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: >-
+      helm install airflow-redis
+      --namespace $NAMESPACE
+      --version 17.11.3
+      -f helm-bitnami-redis-values.yaml
+      --repo https://charts.bitnami.com/bitnami redis
+      --wait
+    timeout: 600
+{% endif %}

tests/templates/kuttl/remote-logging/40-assert.yaml.j2

@@ -0,0 +1,81 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: test-available-condition
+timeout: 600
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=available airflowclusters.airflow.stackable.tech/airflow --timeout 301s
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: test-airflow-cluster
+timeout: 1200
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-webserver-default
+spec:
+  template:
+    spec:
+      terminationGracePeriodSeconds: 120
+status:
+  readyReplicas: 1
+  replicas: 1
+{% if test_scenario['values']['executor'] == 'celery' %}
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-worker-default
+spec:
+  template:
+    spec:
+      terminationGracePeriodSeconds: 300
+status:
+  readyReplicas: 2
+  replicas: 2
+{% endif %}
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-scheduler-default
+spec:
+  template:
+    spec:
+      terminationGracePeriodSeconds: 120
+status:
+  readyReplicas: 1
+  replicas: 1
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: airflow-webserver
+status:
+  expectedPods: 1
+  currentHealthy: 1
+  disruptionsAllowed: 1
+{% if test_scenario['values']['executor'] == 'celery' %}
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: airflow-worker
+status:
+  expectedPods: 2
+  currentHealthy: 2
+  disruptionsAllowed: 1
+{% endif %}
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: airflow-scheduler
+status:
+  expectedPods: 1
+  currentHealthy: 1
+  disruptionsAllowed: 1