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 json
  25 import logging
  26 import random
  27 from json import JSONDecodeError
  28
  29 import yaml
  30 from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
  31
  32 from osm_policy_module.core.config import Config
  33
  34 log = logging.getLogger(__name__)
  35
  36
  37 class MonClient:
  38     def __init__(self, config: Config):
  39         self.kafka_server = "{}:{}".format(
  40             config.get("message", "host"), config.get("message", "port")
  41         )
  42
  43     async def create_alarm(
  44         self,
  45         metric_name: str,
  46         ns_id: str,
  47         vdu_name: str,
  48         vnf_member_index: str,
  49         threshold: int,
  50         operation: str,
  51         statistic: str = "AVERAGE",
  52         action: str = "",
  53         vnfr: object = None,
  54         vnfd: object = None,
  55     ):
  56         cor_id = random.randint(1, 10e7)
  57         msg = self._build_create_alarm_payload(
  58             cor_id,
  59             metric_name,
  60             ns_id,
  61             vdu_name,
  62             vnf_member_index,
  63             threshold,
  64             statistic,
  65             operation,
  66             action,
  67             vnfr,
  68             vnfd,
  69         )
  70         log.debug("Sending create_alarm_request %s", msg)
  71         producer = AIOKafkaProducer(
  72             bootstrap_servers=self.kafka_server,
  73             key_serializer=str.encode,
  74             value_serializer=str.encode,
  75         )
  76         await producer.start()
  77         try:
  78             await producer.send_and_wait(
  79                 "alarm_request", key="create_alarm_request", value=json.dumps(msg)
  80             )
  81         finally:
  82             await producer.stop()
  83         log.debug("Waiting for create_alarm_response...")
  84         consumer = AIOKafkaConsumer(
  85             "alarm_response_" + str(cor_id),
  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             bootstrap_servers=self.kafka_server,
 121             key_serializer=str.encode,
 122             value_serializer=str.encode,
 123         )
 124         await producer.start()
 125         try:
 126             await producer.send_and_wait(
 127                 "alarm_request", key="delete_alarm_request", value=json.dumps(msg)
 128             )
 129         finally:
 130             await producer.stop()
 131         log.debug("Waiting for delete_alarm_response...")
 132         consumer = AIOKafkaConsumer(
 133             "alarm_response_" + str(cor_id),
 134             bootstrap_servers=self.kafka_server,
 135             key_deserializer=bytes.decode,
 136             value_deserializer=bytes.decode,
 137             auto_offset_reset="earliest",
 138         )
 139         await consumer.start()
 140         alarm_uuid = None
 141         try:
 142             async for message in consumer:
 143                 try:
 144                     content = json.loads(message.value)
 145                 except JSONDecodeError:
 146                     content = yaml.safe_load(message.value)
 147                 if content["alarm_delete_response"]["correlation_id"] == cor_id:
 148                     log.debug("Received delete_alarm_response %s", content)
 149                     if not content["alarm_delete_response"]["status"]:
 150                         raise ValueError(
 151                             "Error deleting alarm in MON. Response status is False."
 152                         )
 153                     alarm_uuid = content["alarm_delete_response"]["alarm_uuid"]
 154                     break
 155         finally:
 156             await consumer.stop()
 157         if not alarm_uuid:
 158             raise ValueError("No alarm deletion response from MON. Is MON up?")
 159         return alarm_uuid
 160
 161     def _build_create_alarm_payload(
 162         self,
 163         cor_id: int,
 164         metric_name: str,
 165         ns_id: str,
 166         vdu_name: str,
 167         vnf_member_index: str,
 168         threshold: int,
 169         statistic: str,
 170         operation: str,
 171         action: str,
 172         vnfr=None,
 173         vnfd=None,
 174     ):
 175         tags = {
 176             "ns_id": ns_id,
 177             "vdu_name": vdu_name,
 178             "vnf_member_index": vnf_member_index,
 179         }
 180         if vnfr and vnfd:
 181             # TODO: Change for multiple DF support
 182             df = vnfd.get("df", [{}])[0]
 183             if "exporters-endpoints" in df:
 184                 metric_port = df["exporters-endpoints"].get("metric-port", 9100)
 185             if metric_name.startswith("kpi_"):
 186                 metric_name = metric_name.replace("kpi_", "")
 187                 metric_name.strip()
 188                 for vdu in vnfr["vdur"]:
 189                     if vdu["name"] == vdu_name:
 190                         vdu_ip = vdu["ip-address"]
 191                         tags = {"instance": vdu_ip + ":" + str(metric_port)}
 192         alarm_create_request = {
 193             "correlation_id": cor_id,
 194             "alarm_name": "osm_alarm_{}_{}_{}_{}".format(
 195                 ns_id, vnf_member_index, vdu_name, metric_name
 196             ),
 197             "metric_name": metric_name,
 198             "operation": operation,
 199             "severity": "critical",
 200             "threshold_value": threshold,
 201             "statistic": statistic,
 202             "action": action,
 203             "tags": tags,
 204         }
 205         msg = {
 206             "alarm_create_request": alarm_create_request,
 207         }
 208         return msg
 209
 210     def _build_delete_alarm_payload(
 211         self,
 212         cor_id: int,
 213         ns_id: str,
 214         vdu_name: str,
 215         vnf_member_index: str,
 216         alarm_uuid: str,
 217     ):
 218         alarm_delete_request = {
 219             "correlation_id": cor_id,
 220             "alarm_uuid": alarm_uuid,
 221             "tags": {
 222                 "ns_id": ns_id,
 223                 "vdu_name": vdu_name,
 224                 "vnf_member_index": vnf_member_index,
 225             },
 226         }
 227         msg = {
 228             "alarm_delete_request": alarm_delete_request,
 229         }
 230         return msg