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.common.common_db_client import CommonDbClient
  34 from osm_policy_module.common.message_bus_client import MessageBusClient
  35 from osm_policy_module.core.config import Config
  36
  37 log = logging.getLogger(__name__)
  38
  39 ALLOWED_KAFKA_KEYS = ["instantiated", "scaled", "terminated", "notify_alarm"]
  40
  41
  42 class PolicyModuleAgent:
  43     def __init__(self, config: Config, loop=None):
  44         self.conf = config
  45         if not loop:
  46             loop = asyncio.get_event_loop()
  47         self.loop = loop
  48         self.msg_bus = MessageBusClient(config)
  49         self.db_client = CommonDbClient(config)
  50         self.autoscaling_service = AutoscalingService(config, loop)
  51         self.alarming_service = AlarmingService(config, loop)
  52
  53     def run(self):
  54         self.loop.run_until_complete(self.start())
  55
  56     async def start(self):
  57         Path("/tmp/osm_pol_agent_health_flag").touch()
  58         topics = ["ns", "alarm_response"]
  59         await self.msg_bus.aioread(topics, self._process_msg)
  60         log.critical("Exiting...")
  61         if os.path.exists("/tmp/osm_pol_agent_health_flag"):
  62             os.remove("/tmp/osm_pol_agent_health_flag")
  63
  64     async def _process_msg(self, topic, key, msg):
  65         Path("/tmp/osm_pol_agent_health_flag").touch()
  66         log.debug("_process_msg topic=%s key=%s msg=%s", topic, key, msg)
  67         try:
  68             if key in ALLOWED_KAFKA_KEYS:
  69
  70                 if key == "instantiated":
  71                     await self._handle_instantiated(msg)
  72
  73                 if key == "scaled":
  74                     await self._handle_scaled(msg)
  75
  76                 if key == "terminated":
  77                     await self._handle_terminated(msg)
  78
  79                 if key == "notify_alarm":
  80                     await self._handle_alarm_notification(msg)
  81             else:
  82                 log.debug("Key %s is not in ALLOWED_KAFKA_KEYS", key)
  83         except peewee.PeeweeException:
  84             log.exception("Database error consuming message: ")
  85             raise
  86         except Exception:
  87             log.exception("Error consuming message: ")
  88
  89     async def _handle_alarm_notification(self, content):
  90         log.debug("_handle_alarm_notification: %s", content)
  91         alarm_uuid = content["notify_details"]["alarm_uuid"]
  92         status = content["notify_details"]["status"]
  93         await self.autoscaling_service.handle_alarm(alarm_uuid, status)
  94         await self.alarming_service.handle_alarm(alarm_uuid, status, content)
  95
  96     async def _handle_instantiated(self, content):
  97         log.debug("_handle_instantiated: %s", content)
  98         nslcmop_id = content["nslcmop_id"]
  99         nslcmop = self.db_client.get_nslcmop(nslcmop_id)
 100         if (
 101             nslcmop["operationState"] == "COMPLETED"
 102             or nslcmop["operationState"] == "PARTIALLY_COMPLETED"
 103         ):
 104             nsr_id = nslcmop["nsInstanceId"]
 105             log.info("Configuring nsr_id: %s", nsr_id)
 106             await self.autoscaling_service.configure_scaling_groups(nsr_id)
 107             await self.alarming_service.configure_vnf_alarms(nsr_id)
 108         else:
 109             log.info(
 110                 "Network_service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 111                 "Current state is %s. Skipping...",
 112                 nslcmop["operationState"],
 113             )
 114
 115     async def _handle_scaled(self, content):
 116         log.debug("_handle_scaled: %s", content)
 117         nslcmop_id = content["nslcmop_id"]
 118         nslcmop = self.db_client.get_nslcmop(nslcmop_id)
 119         if (
 120             nslcmop["operationState"] == "COMPLETED"
 121             or nslcmop["operationState"] == "PARTIALLY_COMPLETED"
 122         ):
 123             nsr_id = nslcmop["nsInstanceId"]
 124             log.info("Configuring scaled service with nsr_id: %s", nsr_id)
 125             await self.autoscaling_service.configure_scaling_groups(nsr_id)
 126             await self.autoscaling_service.delete_orphaned_alarms(nsr_id)
 127             await self.alarming_service.configure_vnf_alarms(nsr_id)
 128         else:
 129             log.debug(
 130                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 131                 "Current state is %s. Skipping...",
 132                 nslcmop["operationState"],
 133             )
 134
 135     async def _handle_terminated(self, content):
 136         log.debug("_handle_deleted: %s", content)
 137         nsr_id = content["nsr_id"]
 138         if (
 139             content["operationState"] == "COMPLETED"
 140             or content["operationState"] == "PARTIALLY_COMPLETED"
 141         ):
 142             log.info(
 143                 "Deleting scaling groups and alarms for network autoscaling_service with nsr_id: %s",
 144                 nsr_id,
 145             )
 146             await self.autoscaling_service.delete_scaling_groups(nsr_id)
 147             await self.alarming_service.delete_vnf_alarms(nsr_id)
 148         else:
 149             log.info(
 150                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 151                 "Current state is %s. Skipping...",
 152                 content["operationState"],
 153             )