[PYTHON] Raspberry Pi "Lampe de notification Honwaka" Partie 2

Aide à surveiller avec Raspberry Pi

Dans cette série, nous allons créer une lampe de notification à l'aide de Raspberry Pi.

Cette fois, la deuxième fois, c'est la partie Détecteur qui acquiert l'alerte, qui est la source d'information de la "lampe de notification". Dans cet article, nous allons créer un détecteur pour OpsGenie, mais vous pouvez bien sûr utiliser d'autres sources d'alerte. Dans ce cas, créez un détecteur afin que le format de sortie soit le même.

Qu'est-ce qu'OpsGenie?

OpsGenie est un service de gestion des notifications qui reçoit les alarmes par e-mail, WebHook, WebAPI, etc., les classe selon les règles et les émet sous forme d'alertes. Bien qu'il ne soit pas aussi important que PagerDuty, il a une forte affinité avec AWS et une excellente liberté de gestion de l'alerte reçue, donc c'est surtout pour ceux qui font du DevOps. Je le recommande.

――Selon la structure de l'équipe et l'horaire des équipes enregistrés à l'avance, il sera attribué à la personne en charge du «quart» de l'équipe en charge, vous n'aurez donc pas à emmener un nombre inutilement important de personnes avec vous. ――C'est un gars fiable qui contactera constamment la personne responsable jusqu'à ce que vous répondiez par divers moyens tels que le courrier électronique et l'appel vocal, ainsi qu'une application pour smartphone, et si cela ne fonctionne pas, il passera au responsable suivant.

• Integration vous permet de vous connecter à divers services et middleware tels que AWS, Slack, HipChat, Nagios, Zabbix, New Reric ... et de partager le statut et l'historique du support. C'est également utile. ――Le service est payant, mais il y a une période d'essai, vous pouvez donc vous sentir libre de l'essayer. (Si vous êtes un seul utilisateur, il semble que vous puissiez l'utiliser même après le dépassement de la période d'essai.)

Étant donné que l'état de chaque alerte est géré par OpsGenie, la lampe de notification est chargée d'afficher l'état collectivement et d'émettre un son.

Nivellement d'allerm dans Ops Genie

OpsGenie crée une alerte basée sur les informations Allerm reçues. Lorsque vous recevez une alerte urgente, créez une alerte étiquetée «erreur» afin que le «voyant de notification» puisse identifier l'urgence. Des balises peuvent être ajoutées aux alertes en définissant chaque intégration d'OpsGenie.

Travail du détecteur

Le détecteur utilise l'API Alert d'OpsGenie pour obtenir une alerte non fermée. Créez une liste d'alertes au format suivant en fonction des alertes individuelles obtenues et des balises contenues dans les informations détaillées.

Essayez de déplacer le détecteur

Vous pouvez essayer ce module en suivant les étapes ci-dessous.

1. Obtenez un compte OpsGenie
2. Enregistrez une alerte de test depuis le menu Alertes (ajoutez une balise "erreur" si nécessaire)
3. Dans le menu Intégrations, sélectionnez "API" pour obtenir la clé API
4. Exécutez opsgenie.py avec la clé API ci-dessus comme premier argument

Le résultat de l'exécution est le suivant.

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

Et voici le code source Je suis nouveau sur Python, alors j'aimerais avoir de vos nouvelles.

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