Refactors code and adds unit tests
[osm/MON.git] / osm_mon / server / server.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 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact: bdiaz@whitestack.com or glavado@whitestack.com
22 ##
23 """
24 MON component in charge of CRUD operations for vim_accounts and alarms. It uses the message bus to communicate.
25 """
26 import asyncio
27 import json
28 import logging
29
30 from osm_mon.core.config import Config
31 from osm_mon.core.message_bus_client import MessageBusClient
32 from osm_mon.core.response import ResponseBuilder
33 from osm_mon.server.service import ServerService
34
35 log = logging.getLogger(__name__)
36
37
38 class Server:
39
40 def __init__(self, config: Config, loop=None):
41 self.conf = config
42 if not loop:
43 loop = asyncio.get_event_loop()
44 self.loop = loop
45 self.msg_bus = MessageBusClient(config)
46 self.service = ServerService(config)
47
48 def run(self):
49 self.loop.run_until_complete(self.start())
50
51 async def start(self):
52 topics = [
53 "vim_account",
54 "alarm_request"
55 ]
56 await self.msg_bus.aioread(topics, self._process_msg)
57
58 async def _process_msg(self, topic, key, values):
59 log.info("Message arrived: %s", values)
60 try:
61 if topic == "vim_account":
62 if key == "create" or key == "edit":
63 if 'config' not in values:
64 values['config'] = {}
65 self.service.upsert_vim_account(values['_id'],
66 values['name'],
67 values['vim_type'],
68 values['vim_url'],
69 values['vim_user'],
70 values['vim_password'],
71 values['vim_tenant_name'],
72 values['schema_version'],
73 values['config'])
74
75 if key == "delete":
76 self.service.delete_vim_account(values['_id'])
77
78 elif topic == "alarm_request":
79 if key == "create_alarm_request":
80 alarm_details = values['alarm_create_request']
81 cor_id = alarm_details['correlation_id']
82 response_builder = ResponseBuilder()
83 try:
84 alarm = self.service.create_alarm(
85 alarm_details['alarm_name'],
86 alarm_details['threshold_value'],
87 alarm_details['operation'].lower(),
88 alarm_details['severity'].lower(),
89 alarm_details['statistic'].lower(),
90 alarm_details['metric_name'],
91 alarm_details['vdu_name'],
92 alarm_details['vnf_member_index'],
93 alarm_details['ns_id']
94 )
95 response = response_builder.generate_response('create_alarm_response',
96 cor_id=cor_id,
97 status=True,
98 alarm_id=alarm.uuid)
99 except Exception:
100 log.exception("Error creating alarm: ")
101 response = response_builder.generate_response('create_alarm_response',
102 cor_id=cor_id,
103 status=False,
104 alarm_id=None)
105 await self._publish_response('alarm_response_' + str(cor_id), 'create_alarm_response', response)
106
107 if key == "delete_alarm_request":
108 alarm_details = values['alarm_delete_request']
109 alarm_uuid = alarm_details['alarm_uuid']
110 response_builder = ResponseBuilder()
111 cor_id = alarm_details['correlation_id']
112 try:
113 self.service.delete_alarm(alarm_uuid)
114 response = response_builder.generate_response('delete_alarm_response',
115 cor_id=cor_id,
116 status=True,
117 alarm_id=alarm_uuid)
118 except Exception:
119 log.exception("Error deleting alarm: ")
120 response = response_builder.generate_response('delete_alarm_response',
121 cor_id=cor_id,
122 status=False,
123 alarm_id=alarm_uuid)
124 await self._publish_response('alarm_response_' + str(cor_id), 'delete_alarm_response', response)
125
126 except Exception:
127 log.exception("Exception processing message: ")
128
129 async def _publish_response(self, topic: str, key: str, msg: dict):
130 log.info("Sending response %s to topic %s with key %s", json.dumps(msg), topic, key)
131 await self.msg_bus.aiowrite(topic, key, msg)