External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-02-16 10:40:01 -03:30
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2959 from crab86/devel

Browse Source 
Add Grafana notification type

Reviewed-by: https://github.com/softwarefactory-project-zuul[bot]
This commit is contained in:
softwarefactory-project-zuul[bot]
2019-01-21 17:21:14 +00:00
committed by GitHub
parent c48c8c04f4 ebe0ded9c2
commit d021c253aa
9 changed files with 314 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
awx/main/migrations/0055_v340_add_grafana_notification.py Normal file
View File

@@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.11.16 on 2019-01-20 12:00
from __future__ import unicode_literals
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('main', '0054_v340_workflow_convergence'),
]
operations = [
migrations.AlterField(
model_name='notification',
name='notification_type',
field=models.CharField(choices=[('email', 'Email'), ('slack', 'Slack'), ('twilio', 'Twilio'), ('pagerduty', 'Pagerduty'), ('grafana', 'Grafana'), ('hipchat', 'HipChat'), ('webhook', 'Webhook'), ('mattermost', 'Mattermost'), ('rocketchat', 'Rocket.Chat'), ('irc', 'IRC')], max_length=32),
),
migrations.AlterField(
model_name='notificationtemplate',
name='notification_type',
field=models.CharField(choices=[('email', 'Email'), ('slack', 'Slack'), ('twilio', 'Twilio'), ('pagerduty', 'Pagerduty'), ('grafana', 'Grafana'), ('hipchat', 'HipChat'), ('webhook', 'Webhook'), ('mattermost', 'Mattermost'), ('rocketchat', 'Rocket.Chat'), ('irc', 'IRC')], max_length=32),
),
]

2
awx/main/models/notifications.py
View File

@@ -20,6 +20,7 @@ from awx.main.notifications.pagerduty_backend import PagerDutyBackend
from awx.main.notifications.hipchat_backend import HipChatBackend
from awx.main.notifications.webhook_backend import WebhookBackend
from awx.main.notifications.mattermost_backend import MattermostBackend
from awx.main.notifications.grafana_backend import GrafanaBackend
from awx.main.notifications.rocketchat_backend import RocketChatBackend
from awx.main.notifications.irc_backend import IrcBackend
from awx.main.fields import JSONField
@@ -36,6 +37,7 @@ class NotificationTemplate(CommonModelNameNotUnique):
('slack', _('Slack'), SlackBackend),
('twilio', _('Twilio'), TwilioBackend),
('pagerduty', _('Pagerduty'), PagerDutyBackend),
('grafana', _('Grafana'), GrafanaBackend),
('hipchat', _('HipChat'), HipChatBackend),
('webhook', _('Webhook'), WebhookBackend),
('mattermost', _('Mattermost'), MattermostBackend),

66
awx/main/notifications/grafana_backend.py Normal file
View File

