"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.

Blue.GIF

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.

SystemDiagram2.png

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.

OpsGenieWeb

――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.

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

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

Recommended Posts

"Honwaka Notification Lamp" on Raspberry Pi Part 2
"Honwaka Notification Lamp" on Raspberry Pi Part 1
"Honwaka Notification Lamp" on Raspberry Pi Part 3
Matrix multiplication on Raspberry Pi GPU (Part 2)
pigpio on Raspberry pi
Cython on Raspberry Pi
Introduced pyenv on Raspberry Pi
Use NeoPixel on Raspberry Pi
Install OpenCV4 on Raspberry Pi 3
Install TensorFlow 1.15.0 on Raspberry Pi
Testing uart communication on Raspberry Pi
raspberry pi 1 model b, node-red part 17
MQTT on Raspberry Pi and Mac
raspberry pi 4 centos7 install on docker
Install ghoto2 on Raspberry Pi (memo)
Try using ArUco on Raspberry Pi
OpenCV installation procedure on Raspberry Pi
Power on / off Raspberry pi on Arduino
Detect switch status on Raspberry Pi 3
Install OpenMedia Vault 5 on Raspberry Pi 4
L Chika on Raspberry Pi C #
Build wxPython on Ubuntu 20.04 on raspberry pi 4
Phone notification when surveillance camera motion is detected on Raspberry Pi
Detect "brightness" using python on Raspberry Pi 3!
USB boot on Raspberry Pi 4 Model B
Enable UART + serial communication on Raspberry Pi
Adafruit Python BluefruitLE works on Raspberry Pi.
Accelerate Deep Learning on Raspberry Pi 4 CPU
Set swap space on Ubuntu on Raspberry Pi
Programming normally with Node-RED programming on Raspberry Pi 3
Use the Grove sensor on the Raspberry Pi
Install 64-bit OS (bate) on Raspberry Pi
Install docker-compose on 64-bit Raspberry Pi OS
Run servomotor on Raspberry Pi 3 using python
Working with sensors on Mathematica on Raspberry Pi
Build OpenCV-Python environment on Raspberry Pi B +
Detect temperature using python on Raspberry Pi 3!
Mount Windows shared folder on Raspberry Pi
How to install NumPy on Raspberry Pi
I installed OpenCV-Python on my Raspberry Pi
Working with GPS on Raspberry Pi 3 Python
Easy Raspberry Pi GUI App Development Beginner Part 1
Why detectMultiScale () is slow on Raspberry Pi B +
Detect slide switches using python on Raspberry Pi 3!
Build a Django environment on Raspberry Pi (MySQL)
Try using a QR code on a Raspberry Pi
Detect magnet switches using python on Raspberry Pi 3!
Raspberry Pi backup
Enjoy electronic work with GPIO on Raspberry Pi
Power on / off your PC with raspberry pi
Easy Raspberry Pi GUI App Development Beginner Part 2
Use Grove-Temperature & Humidity Sensor (DHT11) on Raspberry Pi
Make DHT11 available on Raspberry Pi + python (memo)
Beginning cross-compilation for Raspberry Pi Zero on Ubuntu
Sound the buzzer using python on Raspberry Pi 3!
Create a visitor notification system using Raspberry Pi
Play with your Ubuntu desktop on your Raspberry Pi 4
Display CPU temperature every 5 seconds on Raspberry Pi 4
Introduced Ceph on Kubernetes on Raspberry Pi 4B (ARM64)
Connect to MySQL with Python on Raspberry Pi
Build a Python development environment on Raspberry Pi