osm_policy_module/common/mon_client.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 json
  26 import logging
  27 import random
  28 from json import JSONDecodeError
  29
  30 import yaml
  31 from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
  32
  33 from osm_policy_module.core.config import Config
  34
  35 log = logging.getLogger(__name__)
  36
  37
  38 class MonClient:
  39     def __init__(self, config: Config, loop=None):
  40         self.kafka_server = '{}:{}'.format(config.get('message', 'host'),
  41                                            config.get('message', 'port'))
  42         if not loop:
  43             loop = asyncio.get_event_loop()
  44         self.loop = loop
  45
  46     async def create_alarm(self, metric_name: str, ns_id: str, vdu_name: str, vnf_member_index: str, threshold: int,
  47                            statistic: str, operation: str):
  48         cor_id = random.randint(1, 10e7)
  49         msg = self._build_create_alarm_payload(cor_id,
  50                                                metric_name,
  51                                                ns_id,
  52                                                vdu_name,
  53                                                vnf_member_index,
  54                                                threshold,
  55                                                statistic,
  56                                                operation)
  57         log.debug("Sending create_alarm_request %s", msg)
  58         producer = AIOKafkaProducer(loop=self.loop,
  59                                     bootstrap_servers=self.kafka_server,
  60                                     key_serializer=str.encode,
  61                                     value_serializer=str.encode)
  62         await producer.start()
  63         try:
  64             await producer.send_and_wait("alarm_request", key="create_alarm_request", value=json.dumps(msg))
  65         finally:
  66             await producer.stop()
  67         log.debug("Waiting for create_alarm_response...")
  68         consumer = AIOKafkaConsumer(
  69             "alarm_response_" + str(cor_id),
  70             loop=self.loop,
  71             bootstrap_servers=self.kafka_server,
  72             key_deserializer=bytes.decode,
  73             value_deserializer=bytes.decode,
  74             auto_offset_reset='earliest')
  75         await consumer.start()
  76         alarm_uuid = None
  77         try:
  78             async for message in consumer:
  79                 try:
  80                     content = json.loads(message.value)
  81                 except JSONDecodeError:
  82                     content = yaml.safe_load(message.value)
  83                 log.debug("Received create_alarm_response %s", content)
  84                 if content['alarm_create_response']['correlation_id'] == cor_id:
  85                     if not content['alarm_create_response']['status']:
  86                         raise ValueError("Error creating alarm in MON")
  87                     alarm_uuid = content['alarm_create_response']['alarm_uuid']
  88                     break
  89         finally:
  90             await consumer.stop()
  91         if not alarm_uuid:
  92             raise ValueError('No alarm deletion response from MON. Is MON up?')
  93         return alarm_uuid
  94
  95     async def delete_alarm(self, ns_id: str, vnf_member_index: str, vdu_name: str, alarm_uuid: str):
  96         cor_id = random.randint(1, 10e7)
  97         msg = self._build_delete_alarm_payload(cor_id, ns_id, vdu_name, vnf_member_index, alarm_uuid)
  98         log.debug("Sending delete_alarm_request %s", msg)
  99         producer = AIOKafkaProducer(loop=self.loop,
 100                                     bootstrap_servers=self.kafka_server,
 101                                     key_serializer=str.encode,
 102                                     value_serializer=str.encode)
 103         await producer.start()
 104         try:
 105             await producer.send_and_wait("alarm_request", key="delete_alarm_request", value=json.dumps(msg))
 106         finally:
 107             await producer.stop()
 108         log.debug("Waiting for delete_alarm_response...")
 109         consumer = AIOKafkaConsumer(
 110             "alarm_response_" + str(cor_id),
 111             loop=self.loop,
 112             bootstrap_servers=self.kafka_server,
 113             key_deserializer=bytes.decode,
 114             value_deserializer=bytes.decode,
 115             auto_offset_reset='earliest')
 116         await consumer.start()
 117         alarm_uuid = None
 118         try:
 119             async for message in consumer:
 120                 try:
 121                     content = json.loads(message.value)
 122                 except JSONDecodeError:
 123                     content = yaml.safe_load(message.value)
 124                 if content['alarm_delete_response']['correlation_id'] == cor_id:
 125                     log.debug("Received delete_alarm_response %s", content)
 126                     if not content['alarm_delete_response']['status']:
 127                         raise ValueError("Error deleting alarm in MON. Response status is False.")
 128                     alarm_uuid = content['alarm_delete_response']['alarm_uuid']
 129                     break
 130         finally:
 131             await consumer.stop()
 132         if not alarm_uuid:
 133             raise ValueError('No alarm deletion response from MON. Is MON up?')
 134         return alarm_uuid
 135
 136     def _build_create_alarm_payload(self, cor_id: int,
 137                                     metric_name: str,
 138                                     ns_id: str,
 139                                     vdu_name: str,
 140                                     vnf_member_index: str,
 141                                     threshold: int,
 142                                     statistic: str,
 143                                     operation: str):
 144
 145         alarm_create_request = {
 146             'correlation_id': cor_id,
 147             'alarm_name': 'osm_alarm_{}_{}_{}_{}'.format(ns_id, vnf_member_index, vdu_name, metric_name),
 148             'metric_name': metric_name,
 149             'operation': operation,
 150             'severity': 'critical',
 151             'threshold_value': threshold,
 152             'statistic': statistic,
 153             'tags': {
 154                 'ns_id': ns_id,
 155                 'vdu_name': vdu_name,
 156                 'vnf_member_index': vnf_member_index,
 157             }
 158         }
 159         msg = {
 160             'alarm_create_request': alarm_create_request,
 161         }
 162         return msg
 163
 164     def _build_delete_alarm_payload(self, cor_id: int, ns_id: str, vdu_name: str,
 165                                     vnf_member_index: str, alarm_uuid: str):
 166         alarm_delete_request = {
 167             'correlation_id': cor_id,
 168             'alarm_uuid': alarm_uuid,
 169             'tags': {
 170                 'ns_id': ns_id,
 171                 'vdu_name': vdu_name,
 172                 'vnf_member_index': vnf_member_index
 173             }
 174         }
 175         msg = {
 176             'alarm_delete_request': alarm_delete_request,
 177         }
 178         return msg