@@ -0,0 +1,66 @@
# Copyright (c) 2016 Ansible, Inc.
# All Rights Reserved.
import datetime
import logging
import requests
import dateutil.parser as dp
from django.utils.encoding import smart_text
from django.utils.translation import ugettext_lazy as _
from awx.main.notifications.base import AWXBaseEmailBackend
logger = logging.getLogger('awx.main.notifications.grafana_backend')
class GrafanaBackend(AWXBaseEmailBackend):
init_parameters = {"grafana_url": {"label": "Grafana URL", "type": "string"},
"grafana_key": {"label": "Grafana API Key", "type": "password"}}
recipient_parameter = "grafana_url"
sender_parameter = None
def __init__(self, grafana_key,dashboardId=None, panelId=None, annotation_tags=None, grafana_no_verify_ssl=False, isRegion=True,
fail_silently=False, **kwargs):
super(GrafanaBackend, self).__init__(fail_silently=fail_silently)
self.grafana_key = grafana_key
self.dashboardId = dashboardId
self.panelId = panelId
self.annotation_tags = annotation_tags if annotation_tags is not None else []
self.grafana_no_verify_ssl = grafana_no_verify_ssl
self.isRegion = isRegion
def format_body(self, body):
return body
def send_messages(self, messages):
sent_messages = 0
for m in messages:
grafana_data = {}
grafana_headers = {}
try:
epoch=datetime.datetime.utcfromtimestamp(0)
grafana_data['time'] = int((dp.parse(m.body['started']).replace(tzinfo=None) - epoch).total_seconds() * 1000)
grafana_data['timeEnd'] = int((dp.parse(m.body['finished']).replace(tzinfo=None) - epoch).total_seconds() * 1000)
except ValueError:
logger.error(smart_text(_("Error converting time {} or timeEnd {} to int.").format(m.body['started'],m.body['finished'])))
if not self.fail_silently:
raise Exception(smart_text(_("Error converting time {} and/or timeEnd {} to int.").format(m.body['started'],m.body['finished'])))
grafana_data['isRegion'] = self.isRegion
grafana_data['dashboardId'] = self.dashboardId
grafana_data['panelId'] = self.panelId
grafana_data['tags'] = self.annotation_tags
grafana_data['text'] = m.subject
grafana_headers['Authorization'] = "Bearer {}".format(self.grafana_key)
grafana_headers['Content-Type'] = "application/json"
r = requests.post("{}/api/annotations".format(m.recipients()[0]),
json=grafana_data,
headers=grafana_headers,
verify=(not self.grafana_no_verify_ssl))
if r.status_code >= 400:
logger.error(smart_text(_("Error sending notification grafana: {}").format(r.text)))
if not self.fail_silently:
raise Exception(smart_text(_("Error sending notification grafana: {}").format(r.text)))
sent_messages += 1
return sent_messages

113
awx/main/tests/unit/notifications/test_grafana.py Normal file
View File

