teuthology/queue: Single command for queue operations

Makes the same teuthology-queue commands work regardless of the queue backend, Paddles or Beanstalk.

Signed-off-by: Aishwarya Mathuria <[email protected]>

docs/docker-compose/docker-compose.yml

@@ -19,7 +19,7 @@ services:
     ports:
       - 5432:5432
   paddles:
-    image: quay.io/ceph-infra/paddles
+    image: quay.io/ceph-infra/paddles:wip-amathuria-removing-beanstalkd
     environment: 
       PADDLES_SERVER_HOST: 0.0.0.0
       PADDLES_SQLALCHEMY_URL: postgresql+psycopg2://admin:password@postgres:5432/paddles

docs/docker-compose/teuthology/teuthology.sh

@@ -24,6 +24,7 @@ if [ -z "$TEUTHOLOGY_WAIT" ]; then
         --ceph-repo https://github.com/ceph/ceph.git \
         --suite-repo https://github.com/ceph/ceph.git \
         -c main \
+	--sha1 552b6d852cbc7384a05805989a675de7d6b4a3fd \
         -m $MACHINE_TYPE \
         --limit 1 \
         -n 100 \
@@ -37,10 +38,12 @@ if [ -z "$TEUTHOLOGY_WAIT" ]; then
         --force-priority \
         $CUSTOM_CONF
     DISPATCHER_EXIT_FLAG='--exit-on-empty-queue'
-    teuthology-queue -m $MACHINE_TYPE -s | \
-      python3 -c "import sys, json; assert json.loads(sys.stdin.read())['count'] > 0, 'queue is empty!'"
+    #teuthology-queue -m $MACHINE_TYPE -s | \
+    #  python3 -c "import sys, json; assert json.loads(sys.stdin.read())['count'] > 0, 'queue is empty!'"
 fi
 teuthology-dispatcher -v \
     --log-dir /teuthology/log \
     --tube $MACHINE_TYPE \
+    --machine-type $MACHINE_TYPE \
     $DISPATCHER_EXIT_FLAG
+

scripts/dispatcher.py

@@ -1,9 +1,9 @@
 """
 usage: teuthology-dispatcher --help
        teuthology-dispatcher --supervisor [-v] --bin-path BIN_PATH --job-config CONFIG --archive-dir DIR
-       teuthology-dispatcher [-v] [--archive-dir DIR] --log-dir LOG_DIR --machine-type MACHINE_TYPE --queue-backend BACKEND
+       teuthology-dispatcher [-v] [--archive-dir DIR] [--exit-on-empty-queue] --log-dir LOG_DIR --tube TUBE --machine-type MACHINE_TYPE
 
-Start a dispatcher for the specified machine type. Grab jobs from a paddles/beanstalk
+Start a dispatcher for the specified tube. Grab jobs from a paddles/beanstalk
 queue and run the teuthology tests they describe as subprocesses. The
 subprocess invoked is a teuthology-dispatcher command run in supervisor
 mode.
@@ -17,12 +17,12 @@
   -v, --verbose                  be more verbose
   -l, --log-dir LOG_DIR          path in which to store logs
   -a DIR, --archive-dir DIR      path to archive results in
-  --machine-type MACHINE_TYPE    the machine type for the job
+  -t, --tube TUBE                which queue to read jobs from
   --supervisor                   run dispatcher in job supervisor mode
   --bin-path BIN_PATH            teuthology bin path
   --job-config CONFIG            file descriptor of job's config file
   --exit-on-empty-queue          if the queue is empty, exit
-  --queue-backend BACKEND        choose between paddles and beanstalk
+  --machine-type MACHINE_TYPE    specify machine type eg. smithi, mira
 """
 
 import docopt

scripts/kill.py

@@ -12,7 +12,7 @@
        teuthology-kill [-p] -o OWNER -m MACHINE_TYPE -r RUN
 
 Kill running teuthology jobs:
-1. Removes any queued jobs from the paddles queue
+1. Removes any queued jobs from the queue
 2. Kills any running jobs
 3. Nukes any machines involved
 

scripts/queue.py

@@ -1,15 +1,16 @@
 import docopt
 
-import teuthology.config
 import teuthology.queue.beanstalk
 import teuthology.queue.paddles
