scheduler/monitor_db_cleanup.py

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