utils.py 文件源码

def check_process_exists_and_amqp_connected(name):
    processes = filter(lambda p: check_process_name(name, p),
                       psutil.process_iter())
    if not processes:
        critical("%s is not running" % name)
    for p in processes:
        try:
            connections = p.get_connections(kind='inet')
        except psutil.NoSuchProcess:
            continue
        found_amqp = (
            len(list(itertools.takewhile(lambda c: len(c.remote_address) <= 1
                                         or c.remote_address[1]
                                         != AMQP_PORT, connections)))
            != len(connections))
        if found_amqp:
            ok("%s is working." % name)
    critical("%s is not connected to AMQP" % name)