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