Repository: dbt-labs/dbt-core
Pull request: Output run hooks (#696)
Fixes #696 

Output information about the on-run-start and on-run-end hooks before and after running dbt.

One wrinkle, the output isn't exactly as described in the issue - it's a bit tricky to move that `Concurrency` line before hooks run. Instead you get this order.

```
Running with dbt=0.13.0
Found 163 macros, 0 analyses, 0 archives, 0 sources, 0 seed files, 0 tests, 2 operations, 1 models

07:58:48 |
07:58:48 | Running 1 on-run-start hook
07:58:48 | 1 of 1 START select * from information_schema.schemata
07:58:48 |
07:58:48 | Concurrency: 2 threads (target='default')
07:58:48 |
07:58:48 | 1 of 1 START view model dbt_postgres.x............................... [RUN]
07:58:48 | 1 of 1 OK created view model dbt_postgres.x.......................... [CREATE VIEW in 0.08s]
07:58:48 |
07:58:48 | Running 1 on-run-end hook
07:58:48 | 1 of 1 START select * from information_schema.schemata
07:58:48 |
07:58:48 |
07:58:48 | Finished running 2 hooks, 1 view models in 0.45s.

Completed successfully

Done. PASS=1 ERROR=0 SKIP=0 TOTAL=1
```

We definitely can move that `Concurrency` line up, and the output would probably look a bit nicer, but it involves moving that into the task level instead of the node runner level and rearranging some things.

Problem statement:
show on-run-start/end + hooks in dbt output
### Feature

dbt doesn't output any info about the start/end hooks that it runs, so it can be hard to understand why the stdout seems to stall. Instead, dbt should log the start/end hooks that it is running to the console.

Example:
```
$ dbt run

Found 3 models, 2 tests, 0 archives, 0 analyses, 96 macros, 2 operations, 0 seed files

12:27:16 | Concurrency: 8 threads (target='dev')
12:27:16 |
12:27:16 | Running 3 on-run-start hooks:
12:27:16 | 1 of 3 START grant usage on ...
12:27:16 | 2 of 3 START grant create on ...
12:27:16 | 3 of 3 START insert into public._audit ...
12:27:16 |
12:27:16 | Running 2 models:
12:27:16 | 1 of 2 START table model demo_schema.dep............................. [RUN]
12:27:16 | 2 of 2 START incremental model demo_schema.local..................... [RUN]
12:27:16 | 1 of 2 OK created table model demo_schema.dep........................ [SELECT 1 in 0.13s]
12:27:16 | 2 of 2 OK created incremental model demo_schema.local................ [INSERT 0 1 in 0.14s]
12:27:16 |
12:27:16 | Running 3 on-run-end hooks:
12:27:16 | 1 of 3 START grant usage on ...
12:27:16 | 2 of 3 START grant create on ...
12:27:16 | 3 of 3 START insert into public._audit ...
12:27:16 |
12:27:16 | Finished running 1 table models, 1 incremental models, 6 run hooks in 0.32s.

Completed successfully

Done. PASS=2 ERROR=0 SKIP=0 TOTAL=2
```

If there are no on-run-start/end hooks, dbt should skip those sections of the info logs. Note that we should also add something like `6 run hooks` to the final line of model run logs.

^ this is an initial idea of how this could look, but I don't know that I _love_ it. Very open to other ideas!

Hints from issue discussion:
I'd like to take a stab at this.
Cool @vijaykiran! Let me know if there's anything I can do to help out when you give it a shot :) 
This looks good

Commit message: add color to hooks, give hooks names

Patch to review:
diff --git a/core/dbt/logger.py b/core/dbt/logger.py
index f658769172f..9840d8c8fc8 100644
--- a/core/dbt/logger.py
+++ b/core/dbt/logger.py
@@ -67,6 +67,10 @@ def notice(self, msg, *args, **kwargs):
 
 # provide this for the cache.
 CACHE_LOGGER = logging.getLogger('dbt.cache')
+# add a dummy handler to avoid `No handlers could be found for logger`
+nothing_handler = logging.StreamHandler()
+nothing_handler.setLevel(CRITICAL)
+CACHE_LOGGER.addHandler(nothing_handler)
 # provide this for RPC connection logging
 RPC_LOGGER = logging.getLogger('dbt.rpc')
 
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index 3e7903a3736..fceebe1aa85 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -1,27 +1,67 @@
 from __future__ import print_function
 
+import time
+
 from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.node_types import NodeType, RunHookType
 from dbt.node_runners import ModelRunner, RPCExecuteRunner
 
 import dbt.exceptions
 import dbt.flags
-import dbt.ui.printer
 from dbt.contracts.graph.parsed import Hook
 from dbt.hooks import get_hook_dict
+from dbt.ui.printer import \
+    print_hook_start_line, \
+    print_hook_end_line, \
+    print_timestamped_line, \
+    print_run_end_messages, \
+    get_counts
 
 from dbt.compilation import compile_node
 from dbt.task.compile import CompileTask, RemoteCompileTask
 from dbt.utils import get_nodes_by_tags
 
 
+class Timer(object):
+    def __init__(self):
+        self.start = None
+        self.end = None
+
+    @property
+    def elapsed(self):
+        if self.start is None or self.end is None:
+            return None
+        return self.end - self.start
+
+    def __enter__(self):
+        self.start = time.time()
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_tracebck):
+        self.end = time.time()
+
+
 class RunTask(CompileTask):
