X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Fjuju_observer.py;fp=n2vc%2Fjuju_observer.py;h=0000000000000000000000000000000000000000;hp=29ae93294e4fbd72c18dc00e9d8489586013a4ef;hb=e8102d9e28e5c502fc66ca842d14e1ad29efbfda;hpb=68b007225cc1fd432dba07fe42a5d1362925bc32 diff --git a/n2vc/juju_observer.py b/n2vc/juju_observer.py deleted file mode 100644 index 29ae932..0000000 --- a/n2vc/juju_observer.py +++ /dev/null @@ -1,318 +0,0 @@ -## -# 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()