diff --git a/python/ray/autoscaler/aws/utils.py b/python/ray/autoscaler/aws/utils.py
index 312bfd8c0..98cab90c5 100644
--- a/python/ray/autoscaler/aws/utils.py
+++ b/python/ray/autoscaler/aws/utils.py
@@ -88,22 +88,28 @@ def handle_boto_error(exc, msg, *args, **kwargs):
         cli_logger.verbose_error(*generic_message_args)
         cli_logger.verbose(vars(exc))
 
-        cli_logger.abort(
-            "Your AWS session has expired.\n\n"
-            "You can request a new one using\n{}\n"
-            "then expose it to Ray by setting\n{}\n{}\n{}\n\n"
-            "You can find a script that automates this at:\n{}",
-            cf.bold(token_command), cf.bold(secret_key_var),
-            cf.bold(session_token_var), cf.bold(access_key_id_var),
-            cf.underlined(aws_session_script_url))
+        cli_logger.panic("Your AWS session has expired.")
+        cli_logger.newline()
+        cli_logger.panic("You can request a new one using")
+        cli_logger.panic(cf.bold(token_command))
+        cli_logger.panic("then expose it to Ray by setting")
+        cli_logger.panic(cf.bold(secret_key_var))
+        cli_logger.panic(cf.bold(session_token_var))
+        cli_logger.panic(cf.bold(access_key_id_var))
+        cli_logger.newline()
+        cli_logger.panic("You can find a script that automates this at:")
+        cli_logger.panic(cf.underlined(aws_session_script_url))
+        # Do not re-raise the exception here because it looks awful
+        # and we already print all the info in verbose
+        cli_logger.abort()
 
     # todo: any other errors that we should catch separately?
 
-    cli_logger.error(*generic_message_args)
+    cli_logger.panic(*generic_message_args)
     cli_logger.newline()
     with cli_logger.verbatim_error_ctx("Boto3 error:"):
         cli_logger.verbose("{}", str(vars(exc)))
-        cli_logger.error("{}", str(exc))
+        cli_logger.panic("{}", str(exc))
     cli_logger.abort()
 
 
