Misc fixes for group info in logging (#11218)

.changes/unreleased/Under the Hood-20250117-152215.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Misc fixes for group info in logging
+time: 2025-01-17T15:22:15.497485Z
+custom:
+    Author: aranke
+    Issue: '11218'

core/dbt/artifacts/resources/v1/owner.py

@@ -1,10 +1,10 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import List, Optional, Union
 
 from dbt_common.contracts.config.properties import AdditionalPropertiesAllowed
 
 
 @dataclass
 class Owner(AdditionalPropertiesAllowed):
-    email: Optional[str] = None
+    email: Union[str, List[str], None] = None
     name: Optional[str] = None

core/dbt/contracts/graph/nodes.py

@@ -1531,7 +1531,7 @@ def to_logging_dict(self) -> Dict[str, Union[str, Dict[str, str]]]:
         return {
             "name": self.name,
             "package_name": self.package_name,
-            "owner": self.owner.to_dict(omit_none=True),
+            "owner": {k: str(v) for k, v in self.owner.to_dict(omit_none=True).items()},
         }
 
 

core/dbt/events/core_types.proto

@@ -1415,6 +1415,7 @@ message LogSnapshotResult {
     float execution_time = 6;
     map<string, string> cfg = 7;
     string result_message = 8;
+    Group group = 9;
 }
 
 message LogSnapshotResultMsg {
@@ -1432,6 +1433,7 @@ message LogSeedResult {
     float execution_time = 6;
     string schema = 7;
     string relation = 8;
+    Group group = 9;
 }
 
 message LogSeedResultMsg {
@@ -1605,6 +1607,7 @@ message SkippingDetails {
     string node_name = 4;
     int32 index = 5;
     int32 total = 6;
+    Group group = 7;
 }
 
 message SkippingDetailsMsg {
@@ -2064,6 +2067,7 @@ message LogSkipBecauseError {
     int32 index = 3;
     int32 total = 4;
     string status = 5;
+    Group group = 6;
 }
 
 message LogSkipBecauseErrorMsg {

core/dbt/task/base.py

@@ -41,6 +41,7 @@
 )
 from dbt.flags import get_flags
 from dbt.graph import Graph
+from dbt.task import group_lookup
 from dbt.task.printer import print_run_result_error
 from dbt_common.events.contextvars import get_node_info
 from dbt_common.events.functions import fire_event
@@ -424,6 +425,8 @@ def on_skip(self):
             # if this model was skipped due to an upstream ephemeral model
             # failure, print a special 'error skip' message.
             # Include skip_cause NodeStatus
+            group = group_lookup.get(self.node.unique_id)
+
             if self._skip_caused_by_ephemeral_failure():
                 fire_event(
                     LogSkipBecauseError(
@@ -432,6 +435,7 @@ def on_skip(self):
                         index=self.node_index,
                         total=self.num_nodes,
                         status=self.skip_cause.status,
+                        group=group,
                     )
                 )
                 # skip_cause here should be the run_result from the ephemeral model
@@ -459,6 +463,7 @@ def on_skip(self):
                         index=self.node_index,
                         total=self.num_nodes,
                         node_info=self.node.node_info,
+                        group=group,
                     )
                 )
 

core/dbt/task/runnable.py

@@ -49,16 +49,16 @@
     parse_difference,
 )
 from dbt.parser.manifest import write_manifest
+from dbt.task import group_lookup
 from dbt.task.base import BaseRunner, ConfiguredTask
+from dbt.task.printer import print_run_end_messages, print_run_result_error
 from dbt_common.context import _INVOCATION_CONTEXT_VAR, get_invocation_context
 from dbt_common.dataclass_schema import StrEnum
 from dbt_common.events.contextvars import log_contextvars, task_contextvars
 from dbt_common.events.functions import fire_event, warn_or_error
 from dbt_common.events.types import Formatting
 from dbt_common.exceptions import NotImplementedError
 
-from .printer import print_run_end_messages, print_run_result_error
-
 RESULT_FILE_NAME = "run_results.json"
 
 
@@ -538,6 +538,8 @@ def execute_with_hooks(self, selected_uids: AbstractSet[str]):
                 res = []
 
                 for index, node in enumerate(self._flattened_nodes or []):
+                    group = group_lookup.get(node.unique_id)
+
                     if node.unique_id not in executed_node_ids:
                         fire_event(
                             SkippingDetails(
@@ -547,6 +549,7 @@ def execute_with_hooks(self, selected_uids: AbstractSet[str]):
                                 index=index + 1,
                                 total=self.num_nodes,
                                 node_info=node.node_info,
+                                group=group,
                             )
                         )
                         skipped_node_result = mark_node_as_skipped(node, executed_node_ids, None)

core/dbt/task/seed.py

@@ -6,15 +6,15 @@
 from dbt.events.types import LogSeedResult, LogStartLine, SeedHeader
 from dbt.graph import ResourceTypeSelector
 from dbt.node_types import NodeType
+from dbt.task import group_lookup
 from dbt.task.base import BaseRunner
+from dbt.task.printer import print_run_end_messages
+from dbt.task.run import ModelRunner, RunTask
 from dbt_common.events.base_types import EventLevel
 from dbt_common.events.functions import fire_event
 from dbt_common.events.types import Formatting
 from dbt_common.exceptions import DbtInternalError
 
-from .printer import print_run_end_messages
-from .run import ModelRunner, RunTask
-
 
 class SeedRunner(ModelRunner):
     def describe_node(self) -> str:
@@ -41,6 +41,7 @@ def compile(self, manifest: Manifest):
 
     def print_result_line(self, result):
         model = result.node
+        group = group_lookup.get(model.unique_id)
         level = EventLevel.ERROR if result.status == NodeStatus.Error else EventLevel.INFO
         fire_event(
             LogSeedResult(
@@ -52,6 +53,7 @@ def print_result_line(self, result):
                 schema=self.node.schema,
                 relation=model.alias,
                 node_info=model.node_info,
+                group=group,
             ),
             level=level,
         )

core/dbt/task/snapshot.py

@@ -4,21 +4,22 @@
 from dbt.events.types import LogSnapshotResult
 from dbt.graph import ResourceTypeSelector
 from dbt.node_types import NodeType
+from dbt.task import group_lookup
 from dbt.task.base import BaseRunner
+from dbt.task.run import ModelRunner, RunTask
 from dbt_common.events.base_types import EventLevel
 from dbt_common.events.functions import fire_event
 from dbt_common.exceptions import DbtInternalError
 from dbt_common.utils import cast_dict_to_dict_of_strings
 
-from .run import ModelRunner, RunTask
-
 
 class SnapshotRunner(ModelRunner):
     def describe_node(self) -> str:
         return "snapshot {}".format(self.get_node_representation())
 
     def print_result_line(self, result):
         model = result.node
+        group = group_lookup.get(model.unique_id)
         cfg = model.config.to_dict(omit_none=True)
         level = EventLevel.ERROR if result.status == NodeStatus.Error else EventLevel.INFO
         fire_event(
@@ -31,6 +32,7 @@ def print_result_line(self, result):
                 execution_time=result.execution_time,
                 node_info=model.node_info,
                 result_message=result.message,
+                group=group,
             ),
             level=level,
         )

