Send email notifications on Supervisor programs died/failed/exited

supervisor_monitor.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import contextlib
import signal
import sys
from email.mime.text import MIMEText
from datetime import datetime
from smtplib import SMTP
from threading import Event

from supervisor import childutils

MAIL_FROM = 'from@example.com'
MAIL_TO = 'to@example.com'
GIS_MTA_MX = 'mta.gis.gehirn.jp'
GIS_API_TOKEN = 'dummy-token'
GIS_API_SECRET = 'dummy-secret'


class EventMonitor:

    def __init__(self):
        self._notification_interval = 30
        self._last_notification_at = None
        self._queued_messages = []
        self._should_stop = Event()

    def _send_notification(self):
        if not len(self._queued_messages):
            return

        body = '\r\n'.join(self._queued_messages) + '\r\n'
        message = MIMEText(body, _charset='utf8')
        message['Subject'] = 'Supervisor Notifications'
        message['From'] = MAIL_FROM
        message['To'] = MAIL_TO

        try:
            with contextlib.closing(SMTP(GIS_MTA_MX)) as conn:
                conn.starttls()
                conn.login(GIS_API_TOKEN, GIS_API_SECRET)
                conn.sendmail(MAIL_FROM,
                              [MAIL_TO],
                              message.as_string())
        except BaseException as why:
            sys.stderr.write('Failed to send email because of unexpected exception %r' % (why, ))  # noqa
            sys.stderr.flush()
        else:
            self._queued_messages[:] = []
            self._last_notification_at = datetime.utcnow()

    def _get_state_change_message(self, headers, payload):
        pheaders, _ = childutils.eventdata(payload+'\n')

        if headers['eventname'] == 'PROCESS_STATE_FATAL':
            return 'Process %(groupname)s:%(processname)s failed to start too many times' % pheaders  # noqa

        if headers['eventname'] == 'PROCESS_STATE_EXITED':
            if int(pheaders['expected']):
                return None
            return 'Process %(groupname)s:%(processname)s (pid %(pid)s) died unexpectedly' % pheaders  # noqa
            
    def _handle_state_event(self, headers, payload):
        message = self._get_state_change_message(headers, payload)
        if not message:
            return

        self._queued_messages.append(message)

    def _handle_tick_event(self, headers, payload):
        should_notify = False
        if not self._last_notification_at:
            should_notify = True
        else:
            now = datetime.utcnow()
            elapsed = (now - self._last_notification_at).total_seconds()
            should_notify = elapsed > self._notification_interval

        if should_notify:
            self._send_notification()

    def _handle_event(self, headers, payload):
        if headers['eventname'].startswith('TICK_'):
            self._handle_tick_event(headers, payload)
        else:
            self._handle_state_event(headers, payload)

    def _run(self):
        headers, payload = childutils.listener.wait(sys.stdin, sys.stdout)
        self._handle_event(headers, payload)
        childutils.listener.ok(sys.stdout)

    def start(self):
        self._queued_messages.append('supervisor monitor started')
        while not self._should_stop.is_set():
            self._run()

    def stop(self):
        self._should_stop.set()


def main():
    monitor = EventMonitor()

    def trap(signo, frame):
        monitor.stop()
    signal.signal(signal.SIGINT, trap)
    signal.signal(signal.SIGTERM, trap)

    monitor.start()


if __name__ == '__main__':
    main()