blob: 2dafc04fa8572fec4cf8c81857b1b6915de9f33a [file] [log] [blame]
# Copyright 2017 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Extra functions for frontend.afe.models.Job objects.
Most of these exist in tightly coupled forms in legacy Autotest code
(e.g., part of methods with completely unrelated names on Task objects
under multiple layers of abstract classes). These are defined here to
sanely reuse without having to commit to a long refactor of legacy code
that is getting deleted soon.
It's not really a good idea to define these on the Job class either;
they are specialized and the Job class already suffers from method
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import os
import time
import urllib
from lucifer import autotest
from lucifer import results
def is_hostless(job):
"""Return True if the job is hostless.
@param job: frontend.afe.models.Job instance
return not hostnames(job)
def hostnames(job):
"""Return a list of hostnames for a job.
@param job: frontend.afe.models.Job instance
hqes = job.hostqueueentry_set.all().prefetch_related('host')
return [ for hqe in hqes if is not None]
def is_aborted(job):
"""Return if the job is aborted.
(This means the job is marked for abortion; the job can still be
@param job: frontend.afe.models.Job instance
return job.hostqueueentry_set.filter(aborted=True).exists()
def is_server_job(job):
"""Return whether the job is a server job.
@param job: frontend.afe.models.Job instance
return not is_client_job(job)
def is_client_job(job):
"""Return whether the job is a client job.
If the job is not a client job, it is a server job.
(In theory a job can be neither. I have no idea what you should do
in that case.)
@param job: frontend.afe.models.Job instance
CONTROL_TYPE = autotest.load('client.common_lib.control_data').CONTROL_TYPE
return CONTROL_TYPE.get_value(job.control_type) == CONTROL_TYPE.CLIENT
def needs_ssp(job):
"""Return whether the job needs SSP.
This also looks up the config for jobs that do not have a value
@param job: frontend.afe.models.Job instance
return (_ssp_enabled()
and is_server_job(job)
# None is the same as True.
and job.require_ssp != False)
def _ssp_enabled():
"""Return whether SSP is enabled in the config."""
global_config = autotest.load('client.common_lib.global_config')
return global_config.global_config.get_config_value(
'AUTOSERV', 'enable_ssp_container', type=bool,
def control_file_path(workdir):
"""Path to control file for a job.
This makes more sense in the old Autotest drone world. The
scheduler has to copy the control file to the drone. It goes to a
temporary path `drone_tmp/attach.N`.
The drone is then able to run `autoserv <args> drone_tmp/attach.N`.
But in the Lucifer world, we are already running on the drone, so we
don't need to rendezvous with a temporary directory through
drone_manager first.
So pick an arbitrary filename to plop into the workdir. autoserv
will later copy this back to the standard control/control.srv.
return os.path.join(workdir, 'lucifer', 'control_attach')
def prepare_control_file(job, workdir):
"""Prepare control file for a job."""
with open(control_file_path(workdir), 'w') as f:
def prepare_keyvals_files(job, workdir):
"""Prepare Autotest keyvals files for a job."""
keyvals = job.keyval_dict()
keyvals['job_queued'] = \
results.write_keyvals(workdir, keyvals)
if is_hostless(job):
for hqe in job.hostqueueentry_set.all().prefetch_related('host'):
workdir,, _host_keyvals(
def write_aborted_keyvals_and_status(job, workdir):
"""Write the keyvals and status for an aborted job."""
aborted_by = 'autotest_system'
aborted_on = int(time.time())
for hqe in job.hostqueueentry_set.all():
if not hasattr(hqe, 'abortedhostqueueentry'):
ahqe = hqe.abortedhostqueueentry
aborted_by = ahqe.aborted_by
aborted_on = int(time.mktime(ahqe.aborted_on.timetuple()))
results.write_keyvals(workdir, {
'aborted_by': aborted_by,
'aborted_on': aborted_on,
workdir, 'Job aborted by %s on %s' % (aborted_by, aborted_on))
def _host_keyvals(host):
"""Return keyvals dict for a host.
@param host: frontend.afe.models.Host instance
labels = list(_host_labels(host))
platform = None
for label in labels:
if label.platform:
platform =
return {
'platform': platform,
'labels': ','.join(urllib.quote( for label in labels),
def _host_labels(host):
"""Return an iterable of labels for a host.
@param host: frontend.afe.models.Host instance
if autotest.load('scheduler.scheduler_models').RESPECT_STATIC_LABELS:
return _host_labels_with_static(host)
return host.labels.all()
def _host_labels_with_static(host):
"""Return a generator of labels for a host, respecting static labels.
@param host: frontend.afe.models.Host instance
models = autotest.load('frontend.afe.models')
replaced_label_ids = frozenset(models.ReplacedLabel.objects.all()
.values_list('label_id', flat=True))
shadowed_labels = set()
for label in host.labels.all():
if in replaced_label_ids:
yield label
for label in host.static_labels.all():
if in shadowed_labels:
yield label