n2vc/juju_observer.py - osm/N2VC - Gitiles

 ##
 # 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()
	##
	# 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()