diff --git a/python/ray/log_monitor.py b/python/ray/log_monitor.py
index 2cd6fc40a..84ed2ce65 100644
--- a/python/ray/log_monitor.py
+++ b/python/ray/log_monitor.py
@@ -14,8 +14,8 @@ from ray.services import get_port
 import ray.utils
 
 # Logger for this module. It should be configured at the entry point
-# into the program using Ray. Ray configures it by default automatically
-# using logging.basicConfig in its entry/init points.
+# into the program using Ray. Ray provides a default configuration at
+# entry/init points.
 logger = logging.getLogger(__name__)
 
 
@@ -154,9 +154,7 @@ if __name__ == "__main__":
         default=ray_constants.LOGGER_FORMAT,
         help=ray_constants.LOGGER_FORMAT_HELP)
     args = parser.parse_args()
-    logging.basicConfig(
-        level=logging.getLevelName(args.logging_level.upper()),
-        format=args.logging_format)
+    ray.utils.setup_logger(args.logging_level, args.logging_format)
 
     redis_ip_address = get_ip_address(args.redis_address)
     redis_port = get_port(args.redis_address)
diff --git a/python/ray/memory_monitor.py b/python/ray/memory_monitor.py
index a52f98d70..1cc42a9b6 100644
--- a/python/ray/memory_monitor.py
+++ b/python/ray/memory_monitor.py
@@ -55,11 +55,10 @@ class MemoryMonitor(object):
         self.last_checked = time.time()
         self.error_threshold = error_threshold
         if not psutil:
-            logger.warning(
-                "WARNING: Not monitoring node memory since `psutil` is not "
-                "installed. Install this with `pip install psutil` "
-                "(or ray[debug]) to enable debugging of memory-related "
-                "crashes.")
+            print("WARNING: Not monitoring node memory since `psutil` is not "
+                  "installed. Install this with `pip install psutil` "
+                  "(or ray[debug]) to enable debugging of memory-related "
+                  "crashes.")
 
     def raise_if_low_memory(self):
         if not psutil:
diff --git a/python/ray/monitor.py b/python/ray/monitor.py
index 782370298..0b6d05d40 100644
--- a/python/ray/monitor.py
+++ b/python/ray/monitor.py
@@ -17,7 +17,8 @@ import ray.gcs_utils
 import ray.utils
 import ray.ray_constants as ray_constants
 from ray.services import get_ip_address, get_port
-from ray.utils import binary_to_hex, binary_to_object_id, hex_to_binary
+from ray.utils import (binary_to_hex, binary_to_object_id, hex_to_binary,
+                       setup_logger)
 
 # Set up logging.
 logger = logging.getLogger(__name__)
@@ -358,8 +359,7 @@ if __name__ == "__main__":
         default=ray_constants.LOGGER_FORMAT,
         help=ray_constants.LOGGER_FORMAT_HELP)
     args = parser.parse_args()
-    level = logging.getLevelName(args.logging_level.upper())
-    logging.basicConfig(level=level, format=args.logging_format)
+    setup_logger(args.logging_level, args.logging_format)
 
     redis_ip_address = get_ip_address(args.redis_address)
     redis_port = get_port(args.redis_address)
diff --git a/python/ray/parameter.py b/python/ray/parameter.py
index 11086604f..f24bdfd6f 100644
--- a/python/ray/parameter.py
+++ b/python/ray/parameter.py
@@ -57,6 +57,9 @@ class RayParams(object):
             Store with hugetlbfs support. Requires plasma_directory.
         include_webui: Boolean flag indicating whether to start the web
             UI, which is a Jupyter notebook.
+        logging_level: Logging level, default will be logging.INFO.
+        logging_format: Logging format, default contains a timestamp,
+            filename, line number, and message. See ray_constants.py.
         plasma_store_socket_name (str): If provided, it will specify the socket
             name used by the plasma store.
         raylet_socket_name (str): If provided, it will specify the socket path
