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