+from teuthology.config import config
 
 doc = """
 usage: teuthology-queue -h
-       teuthology-queue [-s|-d|-f] -m MACHINE_TYPE 
+       teuthology-queue [-s|-d|-f] -m MACHINE_TYPE
        teuthology-queue [-r] -m MACHINE_TYPE
        teuthology-queue -m MACHINE_TYPE -D PATTERN
-       teuthology-queue -p SECONDS [-m MACHINE_TYPE]
+       teuthology-queue -p SECONDS [-m MACHINE_TYPE] [-U USER]
+       teuthology-queue -m MACHINE_TYPE -P PRIORITY [-U USER|-R RUN_NAME]
 
 List Jobs in queue.
 If -D is passed, then jobs with PATTERN in the job name are deleted from the
@@ -29,9 +30,17 @@
   -p, --pause SECONDS   Pause queues for a number of seconds. A value of 0
                         will unpause. If -m is passed, pause that queue,
                         otherwise pause all queues.
+  -P, --priority PRIORITY
+                        Change priority of queued jobs (only in Paddles queues)
+  -U, --user USER       User who owns the jobs
+  -R, --run-name RUN_NAME
+                        Used to change priority of all jobs in the run.
 """
 
 
 def main():
     args = docopt.docopt(doc)
-    teuthology.queue.main(args)
+    if config.queue_backend == 'beanstalk':
+      teuthology.queue.beanstalk.main(args)
+    else:
+      teuthology.queue.paddles.main(args)

scripts/schedule.py

@@ -21,7 +21,6 @@
                                        Queue backend name, use prefix '@'
                                        to append job config to the given
                                        file path as yaml.
-                                       [default: paddles]
   -n <name>, --name <name>             Name of suite run the job is part of
   -d <desc>, --description <desc>      Job description
   -o <owner>, --owner <owner>          Job owner

scripts/worker.py

@@ -9,7 +9,7 @@ def main():
 
 def parse_args():
     parser = argparse.ArgumentParser(description="""
-Grab jobs from a paddles queue and run the teuthology tests they
+Grab jobs from a beanstalk queue and run the teuthology tests they
 describe. One job is run at a time.
 """)
     parser.add_argument(
@@ -29,8 +29,8 @@ def parse_args():
         required=True,
     )
     parser.add_argument(
-        '-m', '--machine-type',
-        help='which machine type the jobs will run on',
+        '-t', '--tube',
+        help='which beanstalk tube to read jobs from',
         required=True,
     )
 

teuthology/config.py

@@ -143,7 +143,7 @@ class TeuthologyConfig(YamlConfig):
         'archive_upload_key': None,
         'archive_upload_url': None,
         'automated_scheduling': False,
-        'backend': 'beanstalk',
+        'queue_backend': 'beanstalk',
         'reserve_machines': 5,
         'ceph_git_base_url': 'https://github.com/ceph/',
         'ceph_git_url': None,

teuthology/dispatcher/__init__.py

@@ -7,7 +7,6 @@
 
 from datetime import datetime
 from typing import Dict, List
