Feature: make the number of columns that are formatted as a table in Teams alerts a cli flag so that users have more control over the formatting in Teams (#1958)

* Introduced a new configuration setting in which one can change the number of columns that are allowed in a result sample that will be displayed as a table. If it exceeds this number of columns than it will be parsed as a JSON

* Use full width for Teams message to allow beter formatting of a table

* Depending on size of table format it differently so that large tables are better readable

* Refactor code + implement a sleep to prevent rate limits from being hit in Teams for scenarios where a lot of alerts need to be sent

* Error handling for the teams webhook alerting to handle cases like network exceptions

* Correct data type in which the message configuration is populating the message that is sent to MS Teams

* Refactor coderabbitai suggestion

* Implemented feedback of Elazar

* Adapt new full widt layout for MS Teams alerting in unit tests

* Fix failing unit tests

---------

Co-authored-by: Frank Tubbing <[email protected]>
Co-authored-by: Itamar Hartstein <[email protected]>

Author: 3 people

elementary/config/config.py

@@ -71,6 +71,7 @@ def __init__(
         azure_container_name: Optional[str] = None,
         report_url: Optional[str] = None,
         teams_webhook: Optional[str] = None,
+        maximum_columns_in_alert_samples: Optional[int] = None,
         env: str = DEFAULT_ENV,
         run_dbt_deps_if_needed: Optional[bool] = None,
         project_name: Optional[str] = None,
@@ -103,6 +104,12 @@ def __init__(
             False,
         )
 
+        self.maximum_columns_in_alert_samples = self._first_not_none(
+            maximum_columns_in_alert_samples,
+            config.get("maximum_columns_in_alert_samples"),
+            4,
+        )
+
         self.timezone = self._first_not_none(
             timezone,
             config.get("timezone"),

elementary/messages/formats/adaptive_cards.py

@@ -144,7 +144,8 @@ def format_table_block(block: TableBlock) -> Dict[str, Any]:
                         "items": [
                             {
                                 "type": "TextBlock",
-                                "text": str(cell),
+                                "text": str(cell) if cell is not None else "",
+                                "wrap": True,
                             }
                         ],
                     }
@@ -256,4 +257,5 @@ def format_adaptive_card(message: MessageBody, version: str = "1.5") -> Dict[str
         "type": "AdaptiveCard",
         "body": format_adaptive_card_body(message),
         "version": version,
+        "msteams": {"width": "Full"},
     }

elementary/messages/messaging_integrations/teams_webhook.py

@@ -1,7 +1,9 @@
 from datetime import datetime
+from http import HTTPStatus
 from typing import Any, Optional
 
 import requests
+from ratelimit import limits, sleep_and_retry
 from typing_extensions import TypeAlias
 
 from elementary.messages.formats.adaptive_cards import format_adaptive_card
@@ -22,6 +24,7 @@
 
 
 Channel: TypeAlias = Optional[str]
+ONE_SECOND = 1
 
 
 def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response:
@@ -42,8 +45,10 @@ def send_adaptive_card(webhook_url: str, card: dict) -> requests.Response:
         headers={"Content-Type": "application/json"},
     )
     response.raise_for_status()
-    if response.status_code == 202:
-        logger.debug("Got 202 response from Teams webhook, assuming success")
+    if response.status_code == HTTPStatus.ACCEPTED:
+        logger.debug(
+            f"Got {HTTPStatus.ACCEPTED} response from Teams webhook, assuming success"
+        )
     return response
 
 
@@ -56,22 +61,36 @@ def __init__(self, url: str) -> None:
     def parse_message_context(self, context: dict[str, Any]) -> EmptyMessageContext:
         return EmptyMessageContext(**context)
 
+    @sleep_and_retry
+    @limits(calls=1, period=ONE_SECOND)
     def send_message(
         self,
         destination: None,
         body: MessageBody,
     ) -> MessageSendResult[EmptyMessageContext]:
         card = format_adaptive_card(body)
         try:
-            send_adaptive_card(self.url, card)
+            response = send_adaptive_card(self.url, card)
+            # For the old teams webhook version of Teams simply returning status code 200
+            # is not indicating that it was successful.
+            # In that version they return some text if it was NOT successful, otherwise
+            # they return the number 1 in the text. For the new teams webhook version they always
+            # return a 202 and nothing else can be used to determine if the message was
+            # sent successfully.
+            if response.status_code not in (HTTPStatus.OK, HTTPStatus.ACCEPTED) or (
+                response.status_code == HTTPStatus.OK and len(response.text) > 1
+            ):
+                raise MessagingIntegrationError(
+                    f"Could not post message to Teams via webhook. Status code: {response.status_code}, Error: {response.text}"
+                )
             return MessageSendResult(
                 message_context=EmptyMessageContext(),
                 timestamp=datetime.utcnow(),
                 message_format="adaptive_cards",
             )
         except requests.RequestException as e:
             raise MessagingIntegrationError(
-                "Failed to send message to Teams webhook"
+                f"An error occurred while posting message to Teams webhook: {str(e)}"
             ) from e
 
     def supports_reply(self) -> bool:

elementary/monitor/alerts/alert_messages/builder.py

@@ -56,6 +56,7 @@
 
 class MessageBuilderConfig(BaseModel):
     alert_groups_subscribers: bool = False
+    maximum_columns_in_alert_samples: int = 4
 
 
 class AlertMessageBuilder:
@@ -343,7 +344,8 @@ def _get_result_blocks(
             elif result_sample:
                 if (
                     isinstance(result_sample, list)
-                    and len(result_sample[0].keys()) <= 4
+                    and len(result_sample[0].keys())
+                    <= self.config.maximum_columns_in_alert_samples
                 ):
                     result_blocks.append(
                         TableBlock.from_dicts(result_sample),

elementary/monitor/cli.py

@@ -285,6 +285,13 @@ def get_cli_properties() -> dict:
     default=None,
     help="A Microsoft Teams webhook URL for sending alerts to a specific channel in Teams.",
 )
+@click.option(
+    "--maximum-columns-in-alert-samples",
+    "-mc",
+    type=int,
+    default=4,
+    help="Maximum number of columns to display as a table in alert samples. Above this, the output is shown as raw JSON.",
+)
 @click.pass_context
 def monitor(
     ctx,
@@ -316,6 +323,7 @@ def monitor(
     filters,
     excludes,
     teams_webhook,
+    maximum_columns_in_alert_samples,
 ):
     """
     Get alerts on failures in dbt jobs.
@@ -347,6 +355,7 @@ def monitor(
         slack_group_alerts_by=group_by,
         report_url=report_url,
         teams_webhook=teams_webhook,
+        maximum_columns_in_alert_samples=maximum_columns_in_alert_samples,
     )
     anonymous_tracking = AnonymousCommandLineTracking(config)
     anonymous_tracking.set_env("use_select", bool(select))

elementary/monitor/data_monitoring/alerts/data_monitoring_alerts.py

@@ -22,7 +22,10 @@
 from elementary.messages.messaging_integrations.exceptions import (
     MessagingIntegrationError,
 )
-from elementary.monitor.alerts.alert_messages.builder import AlertMessageBuilder
+from elementary.monitor.alerts.alert_messages.builder import (
+    AlertMessageBuilder,
+    MessageBuilderConfig,
+)
 from elementary.monitor.alerts.alerts_groups import GroupedByTableAlerts
 from elementary.monitor.alerts.alerts_groups.alerts_group import AlertsGroup
 from elementary.monitor.alerts.grouping_type import GroupingType
@@ -124,6 +127,11 @@ def __init__(
         self.send_test_message_on_success = send_test_message_on_success
         self.override_config_defaults = override_config
         self.alerts_integration = self._get_integration_client()
+        self.alert_message_builder = AlertMessageBuilder(
+            MessageBuilderConfig(
+                maximum_columns_in_alert_samples=self.config.maximum_columns_in_alert_samples
+            )
+        )
 
     def _get_integration_client(
         self,
@@ -346,8 +354,7 @@ def _send_alert(
     ) -> bool:
         if isinstance(self.alerts_integration, BaseIntegration):
             return self.alerts_integration.send_alert(alert)
-        alert_message_builder = AlertMessageBuilder()
-        alert_message_body = alert_message_builder.build(
+        alert_message_body = self.alert_message_builder.build(
             alert=alert,
         )
         alert_message_body = _add_elementary_attribution(alert_message_body)

elementary/monitor/dbt_project/.gitignore

@@ -3,3 +3,4 @@ target/
 dbt_modules/
 dbt_packages*/
 logs/
+dbt_internal_packages/

tests/unit/alerts/alert_messages/fixtures/adaptive_card/alerts_group_model-errors-False_test-failures-False_test-warnings-True_test-errors-True_link-False_env-False_subscribers-True.json

@@ -148,5 +148,8 @@
       ]
     }
   ],
-  "version": "1.5"
+  "version": "1.5",
+  "msteams": {
+    "width": "Full"
+  }
 }

tests/unit/alerts/alert_messages/fixtures/adaptive_card/alerts_group_model-errors-False_test-failures-True_test-warnings-False_test-errors-False_link-False_env-False_subscribers-True.json

@@ -87,5 +87,8 @@
       ]
     }
   ],
-  "version": "1.5"
+  "version": "1.5",
+  "msteams": {
+    "width": "Full"
+  }
 }

tests/unit/alerts/alert_messages/fixtures/adaptive_card/alerts_group_model-errors-False_test-failures-True_test-warnings-False_test-errors-True_link-False_env-False_subscribers-True.json

@@ -148,5 +148,8 @@
       ]
     }
   ],
-  "version": "1.5"
+  "version": "1.5",
+  "msteams": {
+    "width": "Full"
+  }
 }