update for structlog AdapterLogger (dbt-labs#34)

* replace logger with structlog AdapterLogger

* fix flake error

* add changelog

* Update integration tests

Co-authored-by: Jeremy Cohen <[email protected]>

CHANGELOG.md

@@ -8,6 +8,7 @@ N/A
 
 ### Under the hood
 - Remove official support for python 3.6, which is reaching end of life on December 23, 2021 ([dbt-core#4134](https://github.com/dbt-labs/dbt-core/issues/4134), [#38](https://github.com/dbt-labs/dbt-redshift/pull/38))
+- Add support for structured logging [#34](https://github.com/dbt-labs/dbt-redshift/pull/34)
 
 ### Contributors
 N/A

dbt/adapters/redshift/connections.py

@@ -4,7 +4,7 @@
 
 from dbt.adapters.postgres import PostgresConnectionManager
 from dbt.adapters.postgres import PostgresCredentials
-from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt.events import AdapterLogger
 import dbt.exceptions
 import dbt.flags
 
@@ -15,6 +15,8 @@
 from dataclasses import dataclass, field
 from typing import Optional, List
 
+logger = AdapterLogger("Redshift")
+
 drop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()
 
 

dbt/adapters/redshift/impl.py

@@ -7,9 +7,11 @@
 from dbt.adapters.redshift import RedshiftConnectionManager
 from dbt.adapters.redshift import RedshiftColumn
 from dbt.adapters.redshift import RedshiftRelation
-from dbt.logger import GLOBAL_LOGGER as logger  # noqa
+from dbt.events import AdapterLogger
 import dbt.exceptions
 
+logger = AdapterLogger("Redshift")
+
 
 @dataclass
 class RedshiftConfig(AdapterConfig):

dev_requirements.txt

@@ -1,7 +1,7 @@
 # install latest changes in dbt-core + dbt-postgres
 # TODO: how to switch from HEAD to x.y.latest branches after minor releases?
-git+https://github.com/dbt-labs/dbt.git#egg=dbt-core&subdirectory=core
-git+https://github.com/dbt-labs/dbt.git#egg=dbt-postgres&subdirectory=plugins/postgres
+git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-core&subdirectory=core
+git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-postgres&subdirectory=plugins/postgres
 
 bumpversion
 flake8

tests/integration/base.py

@@ -21,10 +21,15 @@
 from dbt.clients.jinja import template_cache
 from dbt.config import RuntimeConfig
 from dbt.context import providers
-from dbt.logger import GLOBAL_LOGGER as logger, log_manager
+from dbt.logger import log_manager
+from dbt.events.functions import (
+    capture_stdout_logs, fire_event, setup_event_logger, stop_capture_stdout_logs
+)
+from dbt.events import AdapterLogger
 from dbt.contracts.graph.manifest import Manifest
 
 
+logger = AdapterLogger("Redshift")
 INITIAL_ROOT = os.getcwd()
 
 
@@ -219,6 +224,7 @@ def setUp(self):
         os.chdir(self.initial_dir)
         # before we go anywhere, collect the initial path info
         self._logs_dir = os.path.join(self.initial_dir, 'logs', self.prefix)
+        setup_event_logger(self._logs_dir)
         _really_makedirs(self._logs_dir)
         self.test_original_source_path = _pytest_get_test_root()
         self.test_root_dir = self._generate_test_root_dir()
@@ -410,16 +416,12 @@ def run_dbt(self, args=None, expect_pass=True, profiles_dir=True):
 
     def run_dbt_and_capture(self, *args, **kwargs):
         try:
-            initial_stdout = log_manager.stdout
-            initial_stderr = log_manager.stderr
-            stringbuf = io.StringIO()
-            log_manager.set_output_stream(stringbuf)
-
+            stringbuf = capture_stdout_logs()
             res = self.run_dbt(*args, **kwargs)
             stdout = stringbuf.getvalue()
 
         finally:
-            log_manager.set_output_stream(initial_stdout, initial_stderr)
+            stop_capture_stdout_logs()
 
         return res, stdout
 

tests/integration/query_comments_test/test_query_comments.py

@@ -2,6 +2,7 @@
 import io
 import json
 import os
+import re
 
 import dbt.exceptions
 from dbt.version import __version__ as dbt_version
@@ -52,28 +53,31 @@ def tearDown(self):
         super().tearDown()
 
     def run_get_json(self, expect_pass=True):
-        self.run_dbt(
+        res, raw_logs = self.run_dbt_and_capture(
             ['--debug', '--log-format=json', 'run'],
             expect_pass=expect_pass
         )
-        logs = []
-        for line in self.stringbuf.getvalue().split('\n'):
+
+        parsed_logs = []
+        for line in raw_logs.split('\n'):
             try:
                 log = json.loads(line)
             except ValueError:
                 continue
 
-            if log['extra'].get('run_state') != 'running':
-                continue
-            logs.append(log)
-        self.assertGreater(len(logs), 0)
-        return logs
+            parsed_logs.append(log)
+
+        # empty lists evaluate as False
+        self.assertTrue(parsed_logs)
+        return parsed_logs
 
     def query_comment(self, model_name, log):
+        # N.B: a temporary string replacement regex to strip the HH:MM:SS from the log line if present.
+        # TODO: make this go away when structured logging is stable 
+        log_msg = re.sub("(?:[01]\d|2[0123]):(?:[012345]\d):(?:[012345]\d \| )", "", log['msg'])
         prefix = 'On {}: '.format(model_name)
-
-        if log['message'].startswith(prefix):
-            msg = log['message'][len(prefix):]
+        if log_msg.startswith(prefix):
+            msg = log_msg[len(prefix):]
             if msg in {'COMMIT', 'BEGIN', 'ROLLBACK'}:
                 return None
             return msg
@@ -88,7 +92,7 @@ def run_assert_comments(self):
             if msg is not None and self.matches_comment(msg):
                 seen = True
 
-        self.assertTrue(seen, 'Never saw a matching log message! Logs:\n{}'.format('\n'.join(l['message'] for l in logs)))
+        self.assertTrue(seen, 'Never saw a matching log message! Logs:\n{}'.format('\n'.join(l['msg'] for l in logs)))
 
     @use_profile('redshift')
     def test_redshift_comments(self):