f7bc0121b15d93f3104d3fe05531f09738f611a1
[osm/NBI.git] / osm_nbi / engine.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 import logging
17 import yaml
18 from osm_common import dbmongo, dbmemory, fslocal, msglocal, msgkafka, version as common_version
19 from osm_common.dbbase import DbException
20 from osm_common.fsbase import FsException
21 from osm_common.msgbase import MsgException
22 from http import HTTPStatus
23
24 from authconn_keystone import AuthconnKeystone
25 from base_topic import EngineException, versiontuple
26 from admin_topics import UserTopic, ProjectTopic, VimAccountTopic, WimAccountTopic, SdnTopic
27 from admin_topics import UserTopicAuth, ProjectTopicAuth, RoleTopicAuth
28 from descriptor_topics import VnfdTopic, NsdTopic, PduTopic, NstTopic
29 from instance_topics import NsrTopic, VnfrTopic, NsLcmOpTopic, NsiTopic, NsiLcmOpTopic
30 from pmjobs_topics import PmJobsTopic
31 from base64 import b64encode
32 from os import urandom, path
33 from threading import Lock
34
35 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
36 min_common_version = "0.1.16"
37
38
39 class Engine(object):
40 map_from_topic_to_class = {
41 "vnfds": VnfdTopic,
42 "nsds": NsdTopic,
43 "nsts": NstTopic,
44 "pdus": PduTopic,
45 "nsrs": NsrTopic,
46 "vnfrs": VnfrTopic,
47 "nslcmops": NsLcmOpTopic,
48 "vim_accounts": VimAccountTopic,
49 "wim_accounts": WimAccountTopic,
50 "sdns": SdnTopic,
51 "users": UserTopic,
52 "projects": ProjectTopic,
53 "nsis": NsiTopic,
54 "nsilcmops": NsiLcmOpTopic
55 # [NEW_TOPIC]: add an entry here
56 # "pm_jobs": PmJobsTopic will be added manually because it needs other parameters
57 }
58
59 map_target_version_to_int = {
60 "1.0": 1000,
61 "1.1": 1001
62 # Add new versions here
63 }
64
65 def __init__(self):
66 self.db = None
67 self.fs = None
68 self.msg = None
69 self.auth = None
70 self.config = None
71 self.operations = None
72 self.logger = logging.getLogger("nbi.engine")
73 self.map_topic = {}
74 self.write_lock = None
75
76 def start(self, config):
77 """
78 Connect to database, filesystem storage, and messaging
79 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
80 :return: None
81 """
82 self.config = config
83 # check right version of common
84 if versiontuple(common_version) < versiontuple(min_common_version):
85 raise EngineException("Not compatible osm/common version '{}'. Needed '{}' or higher".format(
86 common_version, min_common_version))
87
88 try:
89 if not self.db:
90 if config["database"]["driver"] == "mongo":
91 self.db = dbmongo.DbMongo()
92 self.db.db_connect(config["database"])
93 elif config["database"]["driver"] == "memory":
94 self.db = dbmemory.DbMemory()
95 self.db.db_connect(config["database"])
96 else:
97 raise EngineException("Invalid configuration param '{}' at '[database]':'driver'".format(
98 config["database"]["driver"]))
99 if not self.fs:
100 if config["storage"]["driver"] == "local":
101 self.fs = fslocal.FsLocal()
102 self.fs.fs_connect(config["storage"])
103 else:
104 raise EngineException("Invalid configuration param '{}' at '[storage]':'driver'".format(
105 config["storage"]["driver"]))
106 if not self.msg:
107 if config["message"]["driver"] == "local":
108 self.msg = msglocal.MsgLocal()
109 self.msg.connect(config["message"])
110 elif config["message"]["driver"] == "kafka":
111 self.msg = msgkafka.MsgKafka()
112 self.msg.connect(config["message"])
113 else:
114 raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format(
115 config["message"]["driver"]))
116 if not self.auth:
117 if config["authentication"]["backend"] == "keystone":
118 self.auth = AuthconnKeystone(config["authentication"])
119 if not self.operations:
120 if "resources_to_operations" in config["rbac"]:
121 resources_to_operations_file = config["rbac"]["resources_to_operations"]
122 else:
123 possible_paths = (
124 __file__[:__file__.rfind("engine.py")] + "resources_to_operations.yml",
125 "./resources_to_operations.yml"
126 )
127 for config_file in possible_paths:
128 if path.isfile(config_file):
129 resources_to_operations_file = config_file
130 break
131 if not resources_to_operations_file:
132 raise EngineException("Invalid permission configuration: resources_to_operations file missing")
133
134 with open(resources_to_operations_file, 'r') as f:
135 resources_to_operations = yaml.load(f)
136
137 self.operations = []
138
139 for _, value in resources_to_operations["resources_to_operations"].items():
140 if value not in self.operations:
141 self.operations += [value]
142
143 if config["authentication"]["backend"] == "keystone":
144 self.map_from_topic_to_class["users"] = UserTopicAuth
145 self.map_from_topic_to_class["projects"] = ProjectTopicAuth
146 self.map_from_topic_to_class["roles"] = RoleTopicAuth
147
148 self.write_lock = Lock()
149 # create one class per topic
150 for topic, topic_class in self.map_from_topic_to_class.items():
151 if self.auth and topic_class in (UserTopicAuth, ProjectTopicAuth):
152 self.map_topic[topic] = topic_class(self.db, self.fs, self.msg, self.auth)
153 elif self.auth and topic_class == RoleTopicAuth:
154 self.map_topic[topic] = topic_class(self.db, self.fs, self.msg, self.auth,
155 self.operations)
156 else:
157 self.map_topic[topic] = topic_class(self.db, self.fs, self.msg)
158
159 self.map_topic["pm_jobs"] = PmJobsTopic(config["prometheus"].get("host"), config["prometheus"].get("port"))
160 except (DbException, FsException, MsgException) as e:
161 raise EngineException(str(e), http_code=e.http_code)
162
163 def stop(self):
164 try:
165 if self.db:
166 self.db.db_disconnect()
167 if self.fs:
168 self.fs.fs_disconnect()
169 if self.msg:
170 self.msg.disconnect()
171 self.write_lock = None
172 except (DbException, FsException, MsgException) as e:
173 raise EngineException(str(e), http_code=e.http_code)
174
175 def new_item(self, rollback, session, topic, indata=None, kwargs=None, headers=None):
176 """
177 Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry,
178 that must be completed with a call to method upload_content
179 :param rollback: list to append created items at database in case a rollback must to be done
180 :param session: contains the used login username and working project, force to avoid checkins, public
181 :param topic: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds
182 :param indata: data to be inserted
183 :param kwargs: used to override the indata descriptor
184 :param headers: http request headers
185 :return: _id: identity of the inserted data.
186 """
187 if topic not in self.map_topic:
188 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
189 with self.write_lock:
190 return self.map_topic[topic].new(rollback, session, indata, kwargs, headers)
191
192 def upload_content(self, session, topic, _id, indata, kwargs, headers):
193 """
194 Upload content for an already created entry (_id)
195 :param session: contains the used login username and working project
196 :param topic: it can be: users, projects, vnfds, nsds,
197 :param _id: server id of the item
198 :param indata: data to be inserted
199 :param kwargs: used to override the indata descriptor
200 :param headers: http request headers
201 :return: _id: identity of the inserted data.
202 """
203 if topic not in self.map_topic:
204 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
205 with self.write_lock:
206 return self.map_topic[topic].upload_content(session, _id, indata, kwargs, headers)
207
208 def get_item_list(self, session, topic, filter_q=None):
209 """
210 Get a list of items
211 :param session: contains the used login username and working project
212 :param topic: it can be: users, projects, vnfds, nsds, ...
213 :param filter_q: filter of data to be applied
214 :return: The list, it can be empty if no one match the filter_q.
215 """
216 if topic not in self.map_topic:
217 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
218 return self.map_topic[topic].list(session, filter_q)
219
220 def get_item(self, session, topic, _id):
221 """
222 Get complete information on an item
223 :param session: contains the used login username and working project
224 :param topic: it can be: users, projects, vnfds, nsds,
225 :param _id: server id of the item
226 :return: dictionary, raise exception if not found.
227 """
228 if topic not in self.map_topic:
229 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
230 return self.map_topic[topic].show(session, _id)
231
232 def get_file(self, session, topic, _id, path=None, accept_header=None):
233 """
234 Get descriptor package or artifact file content
235 :param session: contains the used login username and working project
236 :param topic: it can be: users, projects, vnfds, nsds,
237 :param _id: server id of the item
238 :param path: artifact path or "$DESCRIPTOR" or None
239 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
240 :return: opened file plus Accept format or raises an exception
241 """
242 if topic not in self.map_topic:
243 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
244 return self.map_topic[topic].get_file(session, _id, path, accept_header)
245
246 def del_item_list(self, session, topic, _filter=None):
247 """
248 Delete a list of items
249 :param session: contains the used login username and working project
250 :param topic: it can be: users, projects, vnfds, nsds, ...
251 :param _filter: filter of data to be applied
252 :return: The deleted list, it can be empty if no one match the _filter.
253 """
254 if topic not in self.map_topic:
255 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
256 with self.write_lock:
257 return self.map_topic[topic].delete_list(session, _filter)
258
259 def del_item(self, session, topic, _id):
260 """
261 Delete item by its internal id
262 :param session: contains the used login username and working project
263 :param topic: it can be: users, projects, vnfds, nsds, ...
264 :param _id: server id of the item
265 :return: dictionary with deleted item _id. It raises exception if not found.
266 """
267 if topic not in self.map_topic:
268 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
269 with self.write_lock:
270 return self.map_topic[topic].delete(session, _id)
271
272 def edit_item(self, session, topic, _id, indata=None, kwargs=None):
273 """
274 Update an existing entry at database
275 :param session: contains the used login username and working project
276 :param topic: it can be: users, projects, vnfds, nsds, ...
277 :param _id: identifier to be updated
278 :param indata: data to be inserted
279 :param kwargs: used to override the indata descriptor
280 :return: dictionary, raise exception if not found.
281 """
282 if topic not in self.map_topic:
283 raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR)
284 with self.write_lock:
285 return self.map_topic[topic].edit(session, _id, indata, kwargs)
286
287 def create_admin_project(self):
288 """
289 Creates a new project 'admin' into database if database is empty. Useful for initialization.
290 :return: _id identity of the inserted data, or None
291 """
292
293 projects = self.db.get_one("projects", fail_on_empty=False, fail_on_more=False)
294 if projects:
295 return None
296 project_desc = {"name": "admin"}
297 fake_session = {"project_id": "admin", "username": "admin", "admin": True, "force": True, "public": None}
298 rollback_list = []
299 _id = self.map_topic["projects"].new(rollback_list, fake_session, project_desc)
300 return _id
301
302 def create_admin_user(self):
303 """
304 Creates a new user admin/admin into database if database is empty. Useful for initialization
305 :return: _id identity of the inserted data, or None
306 """
307 users = self.db.get_one("users", fail_on_empty=False, fail_on_more=False)
308 if users:
309 return None
310 # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED)
311 user_desc = {"username": "admin", "password": "admin", "projects": ["admin"]}
312 fake_session = {"project_id": "admin", "username": "admin", "admin": True, "force": True, "public": None}
313 roolback_list = []
314 _id = self.map_topic["users"].new(roolback_list, fake_session, user_desc)
315 return _id
316
317 def create_admin(self):
318 """
319 Creates new 'admin' user and project into database if database is empty. Useful for initialization.
320 :return: _id identity of the inserted data, or None
321 """
322 project_id = self.create_admin_project()
323 user_id = self.create_admin_user()
324 if not project_id and not user_id:
325 return None
326 else:
327 return {'project_id': project_id, 'user_id': user_id}
328
329 def upgrade_db(self, current_version, target_version):
330 if target_version not in self.map_target_version_to_int.keys():
331 raise EngineException("Wrong database version '{}'. Expected '{}'"
332 ". It cannot be up/down-grade".format(current_version, target_version),
333 http_code=HTTPStatus.INTERNAL_SERVER_ERROR)
334
335 if current_version == target_version:
336 return
337
338 target_version_int = self.map_target_version_to_int[target_version]
339
340 if not current_version:
341 # create database version
342 serial = urandom(32)
343 version_data = {
344 "_id": "version", # Always "version"
345 "version_int": 1000, # version number
346 "version": "1.0", # version text
347 "date": "2018-10-25", # version date
348 "description": "added serial", # changes in this version
349 'status': "ENABLED", # ENABLED, DISABLED (migration in process), ERROR,
350 'serial': b64encode(serial)
351 }
352 self.db.create("admin", version_data)
353 self.db.set_secret_key(serial)
354 current_version = "1.0"
355
356 if current_version == "1.0" and target_version_int >= self.map_target_version_to_int["1.1"]:
357 self.db.del_list("roles_operations")
358
359 version_data = {
360 "_id": "version",
361 "version_int": 1001,
362 "version": "1.1",
363 "date": "2019-05-24",
364 "description": "set new format for roles_operations"
365 }
366
367 self.db.set_one("admin", {"_id": "version"}, version_data)
368 current_version = "1.1"
369 # TODO add future migrations here
370
371 def init_db(self, target_version='1.0'):
372 """
373 Init database if empty. If not empty it checks that database version and migrates if needed
374 If empty, it creates a new user admin/admin at 'users' and a new entry at 'version'
375 :param target_version: check desired database version. Migrate to it if possible or raises exception
376 :return: None if ok, exception if error or if the version is different.
377 """
378
379 version_data = self.db.get_one("admin", {"_id": "version"}, fail_on_empty=False, fail_on_more=True)
380 # check database status is ok
381 if version_data and version_data.get("status") != 'ENABLED':
382 raise EngineException("Wrong database status '{}'".format(
383 version_data["status"]), HTTPStatus.INTERNAL_SERVER_ERROR)
384
385 # check version
386 db_version = None if not version_data else version_data.get("version")
387 if db_version != target_version:
388 self.upgrade_db(db_version, target_version)
389
390 # create user admin if not exist
391 if not self.auth:
392 self.create_admin()
393
394 return