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 = [
  41     "instantiated",
  42     "scaled",
  43     "terminated",
  44     "notify_alarm",
  45     "policy_updated",
  46     "vnf_terminated",
  47 ]
  48
  49
  50 class PolicyModuleAgent:
  51     def __init__(self, config: Config, loop=None):
  52         self.conf = config
  53         if not loop:
  54             loop = asyncio.get_event_loop()
  55         self.loop = loop
  56         self.msg_bus = MessageBusClient(config)
  57         self.db_client = CommonDbClient(config)
  58         self.autoscaling_service = AutoscalingService(config, loop)
  59         self.alarming_service = AlarmingService(config, loop)
  60         self.healing_service = HealingService(config, loop)
  61
  62     def run(self):
  63         self.loop.run_until_complete(self.start())
  64
  65     async def start(self):
  66         Path("/tmp/osm_pol_agent_health_flag").touch()
  67         topics = ["ns", "alarm_response"]
  68         await self.msg_bus.aioread(topics, self._process_msg)
  69         log.critical("Exiting...")
  70         if os.path.exists("/tmp/osm_pol_agent_health_flag"):
  71             os.remove("/tmp/osm_pol_agent_health_flag")
  72
  73     async def _process_msg(self, topic, key, msg):
  74         Path("/tmp/osm_pol_agent_health_flag").touch()
  75         log.debug("_process_msg topic=%s key=%s msg=%s", topic, key, msg)
  76         try:
  77             if key in ALLOWED_KAFKA_KEYS:
  78
  79                 if key == "instantiated":
  80                     await self._handle_instantiated(msg)
  81
  82                 if key == "scaled":
  83                     await self._handle_scaled(msg)
  84
  85                 if key == "terminated":
  86                     await self._handle_terminated(msg)
  87
  88                 if key == "notify_alarm":
  89                     await self._handle_alarm_notification(msg)
  90
  91                 if key == "policy_updated":
  92                     await self._handle_policy_update(msg)
  93
  94                 if key == "vnf_terminated":
  95                     await self._handle_vnf_terminated(msg)
  96             else:
  97                 log.debug("Key %s is not in ALLOWED_KAFKA_KEYS", key)
  98         except peewee.PeeweeException:
  99             log.exception("Database error consuming message: ")
 100             raise
 101         except Exception:
 102             log.exception("Error consuming message: ")
 103
 104     async def _handle_alarm_notification(self, content):
 105         log.debug("_handle_alarm_notification: %s", content)
 106         alarm_uuid = content["notify_details"]["alarm_uuid"]
 107         status = content["notify_details"]["status"]
 108         await self.autoscaling_service.handle_alarm(alarm_uuid, status)
 109         await self.alarming_service.handle_alarm(alarm_uuid, status, content)
 110         await self.healing_service.handle_alarm(alarm_uuid, status)
 111
 112     async def _handle_instantiated(self, content):
 113         log.debug("_handle_instantiated: %s", content)
 114         nslcmop_id = content["nslcmop_id"]
 115         nslcmop = self.db_client.get_nslcmop(nslcmop_id)
 116         if (
 117             nslcmop["operationState"] == "COMPLETED"
 118             or nslcmop["operationState"] == "PARTIALLY_COMPLETED"
 119         ):
 120             nsr_id = nslcmop["nsInstanceId"]
 121             log.info("Configuring nsr_id: %s", nsr_id)
 122             await self.autoscaling_service.configure_scaling_groups(nsr_id)
 123             await self.alarming_service.configure_vnf_alarms(nsr_id)
 124             await self.healing_service.configure_healing_alarms(nsr_id)
 125         else:
 126             log.info(
 127                 "Network_service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 128                 "Current state is %s. Skipping...",
 129                 nslcmop["operationState"],
 130             )
 131
 132     async def _handle_scaled(self, content):
 133         log.debug("_handle_scaled: %s", content)
 134         nslcmop_id = content["nslcmop_id"]
 135         nslcmop = self.db_client.get_nslcmop(nslcmop_id)
 136         if (
 137             nslcmop["operationState"] == "COMPLETED"
 138             or nslcmop["operationState"] == "PARTIALLY_COMPLETED"
 139         ):
 140             nsr_id = nslcmop["nsInstanceId"]
 141             log.info("Configuring scaled service with nsr_id: %s", nsr_id)
 142             await self.autoscaling_service.configure_scaling_groups(nsr_id)
 143             await self.autoscaling_service.delete_orphaned_alarms(nsr_id)
 144             await self.alarming_service.configure_vnf_alarms(nsr_id)
 145             await self.healing_service.configure_healing_alarms(nsr_id)
 146             await self.healing_service.delete_orphaned_healing_alarms(nsr_id)
 147         else:
 148             log.debug(
 149                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 150                 "Current state is %s. Skipping...",
 151                 nslcmop["operationState"],
 152             )
 153
 154     async def _handle_terminated(self, content):
 155         log.debug("_handle_deleted: %s", content)
 156         nsr_id = content["nsr_id"]
 157         if (
 158             content["operationState"] == "COMPLETED"
 159             or content["operationState"] == "PARTIALLY_COMPLETED"
 160         ):
 161             log.info(
 162                 "Deleting scaling groups and alarms for network autoscaling_service with nsr_id: %s",
 163                 nsr_id,
 164             )
 165             await self.autoscaling_service.delete_scaling_groups(nsr_id)
 166             await self.alarming_service.delete_vnf_alarms(nsr_id)
 167             await self.healing_service.delete_healing_alarms(nsr_id)
 168         else:
 169             log.info(
 170                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 171                 "Current state is %s. Skipping...",
 172                 content["operationState"],
 173             )
 174
 175     async def _handle_policy_update(self, content):
 176         log.info("_handle_policy_update: %s", content)
 177         nsr_id = content["nsr_id"]
 178         vnf_member_index = content["vnf_member_index"]
 179         if (
 180             content["operationState"] == "COMPLETED"
 181             or content["operationState"] == "PARTIALLY_COMPLETED"
 182         ):
 183             log.info(
 184                 "Updating policies of VNF with nsr_id: %s and vnf-member-index: %s"
 185                 % (nsr_id, vnf_member_index)
 186             )
 187             await self.autoscaling_service.delete_scaling_groups(
 188                 nsr_id, vnf_member_index
 189             )
 190             await self.alarming_service.delete_vnf_alarms(nsr_id, vnf_member_index)
 191             await self.autoscaling_service.configure_scaling_groups(
 192                 nsr_id, vnf_member_index
 193             )
 194             await self.alarming_service.configure_vnf_alarms(nsr_id, vnf_member_index)
 195         else:
 196             log.info(
 197                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 198                 "Current state is %s. Skipping...",
 199                 content["operationState"],
 200             )
 201
 202     async def _handle_vnf_terminated(self, content):
 203         nsr_id = content["nsr_id"]
 204         vnf_member_index = content["vnf_member_index"]
 205         if (
 206             content["operationState"] == "COMPLETED"
 207             or content["operationState"] == "PARTIALLY_COMPLETED"
 208         ):
 209             log.info(
 210                 "Deleting policies of VNF with nsr_id: %s and vnf-member-index: %s"
 211                 % (nsr_id, vnf_member_index)
 212             )
 213             await self.autoscaling_service.delete_scaling_groups(
 214                 nsr_id, vnf_member_index
 215             )
 216             await self.alarming_service.delete_vnf_alarms(nsr_id, vnf_member_index)
 217         else:
 218             log.info(
 219                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 220                 "Current state is %s. Skipping...",
 221                 content["operationState"],
 222             )