WIP: keybase alert bot (receives hooks from prometheus alertmanager)

keybase_alerts.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from flask import Flask, abort, jsonify, render_template, request, session, Response
import asyncio
from aiohttp import web
import certifi
import os
import sys
import logging
import urllib3
from pykeybasebot import Bot, EventType, ChatClient
import pykeybasebot
import pykeybasebot.types.chat1 as chat1
from celery import Celery

log = logging.getLogger(__name__)
logging.basicConfig(
    stream=sys.stdout,
    level=logging.DEBUG,
    format="%(asctime)s.%(msecs)03d %(levelname)s {%(module)s} [%(funcName)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)


def parse_alert(alert):
    print(alert)
    message = ""
    if alert.get("labels") == None or alert.get("status") == None or alert.get("annotations") == None:
        return None
    message += "*{}*\t".format(alert["labels"]["alertname"])
    message += (
        "\nHost: "
        + alert["labels"]["host"]
        + " ("
        + alert["labels"]["ip"]
        + ") at "
        + alert["labels"]["hosting"]
        + "\n"
    )
    if alert["status"] == "firing":
        message += "Detected: " "" + alert["startsAt"] + " \n"
    if alert["status"] == "resolved":
        message += "Resolved: " "" + alert["endsAt"] + " \n"
    if alert["annotations"].get("description"):
        message += "_" + alert["annotations"]["description"] + "_"
    # if alert["annotations"].get("summary"):
    #    message += alert["annotations"]["summary"] + "\n"
    return message


def check_alert(alert_dict):
    if all(k in alert_dict.keys() for k in ("labels", "status", "annotations")):
        return True
    else:
        return False


def noop_handler(*args, **kwargs):
    pass


port = int(os.environ.get("PORT", "8080"))
host = os.environ.get("ADDRESS", "0.0.0.0")
bot_channels = [{"name": "my_keybase_team.alerts", "topic_type": "chat", "members_type": "team"}]
channel = chat1.ChatChannel(name="my_keybase_team.alerts", topic_name="general", members_type="team")
listen_options = {"wallet": False, "hide-exploding": False, "filter-channels": bot_channels}
loop = asyncio.get_event_loop()
alert_bot = pykeybasebot.Bot(username="alert_bot_username", paperkey="redacted paperkey", handler=noop_handler, loop=loop)


async def handle(request):
    data = await request.json()
    alerts = data["alerts"]
    if len(alerts) >= 1:
        for alert in alerts:
            if check_alert(alert):
                msg = parse_alert(alert)
                print(msg)
                await alert_bot.chat.send(channel, msg)
            else:
                continue
        return web.Response(text=str(alerts))
    else:
        return web.Response(text=None)


log.info("Starting alertmanager-keybase-bot, listening on {}:{}".format(host, port))
app = web.Application(loop=loop)
app.router.add_route("POST", "/", handle)
tasks = [asyncio.ensure_future(alert_bot.start(listen_options)) + asyncio.ensure_future(web.run_app(app))]
loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))