watcher/watcher/decision_engine/audit/continuous.py

# -*- encoding: utf-8 -*-
# Copyright (c) 2016 Servionica LTD
# Copyright (c) 2016 Intel Corp
#
# Authors: Alexander Chadin <a.chadin@servionica.ru>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import datetime

from croniter import croniter
from dateutil import tz
from oslo_utils import timeutils

from watcher.common import context
from watcher.common import scheduling
from watcher.common import utils
from watcher import conf
from watcher.db.sqlalchemy import api as sq_api
from watcher.db.sqlalchemy import job_store
from watcher.decision_engine.audit import base
from watcher import objects


CONF = conf.CONF


class ContinuousAuditHandler(base.AuditHandler):
    def __init__(self):
        super(ContinuousAuditHandler, self).__init__()
        # scheduler for executing audits
        self._audit_scheduler = None
        # scheduler for a periodic task to launch audit
        self._period_scheduler = None
        self.context_show_deleted = context.RequestContext(is_admin=True,
                                                           show_deleted=True)

    @property
    def scheduler(self):
        if self._audit_scheduler is None:
            self._audit_scheduler = scheduling.BackgroundSchedulerService(
                jobstores={
                    'default': job_store.WatcherJobStore(
                        engine=sq_api.get_engine()),
                }
            )
        return self._audit_scheduler

    @property
    def period_scheduler(self):
        if self._period_scheduler is None:
            self._period_scheduler = scheduling.BackgroundSchedulerService()
        return self._period_scheduler

    def _is_audit_inactive(self, audit):
        audit = objects.Audit.get_by_uuid(
            self.context_show_deleted, audit.uuid, eager=True)
        if (objects.audit.AuditStateTransitionManager().is_inactive(audit) or
                (audit.hostname != CONF.host) or
                (self.check_audit_expired(audit))):
            # if audit isn't in active states, audit's job must be removed to
            # prevent using of inactive audit in future.
            jobs = [job for job in self.scheduler.get_jobs()
                    if job.name == 'execute_audit' and
                    job.args[0].uuid == audit.uuid]
            if jobs:
                jobs[0].remove()
            return True

        return False

    def do_execute(self, audit, request_context):
        solution = super(ContinuousAuditHandler, self)\
            .do_execute(audit, request_context)

        if audit.audit_type == objects.audit.AuditType.CONTINUOUS.value:
            a_plan_filters = {'audit_uuid': audit.uuid,
                              'state': objects.action_plan.State.RECOMMENDED}
            action_plans = objects.ActionPlan.list(
                request_context, filters=a_plan_filters, eager=True)
            for plan in action_plans:
                plan.state = objects.action_plan.State.CANCELLED
                plan.save()
        return solution

    @staticmethod
    def _next_cron_time(audit):
        if utils.is_cron_like(audit.interval):
            return croniter(audit.interval, timeutils.utcnow()
                            ).get_next(datetime.datetime)

    @classmethod
    def execute_audit(cls, audit, request_context):
        self = cls()
        if not self._is_audit_inactive(audit):
            try:
                self.execute(audit, request_context)
            except Exception:
                raise
            finally:
                if utils.is_int_like(audit.interval):
                    audit.next_run_time = (
                        timeutils.utcnow() +
                        datetime.timedelta(seconds=int(audit.interval)))
                else:
                    audit.next_run_time = self._next_cron_time(audit)
                audit.save()

    def _add_job(self, trigger, audit, audit_context, **trigger_args):
        time_var = 'next_run_time' if trigger_args.get(
            'next_run_time') else 'run_date'
        # We should convert UTC time to local time without tzinfo
        trigger_args[time_var] = trigger_args[time_var].replace(
            tzinfo=tz.tzutc()).astimezone(tz.tzlocal()).replace(tzinfo=None)
        self.scheduler.add_job(self.execute_audit, trigger,
                               args=[audit, audit_context],
                               name='execute_audit',
                               **trigger_args)

    def check_audit_expired(self, audit):
        current = timeutils.utcnow()
        # Note: if audit still didn't get into the timeframe,
        #       skip it
        if audit.start_time and audit.start_time > current:
            return True
        if audit.end_time and audit.end_time < current:
            if audit.state != objects.audit.State.SUCCEEDED:
                audit.state = objects.audit.State.SUCCEEDED
                audit.save()
            return True

        return False

    def launch_audits_periodically(self):
        # if audit scheduler stop, restart it
        if not self.scheduler.running:
            self.scheduler.start()

        audit_context = context.RequestContext(is_admin=True)
        audit_filters = {
            'audit_type': objects.audit.AuditType.CONTINUOUS.value,
            'state__in': (objects.audit.State.PENDING,
                          objects.audit.State.ONGOING),
        }
        audit_filters['hostname'] = None
        unscheduled_audits = objects.Audit.list(
            audit_context, filters=audit_filters, eager=True)
        for audit in unscheduled_audits:
            # If continuous audit doesn't have a hostname yet,
            # Watcher will set current CONF.host value.
            # TODO(alexchadin): Add scheduling of new continuous audits.
            audit.hostname = CONF.host
            audit.save()
        scheduler_job_args = [
            (job.args[0].uuid, job) for job
            in self.scheduler.get_jobs()
            if job.name == 'execute_audit']
        scheduler_jobs = dict(scheduler_job_args)
        # if audit isn't in active states, audit's job should be removed
        jobs_to_remove = []
        for job in scheduler_jobs.values():
            if self._is_audit_inactive(job.args[0]):
                jobs_to_remove.append(job.args[0].uuid)
        for audit_uuid in jobs_to_remove:
            scheduler_jobs.pop(audit_uuid)
        audit_filters['hostname'] = CONF.host
        audits = objects.Audit.list(
            audit_context, filters=audit_filters, eager=True)
        for audit in audits:
            if self.check_audit_expired(audit):
                continue
            existing_job = scheduler_jobs.get(audit.uuid, None)
            # if audit is not presented in scheduled audits yet,
            # just add a new audit job.
            # if audit is already in the job queue, and interval has changed,
            # we need to remove the old job and add a new one.
            if (existing_job is None) or (
                existing_job and
                    audit.interval != existing_job.args[0].interval):
                if existing_job:
                    self.scheduler.remove_job(existing_job.id)
                # if interval is provided with seconds
                if utils.is_int_like(audit.interval):
                    # if audit has already been provided and we need
                    # to restore it after shutdown
                    if audit.next_run_time is not None:
                        old_run_time = audit.next_run_time
                        current = timeutils.utcnow()
                        if old_run_time < current:
                            delta = datetime.timedelta(
                                seconds=(int(audit.interval) - (
                                    current - old_run_time).seconds %
                                    int(audit.interval)))
                            audit.next_run_time = current + delta
                        next_run_time = audit.next_run_time
                    # if audit is new one
                    else:
                        next_run_time = timeutils.utcnow()
                    self._add_job('interval', audit, audit_context,
                                  seconds=int(audit.interval),
                                  next_run_time=next_run_time)

                else:
                    audit.next_run_time = self._next_cron_time(audit)
                    self._add_job('date', audit, audit_context,
                                  run_date=audit.next_run_time)
                audit.hostname = CONF.host
                audit.save()

    def start(self):
        self.period_scheduler.add_job(
            self.launch_audits_periodically,
            'interval',
            seconds=CONF.watcher_decision_engine.continuous_audit_interval,
            next_run_time=datetime.datetime.now())
        self.period_scheduler.start()
        # audit scheduler start
        self.scheduler.start()