osm_policy_module/autoscaling/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
  27 import peewee
  28
  29 from osm_policy_module.autoscaling.service import Service
  30 from osm_policy_module.common.message_bus_client import MessageBusClient
  31 from osm_policy_module.core.config import Config
  32 from osm_policy_module.core.database import ScalingAlarm
  33
  34 log = logging.getLogger(__name__)
  35
  36 ALLOWED_KAFKA_KEYS = ['instantiated', 'scaled', 'terminated', 'notify_alarm']
  37
  38
  39 class PolicyModuleAgent:
  40     def __init__(self, config: Config, loop=None):
  41         self.conf = config
  42         if not loop:
  43             loop = asyncio.get_event_loop()
  44         self.loop = loop
  45         self.msg_bus = MessageBusClient(config)
  46         self.service = Service(config, loop)
  47
  48     def run(self):
  49         self.loop.run_until_complete(self.start())
  50
  51     async def start(self):
  52         topics = [
  53             "ns",
  54             "alarm_response"
  55         ]
  56         await self.msg_bus.aioread(topics, self._process_msg)
  57         log.critical("Exiting...")
  58
  59     async def _process_msg(self, topic, key, msg):
  60         log.debug("_process_msg topic=%s key=%s msg=%s", topic, key, msg)
  61         log.info("Message arrived: %s", msg)
  62         try:
  63             if key in ALLOWED_KAFKA_KEYS:
  64
  65                 if key == 'instantiated':
  66                     await self._handle_instantiated(msg)
  67
  68                 if key == 'scaled':
  69                     await self._handle_scaled(msg)
  70
  71                 if key == 'terminated':
  72                     await self._handle_terminated(msg)
  73
  74                 if key == 'notify_alarm':
  75                     await self._handle_alarm_notification(msg)
  76             else:
  77                 log.debug("Key %s is not in ALLOWED_KAFKA_KEYS", key)
  78         except peewee.PeeweeException:
  79             log.exception("Database error consuming message: ")
  80             raise
  81         except Exception:
  82             log.exception("Error consuming message: ")
  83
  84     async def _handle_alarm_notification(self, content):
  85         log.debug("_handle_alarm_notification: %s", content)
  86         alarm_uuid = content['notify_details']['alarm_uuid']
  87         metric_name = content['notify_details']['metric_name']
  88         operation = content['notify_details']['operation']
  89         threshold = content['notify_details']['threshold_value']
  90         vdu_name = content['notify_details']['vdu_name']
  91         vnf_member_index = content['notify_details']['vnf_member_index']
  92         nsr_id = content['notify_details']['ns_id']
  93         log.info(
  94             "Received alarm notification for alarm %s, \
  95             metric %s, \
  96             operation %s, \
  97             threshold %s, \
  98             vdu_name %s, \
  99             vnf_member_index %s, \
 100             ns_id %s ",
 101             alarm_uuid, metric_name, operation, threshold, vdu_name, vnf_member_index, nsr_id)
 102         try:
 103             alarm = self.service.get_alarm(alarm_uuid)
 104             await self.service.scale(alarm)
 105         except ScalingAlarm.DoesNotExist:
 106             log.info("There is no action configured for alarm %s.", alarm_uuid)
 107
 108     async def _handle_instantiated(self, content):
 109         log.debug("_handle_instantiated: %s", content)
 110         nslcmop_id = content['nslcmop_id']
 111         nslcmop = self.service.get_nslcmop(nslcmop_id)
 112         if nslcmop['operationState'] == 'COMPLETED' or nslcmop['operationState'] == 'PARTIALLY_COMPLETED':
 113             nsr_id = nslcmop['nsInstanceId']
 114             log.info("Configuring scaling groups for network service with nsr_id: %s", nsr_id)
 115             await self.service.configure_scaling_groups(nsr_id)
 116         else:
 117             log.info(
 118                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 119                 "Current state is %s. Skipping...",
 120                 nslcmop['operationState'])
 121
 122     async def _handle_scaled(self, content):
 123         log.debug("_handle_scaled: %s", content)
 124         nslcmop_id = content['nslcmop_id']
 125         nslcmop = self.service.get_nslcmop(nslcmop_id)
 126         if nslcmop['operationState'] == 'COMPLETED' or nslcmop['operationState'] == 'PARTIALLY_COMPLETED':
 127             nsr_id = nslcmop['nsInstanceId']
 128             log.info("Configuring scaling groups for network service with nsr_id: %s", nsr_id)
 129             await self.service.configure_scaling_groups(nsr_id)
 130             log.info("Checking for orphaned alarms to be deleted for network service with nsr_id: %s", nsr_id)
 131             await self.service.delete_orphaned_alarms(nsr_id)
 132         else:
 133             log.info(
 134                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 135                 "Current state is %s. Skipping...",
 136                 nslcmop['operationState'])
 137
 138     async def _handle_terminated(self, content):
 139         log.debug("_handle_deleted: %s", content)
 140         nsr_id = content['nsr_id']
 141         if content['operationState'] == 'COMPLETED' or content['operationState'] == 'PARTIALLY_COMPLETED':
 142             log.info("Deleting scaling groups and alarms for network service with nsr_id: %s", nsr_id)
 143             await self.service.delete_scaling_groups(nsr_id)
 144         else:
 145             log.info(
 146                 "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
 147                 "Current state is %s. Skipping...",
 148                 content['operationState'])