160 lines
6.2 KiB
Python
160 lines
6.2 KiB
Python
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
|