RaspberryPiで「ほんわか お知らせランプ」Part 2

RaspberryPiで監視をお助け

このシリーズでは、RaspberryPiを使ったお知らせランプを製作します。

第２回の今回は、「お知らせランプ」の情報源となるAlertを取得する Detector部分です。
この記事では、OpsGenie用のDetectorを作りますが、もちろん他のAlert源を使ってもOK。
その場合は、出力形式が同じになるように、Detectorを作成します。

OpsGenieって？

OpsGenieはメールやWebHook, WebAPIなどを通してアラームを受け取って、それをルールに沿って分類のうえ、Alertとして発報する通知管理のサービスです。
PagerDuty程メジャーではありませんが、AWSとの親和性も高く、受け取ったAlertのハンドリングの自由度に優れているのでDevOpsをやっている人には特にオススメです。

• あらかじめ登録した、チーム体制とシフトスケジュールに従って、担当チームの「シフト」の担当者に振り分けてくれるので、不必要に多数の人を道連れにしなくて済みます。
• メールや音声電話はもちろん、スマホアプリなど様々な手段で、応答するまでしつこく担当者に連絡してくるし、それでもダメなら、次の担当者にエスカレーションしてくれる、頼もしいヤツです。
• IntegrationによりAWS、Slack、HipChat、Nagios、Zabbix、NewReric...と様々なサービスやミドルウエアと連携できて、対応状況や履歴のシェアにも役立ちます。
• サービスは有償ですが、トライアル期間があるので、気軽に試す事ができます。（単一ユーザであれば、トライアル期間を超えても使えているみたいです。）

ひとつひとつのAlertの状態はOpsGenieによって管理されているので、お知らせランプはその状態を集約して表示したり、音を鳴らしたりするという役割を担当します。

OpsGenieでのAlermのレベル分け

OpsGenieでは、受け取ったAlermの情報を元にAlertを作ります。
緊急性のあるAlermを受け取ったら「error」というタグ付きのAlertを作成して、「お知らせランプ」が緊急度を識別できる様にします。
AlertへTagの付与は、OpsGenieの各Integrationの設定により可能です。

Detectorの仕事

Detectorは、OpsGenieのAlertAPIを使って、CloseされていないAlertを取得します。
取得した個々のAlertと、その詳細情報に含まれるTagを基にして、以下の形式でAlertのリストを作成します。

| 項目 | 内容 | コメント |
|---|---|---|---|
| tinyId | OpsGenieが付与するAlert番号 |
| acknowledged | True or False | Alertが認知済みならTrue |
| error | True or False | Alertに'error' というTagがセットされていたらTrue |
| new | True or False | 前回無かったAlertならTrue |

Detectorを動かしてみる

以下の手順でこのモジュールを試す事ができます。

1. OpsGenieのアカウントを取得する
2. Alertsメニューからテスト用Alertを登録する（必要に応じて'error'Tagを付ける）
   
3. Integrationsメニューから、「API」を選択して、API Keyを取得する
4. 上記 API Keyを第一引数にして、opsgenie.pyを実行する

実行結果は、以下のようになります。

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

そして、こちらがソースコード
Python初心者なので、暖かいご指摘を希望します。

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