Monitor Elastic MapReduce jobflows for state changes, and announce them using OS X's say command.

monitor-jobflows.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#  monitor-jobflows.py
#

"""
Monitors for changes in any jobflow's status and announces them.
"""

import optparse
import os
import sys
import time
import re

import boto

DEFAULT_POLL = 10   # poll interval in seconds

TERMINAL_STATES = ['FAILED', 'TERMINATED']


def check_status(poll_interval=DEFAULT_POLL):
    emr = boto.connect_emr()
    jobflows = {}

    for j in emr.describe_jobflows():
        jobflows[j.jobflowid] = j.state

    say('Monitoring jobflows.')
    time.sleep(0.25)
    n_active = sum(1 for s in jobflows.itervalues() if s not in
            TERMINAL_STATES)
    say('%d jobflow%s currently active.' % (
            n_active,
            's' if n_active != 1 else ''  # pluralize
        ))

    while True:
        time.sleep(poll_interval)
        for j in emr.describe_jobflows():
            s = j.state
            jid = j.jobflowid

            if jid not in jobflows:
                jobflows[jid] = s
                say('New jobflow %s has state %s.' % (jid, s))

            elif s != jobflows[jid]:
                if s in TERMINAL_STATES:
                    say('Jobflow %s has terminal state %s.' % (jid, s))
                else:
                    say('Jobflow %s has reached state %s.' % (jid, s))

                jobflows[jid] = s


def say(s):
    print s

    # refer to jobflows by the first three letters of their id
    s = re.sub('j-(.)(.)(.)[A-Z0-9]+', r'\1-\2-\3', s)

    os.system("say '%s'" % s)


def _create_option_parser():
    usage = \
"""%prog [options]

Monitors and announces changes in any EMR jobflow status."""

    parser = optparse.OptionParser(usage)

    return parser


def main(argv):
    parser = _create_option_parser()
    (options, args) = parser.parse_args(argv)

    if args:
        parser.print_help()
        sys.exit(1)

    try:
        check_status(*args)
    except KeyboardInterrupt:
        pass


if __name__ == '__main__':
    main(sys.argv[1:])