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