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