diff --git a/python/ray/autoscaler/cli_logger.py b/python/ray/autoscaler/cli_logger.py
index 22bd8b5ef..bd10686b8 100644
--- a/python/ray/autoscaler/cli_logger.py
+++ b/python/ray/autoscaler/cli_logger.py
@@ -61,7 +61,7 @@ def _patched_makeRecord(self,
 logging.Logger.makeRecord = _patched_makeRecord
 
 
-def _parent_frame_info():
+def _external_caller_info():
     """Get the info from the caller frame.
 
     Used to override the logging function and line number with the correct
@@ -69,11 +69,14 @@ def _parent_frame_info():
     """
 
     frame = inspect.currentframe()
-    # we are also in a function, so must go 2 levels up
-    caller = frame.f_back.f_back
+    caller = frame
+    levels = 0
+    while caller.f_code.co_filename == __file__:
+        caller = caller.f_back
+        levels += 1
     return {
         "lineno": caller.f_lineno,
-        "filename": os.path.basename(caller.f_code.co_filename),
+        "filename": os.path.basename(caller.f_code.co_filename)
     }
 
 
@@ -159,26 +162,43 @@ def _format_msg(msg: str,
     return ", ".join(res)
 
 
+# TODO: come up with a plan to unify logging.
+# formatter = logging.Formatter(
+#     # TODO(maximsmol): figure out the required log level padding
+#     #                  width automatically
+#     fmt="[{asctime}] {levelname:6} {message}",
+#     datefmt="%x %X",
+#     # We want alignment on our level names
+#     style="{")
+
+
+def _isatty():
+    """More robust check for interactive terminal/tty."""
+    try:
+        # https://stackoverflow.com/questions/6108330/
+        # checking-for-interactive-shell-in-a-python-script
+        return sys.__stdin__.isatty()
+    except Exception:
+        # sometimes this can fail due to closed output
+        # either way, no-tty is generally safe fallback.
+        return False
+
+
 class _CliLogger():
     """Singleton class for CLI logging.
 
     Attributes:
-        strip (bool):
-            If `strip` is `True`, all TTY control sequences will be
-            removed from the output.
         old_style (bool):
             If `old_style` is `True`, the old logging calls are used instead
-            of the new CLI UX. This is enabled by default and remains for
-            backwards compatibility.
+            of the new CLI UX. This is disabled by default and remains for
+            backwards compatibility. Currently can only be set via env var
+            RAY_LOG_NEWSTYLE="0".
         color_mode (str):
             Can be "true", "false", or "auto".
 
-            Determines the value of `strip` using a human-readable string
-            that can be set from command line arguments.
+            Enables or disables `colorful`.
 
-            Also affects the `colorful` settings.
-
-            If `color_mode` is "auto", `strip` is set to `not stdout.isatty()`
+            If `color_mode` is "auto", is set to `not stdout.isatty()`
         indent_level (int):
             The current indentation level.
 
@@ -188,26 +208,71 @@ class _CliLogger():
 
             Low verbosity will disable `verbose` and `very_verbose` messages.
     """
-    strip: bool
     old_style: bool
     color_mode: str
     # color_mode: Union[Literal["auto"], Literal["false"], Literal["true"]]
     indent_level: int
-    verbosity: int
+    interactive: bool
+    VALID_LOG_STYLES = ("auto", "record", "pretty")
 
     _autodetected_cf_colormode: int
 
     def __init__(self):
-        self.old_style = True
-        self.color_mode = "auto"
+        self.old_style = os.environ.get("RAY_LOG_NEWSTYLE", "1") == "0"
+        self.pretty = True
+        self.interactive = _isatty()
         self.indent_level = 0
 
-        self.verbosity = 0
+        self._verbosity = 0
+        self._color_mode = "auto"
+        self._log_style = "auto"
 
         # store whatever colorful has detected for future use if
         # the color ouput is toggled (colorful detects # of supported colors,
         # so it has some non-trivial logic to determine this)
         self._autodetected_cf_colormode = cf.colorful.colormode
+        self.set_format()
+
+    def set_format(self, format_tmpl=None):
+        if not format_tmpl:
+            import ray.ray_constants as ray_constants
+            format_tmpl = ray_constants.LOGGER_FORMAT
+        self._formatter = logging.Formatter(format_tmpl)
+
+    @property
+    def log_style(self):
+        return self._log_style
+
+    @log_style.setter
+    def log_style(self, x):
+        self._log_style = x.lower()
+
+        if self._log_style == "auto":
+            self.pretty = _isatty()
+        elif self._log_style == "record":
+            self.pretty = False
+            self.color_mode = "false"
+        elif self._log_style == "pretty":
+            self.pretty = True
+
+    @property
+    def color_mode(self):
+        return self._color_mode
+
+    @color_mode.setter
+    def color_mode(self, x):
+        self._color_mode = x.lower()
+        self.detect_colors()
+
+    @property
+    def verbosity(self):
+        if not self.pretty:
+            return 999
+        return self._verbosity
+
+    @verbosity.setter
+    def verbosity(self, x):
+        self._verbosity = x
 
     def detect_colors(self):
         """Update color output settings.
@@ -216,7 +281,6 @@ class _CliLogger():
         color output
         (8-color ANSI if no terminal detected to be safe) in colorful.
         """
-        self.color_mode = self.color_mode.lower()
         if self.color_mode == "true":
             if self._autodetected_cf_colormode != cf.NO_COLORS:
                 cf.colormode = self._autodetected_cf_colormode
@@ -237,7 +301,10 @@ class _CliLogger():
         """
         self.print("")
 
-    def _print(self, msg: str, linefeed: bool = True):
+    def _print(self,
+               msg: str,
+               _level_str: str = "INFO",
+               _linefeed: bool = True):
         """Proxy for printing messages.
 
         Args:
@@ -246,13 +313,31 @@ class _CliLogger():
                 If `linefeed` is `False` no linefeed is printed at the
                 end of the message.
         """
+        if self.pretty:
+            rendered_message = "  " * self.indent_level + msg
+        else:
+            if msg.strip() == "":
+                return
+            caller_info = _external_caller_info()
+            record = logging.LogRecord(
+                name="cli",
+                # We override the level name later
+                # TODO(maximsmol): give approximate level #s to our log levels
+                level=0,
+                # The user-facing logs do not need this information anyway
+                # and it would be very tedious to extract since _print
+                # can be at varying depths in the call stack
+                # TODO(maximsmol): do it anyway to be extra
+                pathname=caller_info["filename"],
+                lineno=caller_info["lineno"],
+                msg=msg,
+                args={},
+                # No exception
+                exc_info=None)
+            record.levelname = _level_str
+            rendered_message = self._formatter.format(record)
 
-        if self.old_style:
-            return
-
-        rendered_message = "  " * self.indent_level + msg
-
-        if not linefeed:
+        if not _linefeed:
             sys.stdout.write(rendered_message)
             sys.stdout.flush()
             return
@@ -335,7 +420,7 @@ class _CliLogger():
         For arguments, see `_format_msg`.
         """
         if self.verbosity > 0:
-            self.print(msg, *args, **kwargs)
+            self.print(msg, *args, _level_str="VINFO", **kwargs)
 
     def verbose_warning(self, msg, *args, **kwargs):
         """Prints a formatted warning if verbosity is not 0.
@@ -343,7 +428,7 @@ class _CliLogger():
         For arguments, see `_format_msg`.
         """
         if self.verbosity > 0:
-            self.warning(msg, *args, **kwargs)
+            self._warning(msg, *args, _level_str="VWARN", **kwargs)
 
     def verbose_error(self, msg: str, *args: Any, **kwargs: Any):
         """Logs an error if verbosity is not 0.
@@ -351,7 +436,7 @@ class _CliLogger():
         For arguments, see `_format_msg`.
         """
         if self.verbosity > 0:
-            self.error(msg, *args, **kwargs)
+            self._error(msg, *args, _level_str="VERR", **kwargs)
 
     def very_verbose(self, msg: str, *args: Any, **kwargs: Any):
         """Prints if verbosity is > 1.
@@ -359,30 +444,56 @@ class _CliLogger():
         For arguments, see `_format_msg`.
         """
         if self.verbosity > 1:
-            self.print(msg, *args, **kwargs)
+            self.print(msg, *args, _level_str="VVINFO", **kwargs)
 
     def success(self, msg: str, *args: Any, **kwargs: Any):
         """Prints a formatted success message.
 
         For arguments, see `_format_msg`.
         """
-        self.print(cf.limeGreen(msg), *args, **kwargs)
+        self.print(cf.limeGreen(msg), *args, _level_str="SUCC", **kwargs)
 
-    def warning(self, msg: str, *args: Any, **kwargs: Any):
+    def _warning(self,
+                 msg: str,
+                 *args: Any,
+                 _level_str: str = None,
+                 **kwargs: Any):
         """Prints a formatted warning message.
 
         For arguments, see `_format_msg`.
         """
-        self.print(cf.orange(msg), *args, **kwargs)
+        if _level_str is None:
+            raise ValueError("Log level not set.")
+        self.print(cf.orange(msg), *args, _level_str=_level_str, **kwargs)
 
-    def error(self, msg: str, *args: Any, **kwargs: Any):
+    def warning(self, *args, **kwargs):
+        self._warning(*args, _level_str="WARN", **kwargs)
+
+    def _error(self,
+               msg: str,
+               *args: Any,
+               _level_str: str = None,
+               **kwargs: Any):
         """Prints a formatted error message.
 
         For arguments, see `_format_msg`.
         """
-        self.print(cf.red(msg), *args, **kwargs)
+        if _level_str is None:
+            raise ValueError("Log level not set.")
+        self.print(cf.red(msg), *args, _level_str=_level_str, **kwargs)
 
-    def print(self, msg: str, *args: Any, **kwargs: Any):
+    def error(self, *args, **kwargs):
+        self._error(*args, _level_str="ERR", **kwargs)
+
+    def panic(self, *args, **kwargs):
+        self._error(*args, _level_str="PANIC", **kwargs)
+
+    # Fine to expose _level_str here, since this is a general log function.
+    def print(self,
+              msg: str,
+              *args: Any,
+              _level_str: str = "INFO",
+              **kwargs: Any):
         """Prints a message.
 
         For arguments, see `_format_msg`.
@@ -390,9 +501,13 @@ class _CliLogger():
         if self.old_style:
             return
 
-        self._print(_format_msg(msg, *args, **kwargs))
+        self._print(_format_msg(msg, *args, **kwargs), _level_str=_level_str)
 
-    def abort(self, msg: Optional[str] = None, *args: Any, **kwargs: Any):
+    def abort(self,
+              msg: Optional[str] = None,
+              *args: Any,
+              exc: Any = None,
+              **kwargs: Any):
         """Prints an error and aborts execution.
 
         Print an error and throw an exception to terminate the program
@@ -402,9 +517,15 @@ class _CliLogger():
             return
 
         if msg is not None:
-            self.error(msg, *args, **kwargs)
+            self._error(msg, *args, _level_str="PANIC", **kwargs)
 
-        raise SilentClickException("Exiting due to cli_logger.abort()")
+        if exc is not None:
+            raise exc
+
+        exc_cls = click.ClickException
+        if self.pretty:
+            exc_cls = SilentClickException
+        raise exc_cls("Exiting due to cli_logger.abort()")
 
     def doassert(self, val: bool, msg: str, *args: Any, **kwargs: Any):
         """Handle assertion without throwing a scary exception.
@@ -418,7 +539,15 @@ class _CliLogger():
             return
 
         if not val:
-            self.abort(msg, *args, **kwargs)
+            exc = None
+            if not self.pretty:
+                exc = AssertionError()
+
+            # TODO(maximsmol): rework asserts so that we get the expression
+            #                  that triggered the assert
+            #                  to do this, install a global try-catch
+            #                  for AssertionError and raise them normally
+            self.abort(msg, *args, exc=exc, **kwargs)
 
     def old_debug(self, logger: logging.Logger, msg: str, *args: Any,
                   **kwargs: Any):
@@ -435,7 +564,8 @@ class _CliLogger():
         """
         if self.old_style:
             logger.debug(
-                _format_msg(msg, *args, **kwargs), extra=_parent_frame_info())
+                _format_msg(msg, *args, **kwargs),
+                extra=_external_caller_info())
             return
 
     def old_info(self, logger: logging.Logger, msg: str, *args: Any,
@@ -453,7 +583,8 @@ class _CliLogger():
         """
         if self.old_style:
             logger.info(
-                _format_msg(msg, *args, **kwargs), extra=_parent_frame_info())
+                _format_msg(msg, *args, **kwargs),
+                extra=_external_caller_info())
             return
 
     def old_warning(self, logger: logging.Logger, msg: str, *args: Any,
@@ -471,7 +602,8 @@ class _CliLogger():
         """
         if self.old_style:
             logger.warning(
-                _format_msg(msg, *args, **kwargs), extra=_parent_frame_info())
+                _format_msg(msg, *args, **kwargs),
+                extra=_external_caller_info())
             return
 
     def old_error(self, logger: logging.Logger, msg: str, *args: Any,
@@ -489,7 +621,8 @@ class _CliLogger():
         """
         if self.old_style:
             logger.error(
-                _format_msg(msg, *args, **kwargs), extra=_parent_frame_info())
+                _format_msg(msg, *args, **kwargs),
+                extra=_external_caller_info())
             return
 
     def old_exception(self, logger: logging.Logger, msg: str, *args: Any,
@@ -507,7 +640,8 @@ class _CliLogger():
         """
         if self.old_style:
             logger.exception(
-                _format_msg(msg, *args, **kwargs), extra=_parent_frame_info())
+                _format_msg(msg, *args, **kwargs),
+                extra=_external_caller_info())
             return
 
     def render_list(self, xs: List[str], separator: str = cf.reset(", ")):
@@ -536,13 +670,18 @@ class _CliLogger():
                 The default action to take if the user just presses enter
                 with no input.
         """
-
         if self.old_style:
             return
 
         should_abort = _abort
         default = _default
 
+        if not self.interactive and not yes:
+            # no formatting around --yes here since this is non-interactive
+            self.error("This command requires user confirmation. "
+                       "When running non-interactively, supply --yes to skip.")
+            raise ValueError("Non-interactive confirm without --yes.")
+
         if default:
             yn_str = cf.limeGreen("Y") + "/" + cf.red("n")
         else:
@@ -563,7 +702,7 @@ class _CliLogger():
                         cf.dimmed("[automatic, due to --yes]"))
             return True
 
-        self._print(complete_str, linefeed=False)
+        self._print(complete_str, _linefeed=False)
 
         res = None
         yes_answers = ["y", "yes", "true", "1"]
@@ -590,7 +729,7 @@ class _CliLogger():
                            "Expected {} or {}", indent, cf.bold(ans.strip()),
                            self.render_list(yes_answers, "/"),
                            self.render_list(no_answers, "/"))
-                self._print(indent + confirm_str, linefeed=False)
+                self._print(indent + confirm_str, _linefeed=False)
         except KeyboardInterrupt:
             self.newline()
             res = default
diff --git a/python/ray/autoscaler/command_runner.py b/python/ray/autoscaler/command_runner.py
index 2a15d528d..6adf114c9 100644
--- a/python/ray/autoscaler/command_runner.py
+++ b/python/ray/autoscaler/command_runner.py
@@ -445,6 +445,7 @@ class SSHCommandRunner(CommandRunnerInterface):
             if not cli_logger.old_style and not with_output:
                 return run_cmd_redirected(
                     final_cmd,
+                    process_runner=self.process_runner,
                     silent=silent,
                     use_login_shells=is_using_login_shells())
             if with_output:
diff --git a/python/ray/autoscaler/commands.py b/python/ray/autoscaler/commands.py
index ffbd211f7..801aed273 100644
--- a/python/ray/autoscaler/commands.py
+++ b/python/ray/autoscaler/commands.py
@@ -128,7 +128,8 @@ def create_or_update_cluster(config_file: str,
         cli_logger.newline()
 
     def handle_yaml_error(e):
-        cli_logger.error("Cluster config invalid\n")
+        cli_logger.error("Cluster config invalid")
+        cli_logger.newline()
         cli_logger.error("Failed to load YAML file " + cf.bold("{}"),
                          config_file)
         cli_logger.newline()
@@ -164,7 +165,7 @@ def create_or_update_cluster(config_file: str,
         raise NotImplementedError("Unsupported provider {}".format(
             config["provider"]))
 
-    cli_logger.success("Cluster configuration valid\n")
+    cli_logger.success("Cluster configuration valid")
 
     printed_overrides = False
 
diff --git a/python/ray/autoscaler/subprocess_output_util.py b/python/ray/autoscaler/subprocess_output_util.py
index 2d9f101d2..067707a1c 100644
--- a/python/ray/autoscaler/subprocess_output_util.py
+++ b/python/ray/autoscaler/subprocess_output_util.py
@@ -173,10 +173,13 @@ def _read_subprocess_stream(f, output_file, is_stdout=False):
 
 def _run_and_process_output(cmd,
                             stdout_file,
+                            process_runner=subprocess,
                             stderr_file=None,
                             use_login_shells=False):
     """Run a command and process its output for special cases.
 
+    Calls a standard 'check_call' if process_runner is not subprocess.
+
     Specifically, run all command output through regex to detect
     error conditions and filter out non-error messages that went to stderr
     anyway (SSH writes ALL of its "system" messages to stderr even if they
@@ -184,6 +187,8 @@ def _run_and_process_output(cmd,
 
     Args:
         cmd (List[str]): Command to run.
+        process_runner: Used for command execution. Assumed to have
+            'check_call' and 'check_output' inplemented.
         stdout_file: File to redirect stdout to.
         stderr_file: File to redirect stderr to.
 
@@ -227,17 +232,18 @@ def _run_and_process_output(cmd,
     # This already should be validated in a higher place of the stack.
     assert not (does_allow_interactive() and is_output_redirected()), (
         "Cannot redirect output while in interactive mode.")
-    if does_allow_interactive() and not is_output_redirected():
+    if process_runner != subprocess or (does_allow_interactive()
+                                        and not is_output_redirected()):
         stdin_overwrite = None
 
     # See implementation note #1
 
-    if use_login_shells:
+    if use_login_shells or process_runner != subprocess:
         if stdout_file is None:
             stdout_file = sys.stdout
         if stderr_file is None:
             stderr_file = sys.stderr
-        return subprocess.check_call(
+        return process_runner.check_call(
             cmd,
             # See implementation note #2
             stdin=stdin_overwrite,
@@ -310,11 +316,15 @@ def _run_and_process_output(cmd,
             return p.returncode
 
 
-def run_cmd_redirected(cmd, silent=False, use_login_shells=False):
+def run_cmd_redirected(cmd,
+                       process_runner=subprocess,
+                       silent=False,
+                       use_login_shells=False):
     """Run a command and optionally redirect output to a file.
 
     Args:
         cmd (List[str]): Command to run.
+        process_runner: Process runner used for executing commands.
         silent (bool): If true, the command output will be silenced completely
                        (redirected to /dev/null), unless verbose logging
                        is enabled. Use this for runnign utility commands like
@@ -322,11 +332,17 @@ def run_cmd_redirected(cmd, silent=False, use_login_shells=False):
     """
     if silent and cli_logger.verbosity < 1:
         return _run_and_process_output(
-            cmd, stdout_file=None, use_login_shells=use_login_shells)
+            cmd,
+            process_runner=process_runner,
+            stdout_file=None,
+            use_login_shells=use_login_shells)
 
     if not is_output_redirected():
         return _run_and_process_output(
-            cmd, stdout_file=sys.stdout, use_login_shells=use_login_shells)
+            cmd,
+            process_runner=process_runner,
+            stdout_file=sys.stdout,
+            use_login_shells=use_login_shells)
     else:
         tmpfile_path = os.path.join(
             tempfile.gettempdir(), "ray-up-{}-{}.txt".format(
@@ -338,12 +354,10 @@ def run_cmd_redirected(cmd, silent=False, use_login_shells=False):
                 buffering=1) as tmp:
             cli_logger.verbose("Command stdout is redirected to {}",
                                cf.bold(tmp.name))
-            cli_logger.verbose(
-                cf.dimmed("Use --dump-command-output to "
-                          "dump to terminal instead."))
 
             return _run_and_process_output(
                 cmd,
+                process_runner=process_runner,
                 stdout_file=tmp,
                 stderr_file=tmp,
                 use_login_shells=use_login_shells)
diff --git a/python/ray/scripts/scripts.py b/python/ray/scripts/scripts.py
index 2fba05a38..02e7ddfe9 100644
--- a/python/ray/scripts/scripts.py
+++ b/python/ray/scripts/scripts.py
@@ -53,24 +53,21 @@ def check_no_existing_redis_clients(node_ip_address, redis_client):
 
 logging_options = [
     click.option(
-        "--log-new-style/--log-old-style",
-        is_flag=True,
-        default=True,
-        envvar="RAY_LOG_NEWSTYLE",
-        help=("Whether to use the old or the new CLI UX. "
-              "You can also toggle this via the env var RAY_LOG_NEWSTYLE. "
-              "The new UX supports colored, formatted output and was "
-              "designed to display only the most important information for "
-              "human users. The old UX uses the standard `logging` module. "
-              "It is most suitable for writing to a file and will include "
-              "timestamps and message level (ERROR/WARNING/INFO).")),
+        "--log-style",
+        required=False,
+        type=click.Choice(cli_logger.VALID_LOG_STYLES, case_sensitive=False),
+        default="auto",
+        help=("If 'pretty', outputs with formatting and color. If 'record', "
+              "outputs record-style without formatting. "
+              "'auto' defaults to 'pretty', and disables pretty logging "
+              "if stdin is *not* a TTY.")),
     click.option(
         "--log-color",
         required=False,
         type=click.Choice(["auto", "false", "true"], case_sensitive=False),
         default="auto",
         help=("Use color logging. "
-              "Valid values are: auto (if stdout is a tty), true, false.")),
+              "Auto enables color logging if stdout is a TTY.")),
     click.option("-v", "--verbose", count=True)
 ]
 
@@ -101,6 +98,7 @@ def add_click_options(options):
 def cli(logging_level, logging_format):
     level = logging.getLevelName(logging_level.upper())
     ray.utils.setup_logger(level, logging_format)
+    cli_logger.set_format(format_tmpl=logging_format)
 
 
 @click.command()
@@ -394,10 +392,10 @@ def start(node_ip_address, redis_address, address, redis_port, port,
           autoscaling_config, no_redirect_worker_output, no_redirect_output,
           plasma_store_socket_name, raylet_socket_name, temp_dir, include_java,
           java_worker_options, load_code_from_local, system_config, lru_evict,
-          enable_object_reconstruction, metrics_export_port, log_new_style,
+          enable_object_reconstruction, metrics_export_port, log_style,
           log_color, verbose):
     """Start Ray processes manually on the local machine."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -565,8 +563,7 @@ def start(node_ip_address, redis_address, address, redis_port, port,
             ray_params, head=True, shutdown_at_exit=block, spawn_reaper=block)
         redis_address = node.redis_address
 
-        # new-style CLI UX (--log-new-style)
-        # this is a noop if that flag is not set, so the old logger calls
+        # this is a noop if new-style is not set, so the old logger calls
         # are still in place
         cli_logger.newline()
         startup_msg = "Ray runtime started."
@@ -760,10 +757,10 @@ def start(node_ip_address, redis_address, address, redis_port, port,
     is_flag=True,
     help="If set, ray will send SIGKILL instead of SIGTERM.")
 @add_click_options(logging_options)
-def stop(force, verbose, log_new_style, log_color):
+def stop(force, verbose, log_style, log_color):
     """Stop Ray processes manually on the local machine."""
 
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -929,9 +926,9 @@ def stop(force, verbose, log_new_style, log_color):
 @add_click_options(logging_options)
 def up(cluster_config_file, min_workers, max_workers, no_restart, restart_only,
        yes, cluster_name, no_config_cache, redirect_command_output,
-       use_login_shells, log_new_style, log_color, verbose):
+       use_login_shells, log_style, log_color, verbose):
     """Create or update a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -995,9 +992,9 @@ def up(cluster_config_file, min_workers, max_workers, no_restart, restart_only,
     help="Retain the minimal amount of workers specified in the config.")
 @add_click_options(logging_options)
 def down(cluster_config_file, yes, workers_only, cluster_name,
-         keep_min_workers, log_new_style, log_color, verbose):
+         keep_min_workers, log_style, log_color, verbose):
     """Tear down a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -1047,10 +1044,10 @@ def kill_random_node(cluster_config_file, yes, hard, cluster_name):
     type=str,
     help="Override the configured cluster name.")
 @add_click_options(logging_options)
-def monitor(cluster_config_file, lines, cluster_name, log_new_style, log_color,
+def monitor(cluster_config_file, lines, cluster_name, log_style, log_color,
             verbose):
     """Tails the autoscaler logs of a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -1091,10 +1088,9 @@ def monitor(cluster_config_file, lines, cluster_name, log_new_style, log_color,
     help="Port to forward. Use this multiple times to forward multiple ports.")
 @add_click_options(logging_options)
 def attach(cluster_config_file, start, screen, tmux, cluster_name,
-           no_config_cache, new, port_forward, log_new_style, log_color,
-           verbose):
+           no_config_cache, new, port_forward, log_style, log_color, verbose):
     """Create or attach to a SSH session to a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -1122,10 +1118,10 @@ def attach(cluster_config_file, start, screen, tmux, cluster_name,
     type=str,
     help="Override the configured cluster name.")
 @add_click_options(logging_options)
-def rsync_down(cluster_config_file, source, target, cluster_name,
-               log_new_style, log_color, verbose):
+def rsync_down(cluster_config_file, source, target, cluster_name, log_style,
+               log_color, verbose):
     """Download specific files from a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -1151,9 +1147,9 @@ def rsync_down(cluster_config_file, source, target, cluster_name,
     help="Upload to all nodes (workers and head).")
 @add_click_options(logging_options)
 def rsync_up(cluster_config_file, source, target, cluster_name, all_nodes,
-             log_new_style, log_color, verbose):
+             log_style, log_color, verbose):
     """Upload specific files to a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -1213,8 +1209,8 @@ def rsync_up(cluster_config_file, source, target, cluster_name, all_nodes,
 @click.argument("script_args", nargs=-1)
 @add_click_options(logging_options)
 def submit(cluster_config_file, screen, tmux, stop, start, cluster_name,
-           no_config_cache, port_forward, script, args, script_args,
-           log_new_style, log_color, verbose):
+           no_config_cache, port_forward, script, args, script_args, log_style,
+           log_color, verbose):
     """Uploads and runs a script on the specified cluster.
 
     The script is automatically synced to the following location:
