Drive the task handler log level from the airflow.task logger

CHANGELOG.md

@@ -11,11 +11,20 @@
 - Bump stackable-operator to 0.114.0 ([#827]).
 - The reconciler now applies resources and derives the cluster status in discrete
   apply and update_status steps ([#828]).
+- The level configured for the `airflow.task` logger now sets the level of the `task` handler,
+  which is what the Airflow UI displays, instead of the level of the logger itself. It still
+  defaults to `INFO`, matching Airflow's own default. A level above `INFO` is applied to the
+  handler alone and quietens the UI without raising the logger, so the console and file handlers
+  keep receiving records that the UI no longer shows. A level below `INFO` has to be applied to
+  the logger as well, because a logger discards records before any handler can filter them, so in
+  that direction the UI and the other destinations open up together and cannot be set apart
+  ([#829]).
 
 [#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
 
 ## [26.7.0] - 2026-07-21
 

docs/modules/airflow/pages/usage-guide/logging.adoc

@@ -4,9 +4,9 @@
 The logs can be forwarded to a Vector log aggregator by providing a discovery ConfigMap for the aggregator and by enabling the log agent:
 
 NOTE: The `task` handler is responsible for showing the task logs in the UI.
-Unfortunately, the log level of the `task` handler cannot be specified.
-To avoid that all logs are emitted to the UI, the log level of the `airflow.task` logger is set explicitly to `INFO`.
-You can change the log level as shown below.
+Its log level is taken from the level configured for the `airflow.task` logger and defaults to `INFO`, as it does in Airflow itself.
+A level above `INFO` applies to the handler only, so it quietens the UI while the other destinations keep receiving the records they received before.
+A level below `INFO` cannot apply to the handler alone: a logger discards records before any of its handlers can filter them, so the logger has to be opened up as well and the additional records then reach every destination, not just the UI.
 
 [source,yaml]
 ----
@@ -29,11 +29,12 @@ spec:
         enableVectorAgent: true
         containers:
           airflow:
+            # Show only WARN and above in the UI, while the other destinations keep receiving INFO.
             loggers:
+              "airflow.task":
+                level: WARN
               "airflow.processor":
                 level: INFO
-              "airflow.task":
-                level: DEBUG
   schedulers:
     config:
       logging:

rust/operator-binary/src/controller/build/properties/product_logging/mod.rs

@@ -1,17 +1,24 @@
 //! Renders the logging config files (`log_config.py` and the Vector agent config)
 //! assembled into the rolegroup `ConfigMap`.
 
-use std::fmt::Write;
+use std::{cmp, fmt::Write};
 
 use stackable_operator::{
     commons::product_image_selection::ResolvedProductImage,
-    product_logging::spec::AutomaticContainerLogConfig,
+    product_logging::spec::{AutomaticContainerLogConfig, LogLevel},
     v2::product_logging::framework::ValidatedContainerLogConfigChoice,
 };
 
 /// The rotating log file the generated `log_config.py` writes to (consumed by the Vector agent).
 const LOG_FILE: &str = "airflow.py.json";
 
+/// The logger whose configured level drives the `task` handler.
+///
+/// Airflow keeps the task logs the web UI displays on a dedicated `task` handler, which has no
+/// appender of its own in [`AutomaticContainerLogConfig`]. Its level is therefore taken from the
+/// level configured for this logger.
+const TASK_LOGGER: &str = "airflow.task";
+
 /// The Vector agent configuration (`vector.yaml`).
 const VECTOR_CONFIG: &str = include_str!("vector.yaml");
 
@@ -53,7 +60,12 @@ fn create_airflow_stdlib_config(
     let loggers_config = log_config
         .loggers
         .iter()
-        .filter(|(name, _)| name.as_str() != AutomaticContainerLogConfig::ROOT_LOGGER)
+        // The task logger is rendered explicitly below, because its level also drives the
+        // `task` handler and must not simply be assigned to the logger.
+        .filter(|(name, _)| {
+            name.as_str() != AutomaticContainerLogConfig::ROOT_LOGGER
+                && name.as_str() != TASK_LOGGER
+        })
         .fold(String::new(), |mut output, (name, config)| {
             let _ = writeln!(
                 output,
@@ -97,10 +109,11 @@ for logger_name, logger_config in LOGGING_CONFIG['loggers'].items():
     if logger_name != 'airflow.task':
         logger_config['propagate'] = True
     # The default behavior of airflow is to enforce log level 'INFO' on tasks. (https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#logging-level)
-    # TODO: Make task handler log level configurable through CRDs with default 'INFO'.
-    # e.g. LOGGING_CONFIG['handlers']['task']['level'] = {{task_log_level}}
+    # Records are filtered by the task handler below rather than here, so this level is
+    # only lowered, never raised. A logger drops records before any handler sees them, so
+    # a level above INFO would starve the console and file handlers as well.
     if 'handlers' in logger_config and 'task' in logger_config['handlers']:
-        logger_config['level'] = logging.INFO
+        logger_config['level'] = {task_logger_level}
 
 LOGGING_CONFIG.setdefault('formatters', {{}})
 LOGGING_CONFIG['formatters']['json'] = {{
@@ -119,6 +132,8 @@ LOGGING_CONFIG['handlers']['file'] = {{
     'maxBytes': 1048576,
     'backupCount': 1,
 }}
+LOGGING_CONFIG['handlers'].setdefault('task', {{}})
+LOGGING_CONFIG['handlers']['task']['level'] = {task_log_level}
 
 LOGGING_CONFIG['root'] = {{
     'level': {root_log_level},
@@ -139,17 +154,47 @@ LOGGING_CONFIG['root'] = {{
             .and_then(|file| file.level)
             .unwrap_or_default()
             .to_python_expression(),
+        task_log_level = task_log_level(log_config).to_python_expression(),
+        task_logger_level = task_logger_level(log_config).to_python_expression(),
     )
 }
 
+/// The log level for the `task` handler, which is what the Airflow web UI displays.
+///
+/// Taken from the level configured for the [`TASK_LOGGER`], and defaults to `INFO`, which is
+/// also Airflow's own default, when that logger is not configured.
+fn task_log_level(log_config: &AutomaticContainerLogConfig) -> LogLevel {
+    log_config
+        .loggers
+        .get(TASK_LOGGER)
+        .map(|logger| logger.level)
+        .unwrap_or(LogLevel::INFO)
+}
+
+/// The level to set on the [`TASK_LOGGER`] itself.
+///
+/// Airflow pins it to `INFO`. A logger discards records before any handler can filter them, so
+/// asking for a level below `INFO` has to open the logger up as well, otherwise the handler
+/// would never see the records. Asking for a level above `INFO` must *not* raise the logger,
+/// because the console and file handlers still want those records, so the logger is left at
+/// `INFO` and the `task` handler filters on its own.
+fn task_logger_level(log_config: &AutomaticContainerLogConfig) -> LogLevel {
+    cmp::min(LogLevel::INFO, task_log_level(log_config))
+}
+
 fn create_airflow_structlog_config(
     log_config: &AutomaticContainerLogConfig,
     log_dir: &str,
 ) -> String {
     let loggers_config = log_config
         .loggers
         .iter()
-        .filter(|(name, _)| name.as_str() != AutomaticContainerLogConfig::ROOT_LOGGER)
+        // The task logger is rendered explicitly below, because its level also drives the
+        // `task` handler and must not simply be assigned to the logger.
+        .filter(|(name, _)| {
+            name.as_str() != AutomaticContainerLogConfig::ROOT_LOGGER
+                && name.as_str() != TASK_LOGGER
+        })
         .fold(String::new(), |mut output, (name, config)| {
             let _ = writeln!(
                 output,
@@ -200,6 +245,7 @@ LOGGING_CONFIG = {{
         }},
         'task': {{
             'class': 'airflow.utils.log.file_task_handler.FileTaskHandler',
+            'level': {task_log_level},
             'formatter': 'airflow',
             'base_log_folder': '{log_dir}',
             'filters': ['mask_secrets_core']
@@ -208,7 +254,7 @@ LOGGING_CONFIG = {{
     'loggers': {{
         'airflow.task': {{
             'handlers': ['task'],
-            'level': logging.INFO,
+            'level': {task_logger_level},
             'propagate': True,
             'filters': ['mask_secrets_core']
         }}
@@ -235,11 +281,17 @@ REMOTE_TASK_LOG = airflow_local_settings.REMOTE_TASK_LOG
             .unwrap_or_default()
             .to_python_expression(),
         root_log_level = log_config.root_log_level().to_python_expression(),
+        task_log_level = task_log_level(log_config).to_python_expression(),
+        task_logger_level = task_logger_level(log_config).to_python_expression(),
     )
 }
 
 #[cfg(test)]
 mod tests {
+    use std::collections::BTreeMap;
+
+    use stackable_operator::product_logging::spec::{LogLevel, LoggerConfig};
+
     use super::*;
 
     #[test]
@@ -254,4 +306,198 @@ mod tests {
         assert!(content.contains("${ROLE_NAME}"));
         assert!(content.contains("${VECTOR_AGGREGATOR_ADDRESS}"));
     }
+
+    fn log_config_with_task_logger(level: Option<LogLevel>) -> AutomaticContainerLogConfig {
+        let mut loggers = BTreeMap::new();
+
+        if let Some(level) = level {
+            loggers.insert(TASK_LOGGER.to_string(), LoggerConfig { level });
+        }
+
+        AutomaticContainerLogConfig {
+            loggers,
+            console: None,
+            file: None,
+        }
+    }
+
+    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)
+    }
+
+    /// The requested level paired with the `task` handler level and the `airflow.task` logger
+    /// level that it produces, for every level a user can configure.
+    ///
+    /// The two are deliberately listed together, because the relationship between them is not
+    /// symmetric and that is the part which is easy to get wrong:
+    ///
+    /// - At or above `INFO` the handler follows the request and the logger stays at `INFO`. Only
+    ///   the UI goes quiet; the console and file handlers keep receiving the records it drops.
+    /// - Below `INFO` the two are coupled. A logger discards records before any of its handlers
+    ///   can filter them, so the logger has to be opened up too, which means the extra records
+    ///   reach every destination and not just the UI. Lowering the UI on its own is therefore
+    ///   not possible, whichever way this is implemented.
+    const TASK_LEVELS: [(LogLevel, &str, &str); 7] = [
+        // requested        task handler        airflow.task logger
+        (LogLevel::TRACE, "logging.DEBUG", "logging.DEBUG"),
+        (LogLevel::DEBUG, "logging.DEBUG", "logging.DEBUG"),
+        (LogLevel::INFO, "logging.INFO", "logging.INFO"),
+        (LogLevel::WARN, "logging.WARNING", "logging.INFO"),
+        (LogLevel::ERROR, "logging.ERROR", "logging.INFO"),
+        (LogLevel::FATAL, "logging.CRITICAL", "logging.INFO"),
+        (LogLevel::NONE, "logging.CRITICAL + 1", "logging.INFO"),
+    ];
+
+    // Spells out the handler/logger pair for every configurable level in one place, so the
+    // asymmetry documented on TASK_LEVELS is visible instead of being spread across cases.
+    #[test]
+    fn test_stdlib_config_task_handler_and_logger_levels() {
+        for (requested, handler_level, logger_level) in TASK_LEVELS {
+            let content = stdlib_config(&log_config_with_task_logger(Some(requested)));
+
+            assert!(
+                content.contains(&format!(
+                    "LOGGING_CONFIG['handlers']['task']['level'] = {handler_level}"
+                )),
+                "{requested} must put {handler_level} on the task handler"
+            );
+            assert!(
+                content.contains(&format!("logger_config['level'] = {logger_level}")),
+                "{requested} must put {logger_level} on the airflow.task logger"
+            );
+        }
+    }
+
+    #[test]
+    fn test_structlog_config_task_handler_and_logger_levels() {
+        for (requested, handler_level, logger_level) in TASK_LEVELS {
+            let content = create_airflow_structlog_config(
+                &log_config_with_task_logger(Some(requested)),
+                "/stackable/log",
+            );
+
+            assert_eq!(
+                level_in_block(&content, "'task': {"),
+                handler_level,
+                "{requested} must put {handler_level} on the task handler"
+            );
+            assert_eq!(
+                level_in_block(&content, "'airflow.task': {"),
+                logger_level,
+                "{requested} must put {logger_level} on the airflow.task logger"
+            );
+        }
+    }
+
+    // Lowering the task level cannot quieten or open up the UI alone: the logger moves with it,
+    // so the console and file handlers see the extra records as well. This is the coupling the
+    // docs and the changelog describe, pinned here so it cannot change unnoticed.
+    #[test]
+    fn test_task_level_below_info_is_coupled_to_the_logger() {
+        for requested in [LogLevel::TRACE, LogLevel::DEBUG] {
+            let log_config = log_config_with_task_logger(Some(requested));
+
+            assert_eq!(task_log_level(&log_config), requested);
+            assert_eq!(
+                task_logger_level(&log_config),
+                requested,
+                "{requested} is below INFO, so the logger must be opened up to match the handler"
+            );
+        }
+    }
+
+    // At or above INFO the handler and the logger diverge, which is what lets the UI be
+    // quietened without starving the console and file handlers.
+    #[test]
+    fn test_task_level_above_info_leaves_the_logger_at_info() {
+        for requested in [
+            LogLevel::WARN,
+            LogLevel::ERROR,
+            LogLevel::FATAL,
+            LogLevel::NONE,
+        ] {
+            let log_config = log_config_with_task_logger(Some(requested));
+
+            assert_eq!(task_log_level(&log_config), requested);
+            assert_eq!(
+                task_logger_level(&log_config),
+                LogLevel::INFO,
+                "{requested} must not raise the logger, or other handlers lose records"
+            );
+        }
+    }
+
+    // Nothing configured must land on Airflow's own default rather than on whatever the
+    // clamping happens to produce.
+    #[test]
+    fn test_stdlib_config_defaults_task_handler_to_info() {
+        let content = stdlib_config(&log_config_with_task_logger(None));
+
+        assert!(content.contains("LOGGING_CONFIG['handlers']['task']['level'] = logging.INFO"));
+        assert!(content.contains("logger_config['level'] = logging.INFO"));
+    }
+
+    // The airflow.task entry drives the handler, so the generic loggers block must not also
+    // emit it. If it did, it would assign the requested level straight to the logger and undo
+    // the clamping above.
+    #[test]
+    fn test_stdlib_config_does_not_emit_the_task_logger_generically() {
+        let content = stdlib_config(&log_config_with_task_logger(Some(LogLevel::ERROR)));
+
+        assert!(!content.contains("LOGGING_CONFIG['loggers']['airflow.task']"));
+    }
+
+    #[test]
+    fn test_structlog_config_does_not_emit_the_task_logger_generically() {
+        let content = create_airflow_structlog_config(
+            &log_config_with_task_logger(Some(LogLevel::ERROR)),
+            "/stackable/log",
+        );
+
+        assert!(!content.contains("LOGGING_CONFIG['loggers']['airflow.task']"));
+    }
+
+    /// Returns the level assigned inside the block starting at `block`, e.g. the `task`
+    /// handler or the `airflow.task` logger, so the two cannot be confused for each other.
+    fn level_in_block(content: &str, block: &str) -> String {
+        let start = content
+            .find(block)
+            .unwrap_or_else(|| panic!("the {block} block must be present"));
+        let rest = &content[start..];
+        let level_at = rest.find("'level': ").expect("the block must set a level");
+
+        rest[level_at + "'level': ".len()..]
+            .split([',', '\n'])
+            .next()
+            .expect("the level must be terminated")
+            .trim()
+            .to_string()
+    }
+
+    #[test]
+    fn test_structlog_config_defaults_task_handler_to_info() {
+        let content =
+            create_airflow_structlog_config(&log_config_with_task_logger(None), "/stackable/log");
+
+        assert_eq!(level_in_block(&content, "'task': {"), "logging.INFO");
+        assert_eq!(
+            level_in_block(&content, "'airflow.task': {"),
+            "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,
+        }
+    }
 }