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 json |
25 |
1 |
import logging |
26 |
1 |
import random |
27 |
1 |
from json import JSONDecodeError |
28 |
|
|
29 |
1 |
import yaml |
30 |
1 |
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer |
31 |
|
|
32 |
1 |
from osm_policy_module.core.config import Config |
33 |
|
|
34 |
1 |
log = logging.getLogger(__name__) |
35 |
|
|
36 |
|
|
37 |
1 |
class MonClient: |
38 |
1 |
def __init__(self, config: Config): |
39 |
1 |
self.kafka_server = "{}:{}".format( |
40 |
|
config.get("message", "host"), config.get("message", "port") |
41 |
|
) |
42 |
|
|
43 |
1 |
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 |
0 |
cor_id = random.randint(1, 10e7) |
57 |
0 |
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 |
0 |
log.debug("Sending create_alarm_request %s", msg) |
71 |
0 |
producer = AIOKafkaProducer( |
72 |
|
bootstrap_servers=self.kafka_server, |
73 |
|
key_serializer=str.encode, |
74 |
|
value_serializer=str.encode, |
75 |
|
) |
76 |
0 |
await producer.start() |
77 |
0 |
try: |
78 |
0 |
await producer.send_and_wait( |
79 |
|
"alarm_request", key="create_alarm_request", value=json.dumps(msg) |
80 |
|
) |
81 |
|
finally: |
82 |
0 |
await producer.stop() |
83 |
0 |
log.debug("Waiting for create_alarm_response...") |
84 |
0 |
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 |
0 |
await consumer.start() |
92 |
0 |
alarm_uuid = None |
93 |
0 |
try: |
94 |
0 |
async for message in consumer: |
95 |
0 |
try: |
96 |
0 |
content = json.loads(message.value) |
97 |
0 |
except JSONDecodeError: |
98 |
0 |
content = yaml.safe_load(message.value) |
99 |
0 |
log.debug("Received create_alarm_response %s", content) |
100 |
0 |
if content["alarm_create_response"]["correlation_id"] == cor_id: |
101 |
0 |
if not content["alarm_create_response"]["status"]: |
102 |
0 |
raise ValueError("Error creating alarm in MON") |
103 |
0 |
alarm_uuid = content["alarm_create_response"]["alarm_uuid"] |
104 |
0 |
break |
105 |
|
finally: |
106 |
0 |
await consumer.stop() |
107 |
0 |
if not alarm_uuid: |
108 |
0 |
raise ValueError("No alarm deletion response from MON. Is MON up?") |
109 |
0 |
return alarm_uuid |
110 |
|
|
111 |
1 |
async def delete_alarm( |
112 |
|
self, ns_id: str, vnf_member_index: str, vdu_name: str, alarm_uuid: str |
113 |
|
): |
114 |
0 |
cor_id = random.randint(1, 10e7) |
115 |
0 |
msg = self._build_delete_alarm_payload( |
116 |
|
cor_id, ns_id, vdu_name, vnf_member_index, alarm_uuid |
117 |
|
) |
118 |
0 |
log.debug("Sending delete_alarm_request %s", msg) |
119 |
0 |
producer = AIOKafkaProducer( |
120 |
|
bootstrap_servers=self.kafka_server, |
121 |
|
key_serializer=str.encode, |
122 |
|
value_serializer=str.encode, |
123 |
|
) |
124 |
0 |
await producer.start() |
125 |
0 |
try: |
126 |
0 |
await producer.send_and_wait( |
127 |
|
"alarm_request", key="delete_alarm_request", value=json.dumps(msg) |
128 |
|
) |
129 |
|
finally: |
130 |
0 |
await producer.stop() |
131 |
0 |
log.debug("Waiting for delete_alarm_response...") |
132 |
0 |
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 |
0 |
await consumer.start() |
140 |
0 |
alarm_uuid = None |
141 |
0 |
try: |
142 |
0 |
async for message in consumer: |
143 |
0 |
try: |
144 |
0 |
content = json.loads(message.value) |
145 |
0 |
except JSONDecodeError: |
146 |
0 |
content = yaml.safe_load(message.value) |
147 |
0 |
if content["alarm_delete_response"]["correlation_id"] == cor_id: |
148 |
0 |
log.debug("Received delete_alarm_response %s", content) |
149 |
0 |
if not content["alarm_delete_response"]["status"]: |
150 |
0 |
raise ValueError( |
151 |
|
"Error deleting alarm in MON. Response status is False." |
152 |
|
) |
153 |
0 |
alarm_uuid = content["alarm_delete_response"]["alarm_uuid"] |
154 |
0 |
break |
155 |
|
finally: |
156 |
0 |
await consumer.stop() |
157 |
0 |
if not alarm_uuid: |
158 |
0 |
raise ValueError("No alarm deletion response from MON. Is MON up?") |
159 |
0 |
return alarm_uuid |
160 |
|
|
161 |
1 |
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 |
0 |
tags = { |
176 |
|
"ns_id": ns_id, |
177 |
|
"vdu_name": vdu_name, |
178 |
|
"vnf_member_index": vnf_member_index, |
179 |
|
} |
180 |
0 |
if vnfr and vnfd: |
181 |
|
# TODO: Change for multiple DF support |
182 |
0 |
df = vnfd.get("df", [{}])[0] |
183 |
0 |
metric_port = 9100 |
184 |
0 |
if "exporters-endpoints" in df: |
185 |
0 |
metric_port = df["exporters-endpoints"].get("metric-port", 9100) |
186 |
0 |
if metric_name.startswith("kpi_"): |
187 |
0 |
metric_name = metric_name.replace("kpi_", "") |
188 |
0 |
metric_name.strip() |
189 |
0 |
for vdu in vnfr["vdur"]: |
190 |
0 |
if vdu["name"] == vdu_name: |
191 |
0 |
vdu_ip = vdu["ip-address"] |
192 |
0 |
tags = {"instance": vdu_ip + ":" + str(metric_port)} |
193 |
0 |
alarm_create_request = { |
194 |
|
"correlation_id": cor_id, |
195 |
|
"alarm_name": "osm_alarm_{}_{}_{}_{}".format( |
196 |
|
ns_id, vnf_member_index, vdu_name, metric_name |
197 |
|
), |
198 |
|
"metric_name": metric_name, |
199 |
|
"operation": operation, |
200 |
|
"severity": "critical", |
201 |
|
"threshold_value": threshold, |
202 |
|
"statistic": statistic, |
203 |
|
"action": action, |
204 |
|
"tags": tags, |
205 |
|
} |
206 |
0 |
msg = { |
207 |
|
"alarm_create_request": alarm_create_request, |
208 |
|
} |
209 |
0 |
return msg |
210 |
|
|
211 |
1 |
def _build_delete_alarm_payload( |
212 |
|
self, |
213 |
|
cor_id: int, |
214 |
|
ns_id: str, |
215 |
|
vdu_name: str, |
216 |
|
vnf_member_index: str, |
217 |
|
alarm_uuid: str, |
218 |
|
): |
219 |
0 |
alarm_delete_request = { |
220 |
|
"correlation_id": cor_id, |
221 |
|
"alarm_uuid": alarm_uuid, |
222 |
|
"tags": { |
223 |
|
"ns_id": ns_id, |
224 |
|
"vdu_name": vdu_name, |
225 |
|
"vnf_member_index": vnf_member_index, |
226 |
|
}, |
227 |
|
} |
228 |
0 |
msg = { |
229 |
|
"alarm_delete_request": alarm_delete_request, |
230 |
|
} |
231 |
0 |
return msg |