add log-format flag (#1237)

core/dbt/logger.py

@@ -236,6 +236,7 @@ def __init__(
  logbook.Handler.__init__(self, level, filter, bubble)
  if log_dir is not None:
  self.set_path(log_dir)
+ self._text_format_string = None
 
  def reset(self):
  if self.initialized:

core/dbt/main.py

@@ -180,7 +180,7 @@ def run_from_args(parsed):
  log_cache_events(getattr(parsed, 'log_cache_events', False))
  flags.set_from_args(parsed)
 
- parsed.cls.pre_init_hook()
+ parsed.cls.pre_init_hook(parsed)
  # we can now use the logger for stdout
 
  logger.info("Running with dbt{}".format(dbt.version.installed))
@@ -740,6 +740,13 @@ def parse_args(args):
  '''
  )
 
+ p.add_argument(
+ '--log-format',
+ choices=['text', 'json', None],
+ default=None,
+ help='''Specify the log format, overriding the command's default.'''
+ )
+
  p.add_argument(
  '--no-write-json',
  action='store_false',

core/dbt/task/base.py

@@ -5,7 +5,7 @@
 from dbt.config import RuntimeConfig, Project
 from dbt.config.profile import read_profile, PROFILES_DIR
 from dbt import tracking
-from dbt.logger import GLOBAL_LOGGER as logger
+from dbt.logger import GLOBAL_LOGGER as logger, log_manager
 import dbt.exceptions
 
 
@@ -45,8 +45,12 @@ def __init__(self, args, config):
  self.config = config
 
  @classmethod
- def pre_init_hook(cls):
+ def pre_init_hook(cls, args):
  """A hook called before the task is initialized."""
+ if args.log_format == 'text' or args.log_format is None:
+ log_manager.format_text()
+ else:
+ log_manager.format_json()
 
  @classmethod
  def from_args(cls, args):

core/dbt/task/list.py

@@ -43,9 +43,10 @@ def __init__(self, args, config):
  )
 
  @classmethod
- def pre_init_hook(cls):
+ def pre_init_hook(cls, args):
  """A hook called before the task is initialized."""
  log_manager.stderr_console()
+ super().pre_init_hook(args)
 
  def _iterate_selected_nodes(self):
  nodes = sorted(self.select_nodes())

core/dbt/task/rpc_server.py

@@ -94,6 +94,8 @@ def signhup_replace():
 
 
 class RPCServerTask(ConfiguredTask):
+ DEFAULT_LOG_FORMAT = 'json'
+
  def __init__(self, args, config, tasks=None):
  if os.name == 'nt':
  raise RuntimeException(
@@ -106,6 +108,14 @@ def __init__(self, args, config, tasks=None):
  self._reload_task_manager()
  signal.signal(signal.SIGHUP, self._sighup_handler)
 
+ @classmethod
+ def pre_init_hook(cls, args):
+ """A hook called before the task is initialized."""
+ if args.log_format == 'text':
+ log_manager.format_text()
+ else:
+ log_manager.format_json()
+
  def _reload_task_manager(self):
  """This function can only be running once at a time, as it runs in the
  signal handler we replace
@@ -144,8 +154,7 @@ def _default_tasks():
  def single_threaded(self):
  return SINGLE_THREADED_WEBSERVER or self.args.single_threaded
 
- def run(self):
- log_manager.format_json()
+ def run_forever(self):
  host = self.args.host
  port = self.args.port
  addr = (host, port)
@@ -154,7 +163,6 @@ def run(self):
  if host == '0.0.0.0':
  display_host = 'localhost'
 
- ServerContext().push_application()
  logger.info(
  'Serving RPC server at {}:{}, pid={}'.format(
  *addr, os.getpid()
@@ -182,6 +190,10 @@ def run(self):
  # fast.
  run_simple(host, port, app, threaded=not self.single_threaded)
 
+ def run(self):
+ with ServerContext().applicationbound():
+ self.run_forever()
+
  @Request.application
  def handle_jsonrpc_request(self, request):
  with HTTPRequest(request):