"""Autotest AFE Cleanup used by the scheduler"""
import contextlib
import logging
import random
import time
from autotest_lib.client.common_lib import utils
from autotest_lib.frontend.afe import models
from autotest_lib.scheduler import scheduler_config
from autotest_lib.client.common_lib import global_config
from autotest_lib.client.common_lib import host_protections
try:
from chromite.lib import metrics
except ImportError:
metrics = utils.metrics_mock
_METRICS_PREFIX = 'chromeos/autotest/scheduler/cleanup'
class PeriodicCleanup(object):
"""Base class to schedule periodical cleanup work.
"""
def __init__(self, db, clean_interval_minutes, run_at_initialize=False):
self._db = db
self.clean_interval_minutes = clean_interval_minutes
self._last_clean_time = time.time()
self._run_at_initialize = run_at_initialize
def initialize(self):
"""Method called by scheduler at the startup.
"""
if self._run_at_initialize:
self._cleanup()
def run_cleanup_maybe(self):
"""Test if cleanup method should be called.
"""
should_cleanup = (self._last_clean_time +
self.clean_interval_minutes * 60
< time.time())
if should_cleanup:
self._cleanup()
self._last_clean_time = time.time()
def _cleanup(self):
"""Abrstract cleanup method."""
raise NotImplementedError
class UserCleanup(PeriodicCleanup):
"""User cleanup that is controlled by the global config variable
clean_interval_minutes in the SCHEDULER section.
"""
def __init__(self, db, clean_interval_minutes):
super(UserCleanup, self).__init__(db, clean_interval_minutes)
self._last_reverify_time = time.time()
@metrics.SecondsTimerDecorator(_METRICS_PREFIX + '/user/durations')
def _cleanup(self):
logging.info('Running periodic cleanup')
self._abort_timed_out_jobs()
self._abort_jobs_past_max_runtime()
self._clear_inactive_blocks()
self._check_for_db_inconsistencies()
self._reverify_dead_hosts()
self._django_session_cleanup()
def _abort_timed_out_jobs(self):
logging.info(
'Aborting all jobs that have timed out and are not complete')
query = models.Job.objects.filter(hostqueueentry__complete=False).extra(
where=['created_on + INTERVAL timeout_mins MINUTE < NOW()'])
jobs = query.distinct()
if not jobs:
return
with _cleanup_warning_banner('timed out jobs', len(jobs)):
for job in jobs:
logging.warning('Aborting job %d due to job timeout', job.id)
job.abort()
_report_detected_errors('jobs_timed_out', len(jobs))
def _abort_jobs_past_max_runtime(self):
"""
Abort executions that have started and are past the job's max runtime.
"""
logging.info('Aborting all jobs that have passed maximum runtime')
rows = self._db.execute("""
SELECT hqe.id FROM afe_host_queue_entries AS hqe
WHERE NOT hqe.complete AND NOT hqe.aborted AND EXISTS
(select * from afe_jobs where hqe.job_id=afe_jobs.id and
hqe.started_on + INTERVAL afe_jobs.max_runtime_mins MINUTE < NOW())
""")
query = models.HostQueueEntry.objects.filter(
id__in=[row[0] for row in rows])
hqes = query.distinct()
if not hqes:
return
with _cleanup_warning_banner('hqes past max runtime', len(hqes)):
for queue_entry in hqes:
logging.warning('Aborting entry %s due to max runtime',
queue_entry)
queue_entry.abort()
_report_detected_errors('hqes_past_max_runtime', len(hqes))
def _check_for_db_inconsistencies(self):
logging.info('Cleaning db inconsistencies')
self._check_all_invalid_related_objects()
def _check_invalid_related_objects_one_way(self, invalid_model,
relation_field, valid_model):
if 'invalid' not in invalid_model.get_field_dict():
return
invalid_objects = list(invalid_model.objects.filter(invalid=True))
invalid_model.objects.populate_relationships(
invalid_objects, valid_model, 'related_objects')
if not invalid_objects:
return
num_objects_with_invalid_relations = 0
errors = []
for invalid_object in invalid_objects:
if invalid_object.related_objects:
related_objects = invalid_object.related_objects
related_list = ', '.join(str(x) for x in related_objects)
num_objects_with_invalid_relations += 1
errors.append('Invalid %s is related to: %s' %
(invalid_object, related_list))
related_manager = getattr(invalid_object, relation_field)
related_manager.clear()
# Only log warnings after we're sure we've seen at least one invalid
# model with some valid relations to avoid empty banners from getting
# printed.
if errors:
invalid_model_name = invalid_model.__name__
valid_model_name = valid_model.__name__
banner = 'invalid %s related to valid %s' % (invalid_model_name,
valid_model_name)
with _cleanup_warning_banner(banner, len(errors)):
for error in errors:
logging.warning(error)
_report_detected_errors(
'invalid_related_objects',
num_objects_with_invalid_relations,
fields={'invalid_model': invalid_model_name,
'valid_model': valid_model_name})
_report_detected_errors(
'invalid_related_objects_relations',
len(errors),
fields={'invalid_model': invalid_model_name,
'valid_model': valid_model_name})
def _check_invalid_related_objects(self, first_model, first_field,
second_model, second_field):
self._check_invalid_related_objects_one_way(
first_model,
first_field,
second_model,
)
self._check_invalid_related_objects_one_way(
second_model,
second_field,
first_model,
)
def _check_all_invalid_related_objects(self):
model_pairs = ((models.Host, 'labels', models.Label, 'host_set'),
(models.AclGroup, 'hosts', models.Host, 'aclgroup_set'),
(models.AclGroup, 'users', models.User, 'aclgroup_set'),
(models.Test, 'dependency_labels', models.Label,
'test_set'))
for first_model, first_field, second_model, second_field in model_pairs:
self._check_invalid_related_objects(
first_model,
first_field,
second_model,
second_field,
)
def _clear_inactive_blocks(self):
logging.info('Clear out blocks for all completed jobs.')
# this would be simpler using NOT IN (subquery), but MySQL
# treats all IN subqueries as dependent, so this optimizes much
# better
self._db.execute("""
DELETE ihq FROM afe_ineligible_host_queues ihq
WHERE NOT EXISTS
(SELECT job_id FROM afe_host_queue_entries hqe
WHERE NOT hqe.complete AND hqe.job_id = ihq.job_id)""")
def _should_reverify_hosts_now(self):
reverify_period_sec = (scheduler_config.config.reverify_period_minutes
* 60)
if reverify_period_sec == 0:
return False
return (self._last_reverify_time + reverify_period_sec) <= time.time()
def _choose_subset_of_hosts_to_reverify(self, hosts):
"""Given hosts needing verification, return a subset to reverify."""
max_at_once = scheduler_config.config.reverify_max_hosts_at_once
if (max_at_once > 0 and len(hosts) > max_at_once):
return random.sample(hosts, max_at_once)
return sorted(hosts)
def _reverify_dead_hosts(self):
if not self._should_reverify_hosts_now():
return
self._last_reverify_time = time.time()
logging.info('Checking for dead hosts to reverify')
hosts = models.Host.objects.filter(
status=models.Host.Status.REPAIR_FAILED,
locked=False,
invalid=False)
hosts = hosts.exclude(
protection=host_protections.Protection.DO_NOT_VERIFY)
if not hosts:
return
hosts = list(hosts)
total_hosts = len(hosts)
hosts = self._choose_subset_of_hosts_to_reverify(hosts)
logging.info('Reverifying dead hosts (%d of %d)', len(hosts),
total_hosts)
with _cleanup_warning_banner('reverify dead hosts', len(hosts)):
for host in hosts:
logging.warning(host.hostname)
_report_detected_errors('dead_hosts_triggered_reverify', len(hosts))
_report_detected_errors('dead_hosts_require_reverify', total_hosts)
for host in hosts:
models.SpecialTask.schedule_special_task(
host=host, task=models.SpecialTask.Task.VERIFY)
def _django_session_cleanup(self):
"""Clean up django_session since django doesn't for us.
http://www.djangoproject.com/documentation/0.96/sessions/
"""
logging.info('Deleting old sessions from django_session')
sql = 'TRUNCATE TABLE django_session'
self._db.execute(sql)
class TwentyFourHourUpkeep(PeriodicCleanup):
"""Cleanup that runs at the startup of monitor_db and every subsequent
twenty four hours.
"""
def __init__(self, db, drone_manager, run_at_initialize=True):
"""Initialize TwentyFourHourUpkeep.
@param db: Database connection object.
@param drone_manager: DroneManager to access drones.
@param run_at_initialize: True to run cleanup when scheduler starts.
Default is set to True.
"""
self.drone_manager = drone_manager
clean_interval_minutes = 24 * 60 # 24 hours
super(TwentyFourHourUpkeep, self).__init__(
db, clean_interval_minutes, run_at_initialize=run_at_initialize)
@metrics.SecondsTimerDecorator(_METRICS_PREFIX + '/daily/durations')
def _cleanup(self):
logging.info('Running 24 hour clean up')
self._check_for_uncleanable_db_inconsistencies()
self._cleanup_orphaned_containers()
def _check_for_uncleanable_db_inconsistencies(self):
logging.info('Checking for uncleanable DB inconsistencies')
self._check_for_active_and_complete_queue_entries()
self._check_for_multiple_platform_hosts()
self._check_for_no_platform_hosts()
def _check_for_active_and_complete_queue_entries(self):
query = models.HostQueueEntry.objects.filter(active=True, complete=True)
num_bad_hqes = query.count()
if num_bad_hqes == 0:
return
num_aborted = 0
logging.warning('%d queue entries found with active=complete=1',
num_bad_hqes)
with _cleanup_warning_banner('active and complete hqes', num_bad_hqes):
for entry in query:
if entry.status == 'Aborted':
entry.active = False
entry.save()
recovery_path = 'was also aborted, set active to False'
num_aborted += 1
else:
recovery_path = 'can not recover'
logging.warning('%s (recovery: %s)', entry.get_object_dict(),
recovery_path)
_report_detected_errors('hqes_active_and_complete', num_bad_hqes)
_report_detected_errors('hqes_aborted_set_to_inactive', num_aborted)
def _check_for_multiple_platform_hosts(self):
rows = self._db.execute("""
SELECT afe_hosts.id, hostname, COUNT(1) AS platform_count,
GROUP_CONCAT(afe_labels.name)
FROM afe_hosts
INNER JOIN afe_hosts_labels ON
afe_hosts.id = afe_hosts_labels.host_id
INNER JOIN afe_labels ON afe_hosts_labels.label_id = afe_labels.id
WHERE afe_labels.platform
GROUP BY afe_hosts.id
HAVING platform_count > 1
ORDER BY hostname""")
if rows:
logging.warning('Cleanup found hosts with multiple platforms')
with _cleanup_warning_banner('hosts with multiple platforms',
len(rows)):
for row in rows:
logging.warning(' '.join(str(item) for item in row))
_report_detected_errors('hosts_with_multiple_platforms', len(rows))
def _check_for_no_platform_hosts(self):
rows = self._db.execute("""
SELECT hostname
FROM afe_hosts
LEFT JOIN afe_hosts_labels
ON afe_hosts.id = afe_hosts_labels.host_id
AND afe_hosts_labels.label_id IN (SELECT id FROM afe_labels
WHERE platform)
WHERE NOT afe_hosts.invalid AND afe_hosts_labels.host_id IS NULL""")
if rows:
with _cleanup_warning_banner('hosts with no platform', len(rows)):
for row in rows:
logging.warning(row[0])
_report_detected_errors('hosts_with_no_platform', len(rows))
def _cleanup_orphaned_containers(self):
"""Cleanup orphaned containers in each drone.
The function queues a lxc_cleanup call in each drone without waiting for
the script to finish, as the cleanup procedure could take minutes and the
script output is logged.
"""
ssp_enabled = global_config.global_config.get_config_value(
'AUTOSERV', 'enable_ssp_container')
if not ssp_enabled:
logging.info(
'Server-side packaging is not enabled, no need to clean '
'up orphaned containers.')
return
self.drone_manager.cleanup_orphaned_containers()
def _report_detected_errors(metric_name, count, fields={}):
"""Reports a counter metric for recovered errors
@param metric_name: Name of the metric to report about.
@param count: How many "errors" were fixed this cycle.
@param fields: Optional fields to include with the metric.
"""
m = '%s/errors_recovered/%s' % (_METRICS_PREFIX, metric_name)
metrics.Counter(m).increment_by(count, fields=fields)
def _report_detected_errors(metric_name, gauge, fields={}):
"""Reports a gauge metric for errors detected
@param metric_name: Name of the metric to report about.
@param gauge: Outstanding number of unrecoverable errors of this type.
@param fields: Optional fields to include with the metric.
"""
m = '%s/errors_detected/%s' % (_METRICS_PREFIX, metric_name)
metrics.Gauge(m).set(gauge, fields=fields)
@contextlib.contextmanager
def _cleanup_warning_banner(banner, error_count=None):
"""Put a clear context in the logs around list of errors
@param: banner: The identifying header to print for context.
@param: error_count: If not None, the number of errors detected.
"""
if error_count is not None:
banner += ' (total: %d)' % error_count
logging.warning('#### START: %s ####', banner)
try:
yield
finally:
logging.warning('#### END: %s ####', banner)