X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_policy_module%2Fcore%2Fagent.py;h=2daebdeab0880d3340dd6cfdb56a3789cf35b6d4;hb=e9795318c10224753ef70ac4071ee88283a9dd12;hp=777723bfd25f57c6a5ad7d2d99f6e9fc2fb798b9;hpb=cb5642a2af495af4319beb1dba2d417b78f3200a;p=osm%2FPOL.git diff --git a/osm_policy_module/core/agent.py b/osm_policy_module/core/agent.py index 777723b..2daebde 100644 --- a/osm_policy_module/core/agent.py +++ b/osm_policy_module/core/agent.py @@ -30,13 +30,21 @@ import peewee from osm_policy_module.alarming.service import AlarmingService from osm_policy_module.autoscaling.service import AutoscalingService +from osm_policy_module.healing.service import HealingService from osm_policy_module.common.common_db_client import CommonDbClient from osm_policy_module.common.message_bus_client import MessageBusClient from osm_policy_module.core.config import Config log = logging.getLogger(__name__) -ALLOWED_KAFKA_KEYS = ["instantiated", "scaled", "terminated", "notify_alarm", "policy_updated", "vnf_terminated"] +ALLOWED_KAFKA_KEYS = [ + "instantiated", + "scaled", + "terminated", + "notify_alarm", + "policy_updated", + "vnf_terminated", +] class PolicyModuleAgent: @@ -49,6 +57,7 @@ class PolicyModuleAgent: self.db_client = CommonDbClient(config) self.autoscaling_service = AutoscalingService(config, loop) self.alarming_service = AlarmingService(config, loop) + self.healing_service = HealingService(config, loop) def run(self): self.loop.run_until_complete(self.start()) @@ -66,7 +75,6 @@ class PolicyModuleAgent: log.debug("_process_msg topic=%s key=%s msg=%s", topic, key, msg) try: if key in ALLOWED_KAFKA_KEYS: - if key == "instantiated": await self._handle_instantiated(msg) @@ -98,6 +106,7 @@ class PolicyModuleAgent: status = content["notify_details"]["status"] await self.autoscaling_service.handle_alarm(alarm_uuid, status) await self.alarming_service.handle_alarm(alarm_uuid, status, content) + await self.healing_service.handle_alarm(alarm_uuid, status) async def _handle_instantiated(self, content): log.debug("_handle_instantiated: %s", content) @@ -111,6 +120,7 @@ class PolicyModuleAgent: log.info("Configuring nsr_id: %s", nsr_id) await self.autoscaling_service.configure_scaling_groups(nsr_id) await self.alarming_service.configure_vnf_alarms(nsr_id) + await self.healing_service.configure_healing_alarms(nsr_id) else: log.info( "Network_service is not in COMPLETED or PARTIALLY_COMPLETED state. " @@ -131,6 +141,8 @@ class PolicyModuleAgent: await self.autoscaling_service.configure_scaling_groups(nsr_id) await self.autoscaling_service.delete_orphaned_alarms(nsr_id) await self.alarming_service.configure_vnf_alarms(nsr_id) + await self.healing_service.configure_healing_alarms(nsr_id) + await self.healing_service.delete_orphaned_healing_alarms(nsr_id) else: log.debug( "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. " @@ -151,6 +163,7 @@ class PolicyModuleAgent: ) await self.autoscaling_service.delete_scaling_groups(nsr_id) await self.alarming_service.delete_vnf_alarms(nsr_id) + await self.healing_service.delete_healing_alarms(nsr_id) else: log.info( "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. " @@ -160,39 +173,49 @@ class PolicyModuleAgent: async def _handle_policy_update(self, content): log.info("_handle_policy_update: %s", content) - nsr_id = content['nsr_id'] - vnf_member_index = content['vnf_member_index'] + nsr_id = content["nsr_id"] + vnf_member_index = content["vnf_member_index"] if ( content["operationState"] == "COMPLETED" or content["operationState"] == "PARTIALLY_COMPLETED" ): log.info( "Updating policies of VNF with nsr_id: %s and vnf-member-index: %s" - % (nsr_id, vnf_member_index)) - await self.autoscaling_service.delete_scaling_groups(nsr_id, vnf_member_index) + % (nsr_id, vnf_member_index) + ) + await self.autoscaling_service.delete_scaling_groups( + nsr_id, vnf_member_index + ) await self.alarming_service.delete_vnf_alarms(nsr_id, vnf_member_index) - await self.autoscaling_service.configure_scaling_groups(nsr_id, vnf_member_index) + await self.autoscaling_service.configure_scaling_groups( + nsr_id, vnf_member_index + ) await self.alarming_service.configure_vnf_alarms(nsr_id, vnf_member_index) else: log.info( "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. " "Current state is %s. Skipping...", - content['operationState']) + content["operationState"], + ) async def _handle_vnf_terminated(self, content): - nsr_id = content['nsr_id'] - vnf_member_index = content['vnf_member_index'] + nsr_id = content["nsr_id"] + vnf_member_index = content["vnf_member_index"] if ( content["operationState"] == "COMPLETED" or content["operationState"] == "PARTIALLY_COMPLETED" ): log.info( "Deleting policies of VNF with nsr_id: %s and vnf-member-index: %s" - % (nsr_id, vnf_member_index)) - await self.autoscaling_service.delete_scaling_groups(nsr_id, vnf_member_index) + % (nsr_id, vnf_member_index) + ) + await self.autoscaling_service.delete_scaling_groups( + nsr_id, vnf_member_index + ) await self.alarming_service.delete_vnf_alarms(nsr_id, vnf_member_index) else: log.info( "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. " "Current state is %s. Skipping...", - content['operationState']) + content["operationState"], + )