@@ -0,0 +1,113 @@
from unittest import mock
import datetime as dt
from django.core.mail.message import EmailMessage
import awx.main.notifications.grafana_backend as grafana_backend
def test_send_messages():
with mock.patch('awx.main.notifications.grafana_backend.requests') as requests_mock:
requests_mock.post.return_value.status_code = 200
m={}
m['started'] = dt.datetime.utcfromtimestamp(60).isoformat()
m['finished'] = dt.datetime.utcfromtimestamp(120).isoformat()
m['subject'] = "test subject"
backend = grafana_backend.GrafanaBackend("testapikey")
message = EmailMessage(m['subject'],{"started":m['started'],"finished":m['finished']}, [], ['https://example.com', ])
sent_messages = backend.send_messages([message, ])
requests_mock.post.assert_called_once_with(
'https://example.com/api/annotations',
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer testapikey'},
json={'tags': [], 'text': 'test subject', 'isRegion': True, 'timeEnd': 120000, 'panelId': None, 'time': 60000, 'dashboardId': None},
verify=True)
assert sent_messages == 1
def test_send_messages_with_no_verify_ssl():
with mock.patch('awx.main.notifications.grafana_backend.requests') as requests_mock:
requests_mock.post.return_value.status_code = 200
m={}
m['started'] = dt.datetime.utcfromtimestamp(60).isoformat()
m['finished'] = dt.datetime.utcfromtimestamp(120).isoformat()
m['subject'] = "test subject"
backend = grafana_backend.GrafanaBackend("testapikey",grafana_no_verify_ssl=True)
message = EmailMessage(m['subject'],{"started":m['started'],"finished":m['finished']}, [], ['https://example.com', ])
sent_messages = backend.send_messages([message, ])
requests_mock.post.assert_called_once_with(
'https://example.com/api/annotations',
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer testapikey'},
json={'tags': [], 'text': 'test subject', 'isRegion': True, 'timeEnd': 120000, 'panelId': None,'time': 60000, 'dashboardId': None},
verify=False)
assert sent_messages == 1
def test_send_messages_with_dashboardid():
with mock.patch('awx.main.notifications.grafana_backend.requests') as requests_mock:
requests_mock.post.return_value.status_code = 200
m={}
m['started'] = dt.datetime.utcfromtimestamp(60).isoformat()
m['finished'] = dt.datetime.utcfromtimestamp(120).isoformat()
m['subject'] = "test subject"
backend = grafana_backend.GrafanaBackend("testapikey",dashboardId=42)
message = EmailMessage(m['subject'],{"started":m['started'],"finished":m['finished']}, [], ['https://example.com', ])
sent_messages = backend.send_messages([message, ])
requests_mock.post.assert_called_once_with(
'https://example.com/api/annotations',
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer testapikey'},
json={'tags': [], 'text': 'test subject', 'isRegion': True, 'timeEnd': 120000, 'panelId': None, 'time': 60000, 'dashboardId': 42},
verify=True)
assert sent_messages == 1
def test_send_messages_with_panelid():
with mock.patch('awx.main.notifications.grafana_backend.requests') as requests_mock:
requests_mock.post.return_value.status_code = 200
m={}
m['started'] = dt.datetime.utcfromtimestamp(60).isoformat()
m['finished'] = dt.datetime.utcfromtimestamp(120).isoformat()
m['subject'] = "test subject"
backend = grafana_backend.GrafanaBackend("testapikey",dashboardId=None,panelId=42)
message = EmailMessage(m['subject'],{"started":m['started'],"finished":m['finished']}, [], ['https://example.com', ])
sent_messages = backend.send_messages([message, ])
requests_mock.post.assert_called_once_with(
'https://example.com/api/annotations',
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer testapikey'},
json={'tags': [], 'text': 'test subject', 'isRegion': True, 'timeEnd': 120000, 'panelId': 42, 'time': 60000, 'dashboardId': None},
verify=True)
assert sent_messages == 1
def test_send_messages_with_bothids():
with mock.patch('awx.main.notifications.grafana_backend.requests') as requests_mock:
requests_mock.post.return_value.status_code = 200
m={}
m['started'] = dt.datetime.utcfromtimestamp(60).isoformat()
m['finished'] = dt.datetime.utcfromtimestamp(120).isoformat()
m['subject'] = "test subject"
backend = grafana_backend.GrafanaBackend("testapikey",dashboardId=42,panelId=42)
message = EmailMessage(m['subject'],{"started":m['started'],"finished":m['finished']}, [], ['https://example.com', ])
sent_messages = backend.send_messages([message, ])
requests_mock.post.assert_called_once_with(
'https://example.com/api/annotations',
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer testapikey'},
json={'tags': [], 'text': 'test subject', 'isRegion': True, 'timeEnd': 120000, 'panelId': 42, 'time': 60000, 'dashboardId': 42},
verify=True)
assert sent_messages == 1
def test_send_messages_with_tags():
with mock.patch('awx.main.notifications.grafana_backend.requests') as requests_mock:
requests_mock.post.return_value.status_code = 200
m={}
m['started'] = dt.datetime.utcfromtimestamp(60).isoformat()
m['finished'] = dt.datetime.utcfromtimestamp(120).isoformat()
m['subject'] = "test subject"
backend = grafana_backend.GrafanaBackend("testapikey",dashboardId=None,panelId=None,annotation_tags=["ansible"])
message = EmailMessage(m['subject'],{"started":m['started'],"finished":m['finished']}, [], ['https://example.com', ])
sent_messages = backend.send_messages([message, ])
requests_mock.post.assert_called_once_with(
'https://example.com/api/annotations',
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer testapikey'},
json={'tags': ['ansible'], 'text': 'test subject', 'isRegion': True, 'timeEnd': 120000, 'panelId': None, 'time': 60000, 'dashboardId': None},
verify=True)
assert sent_messages == 1