From 883026ffc8a13ade2d19a7a2886a0afea2302b27 Mon Sep 17 00:00:00 2001 From: MikaKerman Date: Tue, 8 Oct 2024 14:51:47 +0300 Subject: [PATCH] refactored al-in-one alert templates --- .../alerts/integrations/slack/slack.py | 238 ++++++++++-------- 1 file changed, 129 insertions(+), 109 deletions(-) diff --git a/elementary/monitor/data_monitoring/alerts/integrations/slack/slack.py b/elementary/monitor/data_monitoring/alerts/integrations/slack/slack.py index 639af8d9a..225148e49 100644 --- a/elementary/monitor/data_monitoring/alerts/integrations/slack/slack.py +++ b/elementary/monitor/data_monitoring/alerts/integrations/slack/slack.py @@ -1,7 +1,7 @@ import json import re from datetime import datetime, timedelta -from typing import Any, Dict, Generator, List, Optional, Tuple, Union +from typing import Any, Dict, Generator, List, Optional, Sequence, Tuple, Union from slack_sdk.models.blocks import SectionBlock @@ -763,24 +763,24 @@ def _get_source_freshness_template( ) def _get_alert_type_counters_block(self, alert: GroupedAlert) -> dict: - counters_text: List[str] = [] + counters_texts: List[str] = [] if alert.model_errors: - counters_text.append(f":X: Model errors: {len(alert.model_errors)}") + counters_texts.append(f":X: Model errors: {len(alert.model_errors)}") if alert.test_failures: - if counters_text: - counters_text.append(" |") - counters_text.append( + counters_texts.append( f":small_red_triangle: Test failures: {len(alert.test_failures)}" ) if alert.test_warnings: - if counters_text: - counters_text.append(" |") - counters_text.append(f":warning: Test warnings: {len(alert.test_warnings)}") + counters_texts.append( + f":warning: Test warnings: {len(alert.test_warnings)}" + ) if alert.test_errors: - if counters_text: - counters_text.append(" |") - counters_text.append(f":exclamation: Test errors: {len(alert.test_errors)}") - return self.message_builder.create_context_block(counters_text) + counters_texts.append( + f":exclamation: Test errors: {len(alert.test_errors)}" + ) + return self.message_builder.create_context_block( + [" | ".join(counters_texts)] + ) def _get_group_by_table_template( self, alert: GroupedByTableAlerts, *args, **kwargs @@ -886,6 +886,58 @@ def _get_group_by_table_template( title=title_blocks, preview=preview_blocks, details=details_blocks ) + def _add_compact_all_in_one_sub_group_details_block( + self, + details_blocks: list, + alerts: Sequence[ + Union[ + TestAlertModel, + ModelAlertModel, + SourceFreshnessAlertModel, + GroupedByTableAlerts, + ], + ], + sub_title: str, + bullet_icon: str, + ) -> None: + if not alerts: + return + details_blocks.append( + self.message_builder.create_text_section_block( + f":{bullet_icon}: {len(alerts)} {sub_title}" + ) + ) + + def _get_compact_all_in_one_sub_group_details_block( + self, alert: AllInOneAlert + ) -> List[dict]: + details_blocks: List[dict] = [] + self._add_compact_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.model_errors, + sub_title="Model Errors", + bullet_icon="X", + ) + self._add_compact_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.test_failures, + sub_title="Test Failures", + bullet_icon="small_red_triangle", + ) + self._add_compact_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.test_warnings, + sub_title="Test Warnings", + bullet_icon="warning", + ) + self._add_compact_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.test_errors, + sub_title="Test Errors", + bullet_icon="exclamation", + ) + return details_blocks + def _get_all_in_one_compact_template( self, alert: AllInOneAlert ) -> SlackAlertMessageSchema: @@ -897,32 +949,71 @@ def _get_all_in_one_compact_template( ) ] - details_blocks = [] - if alert.model_errors: - details_blocks.append( - self.message_builder.create_text_section_block( - f":X: {len(alert.model_errors)} Model errors" - ) - ) - if alert.test_failures: - details_blocks.append( - self.message_builder.create_text_section_block( - f":small_red_triangle: {len(alert.test_failures)} Test failures" - ) - ) - if alert.test_warnings: - details_blocks.append( - self.message_builder.create_text_section_block( - f":warning: {len(alert.test_warnings)} Test warnings" - ) - ) - if alert.test_errors: - details_blocks.append( - self.message_builder.create_text_section_block( - f":exclamation: {len(alert.test_errors)} Test errors" + details_blocks = self._get_compact_all_in_one_sub_group_details_block(alert) + return SlackAlertMessageSchema(title=title_blocks, details=details_blocks) + + def _add_all_in_one_sub_group_details_block( + self, + details_blocks: list, + alerts: Sequence[ + Union[ + TestAlertModel, + ModelAlertModel, + SourceFreshnessAlertModel, + GroupedByTableAlerts, + ], + ], + sub_title: str, + bullet_icon: str, + ) -> None: + if not alerts: + return + + details_blocks.append( + self.message_builder.create_text_section_block(f"*{sub_title}*") + ) + for alert in alerts: + text = f":{bullet_icon}: {alert.summary}" + section = ( + self.message_builder.create_section_with_button( + text, + button_text="View Details", + url=alert.report_url, ) + if alert.report_url + else self.message_builder.create_text_section_block(text) ) - return SlackAlertMessageSchema(title=title_blocks, details=details_blocks) + details_blocks.append(section) + + def _get_all_in_one_sub_group_details_blocks( + self, alert: AllInOneAlert + ) -> List[dict]: + details_blocks: List[dict] = [] + self._add_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.model_errors, + sub_title="Model Errors", + bullet_icon="X", + ) + self._add_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.test_failures, + sub_title="Test Failures", + bullet_icon="small_red_triangle", + ) + self._add_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.test_warnings, + sub_title="Test Warnings", + bullet_icon="warning", + ) + self._add_all_in_one_sub_group_details_block( + details_blocks=details_blocks, + alerts=alert.test_errors, + sub_title="Test Errors", + bullet_icon="exclamation", + ) + return details_blocks def _get_all_in_one_template( self, alert: AllInOneAlert, *args, **kwargs @@ -931,84 +1022,13 @@ def _get_all_in_one_template( return self._get_all_in_one_compact_template(alert) self.message_builder.add_message_color(self._get_color(alert.status)) - title_blocks = [ self.message_builder.create_header_block( f"{self._get_display_name(alert.status)}: {alert.summary}" ), self._get_alert_type_counters_block(alert), ] - - details_blocks = [] - - if alert.model_errors: - details_blocks.append( - self.message_builder.create_text_section_block("*Model errors*") - ) - for model_error_alert in alert.model_errors: - text = f":X: {model_error_alert.summary}" - section = ( - self.message_builder.create_section_with_button( - text, - button_text="View Details", - url=model_error_alert.report_url, - ) - if model_error_alert.report_url - else self.message_builder.create_text_section_block(text) - ) - details_blocks.append(section) - - if alert.test_failures: - details_blocks.append( - self.message_builder.create_text_section_block("*Test failures*") - ) - for test_failure_alert in alert.test_failures: - text = f":small_red_triangle: {test_failure_alert.summary}" - section = ( - self.message_builder.create_section_with_button( - text, - button_text="View Details", - url=test_failure_alert.report_url, - ) - if test_failure_alert.report_url - else self.message_builder.create_text_section_block(text) - ) - details_blocks.append(section) - - if alert.test_warnings: - details_blocks.append( - self.message_builder.create_text_section_block("*Test warnings*") - ) - for test_warning_alert in alert.test_warnings: - text = f":warning: {test_warning_alert.summary}" - section = ( - self.message_builder.create_section_with_button( - text, - button_text="View Details", - url=test_warning_alert.report_url, - ) - if test_warning_alert.report_url - else self.message_builder.create_text_section_block(text) - ) - details_blocks.append(section) - - if alert.test_errors: - details_blocks.append( - self.message_builder.create_text_section_block("*Test errors*") - ) - for test_error_alert in alert.test_errors: - text = f":exclamation: {test_error_alert.summary}" - section = ( - self.message_builder.create_section_with_button( - text, - button_text="View Details", - url=test_error_alert.report_url, - ) - if test_error_alert.report_url - else self.message_builder.create_text_section_block(text) - ) - details_blocks.append(section) - + details_blocks = self._get_all_in_one_sub_group_details_blocks(alert) return SlackAlertMessageSchema(title=title_blocks, details=details_blocks) @staticmethod