core/dbt/task/test.py

@@ -37,18 +37,17 @@
 from dbt.graph import ResourceTypeSelector
 from dbt.node_types import TEST_NODE_TYPES, NodeType
 from dbt.parser.unit_tests import UnitTestManifestLoader
+from dbt.task import group_lookup
 from dbt.task.base import BaseRunner, resource_types_from_args
+from dbt.task.compile import CompileRunner
+from dbt.task.run import RunTask
 from dbt.utils import _coerce_decimal, strtobool
 from dbt_common.dataclass_schema import dbtClassMixin
 from dbt_common.events.format import pluralize
 from dbt_common.events.functions import fire_event
 from dbt_common.exceptions import DbtBaseException, DbtRuntimeError
 from dbt_common.ui import green, red
 
-from . import group_lookup
-from .compile import CompileRunner
-from .run import RunTask
-
 if TYPE_CHECKING:
     import agate
 

tests/functional/logging/test_logging.py

@@ -119,6 +119,27 @@ def test_invalid_event_value(project, logs_dir):
     access: public
 """
 
+groups_yml_with_multiple_emails = """
+groups:
+  - name: my_group_with_multiple_emails
+    owner:
+      name: my_name
+      email:
+        - [email protected]
+        - [email protected]
+      slack: my_slack
+      other_property: something_else
+
+models:
+  - name: my_model
+    group: my_group_with_multiple_emails
+    access: public
+    columns:
+      - name: my_column
+        tests:
+         - not_null
+"""
+
 
 class TestRunResultErrorNodeInfo:
     @pytest.fixture(scope="class")
@@ -289,3 +310,41 @@ def models(self):
     def test_node_info_on_results(self, project, logs_dir):
         results = run_dbt(["--no-write-json", "run"])
         assert len(results) == 1
+
+
+class TestRunResultGroupWithMultipleEmails:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": "select 1 as id, null as my_column",
+            "groups.yml": groups_yml_with_multiple_emails,
+        }
+
+    def test_node_info_on_results(self, project, logs_dir):
+        results = run_dbt(["--log-format=json", "build"], expect_pass=False)
+        assert len(results) == 2
+
+        log_file = read_file(logs_dir, "dbt.log")
+        run_result_error_count = 0
+
+        for log_line in log_file.split("\n"):
+            if not log_line:
+                continue
+
+            log_json = json.loads(log_line)
+            if log_json["info"]["level"] == EventLevel.DEBUG:
+                continue
+
+            if log_json["info"]["name"] == "RunResultError":
+                assert "group" in log_json["data"]
+                group_data = log_json["data"]["group"]
+                assert group_data["name"] == "my_group_with_multiple_emails"
+                assert group_data["owner"] == {
+                    "name": "my_name",
+                    "email": "['[email protected]', '[email protected]']",
+                    "slack": "my_slack",
+                    "other_property": "something_else",
+                }
+                run_result_error_count += 1
+
+        assert run_result_error_count == 1