Enhance hipchat plugin to color-code messages by alarm severity

Change-Id: Ic63d130c8545144be17f4be2faecf5c3b63375be
This commit is contained in:
Brad Klein 2017-01-18 15:12:56 -07:00
parent ac566b9f84
commit 833aad71e8
2 changed files with 115 additions and 3 deletions

View File

@ -36,6 +36,11 @@ from monasca_notification.plugins import abstract_notifier
"""
SEVERITY_COLORS = {"low": 'green',
'medium': 'gray',
'high': 'yellow',
'critical': 'red'}
class HipChatNotifier(abstract_notifier.AbstractNotifier):
def __init__(self, log):
@ -68,12 +73,15 @@ class HipChatNotifier(abstract_notifier.AbstractNotifier):
'metrics': notification.metrics}
hipchat_request = {}
hipchat_request['color'] = 'green'
hipchat_request['color'] = self._get_color(notification.severity.lower())
hipchat_request['message_format'] = 'text'
hipchat_request['message'] = json.dumps(body, indent=3)
return hipchat_request
def _get_color(self, severity):
return SEVERITY_COLORS.get(severity, 'purple')
def send_notification(self, notification):
"""Send the notification via hipchat
Posts on the given url
@ -109,9 +117,9 @@ class HipChatNotifier(abstract_notifier.AbstractNotifier):
self._log.info("Notification successfully posted.")
return True
else:
msg = "Received an HTTP code {} when trying to send to hipchat on URL {} with response {} ."
msg = "Received an HTTP code {} when trying to send to hipchat on URL {} with response {}."
self._log.error(msg.format(result.status_code, url, result.text))
return False
except Exception:
self._log.exception("Error trying to send to hipchat on URL {}".format(url))
self._log.exception("Error trying to send to hipchat on URL {}".format(url))
return False

View File

@ -0,0 +1,104 @@
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import mock
import Queue
import unittest
from monasca_notification import notification as m_notification
from monasca_notification.plugins import hipchat_notifier
def alarm(metrics):
return {"tenantId": "0",
"alarmId": "0",
"alarmDefinitionId": 0,
"alarmName": "test Alarm",
"alarmDescription": "test Alarm description",
"oldState": "OK",
"newState": "ALARM",
"severity": "CRITICAL",
"link": "some-link",
"lifecycleState": "OPEN",
"stateChangeReason": "I am alarming!",
"timestamp": 1429023453632,
"metrics": metrics}
class requestsResponse(object):
def __init__(self, status):
self.status_code = status
class TestHipchat(unittest.TestCase):
def setUp(self):
self.trap = Queue.Queue()
self.hipchat_config = {'timeout': 50}
def tearDown(self):
self.assertTrue(self.trap.empty())
def _http_post_200(self, url, data, **kwargs):
self.trap.put(url)
self.trap.put(data)
r = requestsResponse(200)
return r
@mock.patch('monasca_notification.plugins.hipchat_notifier.requests')
def notify(self, http_func, mock_requests):
mock_log = mock.MagicMock()
mock_log.warn = self.trap.put
mock_log.error = self.trap.put
mock_log.exception = self.trap.put
mock_requests.post = http_func
hipchat = hipchat_notifier.HipChatNotifier(mock_log)
hipchat.config(self.hipchat_config)
metric = []
metric_data = {'dimensions': {'hostname': 'foo1', 'service': 'bar1'}}
metric.append(metric_data)
alarm_dict = alarm(metric)
notification = m_notification.Notification(0, 'hipchat', 'hipchat notification',
'http://mock:3333/', 0, 0, alarm_dict)
self.trap.put(hipchat.send_notification(notification))
def test_hipchat_success(self):
"""hipchat success
"""
self.notify(self._http_post_200)
url = self.trap.get(timeout=1)
data = self.trap.get(timeout=1)
self.valid_hipchat_message(url, data)
return_value = self.trap.get()
self.assertTrue(return_value)
def valid_hipchat_message(self, url, data):
self.assertEqual(url, "http://mock:3333/")
self.assertEqual(data.get('color'), 'red')
self.assertEqual(data.get('message_format'), 'text')
message = json.loads(data.get('message'))
self.assertEqual(message.get('message'), 'I am alarming!')
self.assertEqual(message.get('alarm_name'), 'test Alarm')
self.assertEqual(message.get('alarm_description'), 'test Alarm description')