-
Notifications
You must be signed in to change notification settings - Fork 171
/
alerta_msteams.py
159 lines (138 loc) · 6.18 KB
/
alerta_msteams.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
import json
import logging
import os
import pymsteams
import requests
from alerta.plugins import PluginBase
try:
from alerta.plugins import app # alerta >= 5.0
except ImportError:
from alerta.app import app # alerta < 5.0
LOG = logging.getLogger('alerta.plugins.msteams')
try:
from jinja2 import Template
except Exception as e:
LOG.error(
'MS Teams: ERROR - Jinja template error: %s, template functionality will be unavailable', e)
MS_TEAMS_COLORS_MAP = app.config.get('MS_TEAMS_COLORS_MAP', {})
MS_TEAMS_DEFAULT_COLORS_MAP = {'security': '000000',
'critical': 'D8122A',
'major': 'EA680F',
'minor': 'FFBE1E',
'warning': '1E90FF'}
MS_TEAMS_DEFAULT_COLOR = '00AA5A'
MS_TEAMS_DEFAULT_TIMEOUT = 7 # pymsteams http_timeout
class SendConnectorCardMessage(PluginBase):
def __init__(self, name=None):
# override user-defined severities(colors)
self._colors = MS_TEAMS_DEFAULT_COLORS_MAP
self._colors.update(MS_TEAMS_COLORS_MAP)
super().__init__(name)
def _load_template(self, templateFmt):
try:
if os.path.exists(templateFmt):
with open(templateFmt) as f:
template = Template(f.read())
else:
template = Template(templateFmt)
return template
except Exception as e:
LOG.error('MS Teams: ERROR - Template init failed: %s', e)
return
def pre_receive(self, alert, **kwargs):
return alert
def post_receive(self, alert, **kwargs):
MS_TEAMS_WEBHOOK_URL = self.get_config(
'MS_TEAMS_WEBHOOK_URL', default='', type=str, **kwargs)
MS_TEAMS_SUMMARY_FMT = self.get_config(
'MS_TEAMS_SUMMARY_FMT', default=None, type=str, **kwargs) # Message summary(title) format
MS_TEAMS_TEXT_FMT = self.get_config(
'MS_TEAMS_TEXT_FMT', default=None, type=str, **kwargs) # Message text format
# json/Jinja2 MS teams messagecard payload
MS_TEAMS_PAYLOAD = self.get_config(
'MS_TEAMS_PAYLOAD', default=None, type=str, **kwargs)
MS_TEAMS_INBOUNDWEBHOOK_URL = self.get_config(
'MS_TEAMS_INBOUNDWEBHOOK_URL', default=None, type=str, **kwargs) # webhook url for connectorcard actions
# X-API-Key (needs webhook.write permission)
MS_TEAMS_APIKEY = self.get_config(
'MS_TEAMS_APIKEY', default=None, type=str, **kwargs)
DASHBOARD_URL = self.get_config(
'DASHBOARD_URL', default='', type=str, **kwargs)
if alert.repeat:
return
color = self._colors.get(alert.severity, MS_TEAMS_DEFAULT_COLOR)
url = '{}/#/alert/{}'.format(DASHBOARD_URL, alert.id)
template_vars = {
'alert': alert,
'config': app.config,
'color': color,
'url': url
}
if MS_TEAMS_INBOUNDWEBHOOK_URL and MS_TEAMS_APIKEY:
# Add X-API-Key header for teams(webhook) HttpPOST actions
template_vars['headers'] = '[ {{ "name": "X-API-Key", "value": "{}" }} ]'.format(
MS_TEAMS_APIKEY)
template_vars['webhook_url'] = MS_TEAMS_INBOUNDWEBHOOK_URL
if MS_TEAMS_PAYLOAD:
# Use "raw" json ms teams message card format
payload_template = self._load_template(MS_TEAMS_PAYLOAD)
try:
card_json = payload_template.render(**template_vars)
LOG.debug('MS Teams payload(json): %s', card_json)
# Try to check that we've valid json
json.loads(card_json)
except Exception as e:
LOG.error(
'MS Teams: ERROR - Template(PAYLOAD) render failed: %s', e)
return
else:
# Use pymsteams to format/construct message card
if MS_TEAMS_SUMMARY_FMT:
template = self._load_template(MS_TEAMS_SUMMARY_FMT)
try:
summary = template.render(**template_vars)
except Exception as e:
LOG.error('MS Teams: ERROR - Template render failed: %s', e)
return
else:
summary = ('<b>[{status}] {environment} {service} {severity} - <i>{event} on {resource}</i></b>').format(
status=alert.status.capitalize(),
environment=alert.environment.upper(),
service=','.join(alert.service),
severity=alert.severity.capitalize(),
event=alert.event,
resource=alert.resource
)
if MS_TEAMS_TEXT_FMT:
txt_template = self._load_template(MS_TEAMS_TEXT_FMT)
try:
text = txt_template.render(**template_vars)
except Exception as e:
LOG.error(
'MS Teams: ERROR - Template(TEXT_FMT) render failed: %s', e)
return
else:
text = alert.text
LOG.debug('MS Teams payload: %s', summary)
try:
if MS_TEAMS_PAYLOAD:
# Use requests.post to send raw json message card
LOG.debug('MS Teams sending(json payload) POST to %s',
MS_TEAMS_WEBHOOK_URL)
r = requests.post(
MS_TEAMS_WEBHOOK_URL, data=card_json, timeout=MS_TEAMS_DEFAULT_TIMEOUT)
LOG.debug('MS Teams response: %s / %s' %
(r.status_code, r.text))
else:
# Use pymsteams to send card
msTeamsMessage = pymsteams.connectorcard(
hookurl=MS_TEAMS_WEBHOOK_URL, http_timeout=MS_TEAMS_DEFAULT_TIMEOUT)
msTeamsMessage.title(summary)
msTeamsMessage.text(text)
msTeamsMessage.addLinkButton('Open in Alerta', url)
msTeamsMessage.color(color)
msTeamsMessage.send()
except Exception as e:
raise RuntimeError('MS Teams: ERROR - %s', e)
def status_change(self, alert, status, text, **kwargs):
return