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( |
41 |
|
config.get("message", "host"), config.get("message", "port") |
42 |
|
) |
43 |
1 |
if not loop: |
44 |
0 |
loop = asyncio.get_event_loop() |
45 |
1 |
self.loop = loop |
46 |
|
|
47 |
1 |
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 |
0 |
cor_id = random.randint(1, 10e7) |
59 |
0 |
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 |
0 |
log.debug("Sending create_alarm_request %s", msg) |
71 |
0 |
producer = AIOKafkaProducer( |
72 |
|
loop=self.loop, |
73 |
|
bootstrap_servers=self.kafka_server, |
74 |
|
key_serializer=str.encode, |
75 |
|
value_serializer=str.encode, |
76 |
|
) |
77 |
0 |
await producer.start() |
78 |
0 |
try: |
79 |
0 |
await producer.send_and_wait( |
80 |
|
"alarm_request", key="create_alarm_request", value=json.dumps(msg) |
81 |
|
) |
82 |
|
finally: |
83 |
0 |
await producer.stop() |
84 |
0 |
log.debug("Waiting for create_alarm_response...") |
85 |
0 |
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 |
0 |
await consumer.start() |
94 |
0 |
alarm_uuid = None |
95 |
0 |
try: |
96 |
0 |
async for message in consumer: |
97 |
0 |
try: |
98 |
0 |
content = json.loads(message.value) |
99 |
0 |
except JSONDecodeError: |
100 |
0 |
content = yaml.safe_load(message.value) |
101 |
0 |
log.debug("Received create_alarm_response %s", content) |
102 |
0 |
if content["alarm_create_response"]["correlation_id"] == cor_id: |
103 |
0 |
if not content["alarm_create_response"]["status"]: |
104 |
0 |
raise ValueError("Error creating alarm in MON") |
105 |
0 |
alarm_uuid = content["alarm_create_response"]["alarm_uuid"] |
106 |
0 |
break |
107 |
|
finally: |
108 |
0 |
await consumer.stop() |
109 |
0 |
if not alarm_uuid: |
110 |
0 |
raise ValueError("No alarm deletion response from MON. Is MON up?") |
111 |
0 |
return alarm_uuid |
112 |
|
|
113 |
1 |
async def delete_alarm( |
114 |
|
self, ns_id: str, vnf_member_index: str, vdu_name: str, alarm_uuid: str |
115 |
|
): |
116 |
0 |
cor_id = random.randint(1, 10e7) |
117 |
0 |
msg = self._build_delete_alarm_payload( |
118 |
|
cor_id, ns_id, vdu_name, vnf_member_index, alarm_uuid |
119 |
|
) |
120 |
0 |
log.debug("Sending delete_alarm_request %s", msg) |
121 |
0 |
producer = AIOKafkaProducer( |
122 |
|
loop=self.loop, |
123 |
|
bootstrap_servers=self.kafka_server, |
124 |
|
key_serializer=str.encode, |
125 |
|
value_serializer=str.encode, |
126 |
|
) |
127 |
0 |
await producer.start() |
128 |
0 |
try: |
129 |
0 |
await producer.send_and_wait( |
130 |
|
"alarm_request", key="delete_alarm_request", value=json.dumps(msg) |
131 |
|
) |
132 |
|
finally: |
133 |
0 |
await producer.stop() |
134 |
0 |
log.debug("Waiting for delete_alarm_response...") |
135 |
0 |
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 |
0 |
await consumer.start() |
144 |
0 |
alarm_uuid = None |
145 |
0 |
try: |
146 |
0 |
async for message in consumer: |
147 |
0 |
try: |
148 |
0 |
content = json.loads(message.value) |
149 |
0 |
except JSONDecodeError: |
150 |
0 |
content = yaml.safe_load(message.value) |
151 |
0 |
if content["alarm_delete_response"]["correlation_id"] == cor_id: |
152 |
0 |
log.debug("Received delete_alarm_response %s", content) |
153 |
0 |
if not content["alarm_delete_response"]["status"]: |
154 |
0 |
raise ValueError( |
155 |
|
"Error deleting alarm in MON. Response status is False." |
156 |
|
) |
157 |
0 |
alarm_uuid = content["alarm_delete_response"]["alarm_uuid"] |
158 |
0 |
break |
159 |
|
finally: |
160 |
0 |
await consumer.stop() |
161 |
0 |
if not alarm_uuid: |
162 |
0 |
raise ValueError("No alarm deletion response from MON. Is MON up?") |
163 |
0 |
return alarm_uuid |
164 |
|
|
165 |
1 |
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 |
|
|
178 |
0 |
alarm_create_request = { |
179 |
|
"correlation_id": cor_id, |
180 |
|
"alarm_name": "osm_alarm_{}_{}_{}_{}".format( |
181 |
|
ns_id, vnf_member_index, vdu_name, metric_name |
182 |
|
), |
183 |
|
"metric_name": metric_name, |
184 |
|
"operation": operation, |
185 |
|
"severity": "critical", |
186 |
|
"threshold_value": threshold, |
187 |
|
"statistic": statistic, |
188 |
|
"action": action, |
189 |
|
"tags": { |
190 |
|
"ns_id": ns_id, |
191 |
|
"vdu_name": vdu_name, |
192 |
|
"vnf_member_index": vnf_member_index, |
193 |
|
}, |
194 |
|
} |
195 |
0 |
msg = { |
196 |
|
"alarm_create_request": alarm_create_request, |
197 |
|
} |
198 |
0 |
return msg |
199 |
|
|
200 |
1 |
def _build_delete_alarm_payload( |
201 |
|
self, |
202 |
|
cor_id: int, |
203 |
|
ns_id: str, |
204 |
|
vdu_name: str, |
205 |
|
vnf_member_index: str, |
206 |
|
alarm_uuid: str, |
207 |
|
): |
208 |
0 |
alarm_delete_request = { |
209 |
|
"correlation_id": cor_id, |
210 |
|
"alarm_uuid": alarm_uuid, |
211 |
|
"tags": { |
212 |
|
"ns_id": ns_id, |
213 |
|
"vdu_name": vdu_name, |
214 |
|
"vnf_member_index": vnf_member_index, |
215 |
|
}, |
216 |
|
} |
217 |
0 |
msg = { |
218 |
|
"alarm_delete_request": alarm_delete_request, |
219 |
|
} |
220 |
0 |
return msg |