#223 Retrying policy

prom2teams/app/configuration.py

@@ -31,6 +31,12 @@ def _config_command_line():
 def _update_application_configuration(application, configuration):
     if 'Microsoft Teams' in configuration:
         application.config['MICROSOFT_TEAMS'] = configuration['Microsoft Teams']
+    if 'Microsoft Teams Client' in configuration:
+        application.config['TEAMS_CLIENT_CONFIG'] = {
+            'RETRY_ENABLE': configuration.getboolean('Microsoft Teams Client', 'RetryEnable'),
+            'RETRY_WAIT_TIME': configuration.getint('Microsoft Teams Client', 'RetryWaitTime'),
+            'MAX_PAYLOAD': configuration.getint('Microsoft Teams Client', 'MaxPayload')
+        }
     if 'Template' in configuration and 'Path' in configuration['Template']:
         application.config['TEMPLATE_PATH'] = configuration['Template']['Path']
     if 'Log' in configuration and 'Level' in configuration['Log']:

prom2teams/app/sender.py

@@ -1,30 +1,26 @@
-import logging
+
 
 from prom2teams.teams.alarm_mapper import map_and_group, map_prom_alerts_to_teams_alarms
 from prom2teams.teams.composer import TemplateComposer
-from .teams_client import post
+from .teams_client import TeamsClient
 
-log = logging.getLogger('prom2teams')
 
 
 class AlarmSender:
-
-    def __init__(self, template_path=None, group_alerts_by=False):
+    def __init__(self, template_path=None, group_alerts_by=False, teams_client_config=None):
+        self.json_composer = TemplateComposer(template_path)
         self.group_alerts_by = group_alerts_by
-        if template_path:
-            self.json_composer = TemplateComposer(template_path)
-        else:
-            self.json_composer = TemplateComposer()
+        self.teams_client = TeamsClient(teams_client_config)
+        self.max_payload = self.teams_client.max_payload_length
 
-    def _create_alarms(self, alerts):
-        if self.group_alerts_by:
-            alarms = map_and_group(alerts, self.group_alerts_by)
+    def _create_alarms(self, alerts, group_alerts_by):
+        if group_alerts_by:
+            alarms = map_and_group(alerts, group_alerts_by, self.json_composer.compose, self.max_payload)
         else:
             alarms = map_prom_alerts_to_teams_alarms(alerts)
         return self.json_composer.compose_all(alarms)
 
     def send_alarms(self, alerts, teams_webhook_url):
-        sending_alarms = self._create_alarms(alerts)
+        sending_alarms = self._create_alarms(alerts, self.group_alerts_by)
         for team_alarm in sending_alarms:
-            log.debug('The message that will be sent is: %s', str(team_alarm))
-            post(teams_webhook_url, team_alarm)
+            self.teams_client.post(teams_webhook_url, team_alarm)

prom2teams/app/teams_client.py

@@ -1,20 +1,55 @@
+import json
+import logging
 import requests
+from tenacity import retry, wait_fixed, after_log
 
 from .exceptions import MicrosoftTeamsRequestException
 
-session = requests.Session()
-session.headers.update({'Content-Type': 'application/json'})
-
-
-def post(teams_webhook_url, message):
-    response = session.post(teams_webhook_url, data=message)
-    if not response.ok or response.text is not '1':
-        exception_msg = 'Error performing request to: {}.\n' \
-                        ' Returned status code: {}.\n' \
-                        ' Returned data: {}\n' \
-                        ' Sent message: {}\n'
-        raise MicrosoftTeamsRequestException(exception_msg.format(teams_webhook_url,
-                                                                  str(response.status_code),
-                                                                  str(response.text),
-                                                                  str(message)),
-                                             code=response.status_code)
+logger = logging.getLogger('prom2teams')
+
+
+class TeamsClient:
+    DEFAULT_CONFIG = {
+        'MAX_PAYLOAD': 24576,
+        'RETRY_ENABLE': False,
+        'RETRY_WAIT_TIME': 60
+    }
+
+    def __init__(self, config=None):
+        self.session = requests.Session()
+        self.session.headers.update({'Content-Type': 'application/json'})
+
+        if config is None:
+            config = {}
+        config = {**TeamsClient.DEFAULT_CONFIG, **config}
+        self.max_payload_length = config['MAX_PAYLOAD']
+        self.retry = config['RETRY_ENABLE']
+        self.wait_time = config['RETRY_WAIT_TIME']
+
+    def post(self, teams_webhook_url, message):
+        @retry(wait=wait_fixed(self.wait_time), after=after_log(logger, logging.WARN))
+        def post_with_retry(teams_webhook_url, message):
+            self._do_post(teams_webhook_url, message)
+
+        def simple_post(teams_webhook_url, message):
+            self._do_post(teams_webhook_url, message)
+
+        logger.debug(f'The message that will be sent is: {message}')
+        if self.retry:
+            post_with_retry(teams_webhook_url, message)
+        else:
+            simple_post(teams_webhook_url, message)
+
+    def _do_post(self, teams_webhook_url, message):
+        response = self.session.post(teams_webhook_url, data=message)
+        if not response.ok or response.text != '1':
+            exception_msg = 'Error performing request to: {}.\n' \
+                ' Returned status code: {}.\n' \
+                ' Returned data: {}\n' \
+                ' Sent message: {}\n'
+            exception_msg.format(teams_webhook_url,
+                                 str(response.status_code),
+                                 str(response.text),
+                                 str(message))
+            raise MicrosoftTeamsRequestException(
+                exception_msg, code=response.status_code)

