Skip to content

Commit

Permalink
teuthology/queue: Single command for queue operations
Browse files Browse the repository at this point in the history
Makes the same teuthology-queue commands work regardless of the queue backend, Paddles or Beanstalk.

Signed-off-by: Aishwarya Mathuria <amathuri@redhat.com>
  • Loading branch information
amathuria committed Jun 15, 2022
1 parent 172089c commit 74d1650
Show file tree
Hide file tree
Showing 17 changed files with 182 additions and 352 deletions.
35 changes: 0 additions & 35 deletions scripts/beanstalk_queue.py

This file was deleted.

9 changes: 5 additions & 4 deletions scripts/dispatcher.py
Original file line number Diff line number Diff line change
@@ -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] [--queue-backend BACKEND] --log-dir LOG_DIR --tube TUBE
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.
Expand All @@ -17,12 +17,13 @@
-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
--queue-backend BACKEND which backend will be used for the queue
[default: beanstalk]
"""

import docopt
Expand Down
2 changes: 1 addition & 1 deletion scripts/kill.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
45 changes: 0 additions & 45 deletions scripts/paddles_queue.py

This file was deleted.

17 changes: 13 additions & 4 deletions scripts/queue.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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.backend == 'beanstalk':
teuthology.queue.beanstalk.main(args)
else:
teuthology.queue.paddles.main(args)
2 changes: 1 addition & 1 deletion scripts/schedule.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
Queue backend name, use prefix '@'
to append job config to the given
file path as yaml.
[default: paddles]
[default: beanstalk]
-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
Expand Down
6 changes: 3 additions & 3 deletions scripts/worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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,
)

Expand Down
31 changes: 10 additions & 21 deletions teuthology/dispatcher/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,31 +70,32 @@ 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']

if backend is None:
backend = 'beanstalk'

if archive_dir is None:
archive_dir = teuth_config.archive_base

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
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

Expand Down Expand Up @@ -129,8 +130,11 @@ def main(args):
log.info('Reserved job %s', job_id)
log.info('Config is: %s', job.body)
else:
job = report.get_queued_job(machine_type)
job = report.get_queued_job(tube)
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'))
Expand Down Expand Up @@ -220,18 +224,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)
7 changes: 3 additions & 4 deletions teuthology/dispatcher/supervisor.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,9 +70,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'),
Expand Down Expand Up @@ -130,7 +129,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])
Expand Down
1 change: 1 addition & 0 deletions teuthology/orchestra/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
106 changes: 0 additions & 106 deletions teuthology/queue/__init__.py
Original file line number Diff line number Diff line change
@@ -1,106 +0,0 @@
import logging
import pprint
import sys
from collections import OrderedDict

from teuthology import report
from teuthology.config import config

log = logging.getLogger(__name__)

def print_progress(index, total, message=None):
msg = "{m} ".format(m=message) if message else ''
sys.stderr.write("{msg}{i}/{total}\r".format(
msg=msg, i=index, total=total))
sys.stderr.flush()

def end_progress():
sys.stderr.write('\n')
sys.stderr.flush()

class JobProcessor(object):
def __init__(self):
self.jobs = OrderedDict()

def add_job(self, job_id, job_config, job_obj=None):
job_id = str(job_id)

job_dict = dict(
index=(len(self.jobs) + 1),
job_config=job_config,
)
if job_obj:
job_dict['job_obj'] = job_obj
self.jobs[job_id] = job_dict

self.process_job(job_id)

def process_job(self, job_id):
pass

def complete(self):
pass


class JobPrinter(JobProcessor):
def __init__(self, show_desc=False, full=False):
super(JobPrinter, self).__init__()
self.show_desc = show_desc
self.full = full

def process_job(self, job_id):
job_config = self.jobs[job_id]['job_config']
job_index = self.jobs[job_id]['index']
job_priority = job_config['priority']
job_name = job_config['name']
job_desc = job_config['description']
print('Job: {i:>4} priority: {pri:>4} {job_name}/{job_id}'.format(
i=job_index,
pri=job_priority,
job_id=job_id,
job_name=job_name,
))
if self.full:
pprint.pprint(job_config)
elif job_desc and self.show_desc:
for desc in job_desc.split():
print('\t {}'.format(desc))


class RunPrinter(JobProcessor):
def __init__(self):
super(RunPrinter, self).__init__()
self.runs = list()

def process_job(self, job_id):
run = self.jobs[job_id]['job_config']['name']
if run not in self.runs:
self.runs.append(run)
print(run)


class JobDeleter(JobProcessor):
def __init__(self, pattern):
self.pattern = pattern
super(JobDeleter, self).__init__()

def add_job(self, job_id, job_config, job_obj=None):
job_name = job_config['name']
if self.pattern in job_name:
super(JobDeleter, self).add_job(job_id, job_config, job_obj)

def process_job(self, job_id):
job_config = self.jobs[job_id]['job_config']
job_name = job_config['name']
print('Deleting {job_name}/{job_id}'.format(
job_id=job_id,
job_name=job_name,
))
report.try_delete_jobs(job_name, job_id)


def main(args):
if config.backend == 'paddles':
paddles.main(args)
else:
beanstalk.main(args)
Loading

0 comments on commit 74d1650

Please sign in to comment.