[PYTHON] "Honwaka Notification Lamp" on Raspberry Pi Part 2

Help monitor with Raspberry Pi

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.

What is OpsGenie?

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.

Allerm leveling in Ops Genie

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.

Detector work

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.

Try moving the Detector

You can try this module by following the steps below.

1. Get an OpsGenie account
2. Register a test Alert from the Alerts menu (add a'error'Tag if necessary)
3. From the Integrations menu, select "API" to get the API Key
4. Execute opsgenie.py with the above API Key as the first argument

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())