Feature:4005 Docker settings usage warnings

- Warn when used without containerized orchestrator in place

Author: Json-Andriopoulos

src/zenml/config/compiler.py

@@ -28,6 +28,7 @@
 
 from zenml import __version__
 from zenml.config.base_settings import BaseSettings, ConfigurationLevel
+from zenml.config.constants import DOCKER_SETTINGS_KEY
 from zenml.config.pipeline_configurations import PipelineConfiguration
 from zenml.config.pipeline_run_configuration import PipelineRunConfiguration
 from zenml.config.pipeline_spec import OutputSpec, PipelineSpec
@@ -43,6 +44,7 @@
 from zenml.models import PipelineSnapshotBase
 from zenml.pipelines.run_utils import get_default_run_name
 from zenml.utils import pydantic_utils, secret_utils, settings_utils
+from zenml.utils.warnings import WARNING_CONTROLLER, WarningCodes
 
 if TYPE_CHECKING:
     from zenml.pipelines.pipeline_definition import Pipeline
@@ -369,6 +371,37 @@ def _verify_upstream_steps(
                 f"steps in this pipeline: {available_steps}."
             )
 
+    @staticmethod
+    def _validate_docker_settings_usage(
+        docker_settings: "BaseSettings",
+        stack: "Stack",
+    ) -> None:
+        from zenml.orchestrators import (
+            ContainerizedOrchestrator,
+            LocalOrchestrator,
+        )
+
+        if not docker_settings or isinstance(
+            stack.orchestrator, ContainerizedOrchestrator
+        ):
+            return
+
+        warning_message = (
+            "You are specifying docker settings but you are not using a"
+            f"containerized orchestrator: {stack.orchestrator.__class__.__name__}."
+            f"Consider switching stack or using a containerized orchestrator, otherwise"
+            f"your docker settings will be ignored."
+        )
+
+        if isinstance(stack.orchestrator, LocalOrchestrator):
+            WARNING_CONTROLLER.info(
+                warning_code=WarningCodes.ZML002, message=warning_message
+            )
+        else:
+            WARNING_CONTROLLER.warn(
+                warning_code=WarningCodes.ZML002, message=warning_message
+            )
+
     def _filter_and_validate_settings(
         self,
         settings: Dict[str, "BaseSettings"],
@@ -397,9 +430,10 @@ def _filter_and_validate_settings(
             try:
                 settings_instance = resolver.resolve(stack=stack)
             except KeyError:
-                logger.info(
-                    "Not including stack component settings with key `%s`.",
-                    key,
+                WARNING_CONTROLLER.info(
+                    warning_code=WarningCodes.ZML001.value,
+                    message="Not including stack component settings with key {key}",
+                    key=key,
                 )
                 continue
 
@@ -429,6 +463,11 @@ def _filter_and_validate_settings(
 
             validated_settings[key] = settings_instance
 
+        self._validate_docker_settings_usage(
+            stack=stack,
+            docker_settings=settings.get(DOCKER_SETTINGS_KEY),
+        )
+
         return validated_settings
 
     def _get_step_spec(

src/zenml/logger.py

@@ -357,6 +357,9 @@ def init_logging() -> None:
         for handler in root_logger.handlers
     )
 
+    # logging capture warnings
+    logging.captureWarnings(True)
+
     if not has_console_handler:
         console_handler = logging.StreamHandler(sys.stdout)
         console_handler.setFormatter(get_formatter())

src/zenml/utils/warnings/__init__.py

@@ -0,0 +1,28 @@
+#  Copyright (c) ZenML GmbH 2025. All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at:
+#
+#       https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+#  or implied. See the License for the specific language governing
+#  permissions and limitations under the License.
+"""Initialization of the warning utils module.
+
+The `warnings` module contains utilities that help centralize
+and improve warning management.
+"""
+
+from zenml.utils.warnings.controller import WarningController
+from zenml.utils.warnings.registry import WarningCodes
+
+WARNING_CONTROLLER = WarningController()
+
+__all__ = [
+    "WARNING_CONTROLLER",
+    "WarningCodes"
+]

src/zenml/utils/warnings/base.py

@@ -0,0 +1,34 @@
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class WarningSeverity(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+class WarningCategory(str, Enum):
+    USAGE = "USAGE"
+
+
+class WarningVerbosity(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+class WarningConfig(BaseModel, use_enum_values=True):
+    description: str = Field(description="Description of the warning message")
+    severity: WarningSeverity = WarningSeverity.MEDIUM
+    category: WarningCategory
+    code: str
+    is_throttled: bool = Field(
+        description="If the warning is throttled. Throttled warnings should be displayed once per session.",
+        default=False,
+    )
+    verbosity: WarningVerbosity = Field(
+        description="Verbosity of the warning. Low verbosity displays basic details (code & message). Medium displays also call details like module & line number. High displays description and additional info.",
+        default=WarningVerbosity.MEDIUM,
+    )

src/zenml/utils/warnings/controller.py

@@ -0,0 +1,93 @@
+import logging
+from collections import defaultdict
+
+from zenml.enums import LoggingLevels
+from zenml.utils.warnings.base import WarningConfig, WarningVerbosity
+
+logger = logging.getLogger(__name__)
+
+
+class WarningController:
+    def __init__(self) -> None:
+        self._warning_configs: dict[str, WarningConfig] = {}
+        self._warning_statistics: dict[str, int] = defaultdict(int)
+
+    def register(self, warning_configs: dict[str, WarningConfig]) -> None:
+        self._warning_configs.update(warning_configs)
+
+    @staticmethod
+    def _resolve_call_details() -> tuple[str, int]:
+        import inspect
+
+        frame = inspect.stack()[3]  # public methods -> _log -> _resolve
+        module = inspect.getmodule(frame[0])
+        module_name = module.__name__ if module else "<unknown module>"
+        line_number = frame.lineno
+
+        return module_name, line_number
+
+    @staticmethod
+    def _get_display_message(
+        message: str,
+        module_name: str,
+        line_number: int,
+        config: WarningConfig,
+    ):
+        display = f"[{config.code}]({config.category}) - {message}"
+
+        if config.verbosity == WarningVerbosity.MEDIUM:
+            display = f"{module_name}:{line_number} {display}"
+
+        if config.verbosity == WarningVerbosity.HIGH:
+            display = f"{display}\n{config.description}"
+
+        return display
+
+    def _log(
+        self, warning_code: str, message: str, level: LoggingLevels, **kwargs
+    ) -> None:
+        warning_config = self._warning_configs.get(warning_code)
+
+        # resolves the module and line number of the warning call.
+        module_name, line_number = self._resolve_call_details()
+
+        if not warning_config:
+            # If no config is available just follow default behaviour:
+            logger.warning(f"{module_name}:{line_number} - {message}")
+            return
+
+        if warning_config.is_throttled:
+            if warning_code in self._warning_statistics:
+                # Throttled code has already appeared - skip.
+                return
+
+        display_message = self._get_display_message(
+            message=message,
+            module_name=module_name,
+            line_number=line_number,
+            config=warning_config,
+        )
+
+        self._warning_statistics[warning_code] += 1
+
+        if level == LoggingLevels.INFO:
+            logger.info(display_message.format(**kwargs))
+        else:
+            # Assumes warning level is the default if an invalid option is passed.
+            logger.warning(display_message.format(**kwargs))
+
+    def warn(self, *, warning_code: str, message: str, **kwargs) -> None:
+        self._log(warning_code, message, LoggingLevels.WARNING, **kwargs)
+
+    def info(self, *, warning_code: str, message: str, **kwargs) -> None:
+        self._log(warning_code, message, LoggingLevels.INFO, **kwargs)
+
+    @staticmethod
+    def create() -> "WarningController":
+        from zenml.utils.warnings.registry import WARNING_CONFIG_REGISTRY
+
+        registry = WarningController()
+
+        registry.register(warning_configs=WARNING_CONFIG_REGISTRY)
+
+        return registry

src/zenml/utils/warnings/registry.py

@@ -0,0 +1,42 @@
+from enum import Enum
+
+from zenml.utils.warnings.base import (
+    WarningCategory,
+    WarningConfig,
+    WarningSeverity,
+    WarningVerbosity,
+)
+
+
+class WarningCodes(str, Enum):
+    ZML001 = "ZML001"
+    ZML002 = "ZML002"
+
+
+WARNING_CONFIG_REGISTRY = {
+    WarningCodes.ZML001.value: WarningConfig(
+        code=WarningCodes.ZML001.value,
+        category=WarningCategory.USAGE,
+        description="""
+        ## Ignored settings warning
+        
+        The user has provided a settings object without a valid key.
+        """,
+        verbosity=WarningVerbosity.LOW,
+        severity=WarningSeverity.MEDIUM,
+        is_throttled=False,
+    ),
+    WarningCodes.ZML002.value: WarningConfig(
+        code=WarningCodes.ZML002.value,
+        category=WarningCategory.USAGE,
+        description="""
+        ## Unused docker settings warning
+        
+        The user provides a valid docker settings object but the stack
+        does not have a containerised orchestrator and will not make use of it.
+        """,
+        verbosity=WarningVerbosity.LOW,
+        severity=WarningSeverity.MEDIUM,
+        is_throttled=True,
+    ),
+}

tests/unit/config/test_docker_settings.py

@@ -16,7 +16,9 @@
 import pytest
 from pydantic import ValidationError
 
-from zenml.config import DockerSettings
+from zenml.config.docker_settings import (
+    DockerSettings,
+)
 
 
 def test_build_skipping():