osm_policy_module/core/agent.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Whitestack, LLC
   4 # *************************************************************
   5
   6 # This file is part of OSM Monitoring module
   7 # All Rights Reserved to Whitestack, LLC
   8
   9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  10 # not use this file except in compliance with the License. You may obtain
  11 # a copy of the License at
  12
  13 #         http://www.apache.org/licenses/LICENSE-2.0
  14
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  18 # License for the specific language governing permissions and limitations
  19 # under the License.
  20
  21 # For those usages not covered by the Apache License, Version 2.0 please
  22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  23 ##
  24 import asyncio
  25 import logging
  26 from pathlib import Path
  27 import os
  28
  29 import peewee
  30
  31 from osm_policy_module.alarming.service import AlarmingService
  32 from osm_policy_module.autoscaling.service import AutoscalingService
  33 from osm_policy_module.healing.service import HealingService
  34 from osm_policy_module.common.common_db_client import CommonDbClient
  35 from osm_policy_module.common.message_bus_client import MessageBusClient
  36 from osm_policy_module.core.config import Config
  37
  38 log = logging.getLogger(__name__)
  39
  40 ALLOWED_KAFKA_KEYS = ["instantiated", "scaled", "terminated", "notify_alarm", "policy_updated", "vnf_terminated"]
  41
  42
  43 class PolicyModuleAgent:
  44     def __init__(self, config: Config, loop=None):
  45         self.conf = config
  46         if not loop:
  47             loop = asyncio.get_event_loop()
  48         self.loop = loop
  49         self.msg_bus = MessageBusClient(config)
  50         self.db_client = CommonDbClient(config)
  51         self.autoscaling_service = AutoscalingService(config, loop)
  52         self.alarming_service = AlarmingService(config, loop)
  53         self.healing_service = HealingService(config, loop)
  54
  55     def run(self):
  56         self.loop.run_until_complete(self.start())
  57
  58     async def start(self):
  59         Path("/tmp/osm_pol_agent_health_flag").touch()
  60         topics = ["ns", "alarm_response"]
  61         await self.msg_bus.aioread(topics, self._process_msg)
  62         log.critical("Exiting...")
  63         if os.path.exists("/tmp/osm_pol_agent_health_flag"):
  64             os.remove("/tmp/osm_pol_agent_health_flag")
  65
  66     async def _process_msg(self, topic, key, msg):
  67         Path("/tmp/osm_pol_agent_health_flag").touch()
  68         log.debug("_process_msg topic=%s key=%s msg=%s", topic, key, msg)
  69         try:
  70             if key in ALLOWED_KAFKA_KEYS:
  71
  72                 if key == "instantiated":
  73                     await self._handle_instantiated(msg)
  74
  75                 if key == "scaled":
  76                     await self._handle_scaled(msg)
  77
  78                 if key == "terminated":
  79                     await self._handle_terminated(msg)
  80
  81                 if key == "notify_alarm":
  82                     await self._handle_alarm_notification(msg)
  83
  84                 if key == "policy_updated":
  85                     await self._handle_policy_update(msg)
  86
  87                 if key == "vnf_terminated":
  88                     await self._handle_vnf_terminated(msg)
  89             else:
  90                 log.debug("Key %s is not in ALLOWED_KAFKA_KEYS", key)
  91         except peewee.PeeweeException:
  92             log.exception("Database error consuming message: ")
  93             raise
  94         except Exception:
  95             log.exception("Error consuming message: ")
  96
  97     async def _handle_alarm_notification(self, content):
  98         log.debug("_handle_alarm_notification: %s", content)
  99         alarm_uuid = content["notify_details"]["alarm_uuid"]
 100         status = content["notify_details"]["status"]
 101         await self.autoscaling_service.handle_alarm(alarm_uuid, status)
 102         await self.alarming_service.handle_alarm(alarm_uuid, status, content)
 103         await self.healing_service.handle_alarm(alarm_uuid, status)
 104
 105     async def _handle_instantiated(self, content):
 106         log.debug("_handle_instantiated: %s", content)
 107         nslcmop_id = content["nslcmop_id"]
 108         nslcmop = self.db_client.get_nslcmop(nslcmop_id)
 109         if (
 110             nslcmop["operationState"] == "COMPLETED"
 111             or nslcmop["operationState"] == "PARTIALLY_COMPLETED"
 112         ):
 113             nsr_id = nslcmop["nsInstanceId"]
 114             log.info("Configuring nsr_id: %s", nsr_id)
 115             await self.autoscaling_service.configure_scaling_groups(nsr_id)
 116             await self.alarming_service.configure_vnf_alarms(nsr_id)
 117             await self.healing_service.configure_healing_alarms(nsr_id)
 118         else:
 119             log.info(
 120                 "Network_service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 121                 "Current state is %s. Skipping...",
 122                 nslcmop["operationState"],
 123             )
 124
 125     async def _handle_scaled(self, content):
 126         log.debug("_handle_scaled: %s", content)
 127         nslcmop_id = content["nslcmop_id"]
 128         nslcmop = self.db_client.get_nslcmop(nslcmop_id)
 129         if (
 130             nslcmop["operationState"] == "COMPLETED"
 131             or nslcmop["operationState"] == "PARTIALLY_COMPLETED"
 132         ):
 133             nsr_id = nslcmop["nsInstanceId"]
 134             log.info("Configuring scaled service with nsr_id: %s", nsr_id)
 135             await self.autoscaling_service.configure_scaling_groups(nsr_id)
 136             await self.autoscaling_service.delete_orphaned_alarms(nsr_id)
 137             await self.alarming_service.configure_vnf_alarms(nsr_id)
 138             await self.healing_service.configure_healing_alarms(nsr_id)
 139             await self.healing_service.delete_orphaned_healing_alarms(nsr_id)
 140         else:
 141             log.debug(
 142                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 143                 "Current state is %s. Skipping...",
 144                 nslcmop["operationState"],
 145             )
 146
 147     async def _handle_terminated(self, content):
 148         log.debug("_handle_deleted: %s", content)
 149         nsr_id = content["nsr_id"]
 150         if (
 151             content["operationState"] == "COMPLETED"
 152             or content["operationState"] == "PARTIALLY_COMPLETED"
 153         ):
 154             log.info(
 155                 "Deleting scaling groups and alarms for network autoscaling_service with nsr_id: %s",
 156                 nsr_id,
 157             )
 158             await self.autoscaling_service.delete_scaling_groups(nsr_id)
 159             await self.alarming_service.delete_vnf_alarms(nsr_id)
 160             await self.healing_service.delete_healing_alarms(nsr_id)
 161         else:
 162             log.info(
 163                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 164                 "Current state is %s. Skipping...",
 165                 content["operationState"],
 166             )
 167
 168     async def _handle_policy_update(self, content):
 169         log.info("_handle_policy_update: %s", content)
 170         nsr_id = content['nsr_id']
 171         vnf_member_index = content['vnf_member_index']
 172         if (
 173             content["operationState"] == "COMPLETED"
 174             or content["operationState"] == "PARTIALLY_COMPLETED"
 175         ):
 176             log.info(
 177                 "Updating policies of VNF with nsr_id: %s and vnf-member-index: %s"
 178                 % (nsr_id, vnf_member_index))
 179             await self.autoscaling_service.delete_scaling_groups(nsr_id, vnf_member_index)
 180             await self.alarming_service.delete_vnf_alarms(nsr_id, vnf_member_index)
 181             await self.autoscaling_service.configure_scaling_groups(nsr_id, vnf_member_index)
 182             await self.alarming_service.configure_vnf_alarms(nsr_id, vnf_member_index)
 183         else:
 184             log.info(
 185                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 186                 "Current state is %s. Skipping...",
 187                 content['operationState'])
 188
 189     async def _handle_vnf_terminated(self, content):
 190         nsr_id = content['nsr_id']
 191         vnf_member_index = content['vnf_member_index']
 192         if (
 193             content["operationState"] == "COMPLETED"
 194             or content["operationState"] == "PARTIALLY_COMPLETED"
 195         ):
 196             log.info(
 197                 "Deleting policies of VNF with nsr_id: %s and vnf-member-index: %s"
 198                 % (nsr_id, vnf_member_index))
 199             await self.autoscaling_service.delete_scaling_groups(nsr_id, vnf_member_index)
 200             await self.alarming_service.delete_vnf_alarms(nsr_id, vnf_member_index)
 201         else:
 202             log.info(
 203                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 204                 "Current state is %s. Skipping...",
 205                 content['operationState'])