Initial structured logging work with fire_event (#4137)

add event type modeling and fire_event calls

core/dbt/events/README.md

@@ -0,0 +1,9 @@
+# Events Module
+
+The Events module is the implmentation for structured logging. These events represent both a programatic interface to dbt processes as well as human-readable messaging in one centralized place. The centralization allows for leveraging mypy to enforce interface invariants across all dbt events, and the distinct type layer allows for decoupling events and libraries such as loggers.
+
+# Using the Events Module
+The event module provides types that represent what is happening in dbt in `events.types`. These types are intended to represent an exhaustive list of all things happening within dbt that will need to be logged, streamed, or printed. To fire an event, `events.functions::fire_event` is the entry point to the module from everywhere in dbt.
+
+# Adding a New Event
+In `events.types` add a new class that represents the new event. This may be a simple class with no values, or it may be a dataclass with some values to construct downstream messaging. Only include the data necessary to construct this message within this class. You must extend all destinations (e.g. - if your log message belongs on the cli, extend `CliEventABC`) as well as the loglevel this event belongs to.

core/dbt/events/functions.py

@@ -0,0 +1,28 @@
+
+import dbt.logger as logger  # type: ignore # TODO eventually remove dependency on this logger
+from dbt.events.history import EVENT_HISTORY
+from dbt.events.types import CliEventABC, Event
+
+
+# top-level method for accessing the new eventing system
+# this is where all the side effects happen branched by event type
+# (i.e. - mutating the event history, printing to stdout, logging
+# to files, etc.)
+def fire_event(e: Event) -> None:
+    EVENT_HISTORY.append(e)
+    if isinstance(e, CliEventABC):
+        if e.level_tag() == 'test':
+            # TODO after implmenting #3977 send to new test level
+            logger.GLOBAL_LOGGER.debug(logger.timestamped_line(e.cli_msg()))
+        elif e.level_tag() == 'debug':
+            logger.GLOBAL_LOGGER.debug(logger.timestamped_line(e.cli_msg()))
+        elif e.level_tag() == 'info':
+            logger.GLOBAL_LOGGER.info(logger.timestamped_line(e.cli_msg()))
+        elif e.level_tag() == 'warn':
+            logger.GLOBAL_LOGGER.warning()(logger.timestamped_line(e.cli_msg()))
+        elif e.level_tag() == 'error':
+            logger.GLOBAL_LOGGER.error(logger.timestamped_line(e.cli_msg()))
+        else:
+            raise AssertionError(
+                f"Event type {type(e).__name__} has unhandled level: {e.level_tag()}"
+            )

core/dbt/events/history.py

@@ -0,0 +1,7 @@
+from dbt.events.types import Event
+from typing import List
+
+
+# the global history of events for this session
+# TODO this is naive and the memory footprint is likely far too large.
+EVENT_HISTORY: List[Event] = []

core/dbt/events/types.py

@@ -0,0 +1,124 @@
+from abc import ABCMeta, abstractmethod
+from dataclasses import dataclass
+
+
+# types to represent log levels
+
+# in preparation for #3977
+class TestLevel():
+    def level_tag(self) -> str:
+        return "test"
+
+
+class DebugLevel():
+    def level_tag(self) -> str:
+        return "debug"
+
+
+class InfoLevel():
+    def level_tag(self) -> str:
+        return "info"
+
+
+class WarnLevel():
+    def level_tag(self) -> str:
+        return "warn"
+
+
+class ErrorLevel():
+    def level_tag(self) -> str:
+        return "error"
+
+
+# The following classes represent the data necessary to describe a
+# particular event to both human readable logs, and machine reliable
+# event streams. classes extend superclasses that indicate what
+# destinations they are intended for, which mypy uses to enforce
+# that the necessary methods are defined.
+
+
+# top-level superclass for all events
+class Event(metaclass=ABCMeta):
+    # do not define this yourself. inherit it from one of the above level types.
+    @abstractmethod
+    def level_tag(self) -> str:
+        raise Exception("level_tag not implemented for event")
+
+
+class CliEventABC(Event, metaclass=ABCMeta):
+    # Solely the human readable message. Timestamps and formatting will be added by the logger.
+    @abstractmethod
+    def cli_msg(self) -> str:
+        raise Exception("cli_msg not implemented for cli event")
+
+
+class ParsingStart(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Start parsing."
+
+
+class ParsingCompiling(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Compiling."
+
+
+class ParsingWritingManifest(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Writing manifest."
+
+
+class ParsingDone(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Done."
+
+
+class ManifestDependenciesLoaded(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Dependencies loaded"
+
+
+class ManifestLoaderCreated(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "ManifestLoader created"
+
+
+class ManifestLoaded(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Manifest loaded"
+
+
+class ManifestChecked(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Manifest checked"
+
+
+class ManifestFlatGraphBuilt(InfoLevel, CliEventABC):
+    def cli_msg(self) -> str:
+        return "Flat graph built"
+
+
+@dataclass
+class ReportPerformancePath(InfoLevel, CliEventABC):
+    path: str
+
+    def cli_msg(self) -> str:
+        return f"Performance info: {self.path}"
+
+
+# since mypy doesn't run on every file we need to suggest to mypy that every
+# class gets instantiated. But we don't actually want to run this code.
+# making the conditional `if False` causes mypy to skip it as dead code so
+# we need to skirt around that by computing something it doesn't check statically.
+#
+# TODO remove these lines once we run mypy everywhere.
+if 1 == 0:
+    ParsingStart()
+    ParsingCompiling()
+    ParsingWritingManifest()
+    ParsingDone()
+    ManifestDependenciesLoaded()
+    ManifestLoaderCreated()
+    ManifestLoaded()
+    ManifestChecked()
+    ManifestFlatGraphBuilt()
+    ReportPerformancePath(path='')

core/dbt/logger.py

@@ -655,8 +655,12 @@ def get_timestamp():
     return time.strftime("%H:%M:%S")
 
 
+def timestamped_line(msg: str) -> str:
+    return "{} | {}".format(get_timestamp(), msg)
+
+
 def print_timestamped_line(msg: str, use_color: Optional[str] = None):
     if use_color is not None:
         msg = dbt.ui.color(msg, use_color)
 
-    GLOBAL_LOGGER.info("{} | {}".format(get_timestamp(), msg))
+    GLOBAL_LOGGER.info(timestamped_line(msg))

core/dbt/task/parse.py

@@ -11,8 +11,14 @@
 from dbt.parser.manifest import (
     Manifest, ManifestLoader, _check_manifest
 )
-from dbt.logger import DbtProcessState, print_timestamped_line
+from dbt.logger import DbtProcessState
 from dbt.clients.system import write_file
+from dbt.events.types import (
+    ManifestDependenciesLoaded, ManifestLoaderCreated, ManifestLoaded, ManifestChecked,
+    ManifestFlatGraphBuilt, ParsingStart, ParsingCompiling, ParsingWritingManifest, ParsingDone,
+    ReportPerformancePath
+)
+from dbt.events.functions import fire_event
 from dbt.graph import Graph
 import time
 from typing import Optional
@@ -40,7 +46,7 @@ def write_perf_info(self):
         path = os.path.join(self.config.target_path, PERF_INFO_FILE_NAME)
         write_file(path, json.dumps(self.loader._perf_info,
                                     cls=dbt.utils.JSONEncoder, indent=4))
-        print_timestamped_line(f"Performance info: {path}")
+        fire_event(ReportPerformancePath(path=path))
 
     # This method takes code that normally exists in other files
     # and pulls it in here, to simplify logging and make the
@@ -58,38 +64,37 @@ def get_full_manifest(self):
         with PARSING_STATE:
             start_load_all = time.perf_counter()
             projects = root_config.load_dependencies()
-            print_timestamped_line("Dependencies loaded")
+            fire_event(ManifestDependenciesLoaded())
             loader = ManifestLoader(root_config, projects, macro_hook)
-            print_timestamped_line("ManifestLoader created")
+            fire_event(ManifestLoaderCreated())
             manifest = loader.load()
-            print_timestamped_line("Manifest loaded")
+            fire_event(ManifestLoaded())
             _check_manifest(manifest, root_config)
-            print_timestamped_line("Manifest checked")
+            fire_event(ManifestChecked())
             manifest.build_flat_graph()
-            print_timestamped_line("Flat graph built")
+            fire_event(ManifestFlatGraphBuilt())
             loader._perf_info.load_all_elapsed = (
                 time.perf_counter() - start_load_all
             )
 
         self.loader = loader
         self.manifest = manifest
-        print_timestamped_line("Manifest loaded")
+        fire_event(ManifestLoaded())
 
     def compile_manifest(self):
         adapter = get_adapter(self.config)
         compiler = adapter.get_compiler()
         self.graph = compiler.compile(self.manifest)
 
     def run(self):
-        events.register(Progress(ParseStart))
-        print_timestamped_line('Start parsing.')
+        fire_event(ParsingStart())
         self.get_full_manifest()
         if self.args.compile:
-            print_timestamped_line('Compiling.')
+            fire_event(ParsingCompiling())
             self.compile_manifest()
         if self.args.write_manifest:
-            print_timestamped_line('Writing manifest.')
+            fire_event(ParsingWritingManifest())
             self.write_manifest()
 
         self.write_perf_info()
-        print_timestamped_line('Done.')
+        fire_event(ParsingDone())