@@ -1224,7 +1220,7 @@ def submit(cluster_config_file, screen, tmux, stop, start, cluster_name,
     Example:
         >>> ray submit [CLUSTER.YAML] experiment.py -- --smoke-test
     """
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
@@ -1345,10 +1341,10 @@ def submit(cluster_config_file, screen, tmux, stop, start, cluster_name,
     help="Port to forward. Use this multiple times to forward multiple ports.")
 @add_click_options(logging_options)
 def exec(cluster_config_file, cmd, run_env, screen, tmux, stop, start,
-         cluster_name, no_config_cache, port_forward, log_new_style, log_color,
+         cluster_name, no_config_cache, port_forward, log_style, log_color,
          verbose):
     """Execute a command via SSH on a Ray cluster."""
-    cli_logger.old_style = not log_new_style
+    cli_logger.log_style = log_style
     cli_logger.color_mode = log_color
     cli_logger.verbosity = verbose
     cli_logger.detect_colors()
diff --git a/python/ray/tests/test_cli.py b/python/ray/tests/test_cli.py
index 22a13ef94..7eee46011 100644
--- a/python/ray/tests/test_cli.py
+++ b/python/ray/tests/test_cli.py
@@ -16,9 +16,6 @@ WARNING: IF YOU MOCK AWS, DON'T FORGET THE AWS_CREDENTIALS FIXTURE.
 
 Note: config cache does not work with AWS mocks since the AWS resource ids are
       randomized each time.
-
-Note: while not strictly necessary for setup commands e.g. ray up,
-      --log-new-style produces much cleaner output if the test fails.
 """
 import glob
 import sys
@@ -99,6 +96,7 @@ def _debug_check_line_by_line(result, expected_lines):
         if i >= len(expected_lines):
             i += 1
             print("!!!!!! Expected fewer lines")
+            print("\n".join(output_lines[i:]))
             break
 
         exp = expected_lines[i]
@@ -156,7 +154,8 @@ DEFAULT_TEST_CONFIG_PATH = str(
 def test_ray_start(configure_lang):
     runner = CliRunner()
     result = runner.invoke(
-        scripts.start, ["--head", "--log-new-style", "--log-color", "False"])
+        scripts.start,
+        ["--head", "--log-style=pretty", "--log-color", "False"])
     _die_on_error(runner.invoke(scripts.stop))
 
     _check_output_via_pattern("test_ray_start.txt", result)
@@ -186,11 +185,40 @@ def test_ray_up(configure_lang, _unlink_test_ssh_key, configure_aws):
         runner = CliRunner()
         result = runner.invoke(scripts.up, [
             DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "-y",
-            "--log-new-style", "--log-color", "False"
+            "--log-style=pretty", "--log-color", "False"
         ])
         _check_output_via_pattern("test_ray_up.txt", result)
 
 
+@pytest.mark.skipif(
+    sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
+    reason=("Mac builds don't provide proper locale support"))
+@mock_ec2
+@mock_iam
+def test_ray_up_record(configure_lang, _unlink_test_ssh_key, configure_aws):
+    def commands_mock(command, stdin):
+        # if we want to have e.g. some commands fail,
+        # we can have overrides happen here.
+        # unfortunately, cutting out SSH prefixes and such
+        # is, to put it lightly, non-trivial
+        if "uptime" in command:
+            return PopenBehaviour(stdout="MOCKED uptime")
+        if "rsync" in command:
+            return PopenBehaviour(stdout="MOCKED rsync")
+        if "ray" in command:
+            return PopenBehaviour(stdout="MOCKED ray")
+        return PopenBehaviour(stdout="MOCKED GENERIC")
+
+    with _setup_popen_mock(commands_mock):
+        # config cache does not work with mocks
+        runner = CliRunner()
+        result = runner.invoke(scripts.up, [
+            DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "-y",
+            "--log-style=record"
+        ])
+        _check_output_via_pattern("test_ray_up_record.txt", result)
+
+
 @pytest.mark.skipif(
     sys.platform == "darwin" and "travis" in os.environ.get("USER", ""),
     reason=("Mac builds don't provide proper locale support"))
@@ -207,13 +235,13 @@ def test_ray_attach(configure_lang, configure_aws, _unlink_test_ssh_key):
         runner = CliRunner()
         result = runner.invoke(scripts.up, [
             DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "-y",
-            "--log-new-style", "--log-color", "False"
+            "--log-style=pretty", "--log-color", "False"
         ])
         _die_on_error(result)
 
         result = runner.invoke(scripts.attach, [
-            DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "--log-new-style",
-            "--log-color", "False"
+            DEFAULT_TEST_CONFIG_PATH, "--no-config-cache",
+            "--log-style=pretty", "--log-color", "False"
         ])
 
         _check_output_via_pattern("test_ray_attach.txt", result)
@@ -235,13 +263,13 @@ def test_ray_exec(configure_lang, configure_aws, _unlink_test_ssh_key):
         runner = CliRunner()
         result = runner.invoke(scripts.up, [
             DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "-y",
-            "--log-new-style", "--log-color", "False"
+            "--log-style=pretty", "--log-color", "False"
         ])
         _die_on_error(result)
 
         result = runner.invoke(scripts.exec, [
-            DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "--log-new-style",
-            "\"echo This is a test!\""
+            DEFAULT_TEST_CONFIG_PATH, "--no-config-cache",
+            "--log-style=pretty", "\"echo This is a test!\""
         ])
 
         _check_output_via_pattern("test_ray_exec.txt", result)
@@ -268,7 +296,7 @@ def test_ray_submit(configure_lang, configure_aws, _unlink_test_ssh_key):
         runner = CliRunner()
         result = runner.invoke(scripts.up, [
             DEFAULT_TEST_CONFIG_PATH, "--no-config-cache", "-y",
-            "--log-new-style", "--log-color", "False"
+            "--log-style=pretty", "--log-color", "False"
         ])
         _die_on_error(result)
 
@@ -279,7 +307,7 @@ def test_ray_submit(configure_lang, configure_aws, _unlink_test_ssh_key):
                 [
                     DEFAULT_TEST_CONFIG_PATH,
                     "--no-config-cache",
-                    "--log-new-style",
+                    "--log-style=pretty",
                     "--log-color",
                     "False",
                     # this is somewhat misleading, since the file
diff --git a/python/ray/tests/test_cli_patterns/test_ray_up.txt b/python/ray/tests/test_cli_patterns/test_ray_up.txt
index 5f41dfd04..5804f5825 100644
--- a/python/ray/tests/test_cli_patterns/test_ray_up.txt
+++ b/python/ray/tests/test_cli_patterns/test_ray_up.txt
@@ -3,7 +3,6 @@ Thus, the output from subcommands will be logged as is\.
 Consider using --use-normal-shells, if you tested your workflow and it is compatible\.
 
 Cluster configuration valid
-
 Cluster: test-cli
 
 Bootstrapping AWS config
diff --git a/python/ray/tests/test_cli_patterns/test_ray_up_record.txt b/python/ray/tests/test_cli_patterns/test_ray_up_record.txt
new file mode 100644
index 000000000..8021628ca
--- /dev/null
+++ b/python/ray/tests/test_cli_patterns/test_ray_up_record.txt
@@ -0,0 +1,78 @@
+.+\.py.*Commands running under a login shell can produce more output than special processing can handle\.
+.+\.py.*Thus, the output from subcommands will be logged as is\.
+.+\.py.*Consider using --use-normal-shells, if you tested your workflow and it is compatible\.
+.+\.py.*Cluster configuration valid
+.+\.py.*Cluster: test-cli
+.+\.py.*Bootstrapping AWS config
+.+\.py.*Creating new IAM instance profile ray-autoscaler-v1 for use as the default\.
+.+\.py.*Creating new IAM role ray-autoscaler-v1 for use as the default instance role\.
+.+\.py.*Creating new key pair __test-cli_key-1 for use as the default\.
+.+\.py.*Created new security group ray-autoscaler-test-cli \[id=sg-.+\]
+.+\.py.*AWS config
+.+\.py.*IAM Profile: ray-autoscaler-v1 \[default\]
+.+\.py.*EC2 Key pair \(head & workers\): __test-cli_key-1 \[default\]
+.+\.py.*VPC Subnets \(head & workers\): subnet-.+ \[default\]
+.+\.py.*EC2 Security groups \(head & workers\): sg-.+ \[default\]
+.+\.py.*EC2 AMI \(head & workers\): ami-dca37ea5 \[dlami\]
+.+\.py.*No head node found\. Launching a new cluster\. Confirm \[y/N\]: y \[automatic, due to --yes\]
+.+\.py.*Acquiring an up-to-date head node
+.+\.py.*Launched 1 nodes \[subnet_id=subnet-.+\]
+.+\.py.*Launched instance i-.+ \[state=pending, info=pending\]
+.+\.py.*Launched a new head node
+.+\.py.*Fetching the new head node
+.+\.py.*<1/1> Setting up head node
+.+\.py.*Prepared bootstrap config
+.+\.py.*New status: waiting-for-ssh
+.+\.py.*\[1/6\] Waiting for SSH to become available
+.+\.py.*Running `uptime` as a test\.
+.+\.py.*Fetched IP: .+
+.+\.py.*Running `uptime`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*Success\.
+.+\.py.*Updating cluster configuration\. \[hash=.+\]
+.+\.py.*New status: syncing-files
+.+\.py.*\[3/6\] Processing file mounts
+.+\.py.*Running `mkdir -p ~/tests`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*Running `rsync.+`
+.+\.py.*`rsync`ed \./ \(local\) to ~/tests/ \(remote\)
+.+\.py.*~/tests/ from \./
+.+\.py.*Running `mkdir -p ~`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*Running `rsync.+`
+.+\.py.*`rsync`ed .+/ray-bootstrap-.+ \(local\) to ~/ray_bootstrap_config\.yaml \(remote\)
+.+\.py.*~/ray_bootstrap_config\.yaml from .+ray-bootstrap-.+
+.+\.py.*Running `mkdir -p ~`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*Running `rsync.+`
+.+\.py.*`rsync`ed .+__test-cli_key-1\.pem \(local\) to ~/ray_bootstrap_key\.pem \(remote\)
+.+\.py.*~/ray_bootstrap_key\.pem from .+__test-cli_key-1\.pem
+.+\.py.*\[4/6\] No worker file mounts to sync
+.+\.py.*New status: setting-up
+.+\.py.*\[3/5\] Running initialization commands
+.+\.py.*Running `echo init`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*\[4/6\] Running setup commands
+.+\.py.*\(0/4\) echo a
+.+\.py.*Running `echo a`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*\(1/4\) echo b
+.+\.py.*Running `echo b`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*\(2/4\) echo \${echo hi}
+.+\.py.*Running `echo \${echo hi}`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*\(3/4\) echo head
+.+\.py.*Running `echo head`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*\[6/6\] Starting the Ray runtime
+.+\.py.*Running `ray stop`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*Running `ray start --head --autoscaling-config=~/ray_bootstrap_config\.yaml`
+.+\.py.*Full command is `ssh.+`
+.+\.py.*New status: up-to-date
+.+\.py.*Useful commands
+.+\.py.*Monitor autoscaling with
+.+\.py.*  ray exec .+ 'tail -n 100 -f .+
+.+\.py.*Connect to a terminal on the cluster head
+.+\.py.*  ray attach .+
diff --git a/python/ray/tests/test_output.py b/python/ray/tests/test_output.py
index da3b479d3..347fc1686 100644
--- a/python/ray/tests/test_output.py
+++ b/python/ray/tests/test_output.py
@@ -40,16 +40,10 @@ def test_output():
     lines = outputs.split("\n")
     for line in lines:
         print(line)
-    assert len(lines) == 3, lines
-    logging_header = r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}\sINFO\s"
-    assert re.match(
-        logging_header + r"resource_spec.py:\d+ -- Starting Ray with [0-9\.]+ "
-        r"GiB memory available for workers and up to [0-9\.]+ GiB "
-        r"for objects. You can adjust these settings with .*?.", lines[0])
-    assert re.match(
-        logging_header +
-        r"services.py:\d+ -- View the Ray dashboard at .*?127.0.0.1:\d+?.*",
-        lines[1])
+    assert len(lines) == 8, lines
+    assert re.match(r".*Workers: .*GiB", lines[1])
+    assert re.match(r".*Objects: .*GiB", lines[2])
+    assert re.match(r".*Dashboard URL: http:.*", lines[6])
 
 
 if __name__ == "__main__":