Repository: ray-project/ray
Issue category: Bug Report

Issue:
[Core] Ray Core / Ray Data logging configuration leads to unexpected behavior 
### What happened + What you expected to happen

We use `logging.config.dictConfig(config)` to configure the ray data logger  ([here](https://github.com/ray-project/ray/blob/4f6a419467885c295700f2db6375a298bc575fc2/python/ray/data/_internal/logging.py#L176)), but this is also how ray core configures the ray logger ([here](https://github.com/ray-project/ray/blob/4f6a419467885c295700f2db6375a298bc575fc2/python/ray/_private/ray_logging/logging_config.py#L124)).

For both of these logging configs, we use `disable_existing_loggers: False`. The behavior for this is described as ([logging docs](https://docs.python.org/3/library/logging.config.html#dictionary-schema-details)):
> disable_existing_loggers - whether any existing non-root loggers are to be disabled. This setting mirrors the parameter of the same name in [fileConfig()](https://docs.python.org/3/library/logging.config.html#logging.config.fileConfig). If absent, this parameter defaults to True. This value is ignored if incremental is True.

This description makes it seem like these two logging calls are commutative (regardless of ordering they will produce the same result), but that is not exactly how the python logging module works. If we configure the ray module logger then the ray.data module logger, the results are expected and both are configured. If we instead configure the ray.data module then configure the ray module logger, then the ray.data logging configuration is clobbered. This happens because when configuring the parent logger of a module (e.g. ray module logger is the parent logger of the ray.data module logger), the various handlers associated with the child logger are not guaranteed to be preserved.

Our end goal should be a state where the call order of the logging configurations should not affect the logging behavior.

### Versions / Dependencies

ray==2.39.0

### Reproduction script

```python
import ray
import logging

def report_logger(logger):
    # Collect this logger and its parents
    loggers = []
    current_logger = logger
    while current_logger:
        loggers.append(current_logger)
        if not current_logger.parent or current_logger.parent == current_logger:
            break
        current_logger = current_logger.parent
    
    # Report the configuration of each logger in the hierarchy
    print(f"Logging configuration for '{logger.name}' and its hierarchy:")
    for log in reversed(loggers):  # Start from the root and go down to the given logger
        print(f"\nLogger: {log.name or 'root'} (Level: {logging.getLevelName(log.level)})")
        if log.handlers:
            print("  Handlers:")
            for handler in log.handlers:
                print(f"    - {handler.__class__.__name__} (Level: {logging.getLevelName(handler.level)})")
        else:
            print("  No handlers configured")

print("BEFORE")
report_logger(logging.getLogger("ray.data"))
print()

import ray.data
ray.init(logging_config=ray.LoggingConfig(encoding="JSON", log_level="INFO"))

print("AFTER:")
report_logger(logging.getLogger("ray.data"))
```

### Issue Severity

High: It blocks me from completing my task.


Hints:
To be clear about the expected behavior some examples / modifications of the repro:

### 1. Ray Data Import then Ray Core Init [UNEXPECTED BEHAVIOR]
Script: 
```python
print("BEFORE")
report_logger(logging.getLogger("ray.data"))
print()

import ray.data
ray.init(logging_config=ray.LoggingConfig(encoding="JSON", log_level="INFO"))

print("AFTER:")
report_logger(logging.getLogger("ray.data"))
```
Output:
```
BEFORE
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: WARNING)
  No handlers configured

Logger: ray (Level: INFO)
  Handlers:
    - PlainRayHandler (Level: NOTSET)

Logger: ray.data (Level: NOTSET)
  No handlers configured

AFTER:
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: INFO)
  Handlers:
    - StreamHandler (Level: INFO)

Logger: ray (Level: INFO)
  Handlers:
    - StreamHandler (Level: INFO)

Logger: ray.data (Level: NOTSET)
  No handlers configured

```
Notes:
Ray Data configuration ignored, only Ray Core initialization respected.

### 2. Ray Core Init then Ray Data Import [EXPECTED BEHAVIOR]
Script: 
```python
print("BEFORE")
report_logger(logging.getLogger("ray.data"))
print()

ray.init(logging_config=ray.LoggingConfig(encoding="JSON", log_level="INFO"))
import ray.data

print("AFTER:")
report_logger(logging.getLogger("ray.data"))
```
Output:
```
BEFORE
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: WARNING)
  No handlers configured

Logger: ray (Level: INFO)
  Handlers:
    - PlainRayHandler (Level: NOTSET)

Logger: ray.data (Level: NOTSET)
  No handlers configured

AFTER:
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: INFO)
  Handlers:
    - StreamHandler (Level: INFO)

Logger: ray (Level: INFO)
  Handlers:
    - StreamHandler (Level: INFO)

Logger: ray.data (Level: DEBUG)
  Handlers:
    - SessionFileHandler (Level: NOTSET)
    - PlainRayHandler (Level: INFO)
    - SessionFileHandler (Level: ERROR)
```
Notes:
Ray Core configuration is done first, Ray Data configuration is done second, both are respected.

### 3. Only Ray Data Import [EXPECTED BEHAVIOR]
Script: 
```python
print("BEFORE")
report_logger(logging.getLogger("ray.data"))
print()

import ray.data

print("AFTER:")
report_logger(logging.getLogger("ray.data"))
```
Output:
```
BEFORE
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: WARNING)
  No handlers configured

Logger: ray (Level: INFO)
  Handlers:
    - PlainRayHandler (Level: NOTSET)

Logger: ray.data (Level: NOTSET)
  No handlers configured

AFTER:
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: WARNING)
  No handlers configured

Logger: ray (Level: INFO)
  Handlers:
    - PlainRayHandler (Level: NOTSET)

Logger: ray.data (Level: DEBUG)
  Handlers:
    - SessionFileHandler (Level: NOTSET)
    - PlainRayHandler (Level: INFO)
    - SessionFileHandler (Level: ERROR)
```
Notes:
Correctly configures the Ray Data logger.

### 4. Only Ray Core Init [EXPECTED BEHAVIOR]
Script: 
```python
print("BEFORE")
report_logger(logging.getLogger("ray.data"))
print()

ray.init(logging_config=ray.LoggingConfig(encoding="JSON", log_level="INFO"))

print("AFTER:")
report_logger(logging.getLogger("ray.data"))
```
Output:
```
BEFORE
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: WARNING)
  No handlers configured

Logger: ray (Level: INFO)
  Handlers:
    - PlainRayHandler (Level: NOTSET)

Logger: ray.data (Level: NOTSET)
  No handlers configured

AFTER:
Logging configuration for 'ray.data' and its hierarchy:

Logger: root (Level: INFO)
  Handlers:
    - StreamHandler (Level: INFO)

Logger: ray (Level: INFO)
  Handlers:
    - StreamHandler (Level: INFO)

Logger: ray.data (Level: NOTSET)
  No handlers configured

```
Notes:
Correctly configures the Ray Core logger.



@omatthew98 why don't we just do

1. Take the current config
2. Merge in provided one
3. Set the config back
Update from some more offline discussion with @alexeykudinkin:
For now, ensuring that the import / happens after the initialization (e.g. ensuring that the ray module logger is configured before the ray.data module logger), should be an immediate workaround. We should still try to improve the behavior as this is a footgun that will result in unexpected results and missing logs that will hinder debugging efforts for users. Ideally we would do what was suggested above, but unfortunately python's logging module does not provide a native way to retrieve the existing logging configuration as a dictionary, so we would have to manually inspect and reconstruct the dict before merging the provided config which would be somewhat hacky. 

An alternative to this would be to just wrap this configuration at the ray core level by:
1. Persist configuration as root level field of the core logging module
2. Create util `configureLogging` and have it be called from both core and other libraries
3. Within the `configureLogging` function, merge the provided dictionary with the existing dictionary configuration and then update the logging config with the merged config. 
I am trying to understand the structured logging behavior as well. One of my question is 
#1. Ray Data Import then Ray Core Init [UNEXPECTED BEHAVIOR] will wipe out the `ray.data` logger configuration; but why
#2. Ray Core Init then Ray Data Import [EXPECTED BEHAVIOR] maintains the expected logger configurations of both logger?
It seems by going in order #2, we got the correct logger config for all three loggers: i.e., `root` by ray.init, `ray` by ray.init, and finally `ray.data` by import ray data
> I am trying to understand the structured logging behavior as well. One of my question is #1. Ray Data Import then Ray Core Init [UNEXPECTED BEHAVIOR] will wipe out the `ray.data` logger configuration; but why #2. Ray Core Init then Ray Data Import [EXPECTED BEHAVIOR] maintains the expected logger configurations of both logger? It seems by going in order #2, we got the correct logger config for all three loggers: i.e., `root` by ray.init, `ray` by ray.init, and finally `ray.data` by import ray data

Yeah I think that understanding is correct. From what I have read, configuring the parent logger after the child logger (so the order of #1), the result will be only the parent logger is configured. If you configure the child logger after the parent logger (so the order of #2), then the two loggers will be configured as expected. I think this is is just an implementation detail for python's logging module that doesn't seem particularly well documented.

Patch:
diff --git a/python/ray/_private/log.py b/python/ray/_private/log.py
index 73ce2d3ef329..69242c722233 100644
--- a/python/ray/_private/log.py
+++ b/python/ray/_private/log.py
@@ -1,5 +1,4 @@
 import logging
-from logging.config import dictConfig
 import threading
 from typing import Union
 
@@ -74,44 +73,19 @@ def generate_logging_config():
             return
         logger_initialized = True
 
-        formatters = {
-            "plain": {
-                "format": (
-                    "%(asctime)s\t%(levelname)s %(filename)s:%(lineno)s -- %(message)s"
-                ),
-            },
-        }
-
-        handlers = {
-            "default": {
-                "()": PlainRayHandler,
-                "formatter": "plain",
-            }
-        }
-
-        loggers = {
-            # Default ray logger; any log message that gets propagated here will be
-            # logged to the console. Disable propagation, as many users will use
-            # basicConfig to set up a default handler. If so, logs will be
-            # printed twice unless we prevent propagation here.
-            "ray": {
-                "level": "INFO",
-                "handlers": ["default"],
-                "propagate": False,
-            },
-            # Special handling for ray.rllib: only warning-level messages passed through
-            # See https://github.com/ray-project/ray/pull/31858 for related PR
-            "ray.rllib": {
-                "level": "WARN",
-            },
-        }
-
-        dictConfig(
-            {
-                "version": 1,
-                "formatters": formatters,
-                "handlers": handlers,
-                "loggers": loggers,
-                "disable_existing_loggers": False,
-            }
+        plain_formatter = logging.Formatter(
+            "%(asctime)s\t%(levelname)s %(filename)s:%(lineno)s -- %(message)s"
         )
+
+        default_handler = PlainRayHandler()
+        default_handler.setFormatter(plain_formatter)
+
+        ray_logger = logging.getLogger("ray")
+        ray_logger.setLevel(logging.INFO)
+        ray_logger.addHandler(default_handler)
+        ray_logger.propagate = False
+
+        # Special handling for ray.rllib: only warning-level messages passed through
+        # See https://github.com/ray-project/ray/pull/31858 for related PR
+        rllib_logger = logging.getLogger("ray.rllib")
+        rllib_logger.setLevel(logging.WARN)
diff --git a/python/ray/_private/ray_logging/__init__.py b/python/ray/_private/ray_logging/__init__.py
index 5699251e31ce..768bc716c7c3 100644
--- a/python/ray/_private/ray_logging/__init__.py
+++ b/python/ray/_private/ray_logging/__init__.py
@@ -1,6 +1,7 @@
 import colorama
 from dataclasses import dataclass
 import logging
+import logging.handlers
 import os
 import re
 import sys
diff --git a/python/ray/_private/ray_logging/default_impl.py b/python/ray/_private/ray_logging/default_impl.py
index 2d50673c2851..3901baaa956f 100644
--- a/python/ray/_private/ray_logging/default_impl.py
+++ b/python/ray/_private/ray_logging/default_impl.py
@@ -1,4 +1,4 @@
-def get_dict_config_provider():
-    from ray._private.ray_logging.logging_config import DefaultDictConfigProvider
+def get_logging_configurator():
+    from ray._private.ray_logging.logging_config import DefaultLoggingConfigurator
 
-    return DefaultDictConfigProvider()
+    return DefaultLoggingConfigurator()
diff --git a/python/ray/_private/ray_logging/logging_config.py b/python/ray/_private/ray_logging/logging_config.py
index 6788571f9e6a..87cf819abbcf 100644
--- a/python/ray/_private/ray_logging/logging_config.py
+++ b/python/ray/_private/ray_logging/logging_config.py
@@ -12,68 +12,47 @@
 import time
 
 
-class DictConfigProvider(ABC):
+class LoggingConfigurator(ABC):
     @abstractmethod
     def get_supported_encodings(self) -> Set[str]:
         raise NotImplementedError
 
     @abstractmethod
-    def get_dict_config(self, encoding: str, log_level: str) -> dict:
+    def configure_logging(self, encoding: str, log_level: str):
         raise NotImplementedError
 
 
-class DefaultDictConfigProvider(DictConfigProvider):
+class DefaultLoggingConfigurator(LoggingConfigurator):
     def __init__(self):
-        self._dict_configs = {
-            "TEXT": lambda log_level: {
-                "version": 1,
-                "disable_existing_loggers": False,
-                "formatters": {
-                    "text": {
-                        "()": (
-                            f"{TextFormatter.__module__}."
-                            f"{TextFormatter.__qualname__}"
-                        ),
-                    },
-                },
-                "filters": {
-                    "core_context": {
-                        "()": (
-                            f"{CoreContextFilter.__module__}."
-                            f"{CoreContextFilter.__qualname__}"
-                        ),
-                    },
-                },
-                "handlers": {
-                    "console": {
-                        "level": log_level,
-                        "class": "logging.StreamHandler",
-                        "formatter": "text",
-                        "filters": ["core_context"],
-                    },
-                },
-                "root": {
-                    "level": log_level,
-                    "handlers": ["console"],
-                },
-                "loggers": {
-                    "ray": {
-                        "level": log_level,
-                        "handlers": ["console"],
-                        "propagate": False,
-                    }
-                },
-            }
+        self._encoding_to_formatter = {
+            "TEXT": TextFormatter(),
         }
 
     def get_supported_encodings(self) -> Set[str]:
-        return self._dict_configs.keys()
+        return self._encoding_to_formatter.keys()
 
-    def get_dict_config(self, encoding: str, log_level: str) -> dict:
-        return self._dict_configs[encoding](log_level)
+    def configure_logging(self, encoding: str, log_level: str):
+        formatter = self._encoding_to_formatter[encoding]
+        core_context_filter = CoreContextFilter()
+        handler = logging.StreamHandler()
+        handler.setLevel(log_level)
+        handler.setFormatter(formatter)
+        handler.addFilter(core_context_filter)
 
+        root_logger = logging.getLogger()
+        root_logger.setLevel(log_level)
+        root_logger.addHandler(handler)
 
-_dict_config_provider: DictConfigProvider = default_impl.get_dict_config_provider()
+        ray_logger = logging.getLogger("ray")
+        ray_logger.setLevel(log_level)
+        # Remove all existing handlers added by `ray/__init__.py`.
+        for h in ray_logger.handlers[:]:
+            ray_logger.removeHandler(h)
+        ray_logger.addHandler(handler)
+        ray_logger.propagate = False
+
+
+_logging_configurator: LoggingConfigurator = default_impl.get_logging_configurator()
 
 
 @PublicAPI(stability="alpha")
@@ -84,19 +63,16 @@ class LoggingConfig:
     log_level: str = "INFO"
 
     def __post_init__(self):
-        if self.encoding not in _dict_config_provider.get_supported_encodings():
+        if self.encoding not in _logging_configurator.get_supported_encodings():
             raise ValueError(
                 f"Invalid encoding type: {self.encoding}. "
                 "Valid encoding types are: "
-                f"{list(_dict_config_provider.get_supported_encodings())}"
+                f"{list(_logging_configurator.get_supported_encodings())}"
             )
 
-    def _get_dict_config(self) -> dict:
-        """Get the logging configuration based on the encoding type.
-        Returns:
-            dict: The logging configuration.
-        """
-        return _dict_config_provider.get_dict_config(self.encoding, self.log_level)
+    def _configure_logging(self):
+        """Set up the logging configuration for the current process."""
+        _logging_configurator.configure_logging(self.encoding, self.log_level)
 
     def _setup_log_record_factory(self):
         old_factory = logging.getLogRecordFactory()
@@ -121,7 +97,7 @@ def record_factory(*args, **kwargs):
     def _apply(self):
         """Set up both the LogRecord factory and the logging configuration."""
         self._setup_log_record_factory()
-        logging.config.dictConfig(self._get_dict_config())
+        self._configure_logging()
 
 
 LoggingConfig.__doc__ = f"""
@@ -153,7 +129,7 @@ def f():
 
     Args:
         encoding: Encoding type for the logs. The valid values are
-            {list(_dict_config_provider.get_supported_encodings())}
+            {list(_logging_configurator.get_supported_encodings())}
         log_level: Log level for the logs. Defaults to 'INFO'. You can set
             it to 'DEBUG' to receive more detailed debug logs.
     """  # noqa: E501


Test patch:
diff --git a/python/ray/tests/test_logging_2.py b/python/ray/tests/test_logging_2.py
index 9d5be165f9ac..988ed482e2bc 100644
--- a/python/ray/tests/test_logging_2.py
+++ b/python/ray/tests/test_logging_2.py
@@ -207,17 +207,9 @@ def test_record_with_exception(self):
             assert s in formatted
 
 
-class TestLoggingConfig:
-    def test_log_level(self):
-        log_level = "DEBUG"
-        logging_config = LoggingConfig(log_level=log_level)
-        dict_config = logging_config._get_dict_config()
-        assert dict_config["handlers"]["console"]["level"] == log_level
-        assert dict_config["root"]["level"] == log_level
-
-    def test_invalid_dict_config(self):
-        with pytest.raises(ValueError):
-            LoggingConfig(encoding="INVALID")._get_dict_config()
+def test_invalid_encoding():
+    with pytest.raises(ValueError):
+        LoggingConfig(encoding="INVALID")
 
 
 class TestTextModeE2E:
@@ -477,6 +469,27 @@ def print_message(self):
     assert "(MyActor pid=" not in stderr
 
 
+def test_configure_both_structured_logging_and_lib_logging(shutdown_only):
+    """
+    Configure the `ray.test` logger. Then, configure the `root` and `ray`
+    loggers in `ray.init()`. Ensure that the `ray.test` logger is not affected.
+    """
+    script = """
+import ray
+import logging
+
+old_test_logger = logging.getLogger("ray.test")
+assert old_test_logger.getEffectiveLevel() != logging.DEBUG
+old_test_logger.setLevel(logging.DEBUG)
+
+ray.init(logging_config=ray.LoggingConfig(encoding="TEXT", log_level="INFO"))
+
+new_test_logger = logging.getLogger("ray.test")
+assert old_test_logger.getEffectiveLevel() == logging.DEBUG
+"""
+    run_string_as_driver(script)
+
+
 if __name__ == "__main__":
     if os.environ.get("PARALLEL_CI"):
         sys.exit(pytest.main(["-n", "auto", "--boxed", "-vs", __file__]))