# 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.
"""Event handlers."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import datetime
import logging
import time
from lucifer import autotest
from lucifer import jobx
logger = logging.getLogger(__name__)
class EventHandler(object):
"""Event handling dispatcher.
Event handlers are implemented as methods named _handle_<event value>.
Each handler method must handle its exceptions accordingly. If an
exception escapes, the job dies on the spot.
Instances have one public attribute completed. completed is set to
True once the final COMPLETED event is received and the handler
finishes.
"""
def __init__(self, metrics, job, autoserv_exit, results_dir):
"""Initialize instance.
@param metrics: Metrics instance
@param job: frontend.afe.models.Job instance to own
@param hqes: list of HostQueueEntry instances for the job
@param autoserv_exit: autoserv exit status
@param results_dir: Job results directory
"""
self.completed = False
self._metrics = metrics
self._job = job
# TODO(crbug.com/748234): autoserv not implemented yet.
self._autoserv_exit = autoserv_exit
self._results_dir = results_dir
def __call__(self, event, msg):
logger.debug('Received event %r with message %r', event.name, msg)
method_name = '_handle_%s' % event.value
try:
handler = getattr(self, method_name)
except AttributeError:
raise NotImplementedError('%s is not implemented for handling %s',
method_name, event.name)
_retry_db_errors(lambda: handler(msg))
def _handle_starting(self, msg):
# TODO(crbug.com/748234): No event update needed yet.
pass
def _handle_running(self, _msg):
models = autotest.load('frontend.afe.models')
self._job.hostqueueentry_set.all().update(
status=models.HostQueueEntry.Status.RUNNING,
started_on=datetime.datetime.now())
def _handle_gathering(self, _msg):
models = autotest.load('frontend.afe.models')
self._job.hostqueueentry_set.all().update(
status=models.HostQueueEntry.Status.GATHERING)
def _handle_parsing(self, _msg):
models = autotest.load('frontend.afe.models')
self._job.hostqueueentry_set.all().update(
status=models.HostQueueEntry.Status.PARSING)
def _handle_aborted(self, _msg):
for hqe in self._job.hostqueueentry_set.all().prefetch_related('host'):
_mark_hqe_aborted(hqe)
jobx.write_aborted_keyvals_and_status(self._job, self._results_dir)
def _handle_completed(self, _msg):
self._mark_job_complete()
self.completed = True
def _handle_test_passed(self, msg):
if msg == 'autoserv':
self._autoserv_exit = 0
def _handle_test_failed(self, msg):
if msg == 'autoserv':
self._autoserv_exit = 1
def _handle_host_running(self, msg):
models = autotest.load('frontend.afe.models')
host = models.Host.objects.get(hostname=msg)
host.status = models.Host.Status.RUNNING
host.dirty = 1
host.save(update_fields=['status', 'dirty'])
self._metrics.send_host_status(host)
def _handle_host_ready(self, msg):
models = autotest.load('frontend.afe.models')
host = models.Host.objects.get(hostname=msg)
host.status = models.Host.Status.READY
host.save(update_fields=['status'])
self._metrics.send_host_status(host)
def _handle_host_needs_cleanup(self, msg):
models = autotest.load('frontend.afe.models')
host = models.Host.objects.get(hostname=msg)
models.SpecialTask.objects.create(
host_id=host.id,
task=models.SpecialTask.Task.CLEANUP,
requested_by=models.User.objects.get(login=self._job.owner))
def _handle_host_needs_reset(self, msg):
models = autotest.load('frontend.afe.models')
host = models.Host.objects.get(hostname=msg)
models.SpecialTask.objects.create(
host_id=host.id,
task=models.SpecialTask.Task.RESET,
requested_by=models.User.objects.get(login=self._job.owner))
def _handle_x_tests_done(self, msg):
"""Taken from GatherLogsTask.epilog."""
autoserv_exit, failures = (int(x) for x in msg.split(','))
logger.debug('Got autoserv_exit=%d, failures=%d',
autoserv_exit, failures)
success = (autoserv_exit == 0 and failures == 0)
reset_after_failure = not self._job.run_reset and not success
hqes = self._job.hostqueueentry_set.all().prefetch_related('host')
if self._should_reboot_duts(autoserv_exit, failures,
reset_after_failure):
logger.debug('Creating cleanup jobs for hosts')
for entry in hqes:
self._handle_host_needs_cleanup(entry.host.hostname)
else:
logger.debug('Not creating cleanup jobs for hosts')
for entry in hqes:
self._handle_host_ready(entry.host.hostname)
if not reset_after_failure:
logger.debug('Skipping reset because reset_after_failure is False')
return
logger.debug('Creating reset jobs for hosts')
self._metrics.send_reset_after_failure(autoserv_exit, failures)
for entry in hqes:
self._handle_host_needs_reset(entry.host.hostname)
def _should_reboot_duts(self, autoserv_exit, failures, reset_after_failure):
models = autotest.load('frontend.afe.models')
reboot_after = self._job.reboot_after
if self._final_status() == models.HostQueueEntry.Status.ABORTED:
logger.debug('Should reboot because reboot_after=ABORTED')
return True
elif reboot_after == models.Job.RebootAfter.ALWAYS:
logger.debug('Should reboot because reboot_after=ALWAYS')
return True
elif (reboot_after == models.Job.RebootAfter.IF_ALL_TESTS_PASSED
and autoserv_exit == 0 and failures == 0):
logger.debug('Should reboot because'
' reboot_after=IF_ALL_TESTS_PASSED')
return True
else:
return failures > 0 and not reset_after_failure
def _mark_job_complete(self):
"""Perform Autotest operations needed for job completion."""
final_status = self._final_status()
self._mark_hqes_complete(final_status)
self._stop_job_if_necessary(final_status)
self._release_job_if_sharded()
def _mark_hqes_complete(self, final_status):
"""Perform Autotest HQE operations needed for job completion."""
for hqe in self._job.hostqueueentry_set.all():
self._set_completed_status(hqe, final_status)
def _stop_job_if_necessary(self, final_status):
"""Equivalent to scheduler.modes.Job.stop_if_necessary().
The name isn't informative, but this will stop pre-job tasks as
necessary.
"""
models = autotest.load('frontend.afe.models')
if final_status is not models.HostQueueEntry.Status.ABORTED:
_stop_prejob_hqes(self._job)
def _release_job_if_sharded(self):
if self._job.shard_id is not None:
# If shard_id is None, the job will be synced back to the master
self._job.shard_id = None
self._job.save(update_fields=['shard_id'])
def _final_status(self):
models = autotest.load('frontend.afe.models')
Status = models.HostQueueEntry.Status
if jobx.is_aborted(self._job):
return Status.ABORTED
if self._autoserv_exit == 0:
return Status.COMPLETED
return Status.FAILED
def _set_completed_status(self, hqe, status):
"""Set completed status of HQE.
This is a cleaned up version of the one in scheduler_models to work
with Django models.
"""
hqe.status = status
hqe.active = False
hqe.complete = True
if hqe.started_on:
hqe.finished_on = datetime.datetime.now()
hqe.save(update_fields=['status', 'active', 'complete', 'finished_on'])
self._metrics.send_hqe_completion(hqe)
self._metrics.send_hqe_duration(hqe)
class Metrics(object):
"""Class for sending job metrics."""
def __init__(self):
# Metrics
metrics = autotest.chromite_load('metrics')
self._hqe_completion_metric = metrics.Counter(
'chromeos/autotest/scheduler/hqe_completion_count')
self._reset_after_failure_metric = metrics.Counter(
'chromeos/autotest/scheduler/postjob_tasks/'
'reset_after_failure')
self._host_status_metric = metrics.Boolean(
'chromeos/autotest/dut_status', reset_after=True)
def send_host_status(self, host):
"""Send ts_mon metrics for host status.
@param host: frontend.afe.models.Host instance
"""
labellib = autotest.load('utils.labellib')
labels = labellib.LabelsMapping.from_host(host)
fields = {
'dut_host_name': host.hostname,
'board': labels['board'],
'model': labels['model'],
}
# As each device switches state, indicate that it is not in any
# other state. This allows Monarch queries to avoid double counting
# when additional points are added by the Window Align operation.
for s in host.Status.names:
fields['status'] = s
self._host_status_metric.set(s == host.status, fields=fields)
def send_hqe_completion(self, hqe):
"""Send ts_mon metrics for HQE completion."""
fields = {
'status': hqe.status.lower(),
'board': 'NO_HOST',
'pool': 'NO_HOST',
}
if hqe.host:
labellib = autotest.load('utils.labellib')
labels = labellib.LabelsMapping.from_host(hqe.host)
fields['board'] = labels.get('board', '')
fields['pool'] = labels.get('pool', '')
self._hqe_completion_metric.increment(fields=fields)
def send_hqe_duration(self, hqe):
"""Send CloudTrace metrics for HQE duration."""
if not (hqe.started_on and hqe.finished_on):
return
scheduler_models = autotest.load('scheduler.scheduler_models')
cloud_trace = autotest.chromite_load('cloud_trace')
types = autotest.deps_load('google.protobuf.internal.well_known_types')
hqe_trace_id = scheduler_models.hqe_trace_id
span = cloud_trace.Span(
'HQE', spanId='0', traceId=hqe_trace_id(hqe.id))
span.startTime = types.Timestamp()
span.startTime.FromDatetime(hqe.started_on)
span.endTime = types.Timestamp()
span.endTime.FromDatetime(hqe.finished_on)
cloud_trace.LogSpan(span)
def send_reset_after_failure(self, autoserv_exit, failures):
"""Send reset_after_failure metric."""
self._reset_after_failure_metric.increment(
fields={'autoserv_process_success': autoserv_exit == 0,
# Yes, this is a boolean
'num_tests_failed': failures > 0})
def _mark_hqe_aborted(hqe):
"""Perform Autotest operations needed for HQE abortion.
This also operates on the HQE's host, so prefetch it when possible.
This logic is from scheduler_models.HostQueueEntry.abort().
"""
models = autotest.load('frontend.afe.models')
transaction = autotest.deps_load('django.db.transaction')
Status = models.HostQueueEntry.Status
with transaction.commit_on_success():
if hqe.status in (Status.GATHERING, Status.PARSING):
return
if hqe.status in (Status.STARTING, Status.PENDING, Status.RUNNING):
if hqe.host is None:
return
hqe.host.status = models.Host.Status.READY
hqe.host.save(update_fields=['status'])
hqe.status = Status.ABORTED
hqe.save(update_fields=['status'])
def _stop_prejob_hqes(job):
"""Stop pending HQEs for a job (for synch_count)."""
models = autotest.load('frontend.afe.models')
HQEStatus = models.HostQueueEntry.Status
HostStatus = models.Host.Status
not_yet_run = _get_prejob_hqes(job)
if not_yet_run.count() == job.synch_count:
return
entries_to_stop = _get_prejob_hqes(job, include_active=False)
for hqe in entries_to_stop:
if hqe.status == HQEStatus.PENDING:
hqe.host.status = HostStatus.READY
hqe.host.save(update_fields=['status'])
hqe.status = HQEStatus.STOPPED
hqe.save(update_fields=['status'])
def _get_prejob_hqes(job, include_active=True):
"""Return a queryset of not run HQEs for the job (for synch_count)."""
models = autotest.load('frontend.afe.models')
if include_active:
statuses = list(models.HostQueueEntry.PRE_JOB_STATUSES)
else:
statuses = list(models.HostQueueEntry.IDLE_PRE_JOB_STATUSES)
return models.HostQueueEntry.objects.filter(
job=job, status__in=statuses)
def _retry_db_errors(func):
"""Call func, retrying multiple times if database errors are raised.
crbug.com/863504
"""
django = autotest.deps_load('django')
MySQLdb = autotest.deps_load('MySQLdb')
max_retries = 10
# n ... 0 means n + 1 tries, or 1 try plus n retries
for i in xrange(max_retries, -1, -1):
try:
func()
except (django.db.utils.DatabaseError, MySQLdb.OperationalError) as e:
if i == 0:
raise
logger.debug('Got database error %s, retrying', e)
django.db.close_connection()
time.sleep(5)
else:
break