[PYTHON] Raspberry Pi "Honwaka Benachrichtigungslampe" Teil 2

Hilfe bei der Überwachung mit Raspberry Pi

In dieser Serie werden wir eine Benachrichtigungslampe mit Raspberry Pi erstellen.

Dieses Mal, das zweite Mal, ist es der Detektorteil, der den Alarm erfasst, der die Informationsquelle der "Benachrichtigungslampe" ist. In diesem Artikel erstellen wir einen Detektor für OpsGenie, aber Sie können natürlich auch andere Warnquellen verwenden. Erstellen Sie in diesem Fall einen Detektor, damit das Ausgabeformat identisch ist.

Was ist OpsGenie?

OpsGenie ist ein Benachrichtigungsverwaltungsdienst, der Alarme per E-Mail, WebHook, WebAPI usw. empfängt, sie gemäß den Regeln klassifiziert und als Warnungen ausgibt. Obwohl es nicht so wichtig ist wie PagerDuty, hat es eine hohe Affinität zu AWS und eine ausgezeichnete Handhabungsfreiheit für den empfangenen Alert. Dies gilt insbesondere für diejenigen, die DevOps ausführen. Ich empfehle es.

――Es wird der für die "Schicht" des verantwortlichen Teams verantwortlichen Person gemäß der im Voraus registrierten Teamstruktur und dem Schichtplan zugewiesen, sodass Sie nicht unnötig viele Personen mitnehmen müssen. ――Es ist ein zuverlässiger Mitarbeiter, der die verantwortliche Person dauerhaft kontaktiert, bis Sie mit verschiedenen Mitteln wie E-Mail und Sprachanruf sowie einer Smartphone-Anwendung antworten. Wenn dies nicht funktioniert, eskalieren Sie zur nächsten verantwortlichen Person.

• Mit Integration können Sie Links zu verschiedenen Diensten und Middleware wie AWS, Slack, HipChat, Nagios, Zabbix, New Reric ... erstellen und den Supportstatus und den Verlauf teilen. Es ist auch nützlich. ――Der Service ist kostenpflichtig, es gibt jedoch eine Testphase, sodass Sie ihn gerne ausprobieren können. (Wenn Sie ein einzelner Benutzer sind, können Sie ihn anscheinend auch nach Ablauf der Testphase verwenden.)

Da der Status jedes Alarms von OpsGenie verwaltet wird, ist die Benachrichtigungslampe dafür verantwortlich, den Status gemeinsam anzuzeigen und einen Ton zu erzeugen.

Allergennivellierung bei Ops Genie

OpsGenie erstellt einen Alarm basierend auf den erhaltenen Allermeinformationen. Wenn Sie eine dringende Warnung erhalten, erstellen Sie eine Warnung mit dem Tag "Fehler", damit die "Benachrichtigungslampe" die Dringlichkeit erkennen kann. Tags können zu Warnungen hinzugefügt werden, indem jede Integration von OpsGenie festgelegt wird.

Detektorarbeit

Der Detektor verwendet die Alert-API von OpsGenie, um einen nicht geschlossenen Alert zu erhalten. Erstellen Sie eine Liste von Warnungen im folgenden Format, basierend auf den einzelnen erhaltenen Warnungen und den in den detaillierten Informationen enthaltenen Tags.

Versuchen Sie, den Detektor zu bewegen

Sie können dieses Modul ausprobieren, indem Sie die folgenden Schritte ausführen.

1. Holen Sie sich ein OpsGenie-Konto
2. Registrieren Sie eine Testbenachrichtigung über das Menü "Warnungen" (fügen Sie bei Bedarf einen "Fehler" -Tag hinzu).
3. Wählen Sie im Menü "Integrationen" die Option "API", um den API-Schlüssel abzurufen
4. Führen Sie opsgenie.py mit dem obigen API-Schlüssel als erstem Argument aus

Das Ausführungsergebnis ist wie folgt.

$ ./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'}]

Und hier ist der Quellcode Ich bin neu in Python, daher würde ich gerne von Ihnen hören.

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