# 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