Merge "RBAC permission storage in MongoDB"
[osm/NBI.git] / osm_nbi / subscriptions.py
1 # -*- coding: utf-8 -*-
2
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12 # implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15
16 """
17 This module implements a thread that reads from kafka bus implementing all the subscriptions.
18 It is based on asyncio.
19 To avoid race conditions it uses same engine class as the main module for database changes
20 For the moment this module only deletes NS instances when they are terminated with the autoremove flag
21 """
22
23 import logging
24 import threading
25 import asyncio
26 from http import HTTPStatus
27 from osm_common import dbmongo, dbmemory, msglocal, msgkafka
28 from osm_common.dbbase import DbException
29 from osm_common.msgbase import MsgException
30 from engine import EngineException
31
32 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
33
34
35 class SubscriptionException(Exception):
36
37 def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
38 self.http_code = http_code
39 Exception.__init__(self, message)
40
41
42 class SubscriptionThread(threading.Thread):
43
44 def __init__(self, config, engine):
45 """
46 Constructor of class
47 :param config: configuration parameters of database and messaging
48 :param engine: an instance of Engine class, used for deleting instances
49 """
50 threading.Thread.__init__(self)
51
52 self.config = config
53 self.db = None
54 self.msg = None
55 self.engine = engine
56 self.loop = None
57 self.logger = logging.getLogger("nbi.subscriptions")
58 self.aiomain_task = None # asyncio task for receiving kafka bus
59 self.internal_session = { # used for a session to the engine methods
60 "_id": "subscription",
61 "id": "subscription",
62 "project_id": "admin",
63 "admin": True
64 }
65
66 def run(self):
67 """
68 Start of the thread
69 :return: None
70 """
71 self.loop = asyncio.new_event_loop()
72 try:
73 if not self.db:
74 if self.config["database"]["driver"] == "mongo":
75 self.db = dbmongo.DbMongo()
76 self.db.db_connect(self.config["database"])
77 elif self.config["database"]["driver"] == "memory":
78 self.db = dbmemory.DbMemory()
79 self.db.db_connect(self.config["database"])
80 else:
81 raise SubscriptionException("Invalid configuration param '{}' at '[database]':'driver'".format(
82 self.config["database"]["driver"]))
83 if not self.msg:
84 config_msg = self.config["message"].copy()
85 config_msg["loop"] = self.loop
86 if config_msg["driver"] == "local":
87 self.msg = msglocal.MsgLocal()
88 self.msg.connect(config_msg)
89 elif config_msg["driver"] == "kafka":
90 self.msg = msgkafka.MsgKafka()
91 self.msg.connect(config_msg)
92 else:
93 raise SubscriptionException("Invalid configuration param '{}' at '[message]':'driver'".format(
94 config_msg["driver"]))
95
96 except (DbException, MsgException) as e:
97 raise SubscriptionException(str(e), http_code=e.http_code)
98
99 self.logger.debug("Starting")
100 while True:
101 try:
102 self.aiomain_task = asyncio.ensure_future(self.msg.aioread("ns", loop=self.loop,
103 callback=self._msg_callback),
104 loop=self.loop)
105 self.loop.run_until_complete(self.aiomain_task)
106 except asyncio.CancelledError:
107 break # if cancelled it should end, breaking loop
108 except Exception as e:
109 self.logger.exception("Exception '{}' at messaging read loop".format(e), exc_info=True)
110
111 self.logger.debug("Finishing")
112 self._stop()
113 self.loop.close()
114
115 def _msg_callback(self, topic, command, params):
116 """
117 Callback to process a received message from kafka
118 :param topic: topic received
119 :param command: command received
120 :param params: rest of parameters
121 :return: None
122 """
123 try:
124 if topic == "ns":
125 if command == "terminated":
126 self.logger.debug("received ns terminated {}".format(params))
127 if params.get("autoremove"):
128 self.engine.del_item(self.internal_session, "nsrs", _id=params["nsr_id"])
129 self.logger.debug("ns={} deleted from database".format(params["nsr_id"]))
130 return
131 except (EngineException, DbException, MsgException) as e:
132 self.logger.error("Error while processing topic={} command={}: {}".format(topic, command, e))
133 except Exception as e:
134 self.logger.exception("Exception while processing topic={} command={}: {}".format(topic, command, e),
135 exc_info=True)
136
137 def _stop(self):
138 """
139 Close all connections
140 :return: None
141 """
142 try:
143 if self.db:
144 self.db.db_disconnect()
145 if self.msg:
146 self.msg.disconnect()
147 except (DbException, MsgException) as e:
148 raise SubscriptionException(str(e), http_code=e.http_code)
149
150 def terminate(self):
151 """
152 This is a threading safe method to terminate this thread. Termination is done asynchronous afterwards,
153 but not immediately.
154 :return: None
155 """
156 self.loop.call_soon_threadsafe(self.aiomain_task.cancel)