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(
  41             config.get("message", "host"), config.get("message", "port")
  42         )
  43         if not loop:
  44             loop = asyncio.get_event_loop()
  45         self.loop = loop
  46
  47     async def create_alarm(
  48         self,
  49         metric_name: str,
  50         ns_id: str,
  51         vdu_name: str,
  52         vnf_member_index: str,
  53         threshold: int,
  54         operation: str,
  55         statistic: str = "AVERAGE",
  56     ):
  57         cor_id = random.randint(1, 10e7)
  58         msg = self._build_create_alarm_payload(
  59             cor_id,
  60             metric_name,
  61             ns_id,
  62             vdu_name,
  63             vnf_member_index,
  64             threshold,
  65             statistic,
  66             operation,
  67         )
  68         log.debug("Sending create_alarm_request %s", msg)
  69         producer = AIOKafkaProducer(
  70             loop=self.loop,
  71             bootstrap_servers=self.kafka_server,
  72             key_serializer=str.encode,
  73             value_serializer=str.encode,
  74         )
  75         await producer.start()
  76         try:
  77             await producer.send_and_wait(
  78                 "alarm_request", key="create_alarm_request", value=json.dumps(msg)
  79             )
  80         finally:
  81             await producer.stop()
  82         log.debug("Waiting for create_alarm_response...")
  83         consumer = AIOKafkaConsumer(
  84             "alarm_response_" + str(cor_id),
  85             loop=self.loop,
  86             bootstrap_servers=self.kafka_server,
  87             key_deserializer=bytes.decode,
  88             value_deserializer=bytes.decode,
  89             auto_offset_reset="earliest",
  90         )
  91         await consumer.start()
  92         alarm_uuid = None
  93         try:
  94             async for message in consumer:
  95                 try:
  96                     content = json.loads(message.value)
  97                 except JSONDecodeError:
  98                     content = yaml.safe_load(message.value)
  99                 log.debug("Received create_alarm_response %s", content)
 100                 if content["alarm_create_response"]["correlation_id"] == cor_id:
 101                     if not content["alarm_create_response"]["status"]:
 102                         raise ValueError("Error creating alarm in MON")
 103                     alarm_uuid = content["alarm_create_response"]["alarm_uuid"]
 104                     break
 105         finally:
 106             await consumer.stop()
 107         if not alarm_uuid:
 108             raise ValueError("No alarm deletion response from MON. Is MON up?")
 109         return alarm_uuid
 110
 111     async def delete_alarm(
 112         self, ns_id: str, vnf_member_index: str, vdu_name: str, alarm_uuid: str
 113     ):
 114         cor_id = random.randint(1, 10e7)
 115         msg = self._build_delete_alarm_payload(
 116             cor_id, ns_id, vdu_name, vnf_member_index, alarm_uuid
 117         )
 118         log.debug("Sending delete_alarm_request %s", msg)
 119         producer = AIOKafkaProducer(
 120             loop=self.loop,
 121             bootstrap_servers=self.kafka_server,
 122             key_serializer=str.encode,
 123             value_serializer=str.encode,
 124         )
 125         await producer.start()
 126         try:
 127             await producer.send_and_wait(
 128                 "alarm_request", key="delete_alarm_request", value=json.dumps(msg)
 129             )
 130         finally:
 131             await producer.stop()
 132         log.debug("Waiting for delete_alarm_response...")
 133         consumer = AIOKafkaConsumer(
 134             "alarm_response_" + str(cor_id),
 135             loop=self.loop,
 136             bootstrap_servers=self.kafka_server,
 137             key_deserializer=bytes.decode,
 138             value_deserializer=bytes.decode,
 139             auto_offset_reset="earliest",
 140         )
 141         await consumer.start()
 142         alarm_uuid = None
 143         try:
 144             async for message in consumer:
 145                 try:
 146                     content = json.loads(message.value)
 147                 except JSONDecodeError:
 148                     content = yaml.safe_load(message.value)
 149                 if content["alarm_delete_response"]["correlation_id"] == cor_id:
 150                     log.debug("Received delete_alarm_response %s", content)
 151                     if not content["alarm_delete_response"]["status"]:
 152                         raise ValueError(
 153                             "Error deleting alarm in MON. Response status is False."
 154                         )
 155                     alarm_uuid = content["alarm_delete_response"]["alarm_uuid"]
 156                     break
 157         finally:
 158             await consumer.stop()
 159         if not alarm_uuid:
 160             raise ValueError("No alarm deletion response from MON. Is MON up?")
 161         return alarm_uuid
 162
 163     def _build_create_alarm_payload(
 164         self,
 165         cor_id: int,
 166         metric_name: str,
 167         ns_id: str,
 168         vdu_name: str,
 169         vnf_member_index: str,
 170         threshold: int,
 171         statistic: str,
 172         operation: str,
 173     ):
 174
 175         alarm_create_request = {
 176             "correlation_id": cor_id,
 177             "alarm_name": "osm_alarm_{}_{}_{}_{}".format(
 178                 ns_id, vnf_member_index, vdu_name, metric_name
 179             ),
 180             "metric_name": metric_name,
 181             "operation": operation,
 182             "severity": "critical",
 183             "threshold_value": threshold,
 184             "statistic": statistic,
 185             "tags": {
 186                 "ns_id": ns_id,
 187                 "vdu_name": vdu_name,
 188                 "vnf_member_index": vnf_member_index,
 189             },
 190         }
 191         msg = {
 192             "alarm_create_request": alarm_create_request,
 193         }
 194         return msg
 195
 196     def _build_delete_alarm_payload(
 197         self,
 198         cor_id: int,
 199         ns_id: str,
 200         vdu_name: str,
 201         vnf_member_index: str,
 202         alarm_uuid: str,
 203     ):
 204         alarm_delete_request = {
 205             "correlation_id": cor_id,
 206             "alarm_uuid": alarm_uuid,
 207             "tags": {
 208                 "ns_id": ns_id,
 209                 "vdu_name": vdu_name,
 210                 "vnf_member_index": vnf_member_index,
 211             },
 212         }
 213         msg = {
 214             "alarm_delete_request": alarm_delete_request,
 215         }
 216         return msg