[Data] Adding in better framework for substituting logging handlers (#…

…48056)

## Why are these changes needed?
This provides a more extensible framework around substituting logging
handlers within the ray data logging configuration. This will allow more
substitutions to be added with more ease and less cruft.

## Related issue number

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] 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
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: Matthew Owen <[email protected]>

python/ray/data/_internal/logging.py

@@ -11,6 +11,10 @@
     os.path.join(os.path.dirname(__file__), "logging.yaml")
 )
 
+# Dictionary of substitutions to be performed when using JSON mode. Handlers with names
+# corresponding to keys will be replaced by those corresponding to values.
+RAY_DATA_LOG_HANDLER_JSON_SUBSTITUTIONS = {"file": "file_json"}
+
 # Env. variable to specify the encoding of the file logs when using the default config.
 RAY_DATA_LOG_ENCODING_ENV_VAR_NAME = "RAY_DATA_LOG_ENCODING"
 
@@ -118,8 +122,12 @@ def _load_logging_config(config_path: str):
         config = _load_logging_config(DEFAULT_CONFIG_PATH)
         if log_encoding is not None and log_encoding.upper() == "JSON":
             for logger in config["loggers"].values():
-                logger["handlers"].remove("file")
-                logger["handlers"].append("file_json")
+                for (
+                    old_handler_name,
+                    new_handler_name,
+                ) in RAY_DATA_LOG_HANDLER_JSON_SUBSTITUTIONS.items():
+                    logger["handlers"].remove(old_handler_name)
+                    logger["handlers"].append(new_handler_name)
 
     logging.config.dictConfig(config)