[Data] - Preserve existing handlers for logging - don't use dictConfig (#57169)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?

So the way the logger works is log() -> buffer -> flush() -> target
(file/console/remote_uri)
On shutdown (triggered by logging.config.dictConfig), the MemoryHandler
is closed which sets the output target to None
When the target is set to None, the logs in the buffer have nowhere to
be flushed, so they're retained in memory, hence it accumulates to no
end.

Solution:
Temporarily detach the handlers for loggers (preserve attributes like
`level`, `target`, `filter` and `formatters`) and restore them after
calling `dictConfig()`

Based off of
https://github.com/ray-project/ray/pull/48958/files#diff-d5e32f872cf7b123f0899714728ac7fc0916a5609844a7d1bf7317d98094778c
(Ray Core - Ray Data log fix)


## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [x] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [x] I've run pre-commit jobs to lint the changes in this PR.
([pre-commit
setup](https://docs.ray.io/en/latest/ray-contribute/getting-involved.html#lint-and-formatting))
- [ ] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [x] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: Goutam V. <goutam@anyscale.com>
Signed-off-by: Goutam <goutam@anyscale.com>

Author: goutamvenkat-anyscale

python/ray/data/_internal/logging.py

@@ -180,20 +180,124 @@ def _get_logger_names() -> List[str]:
 def configure_logging() -> None:
     """Configure the Python logger named 'ray.data'.
 
-    This function loads the configration YAML specified by "RAY_DATA_LOGGING_CONFIG"
+    This function loads the configuration YAML specified by "RAY_DATA_LOGGING_CONFIG"
     environment variable. If the variable isn't set, this function loads the default
     "logging.yaml" file that is adjacent to this module.
 
     If "RAY_DATA_LOG_ENCODING" is specified as "JSON" we will enable JSON logging mode
     if using the default logging config.
     """
+    config = _get_logging_config()
 
-    # Dynamically load env vars
+    # Create formatters, filters, and handlers from config
+    formatters = _create_formatters(config)
+    filters = _create_filters(config)
+    handlers = _create_handlers(config, formatters, filters)
+
+    # Configure each logger defined in the config
+    _configure_loggers(config, handlers)
+
+    # Warn if both env vars are set (incompatible)
+    _warn_if_incompatible_env_vars()
+
+
+def _import_class(class_path: str):
+    """Dynamically import a class from a fully qualified path."""
+    import importlib
+
+    if "." not in class_path:
+        raise ValueError(f"Invalid class path: {class_path}")
+
+    module_name, class_name = class_path.rsplit(".", 1)
+    module = importlib.import_module(module_name)
+    return getattr(module, class_name)
+
+
+def _create_formatters(config: dict) -> dict:
+    """Create formatter instances from config."""
+    formatters = {}
+
+    for name, fmt_config in config.get("formatters", {}).items():
+        if "class" in fmt_config:
+            formatter_class = _import_class(fmt_config["class"])
+            formatters[name] = formatter_class()
+        elif "format" in fmt_config:
+            formatters[name] = logging.Formatter(fmt_config["format"])
+
+    return formatters
+
+
+def _create_filters(config: dict) -> dict:
+    """Create filter instances from config."""
+    filters = {}
+
+    for name, filter_config in config.get("filters", {}).items():
+        # https://docs.python.org/3/library/logging.config.html#dictionary-schema-details
+        if "()" in filter_config:
+            filter_class = _import_class(filter_config["()"])
+            filters[name] = filter_class()
+
+    return filters
+
+
+def _create_handlers(config: dict, formatters: dict, filters: dict) -> dict:
+    """Create and configure handler instances from config."""
+    handlers = {}
+
+    # Keys that are not passed to handler constructor
+    HANDLER_CONFIG_KEYS = {"class", "level", "formatter", "filters"}
+
+    for name, handler_config in config.get("handlers", {}).items():
+        # Instantiate handler with all keys except config-only keys
+        handler_class = _import_class(handler_config["class"])
+        handler_kwargs = {
+            k: v for k, v in handler_config.items() if k not in HANDLER_CONFIG_KEYS
+        }
+        handler = handler_class(**handler_kwargs)
+        handler.name = name
+
+        # Configure handler
+        if "level" in handler_config:
+            handler.setLevel(handler_config["level"])
+
+        if "formatter" in handler_config:
+            formatter = formatters.get(handler_config["formatter"])
+            if formatter:
+                handler.setFormatter(formatter)
+
+        for filter_name in handler_config.get("filters", []):
+            filter_obj = filters.get(filter_name)
+            if filter_obj:
+                handler.addFilter(filter_obj)
+
+        handlers[name] = handler
+
+    return handlers
+
+
+def _configure_loggers(config: dict, handlers: dict) -> None:
+    """Configure logger instances from config."""
+    for logger_name, logger_config in config.get("loggers", {}).items():
+        logger = logging.getLogger(logger_name)
+        logger.setLevel(logger_config.get("level", logging.NOTSET))
+
+        # Clear existing handlers
+        for handler in logger.handlers[:]:
+            logger.removeHandler(handler)
+
+        # Add configured handlers
+        for handler_name in logger_config.get("handlers", []):
+            handler = handlers.get(handler_name)
+            if handler:
+                logger.addHandler(handler)
+
+        logger.propagate = logger_config.get("propagate", True)
+
+
+def _warn_if_incompatible_env_vars() -> None:
+    """Warn if both RAY_DATA_LOGGING_CONFIG and RAY_DATA_LOG_ENCODING are set."""
     config_path = os.environ.get(RAY_DATA_LOGGING_CONFIG_ENV_VAR_NAME)
     log_encoding = os.environ.get(RAY_DATA_LOG_ENCODING_ENV_VAR_NAME)
-    config = _get_logging_config()
-
-    logging.config.dictConfig(config)
 
     # After configuring logger, warn if RAY_DATA_LOGGING_CONFIG is used with
     # RAY_DATA_LOG_ENCODING, because they are not both supported together.
@@ -330,5 +434,4 @@ def unregister_dataset_logger(dataset_id: str) -> Optional[int]:
         for logger in loggers:
             logger.removeHandler(log_handler)
         log_handler.close()
-
     return _ACTIVE_DATASET

python/ray/data/tests/test_logging.py

@@ -167,6 +167,47 @@ def test_json_logging_configuration(
     assert "turkey" not in console_log_output
 
 
+def test_configure_logging_preserves_existing_handlers(reset_logging, shutdown_only):
+    """Test that configure_logging() preserves existing handlers.
+
+    When configure_logging() is called, it should not remove existing handlers
+    like MemoryHandler that were added to loggers before configuration.
+    """
+    ray.init()
+
+    # Create a logger and add a MemoryHandler with a target before configuring Ray Data logging
+    test_logger = logging.getLogger("ray.serve.test_preserve")
+    target_handler = logging.StreamHandler()
+    memory_handler = logging.handlers.MemoryHandler(capacity=100, target=target_handler)
+    test_logger.addHandler(memory_handler)
+
+    try:
+        # Verify the memory handler is there and target is set
+        assert memory_handler in test_logger.handlers
+        assert memory_handler.target is not None
+        assert memory_handler.target is target_handler
+
+        # Configure Ray Data logging
+        configure_logging()
+
+        # Verify the memory handler is still present after configuration
+        assert memory_handler in test_logger.handlers
+
+        # Verify the target is still set (would be None if handler was closed/recreated)
+        assert memory_handler.target is not None
+        assert memory_handler.target is target_handler
+
+        # Verify the memory handler still works
+        test_logger.info("test message")
+        assert len(memory_handler.buffer) == 1
+        assert "test message" in memory_handler.buffer[0].getMessage()
+    finally:
+        # Clean up handlers to avoid logging errors during teardown
+        test_logger.removeHandler(memory_handler)
+        memory_handler.close()
+        target_handler.close()
+
+
 if __name__ == "__main__":
     import sys