+    def __init__(self, args, config):
+        super(RunTask, self).__init__(args, config)
+        self.ran_hooks = []
+
     def raise_on_first_error(self):
         return False
 
     def populate_adapter_cache(self, adapter):
         adapter.set_relations_cache(self.manifest)
 
+    def get_hook_sql(self, adapter, hook, idx, num_hooks, extra_context):
+        compiled = compile_node(adapter, self.config, hook, self.manifest,
+                                extra_context)
+        statement = compiled.wrapped_sql
+        hook_index = hook.get('index', num_hooks)
+        hook_dict = get_hook_dict(statement, index=hook_index)
+        if dbt.flags.STRICT_MODE:
+            Hook(**hook_dict)
+        return hook_dict.get('sql', '')
+
     def run_hooks(self, adapter, hook_type, extra_context):
 
         nodes = self.manifest.nodes.values()
@@ -33,22 +73,35 @@ def run_hooks(self, adapter, hook_type, extra_context):
         # b/c psycopg2 automatically begins a transaction when a connection
         # is created.
         adapter.clear_transaction()
+        if not ordered_hooks:
+            return
+        num_hooks = len(ordered_hooks)
+
+        plural = 'hook' if num_hooks == 1 else 'hooks'
+        print_timestamped_line("")
+        print_timestamped_line(
+            'Running {} {} {}'.format(num_hooks, hook_type, plural)
+        )
 
-        for i, hook in enumerate(ordered_hooks):
-            compiled = compile_node(adapter, self.config, hook,
-                                    self.manifest, extra_context)
-            statement = compiled.wrapped_sql
+        for idx, hook in enumerate(ordered_hooks, start=1):
+            sql = self.get_hook_sql(adapter, hook, idx, num_hooks,
+                                    extra_context)
 
-            hook_index = hook.get('index', len(hooks))
-            hook_dict = get_hook_dict(statement, index=hook_index)
+            # convert all whitespace sequences to single spaces
+            hook_text = ' '.join(sql.split())
+            hook_text = '{}.{}.{}'.format(hook.package_name, hook_type, idx)
+            print_hook_start_line(hook_text, idx, num_hooks)
 
-            if dbt.flags.STRICT_MODE:
-                Hook(**hook_dict)
+            with Timer() as timer:
+                if len(sql.strip()) > 0:
+                    status, _ = adapter.execute(sql, auto_begin=False,
+                                                fetch=False)
+            self.ran_hooks.append(hook)
 
-            sql = hook_dict.get('sql', '')
+            print_hook_end_line(hook_text, status, idx, num_hooks,
+                                timer.elapsed)
 
-            if len(sql.strip()) > 0:
-                adapter.execute(sql, auto_begin=False, fetch=False)
+        print_timestamped_line("")
 
     def safe_run_hooks(self, adapter, hook_type, extra_context):
         try:
@@ -57,10 +110,9 @@ def safe_run_hooks(self, adapter, hook_type, extra_context):
             logger.info("Database error while running {}".format(hook_type))
             raise
 
