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