diff --git a/python/ray/ray_constants.py b/python/ray/ray_constants.py
index bc7fa76c2..7e9f5dc04 100644
--- a/python/ray/ray_constants.py
+++ b/python/ray/ray_constants.py
@@ -78,9 +78,9 @@ AUTOSCALER_HEARTBEAT_TIMEOUT_S = env_integer("AUTOSCALER_HEARTBEAT_TIMEOUT_S",
 # Max number of retries to AWS (default is 5, time increases exponentially)
 BOTO_MAX_RETRIES = env_integer("BOTO_MAX_RETRIES", 12)
 
-# Default logger format: only contains the message.
-LOGGER_FORMAT = "%(message)s"
-LOGGER_FORMAT_HELP = "The logging format. default='%(message)s'"
+LOGGER_FORMAT = (
+    "%(asctime)s\t%(levelname)s %(filename)s:%(lineno)s -- %(message)s")
+LOGGER_FORMAT_HELP = "The logging format. default='{}'".format(LOGGER_FORMAT)
 LOGGER_LEVEL = "info"
 LOGGER_LEVEL_CHOICES = ['debug', 'info', 'warning', 'error', 'critical']
 LOGGER_LEVEL_HELP = ("The logging level threshold, choices=['debug', 'info',"
diff --git a/python/ray/scripts/scripts.py b/python/ray/scripts/scripts.py
index 4bf1249c0..65b33423e 100644
--- a/python/ray/scripts/scripts.py
+++ b/python/ray/scripts/scripts.py
@@ -57,8 +57,7 @@ def check_no_existing_redis_clients(node_ip_address, redis_client):
     help=ray_constants.LOGGER_FORMAT_HELP)
 def cli(logging_level, logging_format):
     level = logging.getLevelName(logging_level.upper())
-    logging.basicConfig(level=level, format=logging_format)
-    logger.setLevel(level)
+    ray.utils.setup_logger(level, logging_format)
 
 
 @cli.command()
diff --git a/python/ray/services.py b/python/ray/services.py
index b0110998f..e7f801c09 100644
--- a/python/ray/services.py
+++ b/python/ray/services.py
@@ -55,8 +55,8 @@ RAYLET_EXECUTABLE = os.path.join(
     os.path.abspath(os.path.dirname(__file__)), "core/src/ray/raylet/raylet")
 
 # Logger for this module. It should be configured at the entry point
-# into the program using Ray. Ray configures it by default automatically
-# using logging.basicConfig in its entry/init points.
+# into the program using Ray. Ray provides a default configuration at
+# entry/init points.
 logger = logging.getLogger(__name__)
 
 ProcessInfo = collections.namedtuple("ProcessInfo", [
@@ -849,9 +849,9 @@ def start_ui(redis_address, stdout_file=None, stderr_file=None):
         logger.warning("Failed to start the UI, you may need to run "
                        "'pip install jupyter'.")
     else:
-        logger.info("\n" + "=" * 70)
-        logger.info("View the web UI at {}".format(webui_url))
-        logger.info("=" * 70 + "\n")
+        print("\n" + "=" * 70)
+        print("View the web UI at {}".format(webui_url))
+        print("=" * 70 + "\n")
         return webui_url, process_info
     return None, None
 
diff --git a/python/ray/signature.py b/python/ray/signature.py
index d155e55d8..41d174512 100644
--- a/python/ray/signature.py
+++ b/python/ray/signature.py
@@ -10,8 +10,8 @@ import logging
 from ray.utils import is_cython
 
 # Logger for this module. It should be configured at the entry point
-# into the program using Ray. Ray configures it by default automatically
-# using logging.basicConfig in its entry/init points.
+# into the program using Ray. Ray provides a default configuration at
+# entry/init points.
 logger = logging.getLogger(__name__)
 
 FunctionSignature = namedtuple("FunctionSignature", [
diff --git a/python/ray/tune/__init__.py b/python/ray/tune/__init__.py
index 17c5fa2db..1e341b265 100644
--- a/python/ray/tune/__init__.py
+++ b/python/ray/tune/__init__.py
@@ -2,8 +2,6 @@ from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
 
-import logging
-
 from ray.tune.error import TuneError
 from ray.tune.tune import run_experiments
 from ray.tune.experiment import Experiment
@@ -11,20 +9,6 @@ from ray.tune.registry import register_env, register_trainable
 from ray.tune.trainable import Trainable
 from ray.tune.suggest import grid_search, function, sample_from
 
-
-def _setup_logger():
-    logger = logging.getLogger("ray.tune")
-    handler = logging.StreamHandler()
-    handler.setFormatter(
-        logging.Formatter(
-            "%(asctime)s\t%(levelname)s %(filename)s:%(lineno)s -- %(message)s"
-        ))
-    logger.addHandler(handler)
-    logger.propagate = False
-
-
-_setup_logger()
-
 __all__ = [
     "Trainable",
     "TuneError",
diff --git a/python/ray/utils.py b/python/ray/utils.py
index 87e2ec121..b814fc86f 100644
--- a/python/ray/utils.py
+++ b/python/ray/utils.py
@@ -6,6 +6,7 @@ import binascii
 import functools
 import hashlib
 import inspect
+import logging
 import numpy as np
 import os
 import subprocess
@@ -276,6 +277,33 @@ def resources_from_resource_arguments(default_num_cpus, default_num_gpus,
     return resources
 
 
+_default_handler = None
+
+
+def setup_logger(logging_level, logging_format):
+    """Setup default logging for ray."""
+    logger = logging.getLogger("ray")
+    if type(logging_level) is str:
+        logging_level = logging.getLevelName(logging_level.upper())
+    logger.setLevel(logging_level)
+    global _default_handler
+    _default_handler = logging.StreamHandler()
+    _default_handler.setFormatter(logging.Formatter(logging_format))
+    logger.addHandler(_default_handler)
+    logger.propagate = False
+
+
+def try_update_handler(new_stream):
+    global _default_handler
+    logger = logging.getLogger("ray")
+    if _default_handler:
+        new_handler = logging.StreamHandler(stream=new_stream)
+        new_handler.setFormatter(_default_handler.formatter)
+        _default_handler.close()
+        _default_handler = new_handler
+        logger.addHandler(_default_handler)
+
+
 # This function is copied and modified from
 # https://github.com/giampaolo/psutil/blob/5bd44f8afcecbfb0db479ce230c790fc2c56569a/psutil/tests/test_linux.py#L132-L138  # noqa: E501
 def vmstat(stat):
diff --git a/python/ray/worker.py b/python/ray/worker.py
index 9aa0f8979..c50f50c07 100644
--- a/python/ray/worker.py
+++ b/python/ray/worker.py
@@ -38,12 +38,8 @@ from ray import ObjectID, DriverID, ActorID, ActorHandleID, ClientID, TaskID
 from ray import profiling
 from ray.function_manager import (FunctionActorManager, FunctionDescriptor)
 import ray.parameter
-from ray.utils import (
-    check_oversized_pickle,
-    is_cython,
-    random_string,
-    thread_safe_client,
-)
+from ray.utils import (check_oversized_pickle, is_cython, random_string,
+                       thread_safe_client, setup_logger, try_update_handler)
 
 SCRIPT_MODE = 0
 WORKER_MODE = 1
@@ -62,8 +58,8 @@ DEFAULT_ACTOR_METHOD_CPUS_SPECIFIED_CASE = 0
 DEFAULT_ACTOR_CREATION_CPUS_SPECIFIED_CASE = 1
 
 # Logger for this module. It should be configured at the entry point
-# into the program using Ray. Ray configures it by default automatically
-# using logging.basicConfig in its entry/init points.
+# into the program using Ray. Ray provides a default configuration at
+# entry/init points.
 logger = logging.getLogger(__name__)
 
 try:
@@ -1347,8 +1343,8 @@ def init(redis_address=None,
         configure_logging: True if allow the logging cofiguration here.
             Otherwise, the users may want to configure it by their own.
         logging_level: Logging level, default will be logging.INFO.
-        logging_format: Logging format, default will be "%(message)s"
-            which means only contains the message.
+        logging_format: Logging format, default contains a timestamp,
+            filename, line number, and message. See ray_constants.py.
         plasma_store_socket_name (str): If provided, it will specify the socket
             name used by the plasma store.
         raylet_socket_name (str): If provided, it will specify the socket path
@@ -1367,7 +1363,7 @@ def init(redis_address=None,
     """
 
     if configure_logging:
-        logging.basicConfig(level=logging_level, format=logging_format)
+        setup_logger(logging_level, logging_format)
 
     # Add the use_raylet option for backwards compatibility.
     if use_raylet is not None:
@@ -1837,6 +1833,7 @@ def connect(info,
                     worker.worker_id))
             sys.stdout = log_stdout_file
             sys.stderr = log_stderr_file
+            try_update_handler(sys.stderr)
             services.record_log_files_in_redis(
                 info["redis_address"],
                 info["node_ip_address"], [log_stdout_file, log_stderr_file],
diff --git a/python/ray/workers/default_worker.py b/python/ray/workers/default_worker.py
index 47c2532dc..ffb1d29f0 100644
--- a/python/ray/workers/default_worker.py
+++ b/python/ray/workers/default_worker.py
@@ -3,13 +3,13 @@ from __future__ import division
 from __future__ import print_function
 
 import argparse
-import logging
 import traceback
 
 import ray
 import ray.actor
 import ray.ray_constants as ray_constants
 import ray.tempfile_services as tempfile_services
+import ray.utils
 
 parser = argparse.ArgumentParser(
     description=("Parse addresses for the worker "
@@ -68,9 +68,7 @@ if __name__ == "__main__":
         "raylet_socket_name": args.raylet_name,
     }
 
-    logging.basicConfig(
-        level=logging.getLevelName(args.logging_level.upper()),
-        format=args.logging_format)
+    ray.utils.setup_logger(args.logging_level, args.logging_format)
 
     # Override the temporary directory.
     tempfile_services.set_temp_root(args.temp_dir)
diff --git a/test/runtest.py b/test/runtest.py
index 4c7310bd5..e081b596d 100644
--- a/test/runtest.py
+++ b/test/runtest.py
@@ -2386,34 +2386,41 @@ def test_global_state_api(shutdown_only):
     os.environ.get("RAY_USE_NEW_GCS") == "on",
     reason="New GCS API doesn't have a Python API yet.")
 def test_log_file_api(shutdown_only):
+    """Tests that stderr and stdout are redirected appropriately."""
     ray.init(num_cpus=1, redirect_worker_output=True)
 
-    message = "unique message"
+    message_1 = "unique message"
+    message_2 = "message unique"
 
     @ray.remote
     def f():
-        logger.info(message)
+        print(message_1, file=sys.stdout)
+        print(message_2, file=sys.stderr)
         # The call to sys.stdout.flush() seems to be necessary when using
         # the system Python 2.7 on Ubuntu.
         sys.stdout.flush()
+        sys.stderr.flush()
 
     ray.get(f.remote())
 
     # Make sure that the message appears in the log files.
     start_time = time.time()
-    found_message = False
+    found_message_1 = False
+    found_message_2 = False
     while time.time() - start_time < 10:
         log_files = ray.global_state.log_files()
         for ip, innerdict in log_files.items():
             for filename, contents in innerdict.items():
                 contents_str = "".join(contents)
-                if message in contents_str:
-                    found_message = True
-        if found_message:
+                if message_1 in contents_str:
+                    found_message_1 = True
+                if message_2 in contents_str:
+                    found_message_2 = True
+        if found_message_1 and found_message_2:
             break
         time.sleep(0.1)
 
-    assert found_message is True
+    assert found_message_1 and found_message_2
 
 
 @pytest.mark.skipif(