Troubles using custom notification client - Lava-users

8 Dec 2017


      Hi,
...
From a client side, I'd like to be notified that jobs are achieved in a
distant lava server. I started with the example provided at
https://validation.linaro.org/static/docs/v2/data-export.html#write-your-own...
1. In a 1st try I've changed the example - because I'm using port 10080 -
   and it works without the <lookup_publisher> . For that, I've hard coded
   the url returned by lookup_publisher but this only prints out one status at
   a time, ie :I needed to restart the script each time to have updates :
   "Submited"->*Ctrl-C*->Running->*Ctrl-C*->complete
   2. In a 2nd time I've tried to implement lookup_publisher thinking the
   status would be updated automatically. In that case it wants to connect
   to 5500 and obviously fails after timeout
import xmlrpclib
...
import argparse
import yaml
import signal
import zmq
import xmlrpclib
from urlparse import urlsplit
FINISHED_JOB_STATUS = ["Complete", "Incomplete", "Canceled"]
token = "mytoken"
username = "username"
#hostname = "lava-server:10080"
class JobEndTimeoutError(Exception):
    """ Raise when the specified job does not finish in certain timeframe.
"""
class Timeout(object):
    """ Timeout error class with ALARM signal. Accepts time in seconds. """
    class TimeoutError(Exception):
        pass
def __init__(self, sec):
    self.sec = sec

def __enter__(self):
    signal.signal(signal.SIGALRM, self.timeout_raise)
    signal.alarm(self.sec)

def __exit__(self, *args):
    signal.alarm(0)

def timeout_raise(self, *args):
    raise Timeout.TimeoutError()


class JobListener(object):
def __init__(self, url):
    self.context = zmq.Context.instance()
    self.sock = self.context.socket(zmq.SUB)

    self.sock.setsockopt(zmq.SUBSCRIBE, b"")
    self.sock.connect(url)

def wait_for_job_end(self, job_id, timeout=None):

    try:
        with Timeout(timeout):
            while True:
                msg = self.sock.recv_multipart()
                try:
                    (topic, uuid, dt, username, data) = msg[:]
                except IndexError:
                    # Droping invalid message
                    continue

                data = yaml.safe_load(data)
                if "job" in data:
                    if data["job"] == job_id:
                        if data["status"] in FINISHED_JOB_STATUS:
                            return data

    except Timeout.TimeoutError:
        raise JobEndTimeoutError(
            "JobListener timed out after %s seconds." % timeout)


def lookup_publisher(hostname):
    """
    Lookup the publisher details using XML-RPC
    on the specified hostname.
    """
    xmlrpc_url = "http://%s:10080/RPC2" % (hostname)
    server = xmlrpclib.ServerProxy(xmlrpc_url)
    socket = server.scheduler.get_publisher_event_socket()
    port = urlsplit(socket).port
    listener_url = 'tcp://%s:%s' % (hostname,port)
    print("Using %s" % listener_url)
    return listener_url
if __name__ == '__main__':
#    timeout=1200
parser = argparse.ArgumentParser()
parser.add_argument("-j", "--job-id", type=int, help="Job ID to wait

for")
    parser.add_argument("-t", "--timeout", type=int, help="Timeout in
seconds")
    parser.add_argument("--hostname", help="hostname of the instance")
options = parser.parse_args()


#    server = xmlrpclib.ServerProxy("http://%s:%s@%s/RPC2" % (username,
token, hostname))
#print(server.system.listMethods())

# ret_status=server.scheduler.job_status(options.job_id)
   # print (ret_status['job_status'])
#publisher = 'tcp://%s' % (hostname)
publisher = lookup_publisher(options.hostname)

listener = JobListener(publisher)
listener.wait_for_job_end(options.job_id, options.timeout)