-    @classmethod
-    def print_results_line(cls, results, execution_time):
-        nodes = [r.node for r in results]
-        stat_line = dbt.ui.printer.get_counts(nodes)
+    def print_results_line(self, results, execution_time):
+        nodes = [r.node for r in results] + self.ran_hooks
+        stat_line = get_counts(nodes)
 
         execution = ""
 
@@ -68,8 +120,8 @@ def print_results_line(cls, results, execution_time):
             execution = " in {execution_time:0.2f}s".format(
                 execution_time=execution_time)
 
-        dbt.ui.printer.print_timestamped_line("")
-        dbt.ui.printer.print_timestamped_line(
+        print_timestamped_line("")
+        print_timestamped_line(
             "Finished running {stat_line}{execution}."
             .format(stat_line=stat_line, execution=execution))
 
@@ -107,7 +159,7 @@ def get_runner_type(self):
 
     def task_end_messages(self, results):
         if results:
-            dbt.ui.printer.print_run_end_messages(results)
+            print_run_end_messages(results)
 
 
 class RemoteRunTask(RemoteCompileTask, RunTask):
diff --git a/core/dbt/ui/printer.py b/core/dbt/ui/printer.py
index cc02bd50d7a..8bbdd294ae8 100644
--- a/core/dbt/ui/printer.py
+++ b/core/dbt/ui/printer.py
@@ -49,7 +49,8 @@ def print_timestamped_line(msg, use_color=None):
     logger.info("{} | {}".format(get_timestamp(), msg))
 
 
-def print_fancy_output_line(msg, status, index, total, execution_time=None):
+def print_fancy_output_line(msg, status, index, total, execution_time=None,
+                            truncate=False):
     if index is None or total is None:
         progress = ''
     else:
@@ -60,6 +61,8 @@ def print_fancy_output_line(msg, status, index, total, execution_time=None):
         message=msg)
 
     justified = prefix.ljust(80, ".")
+    if truncate and len(justified) > 77:
+        justified = justified[:77] + '...'
 
     if execution_time is None:
         status_time = ""
@@ -83,6 +86,8 @@ def get_counts(flat_nodes):
 
         if node.get('resource_type') == NodeType.Model:
             t = '{} {}'.format(get_materialization(node), t)
+        elif node.get('resource_type') == NodeType.Operation:
+            t = 'hook'
 
         counts[t] = counts.get(t, 0) + 1
 
@@ -97,6 +102,18 @@ def print_start_line(description, index, total):
     print_fancy_output_line(msg, 'RUN', index, total)
 
 
+def print_hook_start_line(statement, index, total):
+    msg = 'START hook: {}'.format(statement)
+    print_fancy_output_line(msg, 'RUN', index, total, truncate=True)
+
+
+def print_hook_end_line(statement, status, index, total, execution_time):
+    msg = 'OK hook: {}'.format(statement)
+    # hooks don't fail into this path, so always green
+    print_fancy_output_line(msg, green(status), index, total,
+                            execution_time=execution_time, truncate=True)
+
+
 def print_skip_line(model, schema, relation, index, num_models):
     msg = 'SKIP relation {}.{}'.format(schema, relation)
     print_fancy_output_line(msg, yellow('SKIP'), index, num_models)


Code review comments:
diff_hunk: @@ -33,22 +73,35 @@ def run_hooks(self, adapter, hook_type, extra_context):
         # b/c psycopg2 automatically begins a transaction when a connection
         # is created.
         adapter.clear_transaction()
+        if not ordered_hooks:
+            return
+        num_hooks = len(ordered_hooks)
+
+        plural = 'hook' if num_hooks == 1 else 'hooks'
+        print_timestamped_line("")
+        print_timestamped_line(
+            'Running {} {} {}'.format(num_hooks, hook_type, plural)
+        )
 
-        for i, hook in enumerate(ordered_hooks):
-            compiled = compile_node(adapter, self.config, hook,
-                                    self.manifest, extra_context)
-            statement = compiled.wrapped_sql
+        for idx, hook in enumerate(ordered_hooks, start=1):
+            sql = self.get_hook_sql(adapter, hook, idx, num_hooks,
+                                    extra_context)
 
-            hook_index = hook.get('index', len(hooks))
-            hook_dict = get_hook_dict(statement, index=hook_index)
+            # convert all whitespace sequences to single spaces
+            hook_text = ' '.join(sql.split())
path: core/dbt/task/run.py
text: @user1:
```suggestion
```