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