Merge pull request #226 from idealista/features/223

Retrying and limiting messages length when sending to teams

CHANGELOG.md

@@ -7,6 +7,7 @@ This project adheres to [Semantic Versioning](http://semver.org/) and [Keep a ch
 - *[#219](https://github.com/idealista/prom2teams/pull/219) Add timeouts to webhook request to prevent hanging tcp connections in case of network errors* @DanSipola
 ### Added
 - *[#222](https://github.com/idealista/prom2teams/pull/222) Add restrictive security context since the workload doesn't need more permissions to work.* @azman0101
+- *[#226](https://github.com/idealista/prom2teams/pull/226) Retrying policy* @blalop
 
 
 ## [3.0.0](https://github.com/idealista/prom2teams/tree/3.0.0)

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.7-alpine AS builder
+FROM python:3.8-alpine AS builder
 WORKDIR /prom2teams
 COPY LICENSE \
         MANIFEST.in \
@@ -11,7 +11,7 @@ COPY bin/ bin
 RUN apk add gcc libc-dev yaml-dev linux-headers --no-cache \
         && python setup.py bdist_wheel
 
-FROM python:3.7-alpine
+FROM python:3.8-alpine
 LABEL maintainer="labs@idealista.com"
 EXPOSE 8089
 WORKDIR /opt/prom2teams

README.md

@@ -10,6 +10,9 @@
 
 **prom2teams** is a Web server built with Python that receives alert notifications from a previously configured [Prometheus Alertmanager](https://github.com/prometheus/alertmanager) instance and forwards it to [Microsoft Teams](https://teams.microsoft.com/) using defined connectors.
 
+It presents grouping of alerts, labels/annotations exclusion and a Teams' alarm retry policy among its key features.
+
+
 - [Getting Started](#getting-started)
 	- [Prerequisities](#prerequisites)
 	- [Installing](#installing)
@@ -32,7 +35,7 @@
 
 ### Prerequisites
 
-The application has been tested with _Prometheus 2.2.1_, _Python 3.7.0_ and _pip 9.0.1_.
+The application has been tested with _Prometheus 2.2.1_, _Python 3.8.0_ and _pip 9.0.1_.
 
 Newer versions of _Prometheus/Python/pip_ should work but could also present issues.
 
@@ -188,7 +191,7 @@ Another approach is to provide yourself the `module` file [module example](bin/w
 
 The config file is an [INI file](https://docs.python.org/3/library/configparser.html#supported-ini-file-structure) and should have the structure described below:
 
-```
+```ini
 [Microsoft Teams]
 # At least one connector is required here
 Connector: <webhook url>
@@ -214,6 +217,11 @@ Excluded: <Coma separated list of labels to ignore>
 
 [Annotations]
 Excluded: <Comma separated list of annotations to ignore>
+
+[Teams Client]
+RetryEnable: <Enables teams client retry policy> # defaults to false
+RetryWaitTime: <Wait time between retries> # default: 60 secs
+MaxPayload: <Teams client payload limit in bytes> # default: 24KB
 ```
 
 **Note:** Grouping alerts works since v2.2.0
@@ -273,7 +281,7 @@ $ ./test.sh
 ```
 
 ## Built With
-![Python 3.6.2](https://img.shields.io/badge/Python-3.6.2-green.svg)
+![Python 3.8.0](https://img.shields.io/badge/Python-3.8.0-green.svg)
 ![pip 9.0.1](https://img.shields.io/badge/pip-9.0.1-green.svg)
 
 ## Versioning

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)
+            alarms = map_and_group(alerts, self.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)
         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, timeout=(5,20))
-    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)
+log = 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(log, 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)
+
+        log.debug('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, timeout=(5,20))
+        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,10 +17,8 @@ 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):

prom2teams/app/versions/v2/namespace.py

@@ -15,11 +15,11 @@ 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):

prom2teams/teams/alarm_mapper.py

@@ -1,9 +1,15 @@
+from collections import defaultdict, OrderedDict
+
 from prom2teams.teams.teams_alarm_schema import TeamsAlarm, TeamsAlarmSchema
-from collections import defaultdict
+
+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 +23,65 @@ 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}
-
-            alarm = TeamsAlarm(name, status.lower(), severity, summary,
-                               instance, description, fingerprint, extra_labels,
-                               extra_annotations)
-            json_alarm = schema.dump(alarm)
-            teams_alarms.append(json_alarm)
+        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
+
+
+def _map_group(alert_group, compose, payload_limit):
+    schema = TeamsAlarmSchema()
+    combined_alerts = []
+    teams_alarms = []
+    for alert in alert_group:
+        json_alarm = schema.dump(_combine_alerts_to_alarm([*combined_alerts, alert]))
+        if len(compose(json_alarm).encode('utf-8')) > payload_limit:
+            teams_alarms.append(schema.dump(_combine_alerts_to_alarm([alert])))
+            teams_alarms.append(schema.dump(_combine_alerts_to_alarm(combined_alerts)))
+            combined_alerts.clear()
+            json_alarm = None
+        else:
+            combined_alerts.append(alert)
+
+    if json_alarm:
+        teams_alarms.append(json_alarm)
     return teams_alarms
 
 
-def teams_visualization(feature):
-    feature.sort()
+def _combine_alerts_to_alarm(alerts):
+    dicts = list(map(vars, alerts))
+    groupable = _combine_groupable_fields(dicts)
+    extra = _combine_extra_fields(dicts)
+    return _map_dict_alert_to_alarm({**groupable, **extra})
+
+
+def _map_dict_alert_to_alarm(alert):
+    return TeamsAlarm(alert['name'], alert['status'].lower(), 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 _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(OrderedDict.fromkeys(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]

requirements.txt

@@ -11,3 +11,4 @@ DeepDiff==4.3.0
 zipp==1.2.0
 MarkupSafe==1.1.1
 pyrsistent==0.16.0
+tenacity==6.2.0