prom2teams/app/versions/v1/namespace.py

@@ -17,17 +17,16 @@ class AlertReceiver(Resource):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.schema = MessageSchema()
-        if 'TEMPLATE_PATH' in app.config:
-            self.sender = AlarmSender(app.config['TEMPLATE_PATH'])
-        else:
-            self.sender = AlarmSender()
+        self.sender = AlarmSender(template_path=app.config.get('TEMPLATE_PATH'),
+                                  teams_client_config=app.config.get('TEAMS_CLIENT_CONFIG'))
 
     @api_v1.expect(message)
     def post(self):
         _show_deprecated_warning("Call to deprecated function. It will be removed in future versions. "
                                  "Please view the README file.")
         alerts = self.schema.load(request.get_json())
-        self.sender.send_alarms(alerts, app.config['MICROSOFT_TEAMS']['Connector'])
+        self.sender.send_alarms(
+            alerts, app.config['MICROSOFT_TEAMS']['Connector'])
         return 'OK', 201
 
 

prom2teams/app/versions/v2/namespace.py

@@ -15,14 +15,15 @@ class AlertReceiver(Resource):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.schema = MessageSchema(exclude_fields=app.config['LABELS_EXCLUDED'], exclude_annotations=app.config['ANNOTATIONS_EXCLUDED'])
-        if app.config['TEMPLATE_PATH']:
-            self.sender = AlarmSender(app.config['TEMPLATE_PATH'], app.config['GROUP_ALERTS_BY'])
-        else:
-            self.sender = AlarmSender(group_alerts_by=app.config['GROUP_ALERTS_BY'])
+        self.schema = MessageSchema(exclude_fields=app.config['LABELS_EXCLUDED'],
+                                    exclude_annotations=app.config['ANNOTATIONS_EXCLUDED'])
+        self.sender = AlarmSender(template_path=app.config.get('TEMPLATE_PATH'),
+                                  group_alerts_by=app.config['GROUP_ALERTS_BY'],
+                                  teams_client_config=app.config.get('TEAMS_CLIENT_CONFIG'))
 
     @api_v2.expect(message)
     def post(self, connector):
         alerts = self.schema.load(request.get_json())
-        self.sender.send_alarms(alerts, app.config['MICROSOFT_TEAMS'][connector])
+        self.sender.send_alarms(
+            alerts, app.config['MICROSOFT_TEAMS'][connector])
         return 'OK', 201

prom2teams/teams/alarm_mapper.py

@@ -1,9 +1,14 @@
-from prom2teams.teams.teams_alarm_schema import TeamsAlarm, TeamsAlarmSchema
 from collections import defaultdict
 
+from prom2teams.teams.teams_alarm_schema import TeamsAlarm, TeamsAlarmSchema
+
+GROUPABLE_FIELDS = ['name', 'description', 'instance', 'severity', 'status', 'summary', 'fingerprint']
+EXTRA_FIELDS = ['extra_labels', 'extra_annotations']
+FIELD_SEPARATOR = ',\n\n\n'
+
 
 def map_prom_alerts_to_teams_alarms(alerts):
-    alerts = group_alerts(alerts, 'status')
+    alerts = _group_alerts(alerts, 'status')
     teams_alarms = []
     schema = TeamsAlarmSchema()
     for same_status_alerts in alerts:
@@ -17,51 +22,56 @@ def map_prom_alerts_to_teams_alarms(alerts):
     return teams_alarms
 
 
-def map_and_group(alerts, group_alerts_by):
-    alerts = group_alerts(alerts, 'status')
+def map_and_group(alerts, group_alerts_by, compose, payload_limit):
+    alerts = _group_alerts(alerts, 'status')
     teams_alarms = []
     schema = TeamsAlarmSchema()
     for same_status_alerts in alerts:
