--- /dev/null
+##
+# 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()
projects / osm / N2VC.git / blobdiff