-from time import sleep
 
 from teuthology import (
     # non-modules
@@ -78,11 +77,15 @@ def main(args):
         return supervisor.main(args)
 
     verbose = args["--verbose"]
-    machine_type = args["--machine-type"]
+    tube = args["--tube"]
     log_dir = args["--log-dir"]
     archive_dir = args["--archive-dir"]
     exit_on_empty_queue = args["--exit-on-empty-queue"]
-    backend = args['--queue-backend']
+    machine_type = args['--machine-type']
+    backend = teuth_config.queue_backend
+
+    if backend is None:
+        backend = 'beanstalk'
 
     if archive_dir is None:
         archive_dir = teuth_config.archive_base
@@ -97,21 +100,22 @@ def main(args):
 
     if machine_type is None and teuth_config.machine_type is None:
         return
+
     # setup logging for disoatcher in {log_dir}
     loglevel = logging.INFO
     if verbose:
         loglevel = logging.DEBUG
     logging.getLogger().setLevel(loglevel)
     log.setLevel(loglevel)
-    log_file_path = os.path.join(log_dir, f"dispatcher.{machine_type}.{os.getpid()}")
+    log_file_path = os.path.join(log_dir, f"dispatcher.{tube}.{os.getpid()}")
     setup_log_file(log_file_path)
     install_except_hook()
 
     load_config(archive_dir=archive_dir)
 
     if backend == 'beanstalk':
         connection = beanstalk.connect()
-        beanstalk.watch_tube(connection, machine_type)
+        beanstalk.watch_tube(connection, tube)
 
     result_proc = None
 
@@ -141,21 +145,25 @@ def main(args):
             if rc is not None:
                 worst_returncode = max([worst_returncode, rc])
                 job_procs.remove(proc)
-        if config.backend == 'beanstalk':
+        if teuth_config.queue_backend == 'beanstalk':
             job = connection.reserve(timeout=60)
+            if job is not None:
+                job_id = job.jid
+                job_config = yaml.safe_load(job.body)
         else:
             job = report.get_queued_job(machine_type)
+            if job is not None:
+                job = clean_config(job)
+                job_id = job.get('job_id')
+                job_config = job
         if job is None:
             if exit_on_empty_queue and not job_procs:
                 log.info("Queue is empty and no supervisor processes running; exiting!")
                 break
             continue
-        job = clean_config(job)
-        report.try_push_job_info(job, dict(status='running'))
-        job_id = job.get('job_id')
+        report.try_push_job_info(job_config, dict(status='running'))
         log.info('Reserved job %s', job_id)
-        log.info('Config is: %s', job)
-        job_config = job
+        log.info('Config is: %s', job_config)
 
         if job_config.get('stop_worker'):
             keep_running = False
@@ -268,18 +276,3 @@ def create_job_archive(job_name, job_archive_path, archive_dir):
     if not os.path.exists(run_archive):
         safepath.makedirs('/', run_archive)
     safepath.makedirs('/', job_archive_path)
-
-
-def pause_queue(machine_type, paused, paused_by, pause_duration=None):
-    if paused:
-        report.pause_queue(machine_type, paused, paused_by, pause_duration)
-        '''
-        If there is a pause duration specified
-        un-pause the queue after the time elapses
-        '''
-        if pause_duration is not None:
-            sleep(int(pause_duration))
-            paused = False
-            report.pause_queue(machine_type, paused, paused_by)
-    elif not paused:
-        report.pause_queue(machine_type, paused, paused_by)

teuthology/dispatcher/supervisor.py

@@ -88,9 +88,8 @@ def run_job(job_config, teuth_bin_path, archive_dir, verbose):
         if teuth_config.results_server:
             try:
                 report.try_delete_jobs(job_config['name'], job_config['job_id'])
-            except Exception as e:
-                log.warning("Unable to delete job %s, exception occurred: %s",
-                            job_config['job_id'], e)
+            except Exception:
+                log.exception("Unable to delete job %s", job_config['job_id'])
         job_archive = os.path.join(archive_dir, safe_archive)
         args = [
             os.path.join(teuth_bin_path, 'teuthology-results'),
@@ -148,7 +147,7 @@ def run_job(job_config, teuth_bin_path, archive_dir, verbose):
         '--archive', job_config['archive_path'],
         '--name', job_config['name'],
     ])
-    if 'description' in job_config:
+    if job_config.get('description') is not None:
         arg.extend(['--description', job_config['description']])
     job_archive = os.path.join(job_config['archive_path'], 'orig.config.yaml')
     arg.extend(['--', job_archive])

teuthology/exporter.py

@@ -6,10 +6,10 @@
 
 from pathlib import Path
 
-import teuthology.beanstalk as beanstalk
 import teuthology.dispatcher
 from teuthology.config import config
 from teuthology.lock.query import list_locks
+from teuthology.queue import beanstalk
 
 log = logging.getLogger(__name__)
 

teuthology/kill.py

@@ -117,7 +117,7 @@ def find_run_info(serializer, run_name):
         if not os.path.isdir(job_dir):
             continue
         job_num += 1
-        if config.backend == 'beanstalk':
+        if config.queue_backend == 'beanstalk':
             beanstalk.print_progress(job_num, job_total, 'Reading Job: ')
         job_info = serializer.job_info(run_name, job_id, simple=True)
         for key in job_info.keys():

teuthology/orchestra/run.py

@@ -182,6 +182,7 @@ def _raise_for_status(self):
                     command=self.command, exitstatus=self.returncode,
                     node=self.hostname, label=self.label
                 )
+
     def _get_exitstatus(self):
         """
         :returns: the remote command's exit status (return code). Note that