-        grouped_alerts = group_alerts(alerts[same_status_alerts], group_alerts_by)
-        for alert in grouped_alerts:
-            features = group_features(grouped_alerts[alert])
-            name, description, instance, severity, status, summary = (teams_visualization(features["name"]),
-                                                                      teams_visualization(features["description"]),
-                                                                      teams_visualization(features["instance"]),
-                                                                      teams_visualization(features["severity"]),
-                                                                      teams_visualization(features["status"]),
-                                                                      teams_visualization(features["summary"]))
-            fingerprint = teams_visualization(features["fingerprint"])
-            extra_labels = dict()
-            extra_annotations = dict()
-            for element in grouped_alerts[alert]:
-                if hasattr(element, 'extra_labels'):
-                    extra_labels = {**extra_labels, **element.extra_labels}
-                if hasattr(element, 'extra_annotations'):
-                    extra_annotations = {**extra_annotations, **element.extra_annotations}
+        grouped_alerts = _group_alerts(alerts[same_status_alerts], group_alerts_by)
+        for alert_group in grouped_alerts.values():
+            json_alarms = _map_group(alert_group, compose, payload_limit)
+            teams_alarms.extend(json_alarms)
+    return teams_alarms
 
-            alarm = TeamsAlarm(name, status.lower(), severity, summary,
-                               instance, description, fingerprint, extra_labels,
-                               extra_annotations)
-            json_alarm = schema.dump(alarm)
+def _map_group(alert_group, compose, payload_limit):
+    schema = TeamsAlarmSchema()
+    combined_alerts = []
+    teams_alarms = []
+    for alert in alert_group:
+        combined_alerts.append(alert)
+        json_alarm = schema.dump(_combine_alerts_to_alarm(combined_alerts))
+        if len(compose(json_alarm).encode('utf-8')) > payload_limit:
             teams_alarms.append(json_alarm)
+            combined_alerts.clear()
+
+    teams_alarms.append(json_alarm)
     return teams_alarms
 
+def _combine_alerts_to_alarm(alerts):
+    dicts = list(map(vars, alerts))
+    groupable = _combine_groupable_fields(dicts)
+    extra = _combine_extra_fields(dicts)
+    return _dict_alert_to_alarm({**groupable, **extra})
+
+def _dict_alert_to_alarm(alert):
+    return TeamsAlarm(alert['name'], alert['status'], alert['severity'], alert['summary'],
+                      alert['instance'], alert['description'], alert['fingerprint'], 
+                      alert['extra_labels'], alert['extra_annotations'])
+
+def _combine_groupable_fields(alerts):
+    return {field: _teams_visualization([alert[field] for alert in alerts]) for field in GROUPABLE_FIELDS}
 
-def teams_visualization(feature):
-    feature.sort()
+def _combine_extra_fields(alerts):
+    return {field: {k: v for alert in alerts for k,v in alert[field].items()} for field in EXTRA_FIELDS}
+
+def _teams_visualization(field):
+    field.sort()
     # Teams won't print just one new line
-    return ',\n\n\n'.join(feature) if feature else 'unknown'
+    return FIELD_SEPARATOR.join(field) if field else 'unknown'
 
 
-def group_alerts(alerts, group_alerts_by):
+def _group_alerts(alerts, group_alerts_by):
     groups = defaultdict(list)
     for alert in alerts:
         groups[alert.__dict__[group_alerts_by]].append(alert)
     return dict(groups)
-
-
-def group_features(alerts):
-    grouped_features = {feature: list(set([individual_alert.__dict__[feature] for individual_alert in alerts]))
-                        for feature in ["name", "description", "instance", "severity", "status", "summary", "fingerprint"]}
-    return grouped_features

prom2teams/teams/composer.py

@@ -26,8 +26,10 @@ class TemplateComposer(metaclass=_Singleton):
 
     DEFAULT_TEMPLATE_PATH = os.path.abspath(os.path.join(root, 'resources/templates/teams.j2'))
 
-    def __init__(self, template_path=DEFAULT_TEMPLATE_PATH):
+    def __init__(self, template_path=None):
         log.info(template_path)
+        if template_path is None:
+            template_path = TemplateComposer.DEFAULT_TEMPLATE_PATH
         if not os.path.isfile(template_path):
             raise MissingTemplatePathException('Template {} not exists'.format(template_path))
 
@@ -37,7 +39,8 @@ def __init__(self, template_path=DEFAULT_TEMPLATE_PATH):
         environment = Environment(loader=loader, trim_blocks=True)
         self.template = environment.get_template(template_name)
 
-    def compose_all(self, alarms_json):
-        rendered_templates = [self.template.render(status=json_alarm['status'], msg_text=json_alarm)
-                              for json_alarm in alarms_json]
-        return rendered_templates
+    def compose(self, json_alert):
+        return self.template.render(status=json_alert['status'], msg_text=json_alert)
+
+    def compose_all(self, json_alerts):
+        return [self.compose(json_alert) for json_alert in json_alerts]