Refactor EventManager setup and interaction (#9180)

* moving types_pb2.py to common/events

* move event manager setup back to core, remove ref to global EVENT_MANAGER and clean up event manager functions

* move invocation_id from events to first class common concept

* move lowercase utils to common

* move lowercase utils to common

* ref CAPTURE_STREAM through method

* add changie

.changes/unreleased/Breaking Changes-20231130-135348.yaml

@@ -0,0 +1,7 @@
+kind: Breaking Changes
+body: move event manager setup back to core, remove ref to global EVENT_MANAGER and
+  clean up event manager functions
+time: 2023-11-30T13:53:48.645192-08:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "9150"

.changes/unreleased/Under the Hood-20231130-135432.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Move lowercase utils method to common
+time: 2023-11-30T13:54:32.561673-08:00
+custom:
+  Author: colin-rogers-dbt
+  Issue: "9180"

core/dbt/adapters/cache.py

@@ -16,7 +16,7 @@
 )
 from dbt.common.events.functions import fire_event, fire_event_if
 from dbt.adapters.events.types import CacheAction, CacheDumpGraph
-from dbt.utils import lowercase
+from dbt.common.utils.formatting import lowercase
 
 
 def dot_separated(key: _ReferenceKey) -> str:

core/dbt/adapters/events/logging.py

@@ -7,9 +7,10 @@
     AdapterEventWarning,
     AdapterEventError,
 )
+from dbt.common.events import get_event_manager
 from dbt.common.events.contextvars import get_node_info
 from dbt.common.events.event_handler import set_package_logging
-from dbt.common.events.functions import fire_event, EVENT_MANAGER
+from dbt.common.events.functions import fire_event
 
 
 @dataclass
@@ -63,4 +64,4 @@ def set_adapter_dependency_log_level(package_name, level):
         """By default, dbt suppresses non-dbt package logs. This method allows
         you to set the log level for a specific package.
         """
-        set_package_logging(package_name, level, EVENT_MANAGER)
+        set_package_logging(package_name, level, get_event_manager())

core/dbt/cli/requires.py

@@ -1,4 +1,5 @@
 import dbt.tracking
+from dbt.common.invocation import reset_invocation_id
 from dbt.version import installed as installed_version
 from dbt.adapters.factory import adapter_management, register_adapter
 from dbt.flags import set_flags, get_flag_dict
@@ -14,9 +15,8 @@
 from dbt.common.events.functions import (
     fire_event,
     LOG_VERSION,
-    set_invocation_id,
-    setup_event_logger,
 )
+from dbt.events.logging import setup_event_logger
 from dbt.common.events.types import (
     CommandCompleted,
     MainReportVersion,
@@ -52,9 +52,11 @@ def wrapper(*args, **kwargs):
         ctx.obj["flags"] = flags
         set_flags(flags)
 
+        # Reset invocation_id for each 'invocation' of a dbt command (can happen multiple times in a single process)
+        reset_invocation_id()
+
         # Logging
         callbacks = ctx.obj.get("callbacks", [])
-        set_invocation_id()
         setup_event_logger(flags=flags, callbacks=callbacks)
 
         # Tracking

core/dbt/common/events/__init__.py

@@ -0,0 +1,9 @@
+from dbt.common.events.base_types import EventLevel
+from dbt.common.events.event_manager_client import get_event_manager
+from dbt.common.events.functions import get_stdout_config
+from dbt.common.events.logger import LineFormat
+
+# make sure event manager starts with a logger
+get_event_manager().add_logger(
+    get_stdout_config(LineFormat.PlainText, True, EventLevel.INFO, False)
+)

core/dbt/common/events/base_types.py

@@ -8,6 +8,8 @@
 from dbt.common.events.helpers import get_json_string_utcnow
 from typing import Optional
 
+from dbt.common.invocation import get_invocation_id
+
 if sys.version_info >= (3, 8):
     from typing import Protocol
 else:
@@ -26,12 +28,6 @@ def get_global_metadata_vars() -> dict:
     return get_metadata_vars()
 
 
-def get_invocation_id() -> str:
-    from dbt.common.events.functions import get_invocation_id
-
-    return get_invocation_id()
-
-
 # exactly one pid per concrete event
 def get_pid() -> int:
     return os.getpid()

core/dbt/common/events/event_handler.py

@@ -3,7 +3,7 @@
 
 from dbt.common.events.base_types import EventLevel
 from dbt.common.events.types import Note
-from dbt.common.events.eventmgr import IEventManager
+from dbt.common.events.event_manager import IEventManager
 
 
 _log_level_to_event_level_map = {

core/dbt/common/events/eventmgr.py → core/dbt/common/events/event_manager.py

@@ -1,7 +1,6 @@
 import os
 import traceback
 from typing import Callable, List, Optional, Protocol, Tuple
-from uuid import uuid4
 
 from dbt.common.events.base_types import BaseEvent, EventLevel, msg_from_base_event, EventMsg
 from dbt.common.events.logger import LoggerConfig, _Logger, _TextLogger, _JsonLogger, LineFormat
@@ -11,7 +10,6 @@ class EventManager:
     def __init__(self) -> None:
         self.loggers: List[_Logger] = []
         self.callbacks: List[Callable[[EventMsg], None]] = []
-        self.invocation_id: str = str(uuid4())
 
     def fire_event(self, e: BaseEvent, level: Optional[EventLevel] = None) -> None:
         msg = msg_from_base_event(e, level=level)
@@ -45,7 +43,6 @@ def flush(self) -> None:
 
 class IEventManager(Protocol):
     callbacks: List[Callable[[EventMsg], None]]
-    invocation_id: str
     loggers: List[_Logger]
 
     def fire_event(self, e: BaseEvent, level: Optional[EventLevel] = None) -> None:

core/dbt/common/events/event_manager_client.py

@@ -0,0 +1,29 @@
+# Since dbt-rpc does not do its own log setup, and since some events can
+# currently fire before logs can be configured by setup_event_logger(), we
+# create a default configuration with default settings and no file output.
+from dbt.common.events.event_manager import IEventManager, EventManager
+
+_EVENT_MANAGER: IEventManager = EventManager()
+
+
+def get_event_manager() -> IEventManager:
+    global _EVENT_MANAGER
+    return _EVENT_MANAGER
+
+
+def add_logger_to_manager(logger) -> None:
+    global _EVENT_MANAGER
+    _EVENT_MANAGER.add_logger(logger)
+
+
+def ctx_set_event_manager(event_manager: IEventManager) -> None:
+    global _EVENT_MANAGER
+    _EVENT_MANAGER = event_manager
+
+
+def cleanup_event_logger() -> None:
+    # Reset to a no-op manager to release streams associated with logs. This is
+    # especially important for tests, since pytest replaces the stdout stream
+    # during test runs, and closes the stream after the test is over.
+    _EVENT_MANAGER.loggers.clear()
+    _EVENT_MANAGER.callbacks.clear()