In this series, we will make a notification lamp using Raspberry Pi.
This time, the second time, it is the Detector part that acquires the Alert, which is the information source of the "notification lamp". In this article, we will create a Detector for OpsGenie, but of course you can use other Alert sources. In that case, create a Detector so that the output format is the same.
OpsGenie is a notification management service that receives alarms via email, WebHook, WebAPI, etc., classifies them according to the rules, and issues them as Alerts. Although it is not as major as PagerDuty, it has a high affinity with AWS and has excellent handling freedom of the received Alert, so it is especially for those who are doing DevOps I recommend it.
――According to the team structure and shift schedule registered in advance, it will be assigned to the person in charge of "shift" of the team in charge, so you do not have to take an unnecessarily large number of people with you. ――It is a reliable person who will contact the person in charge persistently until you answer by various means such as e-mail and voice call, as well as a smartphone application, and if that does not work, escalate to the next person in charge. -Integration allows you to link with various services and middleware such as AWS, Slack, HipChat, Nagios, Zabbix, New Reric ... and share the support status and history. It is also useful. ――The service is charged, but there is a trial period, so you can feel free to try it. (If you are a single user, it seems that you can use it even after the trial period is exceeded.)
Since the status of each Alert is managed by OpsGenie, the notification lamp is responsible for displaying the status collectively and making a sound.
OpsGenie creates an Alert based on the received Allerm information. When you receive an urgent Alert, create an Alert tagged with "error" so that the "Notification Lamp" can identify the urgency. Tag can be added to Alert by setting each Integral of OpsGenie.
The Detector uses OpsGenie's Alert API to get an unclosed Alert. Create a list of Alerts in the following format based on the individual Alerts obtained and the Tags contained in the detailed information.
item | Contents | comment |
---|---|---|
tinyId | Alert number given by Ops Genie | |
acknowledged | True or False | True if Alert is known |
error | True or False | To Alert'error'True if the Tag is set |
new | True or False | True if there was no Alert last time |
You can try this module by following the steps below.
The execution result is as follows.
$ ./opsgenie.py ********-****-****-****-************
-- Alert List (1st) ----------------------------------
[{'new': True, 'error': False, 'acknowledged': False, 'tinyId': u'15'}, {'new': True, 'error': True, 'acknowledged': False, 'tinyId': u'14'}]
-- Alert List (2nd) ----------------------------------
[{'new': False, 'error': False, 'acknowledged': False, 'tinyId': u'15'}, {'new': False, 'error': True, 'acknowledged': False, 'tinyId': u'14'}]
And here is the source code I'm new to Python, so I'd like to hear from you.
opsgenie.py
#!/usr/bin/env python
#coding:utf-8
import urllib
import urllib2
import json
import time
class OpsGenie:
"""OpsGenie access
We should use If-Modified-Since header if OpsGenie supports."""
def __init__(self, url, apikey):
self.url = url
self.apikey = apikey
self.alert_list = []
# [{"tinyId":0, "acknowledged":False, "error":False, "new": False}, ...]
def _get(self, uri):
# print("url + params: %s" % uri)
try:
response = urllib2.urlopen(uri)
except urllib2.URLError, e:
error = "Error "
if hasattr(e, 'code'):
error = str(e.code)
if hasattr(e, 'reason'):
error = error + ": " + str(e.reason)
print(error)
else:
return response.read()
return None
def get_alerts(self, status="open"):
params = urllib.urlencode(
{
"apiKey": self.apikey,
"status": "open",
"limit": "100"
})
resp = self._get(self.url + params)
if resp is None:
return None
else:
data = json.loads(resp)
return data["alerts"]
def get_alert_detail(self, tiny_id):
params = urllib.urlencode(
{
"apiKey": self.apikey,
"tinyId": tiny_id
})
resp = self._get(self.url + params)
if resp is None:
return None
else:
data = json.loads(resp)
return data
def get_alert_tags(self, tiny_id):
data = self.get_alert_detail(tiny_id)
return data["tags"]
def detector(self):
# get open alerts
alerts = self.get_alerts()
latest_list = []
for a in alerts:
latest_list.append({"tinyId": a["tinyId"], "acknowledged": a["acknowledged"], "error": "", "new": ""})
# set new flags and error flags
for latest in latest_list:
for previous in self.alert_list:
if previous["tinyId"] == latest["tinyId"]:
# existing
latest["new"] = False
latest["error"] = previous["error"]
break
else:
# new
latest["new"] = True
# set error flag if error tag is found
tags = self.get_alert_tags(int(latest["tinyId"]))
latest["error"] = ("error" in tags)
# print("latest_list =" + str(latest_list))
self.alert_list = latest_list
return latest_list
if __name__ == '__main__':
import sys
URL = 'https://api.opsgenie.com/v1/json/alert?'
if len(sys.argv) != 2:
print("Usage: %s 'api-key for your OpsGenie account'." % sys.argv[0])
exit()
API_KEY = sys.argv[1]
o = OpsGenie(URL, API_KEY)
alerts = o.get_alerts()
if alerts is None:
print("Check your api_key or network connection.")
exit()
# print("-- Alerts --------------------------------------------")
# print(json.dumps(alerts, sort_keys=True, indent=2))
for a in alerts:
tiny_id = a["tinyId"]
detail = o.get_alert_detail(tiny_id)
# print("-- Alerts Detail -------------------------------------------")
# print(json.dumps(detail, sort_keys=True, indent=2))
print("-- Alert List (1st) ----------------------------------")
print(o.detector())
time.sleep(3)
print("-- Alert List (2nd) ----------------------------------")
# All new flag should be "False"
print(o.detector())
Recommended Posts