n2vc/juju_observer.py

##
# Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
# This file is part of OSM
# All Rights Reserved.
#
# 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.
#
# For those usages not covered by the Apache License, Version 2.0 please
# contact with: nfvlabs@tid.es
##

import asyncio
import time

from juju.action import Action
from juju.application import Application
from juju.machine import Machine
from juju.model import ModelObserver, Model

from n2vc.exceptions import N2VCTimeoutException
from n2vc.n2vc_conn import N2VCConnector, juju_status_2_osm_status


class _Entity:
    def __init__(self, entity_id: str, entity_type: str, obj: object, db_dict: dict):
        self.entity_id = entity_id
        self.entity_type = entity_type
        self.obj = obj
        self.event = asyncio.Event()
        self.db_dict = db_dict


class JujuModelObserver(ModelObserver):
    def __init__(self, n2vc: N2VCConnector, model: Model):
        self.n2vc = n2vc
        self.model = model
        model.add_observer(self)
        self.machines = dict()
        self.applications = dict()
        self.actions = dict()

    def register_machine(self, machine: Machine, db_dict: dict):
        try:
            entity_id = machine.entity_id
        except Exception:
            # no entity_id aatribute, try machine attribute
            entity_id = machine.machine
        # self.n2vc.debug(
        #   msg='Registering machine for change notifications: {}'.format(entity_id))
        entity = _Entity(
            entity_id=entity_id, entity_type="machine", obj=machine, db_dict=db_dict
        )
        self.machines[entity_id] = entity

    def unregister_machine(self, machine_id: str):
        if machine_id in self.machines:
            del self.machines[machine_id]

    def is_machine_registered(self, machine_id: str):
        return machine_id in self.machines

    def register_application(self, application: Application, db_dict: dict):
        entity_id = application.entity_id
        # self.n2vc.debug(
        #  msg='Registering application for change notifications: {}'.format(entity_id))
        entity = _Entity(
            entity_id=entity_id,
            entity_type="application",
            obj=application,
            db_dict=db_dict,
        )
        self.applications[entity_id] = entity

    def unregister_application(self, application_id: str):
        if application_id in self.applications:
            del self.applications[application_id]

    def is_application_registered(self, application_id: str):
        return application_id in self.applications

    def register_action(self, action: Action, db_dict: dict):
        entity_id = action.entity_id
        # self.n2vc.debug(
        #    msg='Registering action for changes notifications: {}'.format(entity_id))
        entity = _Entity(
            entity_id=entity_id, entity_type="action", obj=action, db_dict=db_dict
        )
        self.actions[entity_id] = entity

    def unregister_action(self, action_id: str):
        if action_id in self.actions:
            del self.actions[action_id]

    def is_action_registered(self, action_id: str):
        return action_id in self.actions

    async def wait_for_machine(
        self,
        machine_id: str,
        progress_timeout: float = None,
        total_timeout: float = None,
    ) -> int:

        if not self.is_machine_registered(machine_id):
            return

        self.n2vc.debug("Waiting for machine completed: {}".format(machine_id))

        # wait for a final state
        entity = self.machines[machine_id]
        return await self._wait_for_entity(
            entity=entity,
            field_to_check="agent_status",
            final_states_list=["started"],
            progress_timeout=progress_timeout,
            total_timeout=total_timeout,
        )

    async def wait_for_application(
        self,
        application_id: str,
        progress_timeout: float = None,
        total_timeout: float = None,
    ) -> int:

        if not self.is_application_registered(application_id):
            return

        self.n2vc.debug("Waiting for application completed: {}".format(application_id))

        # application statuses: unknown, active, waiting
        # wait for a final state
        entity = self.applications[application_id]
        return await self._wait_for_entity(
            entity=entity,
            field_to_check="status",
            final_states_list=["active", "blocked"],
            progress_timeout=progress_timeout,
            total_timeout=total_timeout,
        )

    async def wait_for_action(
        self,
        action_id: str,
        progress_timeout: float = None,
        total_timeout: float = None,
    ) -> int:

        if not self.is_action_registered(action_id):
            return

        self.n2vc.debug("Waiting for action completed: {}".format(action_id))

        # action statuses: pending, running, completed, failed, cancelled
        # wait for a final state
        entity = self.actions[action_id]
        return await self._wait_for_entity(
            entity=entity,
            field_to_check="status",
            final_states_list=["completed", "failed", "cancelled"],
            progress_timeout=progress_timeout,
            total_timeout=total_timeout,
        )

    async def _wait_for_entity(
        self,
        entity: _Entity,
        field_to_check: str,
        final_states_list: list,
        progress_timeout: float = None,
        total_timeout: float = None,
    ) -> int:

        # default values for no timeout
        if total_timeout is None:
            total_timeout = 3600
        if progress_timeout is None:
            progress_timeout = 3600

        # max end time
        now = time.time()
        total_end = now + total_timeout

        if now >= total_end:
            raise N2VCTimeoutException(
                message="Total timeout {} seconds, {}: {}".format(
                    total_timeout, entity.entity_type, entity.entity_id
                ),
                timeout="total",
            )

        # update next progress timeout
        progress_end = now + progress_timeout  # type: float

        # which is closest? progress or end timeout?
        closest_end = min(total_end, progress_end)

        next_timeout = closest_end - now

        retries = 0

        while entity.obj.__getattribute__(field_to_check) not in final_states_list:
            retries += 1
            if await _wait_for_event_or_timeout(entity.event, next_timeout):
                entity.event.clear()
            else:
                message = "Progress timeout {} seconds, {}: {}".format(
                    progress_timeout, entity.entity_type, entity.entity_id
                )
                self.n2vc.debug(message)
                raise N2VCTimeoutException(message=message, timeout="progress")
        # self.n2vc.debug('End of wait. Final state: {}, retries: {}'
        #                 .format(entity.obj.__getattribute__(field_to_check), retries))
        return retries

    async def on_change(self, delta, old, new, model):

        if new is None:
            return

        # log
        # self.n2vc.debug('on_change(): type: {}, entity: {}, id: {}'
        #                 .format(delta.type, delta.entity, new.entity_id))

        if delta.entity == "machine":

            # check registered machine
            if new.entity_id not in self.machines:
                return

            # write change in database
            await self.n2vc.write_app_status_to_db(
                db_dict=self.machines[new.entity_id].db_dict,
                status=juju_status_2_osm_status(delta.entity, new.agent_status),
                detailed_status=new.status_message,
                vca_status=new.status,
                entity_type="machine",
            )

            # set event for this machine
            self.machines[new.entity_id].event.set()

        elif delta.entity == "application":

            # check registered application
            if new.entity_id not in self.applications:
                return

            # write change in database
            await self.n2vc.write_app_status_to_db(
                db_dict=self.applications[new.entity_id].db_dict,
                status=juju_status_2_osm_status(delta.entity, new.status),
                detailed_status=new.status_message,
                vca_status=new.status,
                entity_type="application",
            )

            # set event for this application
            self.applications[new.entity_id].event.set()

        elif delta.entity == "unit":

            # get the application for this unit
            application_id = delta.data["application"]

            # check registered application
            if application_id not in self.applications:
                return

            # write change in database
            if not new.dead:
                await self.n2vc.write_app_status_to_db(
                    db_dict=self.applications[application_id].db_dict,
                    status=juju_status_2_osm_status(delta.entity, new.workload_status),
                    detailed_status=new.workload_status_message,
                    vca_status=new.workload_status,
                    entity_type="unit",
                )

            # set event for this application
            self.applications[application_id].event.set()

        elif delta.entity == "action":

            # check registered action
            if new.entity_id not in self.actions:
                return

            # write change in database
            await self.n2vc.write_app_status_to_db(
                db_dict=self.actions[new.entity_id].db_dict,
                status=juju_status_2_osm_status(delta.entity, new.status),
                detailed_status=new.status,
                vca_status=new.status,
                entity_type="action",
            )

            # set event for this application
            self.actions[new.entity_id].event.set()


async def _wait_for_event_or_timeout(event: asyncio.Event, timeout: float = None):
    try:
        await asyncio.wait_for(fut=event.wait(), timeout=timeout)
    except asyncio.TimeoutError:
        pass
    return event.is_set()