Merge pull request #210 from ceph/wip-queue

Add teuthology-queue command for beanstalk Managmeent.

scripts/queue.py

@@ -0,0 +1,27 @@
+import docopt
+
+import teuthology.config
+import teuthology.beanstalk
+
+doc = """
+usage: teuthology-queue [-h] -m MACHINE_TYPE
+       teuthology-queue [-h] -m MACHINE_TYPE -d JOB
+
+List Jobs in queue:
+  If -d then jobs with JOB in the job name are deleted from the queue.
+
+Arguments:
+  -m, --machine_type MACHINE_TYPE
+                        Which machine type queue to work on.
+
+optional arguments:
+  -h, --help            show this help message and exit
+
+  -d, --delete JOB      Delete Jobs JOB in their name.
+
+""".format(archive_base=teuthology.config.config.archive_base)
+
+
+def main():
+    args = docopt.docopt(doc)
+    teuthology.beanstalk.main(args)

setup.py

@@ -28,6 +28,7 @@ setup(
             'teuthology-results = scripts.results:main',
             'teuthology-report = scripts.report:main',
             'teuthology-kill = scripts.kill:main',
+            'teuthology-queue = scripts.queue:main',
             ],
         },
 

teuthology/beanstalk.py

@@ -0,0 +1,63 @@
+import beanstalkc
+import yaml
+import logging
+
+from .config import config
+
+log = logging.getLogger(__name__)
+
+def beanstalk_connect(machine_type):
+    qhost = config.queue_host
+    qport = config.queue_port
+    if qhost is None or qport is None:
+        raise RuntimeError(
+            'Beanstalk queue information not found in {conf_path}'.format(
+                conf_path=config.teuthology_yaml))
+    log.info("Checking Beanstalk Queue...")
+    beanstalk = beanstalkc.Connection(host=qhost, port=qport)
+    beanstalk.watch(machine_type)
+    beanstalk.ignore('default')
+    return beanstalk
+
+def walk_jobs(beanstalk, machine_type, delete=None):
+    curjobs = beanstalk.stats_tube(machine_type)['current-jobs-ready']
+    if curjobs != 0:
+        x=1
+        while x != curjobs:
+            x += 1
+            job = beanstalk.reserve(timeout=20)
+            if job is not None:
+                if job.body is not None:
+                    job_config = yaml.safe_load(job.body)
+                    job_name=job_config['name']
+                    job_id = job.stats()['id']
+                    job_description = job_config['description']
+                    if delete:
+                        if delete in job_name:
+                            print 'Deleted Job: {job_id} from queue. Name: {job_name}'.format(
+                            job_id = job_id,
+                            job_name = job_name,
+                            )
+                            job.delete()
+                        else:
+                            print "Searching queue... Checked " + str(x) + "/" + str(curjobs)," Jobs\r",
+                    else:
+                        print 'Job: {x}/{curjobs} ID: {job_id} Name: {job_name}'.format(
+                        x=x,
+                        curjobs=curjobs,
+                        job_id=job_id,
+                        job_name=job_name,
+                        )
+                        if job_description:
+                            for desc in job_description.split():
+                                print '\t {desc}'.format(desc=desc)
+        log.info("Finished checking Beanstalk Queue.")
+    else:
+        log.info('No jobs in Beanstalk Queue')
+
+def main(args):
+    machine_type =  args['--machine_type']
+    delete = args['--delete']
+    beanstalk = beanstalk_connect(machine_type)
+    walk_jobs(beanstalk, machine_type, delete)
+    beanstalk.close()