Ensure logging is configured on remote task workers (#21379)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Alex Streed <desertaxle@users.noreply.github.com>
Co-authored-by: alex.s <ajstreed1@gmail.com>

Author: devin-ai-integration[bot]

src/prefect/context.py

@@ -37,6 +37,7 @@
 from prefect.client.schemas.objects import RunType
 from prefect.events.worker import EventsWorker
 from prefect.exceptions import MissingContextError
+from prefect.logging.configuration import ensure_logging_setup
 from prefect.results import (
     ResultStore,
     get_default_persist_setting,
@@ -163,6 +164,8 @@ def hydrated_context(
 
     with ExitStack() as stack:
         if serialized_context:
+            ensure_logging_setup()
+
             # Set up settings context
             if settings_context := serialized_context.get("settings_context"):
                 stack.enter_context(SettingsContext(**settings_context))

src/prefect/logging/configuration.py

@@ -65,6 +65,19 @@ def load_logging_config(path: Path) -> dict[str, Any]:
     return flatdict_to_dict(flat_config)
 
 
+def ensure_logging_setup() -> None:
+    """
+    Ensure Prefect logging is configured in this process, calling
+    `setup_logging` only if it has not already been called.
+
+    Use this in remote execution environments (e.g. Dask/Ray workers) where
+    the normal SDK entry point (`import prefect`) may not have triggered
+    logging configuration.
+    """
+    if not PROCESS_LOGGING_CONFIG:
+        setup_logging()
+
+
 def setup_logging(incremental: bool | None = None) -> dict[str, Any]:
     """
     Sets up logging.

tests/test_logging.py

@@ -33,6 +33,7 @@
 from prefect.logging import LogEavesdropper
 from prefect.logging.configuration import (
     DEFAULT_LOGGING_SETTINGS_PATH,
+    ensure_logging_setup,
     load_logging_config,
     setup_logging,
 )
@@ -285,6 +286,22 @@ def test_setup_logging_applies_root_config_when_no_prior_configuration(
     assert called_config["root"]["handlers"] == ["console"]
 
 
+def test_ensure_logging_setup_calls_setup_logging_when_not_configured(
+    dictConfigMock: MagicMock,
+):
+    ensure_logging_setup()
+    dictConfigMock.assert_called_once()
+
+
+def test_ensure_logging_setup_is_idempotent(dictConfigMock: MagicMock):
+    ensure_logging_setup()
+    ensure_logging_setup()
+    ensure_logging_setup()
+    # setup_logging should only be called once since PROCESS_LOGGING_CONFIG
+    # is populated after the first call
+    dictConfigMock.assert_called_once()
+
+
 def test_setting_aliases_respected_for_logging_config(tmp_path: Path):
     logging_config_content = """
 loggers: