projects / osm / RO.git / blob
? search:


455c625b40aabd8c2faa99785e8322bb8f5833c2
[osm/RO.git] / osm_ro / vim_thread.py
1 # -*- coding: utf-8 -*-
2
3 ##
4 # Copyright 2015 Telefonica Investigacion y Desarrollo, S.A.U.
5 # This file is part of openvim
6 # All Rights Reserved.
7 #
8 # Licensed under the Apache License, Version 2.0 (the "License"); you may
9 # not use this file except in compliance with the License. You may obtain
10 # a copy of the License at
11 #
12 # http://www.apache.org/licenses/LICENSE-2.0
13 #
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
16 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17 # License for the specific language governing permissions and limitations
18 # under the License.
19 #
20 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact with: nfvlabs@tid.es
22 ##
23
24 """"
25 This is thread that interacts with a VIM. It processes TASKs sequentially against a single VIM.
26 The tasks are stored at database in table vim_wim_actions
27 Several vim_wim_actions can refer to the same element at VIM (flavor, network, ...). This is somethng to avoid if RO
28 is migrated to a non-relational database as mongo db. Each vim_wim_actions reference a different instance_Xxxxx
29 In this case "related" colunm contains the same value, to know they refer to the same vim. In case of deletion, it
30 there is related tasks using this element, it is not deleted, The vim_info needed to delete is transfered to other task
31
32 The task content is (M: stored at memory, D: stored at database):
33 MD instance_action_id: reference a global action over an instance-scenario: database instance_actions
34 MD task_index: index number of the task. This together with the previous forms a unique key identifier
35 MD datacenter_vim_id: should contain the uuid of the VIM managed by this thread
36 MD vim_id: id of the vm,net,etc at VIM
37 MD item: database table name, can be instance_vms, instance_nets, TODO: datacenter_flavors, datacenter_images
38 MD item_id: uuid of the referenced entry in the previous table
39 MD action: CREATE, DELETE, FIND
40 MD status: SCHEDULED: action need to be done
41 BUILD: not used
42 DONE: Done and it must be polled to VIM periodically to see status. ONLY for action=CREATE or FIND
43 FAILED: It cannot be created/found/deleted
44 FINISHED: similar to DONE, but no refresh is needed anymore. Task is maintained at database but
45 it is never processed by any thread
46 SUPERSEDED: similar to FINSISHED, but nothing has been done to completed the task.
47 MD extra: text with yaml format at database, dict at memory with:
48 params: list with the params to be sent to the VIM for CREATE or FIND. For DELETE the vim_id is taken
49 from other related tasks
50 find: (only for CREATE tasks) if present it should FIND before creating and use if existing. Contains
51 the FIND params
52 depends_on: list with the 'task_index'es of tasks that must be completed before. e.g. a vm creation depends
53 on a net creation
54 can contain an int (single index on the same instance-action) or str (compete action ID)
55 sdn_net_id: used for net.
56 interfaces: used for VMs. Each key is the uuid of the instance_interfaces entry at database
57 iface_id: uuid of intance_interfaces
58 sdn_port_id:
59 sdn_net_id:
60 vim_info
61 created_items: dictionary with extra elements created that need to be deleted. e.g. ports, volumes,...
62 created: False if the VIM element is not created by other actions, and it should not be deleted
63 vim_status: VIM status of the element. Stored also at database in the instance_XXX
64 vim_info: Detailed information of a vm/net from the VIM. Stored at database in the instance_XXX but not at
65 vim_wim_actions
66 M depends: dict with task_index(from depends_on) to dependency task
67 M params: same as extra[params]
68 MD error_msg: descriptive text upon an error.Stored also at database instance_XXX
69 MD created_at: task creation time. The task of creation must be the oldest
70 MD modified_at: next time task need to be processed. For example, for a refresh, it contain next time refresh must
71 be done
72 MD related: All the tasks over the same VIM element have same "related". Note that other VIMs can contain the
73 same value of related, but this thread only process those task of one VIM. Also related can be the
74 same among several NS os isntance-scenarios
75 MD worker: Used to lock in case of several thread workers.
76
77 """
78
79 import threading
80 import time
81 import Queue
82 import logging
83 import vimconn
84 import vimconn_openvim
85 import vimconn_aws
86 import vimconn_opennebula
87 import vimconn_openstack
88 import vimconn_vmware
89 import vimconn_fos
90 import vimconn_azure
91 import yaml
92 from db_base import db_base_Exception
93 from lib_osm_openvim.ovim import ovimException
94 from copy import deepcopy
95
96 __author__ = "Alfonso Tierno, Pablo Montes"
97 __date__ = "$28-Sep-2017 12:07:15$"
98
99 vim_module = {
100 "openvim": vimconn_openvim,
101 "aws": vimconn_aws,
102 "opennebula": vimconn_opennebula,
103 "openstack": vimconn_openstack,
104 "vmware": vimconn_vmware,
105 "fos": vimconn_fos,
106 "azure": vimconn_azure,
107 }
108
109
110 def is_task_id(task_id):
111 return task_id.startswith("TASK-")
112
113
114 class VimThreadException(Exception):
115 pass
116
117
118 class VimThreadExceptionNotFound(VimThreadException):
119 pass
120
121
122 class vim_thread(threading.Thread):
123 REFRESH_BUILD = 5 # 5 seconds
124 REFRESH_ACTIVE = 60 # 1 minute
125 REFRESH_ERROR = 600
126 REFRESH_DELETE = 3600 * 10
127
128 def __init__(self, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None,
129 db=None, db_lock=None, ovim=None):
130 """Init a thread.
131 Arguments:
132 'id' number of thead
133 'name' name of thread
134 'host','user': host ip or name to manage and user
135 'db', 'db_lock': database class and lock to use it in exclusion
136 """
137 threading.Thread.__init__(self)
138 self.vim = None
139 self.error_status = None
140 self.datacenter_name = datacenter_name
141 self.datacenter_tenant_id = datacenter_tenant_id
142 self.ovim = ovim
143 if not name:
144 self.name = vimconn["id"] + "." + vimconn["config"]["datacenter_tenant_id"]
145 else:
146 self.name = name
147 self.vim_persistent_info = {}
148 self.my_id = self.name[:64]
149
150 self.logger = logging.getLogger('openmano.vim.' + self.name)
151 self.db = db
152 self.db_lock = db_lock
153
154 self.task_lock = task_lock
155 self.task_queue = Queue.Queue(2000)
156
157 def get_vimconnector(self):
158 try:
159 from_ = "datacenter_tenants as dt join datacenters as d on dt.datacenter_id=d.uuid"
160 select_ = ('type', 'd.config as config', 'd.uuid as datacenter_id', 'vim_url', 'vim_url_admin',
161 'd.name as datacenter_name', 'dt.uuid as datacenter_tenant_id',
162 'dt.vim_tenant_name as vim_tenant_name', 'dt.vim_tenant_id as vim_tenant_id',
163 'user', 'passwd', 'dt.config as dt_config')
164 where_ = {"dt.uuid": self.datacenter_tenant_id}
165 vims = self.db.get_rows(FROM=from_, SELECT=select_, WHERE=where_)
166 vim = vims[0]
167 vim_config = {}
168 if vim["config"]:
169 vim_config.update(yaml.load(vim["config"]))
170 if vim["dt_config"]:
171 vim_config.update(yaml.load(vim["dt_config"]))
172 vim_config['datacenter_tenant_id'] = vim.get('datacenter_tenant_id')
173 vim_config['datacenter_id'] = vim.get('datacenter_id')
174
175 # get port_mapping
176 with self.db_lock:
177 vim_config["wim_external_ports"] = self.ovim.get_of_port_mappings(
178 db_filter={"region": vim_config['datacenter_id'], "pci": None})
179
180 self.vim = vim_module[vim["type"]].vimconnector(
181 uuid=vim['datacenter_id'], name=vim['datacenter_name'],
182 tenant_id=vim['vim_tenant_id'], tenant_name=vim['vim_tenant_name'],
183 url=vim['vim_url'], url_admin=vim['vim_url_admin'],
184 user=vim['user'], passwd=vim['passwd'],
185 config=vim_config, persistent_info=self.vim_persistent_info
186 )
187 self.error_status = None
188 except Exception as e:
189 self.logger.error("Cannot load vimconnector for vim_account {}: {}".format(self.datacenter_tenant_id, e))
190 self.vim = None
191 self.error_status = "Error loading vimconnector: {}".format(e)
192
193 def _get_db_task(self):
194 """
195 Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
196 :return: None
197 """
198 now = time.time()
199 try:
200 database_limit = 20
201 task_related = None
202 while True:
203 # get 20 (database_limit) entries each time
204 vim_actions = self.db.get_rows(FROM="vim_wim_actions",
205 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
206 "status": ['SCHEDULED', 'BUILD', 'DONE'],
207 "worker": [None, self.my_id], "modified_at<=": now
208 },
209 ORDER_BY=("modified_at", "created_at",),
210 LIMIT=database_limit)
211 if not vim_actions:
212 return None, None
213 # if vim_actions[0]["modified_at"] > now:
214 # return int(vim_actions[0] - now)
215 for task in vim_actions:
216 # block related task
217 if task_related == task["related"]:
218 continue # ignore if a locking has already tried for these task set
219 task_related = task["related"]
220 # lock ...
221 self.db.update_rows("vim_wim_actions", UPDATE={"worker": self.my_id}, modified_time=0,
222 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
223 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
224 "worker": [None, self.my_id],
225 "related": task_related,
226 "item": task["item"],
227 })
228 # ... and read all related and check if locked
229 related_tasks = self.db.get_rows(FROM="vim_wim_actions",
230 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
231 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
232 "related": task_related,
233 "item": task["item"],
234 },
235 ORDER_BY=("created_at",))
236 # check that all related tasks have been locked. If not release and try again. It can happen
237 # for race conditions if a new related task has been inserted by nfvo in the process
238 some_tasks_locked = False
239 some_tasks_not_locked = False
240 creation_task = None
241 for relate_task in related_tasks:
242 if relate_task["worker"] != self.my_id:
243 some_tasks_not_locked = True
244 else:
245 some_tasks_locked = True
246 if not creation_task and relate_task["action"] in ("CREATE", "FIND"):
247 creation_task = relate_task
248 if some_tasks_not_locked:
249 if some_tasks_locked: # unlock
250 self.db.update_rows("vim_wim_actions", UPDATE={"worker": None}, modified_time=0,
251 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
252 "worker": self.my_id,
253 "related": task_related,
254 "item": task["item"],
255 })
256 continue
257
258 # task of creation must be the first in the list of related_task
259 assert(related_tasks[0]["action"] in ("CREATE", "FIND"))
260
261 if task["extra"]:
262 extra = yaml.load(task["extra"])
263 else:
264 extra = {}
265 task["extra"] = extra
266 if extra.get("depends_on"):
267 task["depends"] = {}
268 if extra.get("params"):
269 task["params"] = deepcopy(extra["params"])
270 return task, related_tasks
271 except Exception as e:
272 self.logger.critical("Unexpected exception at _get_db_task: " + str(e), exc_info=True)
273 return None, None
274
275 def _delete_task(self, task):
276 """
277 Determine if this task need to be done or superseded
278 :return: None
279 """
280
281 def copy_extra_created(copy_to, copy_from):
282 copy_to["created"] = copy_from["created"]
283 if copy_from.get("sdn_net_id"):
284 copy_to["sdn_net_id"] = copy_from["sdn_net_id"]
285 if copy_from.get("interfaces"):
286 copy_to["interfaces"] = copy_from["interfaces"]
287 if copy_from.get("created_items"):
288 if not copy_to.get("created_items"):
289 copy_to["created_items"] = {}
290 copy_to["created_items"].update(copy_from["created_items"])
291
292 task_create = None
293 dependency_task = None
294 deletion_needed = False
295 if task["status"] == "FAILED":
296 return # TODO need to be retry??
297 try:
298 # get all related tasks
299 related_tasks = self.db.get_rows(FROM="vim_wim_actions",
300 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
301 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
302 "action": ["FIND", "CREATE"],
303 "related": task["related"],
304 },
305 ORDER_BY=("created_at",),
306 )
307 for related_task in related_tasks:
308 if related_task["item"] == task["item"] and related_task["item_id"] == task["item_id"]:
309 task_create = related_task
310 # TASK_CREATE
311 if related_task["extra"]:
312 extra_created = yaml.load(related_task["extra"])
313 if extra_created.get("created"):
314 deletion_needed = True
315 related_task["extra"] = extra_created
316 elif not dependency_task:
317 dependency_task = related_task
318 if task_create and dependency_task:
319 break
320
321 # mark task_create as FINISHED
322 self.db.update_rows("vim_wim_actions", UPDATE={"status": "FINISHED"},
323 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
324 "instance_action_id": task_create["instance_action_id"],
325 "task_index": task_create["task_index"]
326 })
327 if not deletion_needed:
328 return
329 elif dependency_task:
330 # move create information from task_create to relate_task
331 extra_new_created = yaml.load(dependency_task["extra"]) or {}
332 extra_new_created["created"] = extra_created["created"]
333 copy_extra_created(copy_to=extra_new_created, copy_from=extra_created)
334
335 self.db.update_rows("vim_wim_actions",
336 UPDATE={"extra": yaml.safe_dump(extra_new_created, default_flow_style=True,
337 width=256),
338 "vim_id": task_create.get("vim_id")},
339 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
340 "instance_action_id": dependency_task["instance_action_id"],
341 "task_index": dependency_task["task_index"]
342 })
343 return False
344 else:
345 task["vim_id"] = task_create["vim_id"]
346 copy_extra_created(copy_to=task["extra"], copy_from=task_create["extra"])
347 return True
348
349 except Exception as e:
350 self.logger.critical("Unexpected exception at _delete_task: " + str(e), exc_info=True)
351
352 def _refres_vm(self, task):
353 """Call VIM to get VMs status"""
354 database_update = None
355
356 vim_id = task["vim_id"]
357 vm_to_refresh_list = [vim_id]
358 try:
359 vim_dict = self.vim.refresh_vms_status(vm_to_refresh_list)
360 vim_info = vim_dict[vim_id]
361 except vimconn.vimconnException as e:
362 # Mark all tasks at VIM_ERROR status
363 self.logger.error("task=several get-VM: vimconnException when trying to refresh vms " + str(e))
364 vim_info = {"status": "VIM_ERROR", "error_msg": str(e)}
365
366 task_id = task["instance_action_id"] + "." + str(task["task_index"])
367 self.logger.debug("task={} get-VM: vim_vm_id={} result={}".format(task_id, task["vim_id"], vim_info))
368
369 # check and update interfaces
370 task_warning_msg = ""
371 for interface in vim_info.get("interfaces", ()):
372 vim_interface_id = interface["vim_interface_id"]
373 if vim_interface_id not in task["extra"]["interfaces"]:
374 self.logger.critical("task={} get-VM: Interface not found {} on task info {}".format(
375 task_id, vim_interface_id, task["extra"]["interfaces"]), exc_info=True)
376 continue
377 task_interface = task["extra"]["interfaces"][vim_interface_id]
378 task_vim_interface = task_interface.get("vim_info")
379 if task_vim_interface != interface:
380 # delete old port
381 if task_interface.get("sdn_port_id"):
382 try:
383 with self.db_lock:
384 self.ovim.delete_port(task_interface["sdn_port_id"], idempotent=True)
385 task_interface["sdn_port_id"] = None
386 except ovimException as e:
387 error_text = "ovimException deleting external_port={}: {}".format(
388 task_interface["sdn_port_id"], e)
389 self.logger.error("task={} get-VM: {}".format(task_id, error_text), exc_info=True)
390 task_warning_msg += error_text
391 # TODO Set error_msg at instance_nets instead of instance VMs
392
393 # Create SDN port
394 sdn_net_id = task_interface.get("sdn_net_id")
395 if sdn_net_id and interface.get("compute_node") and interface.get("pci"):
396 sdn_port_name = sdn_net_id + "." + task["vim_id"]
397 sdn_port_name = sdn_port_name[:63]
398 try:
399 with self.db_lock:
400 sdn_port_id = self.ovim.new_external_port(
401 {"compute_node": interface["compute_node"],
402 "pci": interface["pci"],
403 "vlan": interface.get("vlan"),
404 "net_id": sdn_net_id,
405 "region": self.vim["config"]["datacenter_id"],
406 "name": sdn_port_name,
407 "mac": interface.get("mac_address")})
408 task_interface["sdn_port_id"] = sdn_port_id
409 except (ovimException, Exception) as e:
410 error_text = "ovimException creating new_external_port compute_node={} pci={} vlan={} {}".\
411 format(interface["compute_node"], interface["pci"], interface.get("vlan"), e)
412 self.logger.error("task={} get-VM: {}".format(task_id, error_text), exc_info=True)
413 task_warning_msg += error_text
414 # TODO Set error_msg at instance_nets instead of instance VMs
415
416 self.db.update_rows('instance_interfaces',
417 UPDATE={"mac_address": interface.get("mac_address"),
418 "ip_address": interface.get("ip_address"),
419 "vim_interface_id": interface.get("vim_interface_id"),
420 "vim_info": interface.get("vim_info"),
421 "sdn_port_id": task_interface.get("sdn_port_id"),
422 "compute_node": interface.get("compute_node"),
423 "pci": interface.get("pci"),
424 "vlan": interface.get("vlan")},
425 WHERE={'uuid': task_interface["iface_id"]})
426 task_interface["vim_info"] = interface
427
428 # check and update task and instance_vms database
429 vim_info_error_msg = None
430 if vim_info.get("error_msg"):
431 vim_info_error_msg = self._format_vim_error_msg(vim_info["error_msg"] + task_warning_msg)
432 elif task_warning_msg:
433 vim_info_error_msg = self._format_vim_error_msg(task_warning_msg)
434 task_vim_info = task["extra"].get("vim_info")
435 task_error_msg = task.get("error_msg")
436 task_vim_status = task["extra"].get("vim_status")
437 if task_vim_status != vim_info["status"] or task_error_msg != vim_info_error_msg or \
438 (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]):
439 database_update = {"status": vim_info["status"], "error_msg": vim_info_error_msg}
440 if vim_info.get("vim_info"):
441 database_update["vim_info"] = vim_info["vim_info"]
442
443 task["extra"]["vim_status"] = vim_info["status"]
444 task["error_msg"] = vim_info_error_msg
445 if vim_info.get("vim_info"):
446 task["extra"]["vim_info"] = vim_info["vim_info"]
447
448 return database_update
449
450 def _refres_net(self, task):
451 """Call VIM to get network status"""
452 database_update = None
453
454 vim_id = task["vim_id"]
455 net_to_refresh_list = [vim_id]
456 try:
457 vim_dict = self.vim.refresh_nets_status(net_to_refresh_list)
458 vim_info = vim_dict[vim_id]
459 except vimconn.vimconnException as e:
460 # Mark all tasks at VIM_ERROR status
461 self.logger.error("task=several get-net: vimconnException when trying to refresh nets " + str(e))
462 vim_info = {"status": "VIM_ERROR", "error_msg": str(e)}
463
464 task_id = task["instance_action_id"] + "." + str(task["task_index"])
465 self.logger.debug("task={} get-net: vim_net_id={} result={}".format(task_id, task["vim_id"], vim_info))
466
467 task_vim_info = task["extra"].get("vim_info")
468 task_vim_status = task["extra"].get("vim_status")
469 task_error_msg = task.get("error_msg")
470 task_sdn_net_id = task["extra"].get("sdn_net_id")
471
472 vim_info_status = vim_info["status"]
473 vim_info_error_msg = vim_info.get("error_msg")
474 # get ovim status
475 if task_sdn_net_id:
476 try:
477 with self.db_lock:
478 sdn_net = self.ovim.show_network(task_sdn_net_id)
479 except (ovimException, Exception) as e:
480 text_error = "ovimException getting network snd_net_id={}: {}".format(task_sdn_net_id, e)
481 self.logger.error("task={} get-net: {}".format(task_id, text_error), exc_info=True)
482 sdn_net = {"status": "ERROR", "last_error": text_error}
483 if sdn_net["status"] == "ERROR":
484 if not vim_info_error_msg:
485 vim_info_error_msg = str(sdn_net.get("last_error"))
486 else:
487 vim_info_error_msg = "VIM_ERROR: {} && SDN_ERROR: {}".format(
488 self._format_vim_error_msg(vim_info_error_msg, 1024 // 2 - 14),
489 self._format_vim_error_msg(sdn_net["last_error"], 1024 // 2 - 14))
490 vim_info_status = "ERROR"
491 elif sdn_net["status"] == "BUILD":
492 if vim_info_status == "ACTIVE":
493 vim_info_status = "BUILD"
494
495 # update database
496 if vim_info_error_msg:
497 vim_info_error_msg = self._format_vim_error_msg(vim_info_error_msg)
498 if task_vim_status != vim_info_status or task_error_msg != vim_info_error_msg or \
499 (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]):
500 task["extra"]["vim_status"] = vim_info_status
501 task["error_msg"] = vim_info_error_msg
502 if vim_info.get("vim_info"):
503 task["extra"]["vim_info"] = vim_info["vim_info"]
504 database_update = {"status": vim_info_status, "error_msg": vim_info_error_msg}
505 if vim_info.get("vim_info"):
506 database_update["vim_info"] = vim_info["vim_info"]
507 return database_update
508
509 def _proccess_pending_tasks(self, task, related_tasks):
510 old_task_status = task["status"]
511 create_or_find = False # if as result of processing this task something is created or found
512 next_refresh = 0
513
514 try:
515 if task["status"] == "SCHEDULED":
516 # check if tasks that this depends on have been completed
517 dependency_not_completed = False
518 dependency_modified_at = 0
519 for task_index in task["extra"].get("depends_on", ()):
520 task_dependency = self._look_for_task(task["instance_action_id"], task_index)
521 if not task_dependency:
522 raise VimThreadException(
523 "Cannot get depending net task trying to get depending task {}.{}".format(
524 task["instance_action_id"], task_index))
525 # task["depends"]["TASK-" + str(task_index)] = task_dependency #it references another object,so
526 # database must be look again
527 if task_dependency["status"] == "SCHEDULED":
528 dependency_not_completed = True
529 dependency_modified_at = task_dependency["modified_at"]
530 break
531 elif task_dependency["status"] == "FAILED":
532 raise VimThreadException(
533 "Cannot {} {}, (task {}.{}) because depends on failed {}.{}, (task{}.{}): {}".format(
534 task["action"], task["item"],
535 task["instance_action_id"], task["task_index"],
536 task_dependency["instance_action_id"], task_dependency["task_index"],
537 task_dependency["action"], task_dependency["item"], task_dependency.get("error_msg")))
538
539 task["depends"]["TASK-"+str(task_index)] = task_dependency
540 task["depends"]["TASK-{}.{}".format(task["instance_action_id"], task_index)] = task_dependency
541 if dependency_not_completed:
542 # Move this task to the time dependency is going to be modified plus 10 seconds.
543 self.db.update_rows("vim_wim_actions", modified_time=dependency_modified_at + 10,
544 UPDATE={"worker": None},
545 WHERE={"datacenter_vim_id": self.datacenter_tenant_id, "worker": self.my_id,
546 "related": task["related"],
547 })
548 # task["extra"]["tries"] = task["extra"].get("tries", 0) + 1
549 # if task["extra"]["tries"] > 3:
550 # raise VimThreadException(
551 # "Cannot {} {}, (task {}.{}) because timeout waiting to complete {} {}, "
552 # "(task {}.{})".format(task["action"], task["item"],
553 # task["instance_action_id"], task["task_index"],
554 # task_dependency["instance_action_id"], task_dependency["task_index"]
555 # task_dependency["action"], task_dependency["item"]))
556 return
557
558 database_update = None
559 if task["action"] == "DELETE":
560 deleted_needed = self._delete_task(task)
561 if not deleted_needed:
562 task["status"] = "SUPERSEDED" # with FINISHED instead of DONE it will not be refreshing
563 task["error_msg"] = None
564
565 if task["status"] == "SUPERSEDED":
566 # not needed to do anything but update database with the new status
567 database_update = None
568 elif not self.vim:
569 task["status"] = "FAILED"
570 task["error_msg"] = self.error_status
571 database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]}
572 elif task["item_id"] != related_tasks[0]["item_id"] and task["action"] in ("FIND", "CREATE"):
573 # Do nothing, just copy values from one to another and updata database
574 task["status"] = related_tasks[0]["status"]
575 task["error_msg"] = related_tasks[0]["error_msg"]
576 task["vim_id"] = related_tasks[0]["vim_id"]
577 extra = yaml.load(related_tasks[0]["extra"])
578 task["extra"]["vim_status"] = extra["vim_status"]
579 next_refresh = related_tasks[0]["modified_at"] + 0.001
580 database_update = {"status": task["extra"].get("vim_status", "VIM_ERROR"),
581 "error_msg": task["error_msg"]}
582 if task["item"] == 'instance_vms':
583 database_update["vim_vm_id"] = task["vim_id"]
584 elif task["item"] == 'instance_nets':
585 database_update["vim_net_id"] = task["vim_id"]
586 elif task["item"] == 'instance_vms':
587 if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"):
588 database_update = self._refres_vm(task)
589 create_or_find = True
590 elif task["action"] == "CREATE":
591 create_or_find = True
592 database_update = self.new_vm(task)
593 elif task["action"] == "DELETE":
594 self.del_vm(task)
595 else:
596 raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
597 elif task["item"] == 'instance_nets':
598 if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"):
599 database_update = self._refres_net(task)
600 create_or_find = True
601 elif task["action"] == "CREATE":
602 create_or_find = True
603 database_update = self.new_net(task)
604 elif task["action"] == "DELETE":
605 self.del_net(task)
606 elif task["action"] == "FIND":
607 database_update = self.get_net(task)
608 else:
609 raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
610 elif task["item"] == 'instance_sfis':
611 if task["action"] == "CREATE":
612 create_or_find = True
613 database_update = self.new_sfi(task)
614 elif task["action"] == "DELETE":
615 self.del_sfi(task)
616 else:
617 raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
618 elif task["item"] == 'instance_sfs':
619 if task["action"] == "CREATE":
620 create_or_find = True
621 database_update = self.new_sf(task)
622 elif task["action"] == "DELETE":
623 self.del_sf(task)
624 else:
625 raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
626 elif task["item"] == 'instance_classifications':
627 if task["action"] == "CREATE":
628 create_or_find = True
629 database_update = self.new_classification(task)
630 elif task["action"] == "DELETE":
631 self.del_classification(task)
632 else:
633 raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
634 elif task["item"] == 'instance_sfps':
635 if task["action"] == "CREATE":
636 create_or_find = True
637 database_update = self.new_sfp(task)
638 elif task["action"] == "DELETE":
639 self.del_sfp(task)
640 else:
641 raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"]))
642 else:
643 raise vimconn.vimconnException(self.name + "unknown task item {}".format(task["item"]))
644 # TODO
645 except VimThreadException as e:
646 task["error_msg"] = str(e)
647 task["status"] = "FAILED"
648 database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]}
649 if task["item"] == 'instance_vms':
650 database_update["vim_vm_id"] = None
651 elif task["item"] == 'instance_nets':
652 database_update["vim_net_id"] = None
653
654 task_id = task["instance_action_id"] + "." + str(task["task_index"])
655 self.logger.debug("task={} item={} action={} result={}:'{}' params={}".format(
656 task_id, task["item"], task["action"], task["status"],
657 task["vim_id"] if task["status"] == "DONE" else task.get("error_msg"), task["params"]))
658 try:
659 if not next_refresh:
660 if task["status"] == "DONE":
661 next_refresh = time.time()
662 if task["extra"].get("vim_status") == "BUILD":
663 next_refresh += self.REFRESH_BUILD
664 elif task["extra"].get("vim_status") in ("ERROR", "VIM_ERROR"):
665 next_refresh += self.REFRESH_ERROR
666 elif task["extra"].get("vim_status") == "DELETED":
667 next_refresh += self.REFRESH_DELETE
668 else:
669 next_refresh += self.REFRESH_ACTIVE
670 elif task["status"] == "FAILED":
671 next_refresh = time.time() + self.REFRESH_DELETE
672
673 if create_or_find:
674 # modify all related task with action FIND/CREATED non SCHEDULED
675 self.db.update_rows(
676 table="vim_wim_actions", modified_time=next_refresh + 0.001,
677 UPDATE={"status": task["status"], "vim_id": task.get("vim_id"),
678 "error_msg": task["error_msg"],
679 },
680
681 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
682 "worker": self.my_id,
683 "action": ["FIND", "CREATE"],
684 "related": task["related"],
685 "status<>": "SCHEDULED",
686 })
687 # modify own task
688 self.db.update_rows(
689 table="vim_wim_actions", modified_time=next_refresh,
690 UPDATE={"status": task["status"], "vim_id": task.get("vim_id"),
691 "error_msg": task["error_msg"],
692 "extra": yaml.safe_dump(task["extra"], default_flow_style=True, width=256)},
693 WHERE={"instance_action_id": task["instance_action_id"], "task_index": task["task_index"]})
694 # Unlock tasks
695 self.db.update_rows(
696 table="vim_wim_actions", modified_time=0,
697 UPDATE={"worker": None},
698 WHERE={"datacenter_vim_id": self.datacenter_tenant_id,
699 "worker": self.my_id,
700 "related": task["related"],
701 })
702
703 # Update table instance_actions
704 if old_task_status == "SCHEDULED" and task["status"] != old_task_status:
705 self.db.update_rows(
706 table="instance_actions",
707 UPDATE={("number_failed" if task["status"] == "FAILED" else "number_done"): {"INCREMENT": 1}},
708 WHERE={"uuid": task["instance_action_id"]})
709 if database_update:
710 self.db.update_rows(table=task["item"],
711 UPDATE=database_update,
712 WHERE={"related": task["related"]})
713 except db_base_Exception as e:
714 self.logger.error("task={} Error updating database {}".format(task_id, e), exc_info=True)
715
716 def insert_task(self, task):
717 try:
718 self.task_queue.put(task, False)
719 return None
720 except Queue.Full:
721 raise vimconn.vimconnException(self.name + ": timeout inserting a task")
722
723 def del_task(self, task):
724 with self.task_lock:
725 if task["status"] == "SCHEDULED":
726 task["status"] = "SUPERSEDED"
727 return True
728 else: # task["status"] == "processing"
729 self.task_lock.release()
730 return False
731
732 def run(self):
733 self.logger.debug("Starting")
734 while True:
735 self.get_vimconnector()
736 self.logger.debug("Vimconnector loaded")
737 reload_thread = False
738
739 while True:
740 try:
741 while not self.task_queue.empty():
742 task = self.task_queue.get()
743 if isinstance(task, list):
744 pass
745 elif isinstance(task, str):
746 if task == 'exit':
747 return 0
748 elif task == 'reload':
749 reload_thread = True
750 break
751 self.task_queue.task_done()
752 if reload_thread:
753 break
754
755 task, related_tasks = self._get_db_task()
756 if task:
757 self._proccess_pending_tasks(task, related_tasks)
758 else:
759 time.sleep(5)
760
761 except Exception as e:
762 self.logger.critical("Unexpected exception at run: " + str(e), exc_info=True)
763
764 self.logger.debug("Finishing")
765
766 def _look_for_task(self, instance_action_id, task_id):
767 """
768 Look for a concrete task at vim_actions database table
769 :param instance_action_id: The instance_action_id
770 :param task_id: Can have several formats:
771 <task index>: integer
772 TASK-<task index> :backward compatibility,
773 [TASK-]<instance_action_id>.<task index>: this instance_action_id overrides the one in the parameter
774 :return: Task dictionary or None if not found
775 """
776 if isinstance(task_id, int):
777 task_index = task_id
778 else:
779 if task_id.startswith("TASK-"):
780 task_id = task_id[5:]
781 ins_action_id, _, task_index = task_id.rpartition(".")
782 if ins_action_id:
783 instance_action_id = ins_action_id
784
785 tasks = self.db.get_rows(FROM="vim_wim_actions", WHERE={"instance_action_id": instance_action_id,
786 "task_index": task_index})
787 if not tasks:
788 return None
789 task = tasks[0]
790 task["params"] = None
791 task["depends"] = {}
792 if task["extra"]:
793 extra = yaml.load(task["extra"])
794 task["extra"] = extra
795 task["params"] = extra.get("params")
796 else:
797 task["extra"] = {}
798 return task
799
800 @staticmethod
801 def _format_vim_error_msg(error_text, max_length=1024):
802 if error_text and len(error_text) >= max_length:
803 return error_text[:max_length // 2 - 3] + " ... " + error_text[-max_length // 2 + 3:]
804 return error_text
805
806 def new_vm(self, task):
807 task_id = task["instance_action_id"] + "." + str(task["task_index"])
808 try:
809 params = task["params"]
810 depends = task.get("depends")
811 net_list = params[5]
812 for net in net_list:
813 if "net_id" in net and is_task_id(net["net_id"]): # change task_id into network_id
814 network_id = task["depends"][net["net_id"]].get("vim_id")
815 if not network_id:
816 raise VimThreadException(
817 "Cannot create VM because depends on a network not created or found: " +
818 str(depends[net["net_id"]]["error_msg"]))
819 net["net_id"] = network_id
820 params_copy = deepcopy(params)
821 vim_vm_id, created_items = self.vim.new_vminstance(*params_copy)
822
823 # fill task_interfaces. Look for snd_net_id at database for each interface
824 task_interfaces = {}
825 for iface in params_copy[5]:
826 task_interfaces[iface["vim_id"]] = {"iface_id": iface["uuid"]}
827 result = self.db.get_rows(
828 SELECT=('sdn_net_id', 'interface_id'),
829 FROM='instance_nets as ine join instance_interfaces as ii on ii.instance_net_id=ine.uuid',
830 WHERE={'ii.uuid': iface["uuid"]})
831 if result:
832 task_interfaces[iface["vim_id"]]["sdn_net_id"] = result[0]['sdn_net_id']
833 task_interfaces[iface["vim_id"]]["interface_id"] = result[0]['interface_id']
834 else:
835 self.logger.critical("task={} new-VM: instance_nets uuid={} not found at DB".format(task_id,
836 iface["uuid"]),
837 exc_info=True)
838
839 task["vim_info"] = {}
840 task["extra"]["interfaces"] = task_interfaces
841 task["extra"]["created"] = True
842 task["extra"]["created_items"] = created_items
843 task["extra"]["vim_status"] = "BUILD"
844 task["error_msg"] = None
845 task["status"] = "DONE"
846 task["vim_id"] = vim_vm_id
847 instance_element_update = {"status": "BUILD", "vim_vm_id": vim_vm_id, "error_msg": None}
848 return instance_element_update
849
850 except (vimconn.vimconnException, VimThreadException) as e:
851 self.logger.error("task={} new-VM: {}".format(task_id, e))
852 error_text = self._format_vim_error_msg(str(e))
853 task["error_msg"] = error_text
854 task["status"] = "FAILED"
855 task["vim_id"] = None
856 instance_element_update = {"status": "VIM_ERROR", "vim_vm_id": None, "error_msg": error_text}
857 return instance_element_update
858
859 def del_vm(self, task):
860 task_id = task["instance_action_id"] + "." + str(task["task_index"])
861 vm_vim_id = task["vim_id"]
862 interfaces = task["extra"].get("interfaces", ())
863 try:
864 for iface in interfaces.values():
865 if iface.get("sdn_port_id"):
866 try:
867 with self.db_lock:
868 self.ovim.delete_port(iface["sdn_port_id"], idempotent=True)
869 except ovimException as e:
870 self.logger.error("task={} del-VM: ovimException when deleting external_port={}: {} ".format(
871 task_id, iface["sdn_port_id"], e), exc_info=True)
872 # TODO Set error_msg at instance_nets
873
874 self.vim.delete_vminstance(vm_vim_id, task["extra"].get("created_items"))
875 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
876 task["error_msg"] = None
877 return None
878
879 except vimconn.vimconnException as e:
880 task["error_msg"] = self._format_vim_error_msg(str(e))
881 if isinstance(e, vimconn.vimconnNotFoundException):
882 # If not found mark as Done and fill error_msg
883 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
884 return None
885 task["status"] = "FAILED"
886 return None
887
888 def _get_net_internal(self, task, filter_param):
889 """
890 Common code for get_net and new_net. It looks for a network on VIM with the filter_params
891 :param task: task for this find or find-or-create action
892 :param filter_param: parameters to send to the vimconnector
893 :return: a dict with the content to update the instance_nets database table. Raises an exception on error, or
894 when network is not found or found more than one
895 """
896 vim_nets = self.vim.get_network_list(filter_param)
897 if not vim_nets:
898 raise VimThreadExceptionNotFound("Network not found with this criteria: '{}'".format(filter_param))
899 elif len(vim_nets) > 1:
900 raise VimThreadException("More than one network found with this criteria: '{}'".format(filter_param))
901 vim_net_id = vim_nets[0]["id"]
902
903 # Discover if this network is managed by a sdn controller
904 sdn_net_id = None
905 result = self.db.get_rows(SELECT=('sdn_net_id',), FROM='instance_nets',
906 WHERE={'vim_net_id': vim_net_id, 'datacenter_tenant_id': self.datacenter_tenant_id},
907 ORDER="instance_scenario_id")
908 if result:
909 sdn_net_id = result[0]['sdn_net_id']
910
911 task["status"] = "DONE"
912 task["extra"]["vim_info"] = {}
913 task["extra"]["created"] = False
914 task["extra"]["vim_status"] = "BUILD"
915 task["extra"]["sdn_net_id"] = sdn_net_id
916 task["error_msg"] = None
917 task["vim_id"] = vim_net_id
918 instance_element_update = {"vim_net_id": vim_net_id, "created": False, "status": "BUILD",
919 "error_msg": None, "sdn_net_id": sdn_net_id}
920 return instance_element_update
921
922 def get_net(self, task):
923 task_id = task["instance_action_id"] + "." + str(task["task_index"])
924 try:
925 params = task["params"]
926 filter_param = params[0]
927 instance_element_update = self._get_net_internal(task, filter_param)
928 return instance_element_update
929
930 except (vimconn.vimconnException, VimThreadException) as e:
931 self.logger.error("task={} get-net: {}".format(task_id, e))
932 task["status"] = "FAILED"
933 task["vim_id"] = None
934 task["error_msg"] = self._format_vim_error_msg(str(e))
935 instance_element_update = {"vim_net_id": None, "status": "VIM_ERROR",
936 "error_msg": task["error_msg"]}
937 return instance_element_update
938
939 def new_net(self, task):
940 vim_net_id = None
941 sdn_net_id = None
942 task_id = task["instance_action_id"] + "." + str(task["task_index"])
943 action_text = ""
944 try:
945 # FIND
946 if task["extra"].get("find"):
947 action_text = "finding"
948 filter_param = task["extra"]["find"][0]
949 try:
950 instance_element_update = self._get_net_internal(task, filter_param)
951 return instance_element_update
952 except VimThreadExceptionNotFound:
953 pass
954 # CREATE
955 params = task["params"]
956 action_text = "creating VIM"
957 vim_net_id, created_items = self.vim.new_network(*params[0:3])
958
959 net_name = params[0]
960 net_type = params[1]
961 wim_account_name = None
962 if len(params) >= 4:
963 wim_account_name = params[3]
964
965 sdn_controller = self.vim.config.get('sdn-controller')
966 if sdn_controller and (net_type == "data" or net_type == "ptp"):
967 network = {"name": net_name, "type": net_type, "region": self.vim["config"]["datacenter_id"]}
968
969 vim_net = self.vim.get_network(vim_net_id)
970 if vim_net.get('encapsulation') != 'vlan':
971 raise vimconn.vimconnException(
972 "net '{}' defined as type '{}' has not vlan encapsulation '{}'".format(
973 net_name, net_type, vim_net['encapsulation']))
974 network["vlan"] = vim_net.get('segmentation_id')
975 action_text = "creating SDN"
976 with self.db_lock:
977 sdn_net_id = self.ovim.new_network(network)
978
979 if wim_account_name and self.vim.config["wim_external_ports"]:
980 # add external port to connect WIM. Try with compute node __WIM:wim_name and __WIM
981 action_text = "attaching external port to ovim network"
982 sdn_port_name = sdn_net_id + "." + task["vim_id"]
983 sdn_port_name = sdn_port_name[:63]
984 sdn_port_data = {
985 "compute_node": "__WIM:" + wim_account_name[0:58],
986 "pci": None,
987 "vlan": network["vlan"],
988 "net_id": sdn_net_id,
989 "region": self.vim["config"]["datacenter_id"],
990 "name": sdn_port_name,
991 }
992 try:
993 with self.db_lock:
994 sdn_external_port_id = self.ovim.new_external_port(sdn_port_data)
995 except ovimException:
996 sdn_port_data["compute_node"] = "__WIM"
997 with self.db_lock:
998 sdn_external_port_id = self.ovim.new_external_port(sdn_port_data)
999 self.logger.debug("Added sdn_external_port {} to sdn_network {}".format(sdn_external_port_id,
1000 sdn_net_id))
1001 task["status"] = "DONE"
1002 task["extra"]["vim_info"] = {}
1003 task["extra"]["sdn_net_id"] = sdn_net_id
1004 task["extra"]["vim_status"] = "BUILD"
1005 task["extra"]["created"] = True
1006 task["extra"]["created_items"] = created_items
1007 task["error_msg"] = None
1008 task["vim_id"] = vim_net_id
1009 instance_element_update = {"vim_net_id": vim_net_id, "sdn_net_id": sdn_net_id, "status": "BUILD",
1010 "created": True, "error_msg": None}
1011 return instance_element_update
1012 except (vimconn.vimconnException, ovimException) as e:
1013 self.logger.error("task={} new-net: Error {}: {}".format(task_id, action_text, e))
1014 task["status"] = "FAILED"
1015 task["vim_id"] = vim_net_id
1016 task["error_msg"] = self._format_vim_error_msg(str(e))
1017 task["extra"]["sdn_net_id"] = sdn_net_id
1018 instance_element_update = {"vim_net_id": vim_net_id, "sdn_net_id": sdn_net_id, "status": "VIM_ERROR",
1019 "error_msg": task["error_msg"]}
1020 return instance_element_update
1021
1022 def del_net(self, task):
1023 net_vim_id = task["vim_id"]
1024 sdn_net_id = task["extra"].get("sdn_net_id")
1025 try:
1026 if sdn_net_id:
1027 # Delete any attached port to this sdn network. There can be ports associated to this network in case
1028 # it was manually done using 'openmano vim-net-sdn-attach'
1029 with self.db_lock:
1030 port_list = self.ovim.get_ports(columns={'uuid'},
1031 filter={'name': 'external_port', 'net_id': sdn_net_id})
1032 for port in port_list:
1033 self.ovim.delete_port(port['uuid'], idempotent=True)
1034 self.ovim.delete_network(sdn_net_id, idempotent=True)
1035 if net_vim_id:
1036 self.vim.delete_network(net_vim_id, task["extra"].get("created_items"))
1037 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1038 task["error_msg"] = None
1039 return None
1040 except ovimException as e:
1041 task["error_msg"] = self._format_vim_error_msg("ovimException obtaining and deleting external "
1042 "ports for net {}: {}".format(sdn_net_id, str(e)))
1043 except vimconn.vimconnException as e:
1044 task["error_msg"] = self._format_vim_error_msg(str(e))
1045 if isinstance(e, vimconn.vimconnNotFoundException):
1046 # If not found mark as Done and fill error_msg
1047 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1048 return None
1049 task["status"] = "FAILED"
1050 return None
1051
1052 # Service Function Instances
1053 def new_sfi(self, task):
1054 vim_sfi_id = None
1055 try:
1056 # Waits for interfaces to be ready (avoids failure)
1057 time.sleep(1)
1058 dep_id = "TASK-" + str(task["extra"]["depends_on"][0])
1059 task_id = task["instance_action_id"] + "." + str(task["task_index"])
1060 error_text = ""
1061 interfaces = task["depends"][dep_id]["extra"].get("interfaces")
1062
1063 ingress_interface_id = task.get("extra").get("params").get("ingress_interface_id")
1064 egress_interface_id = task.get("extra").get("params").get("egress_interface_id")
1065 ingress_vim_interface_id = None
1066 egress_vim_interface_id = None
1067 for vim_interface, interface_data in interfaces.iteritems():
1068 if interface_data.get("interface_id") == ingress_interface_id:
1069 ingress_vim_interface_id = vim_interface
1070 break
1071 if ingress_interface_id != egress_interface_id:
1072 for vim_interface, interface_data in interfaces.iteritems():
1073 if interface_data.get("interface_id") == egress_interface_id:
1074 egress_vim_interface_id = vim_interface
1075 break
1076 else:
1077 egress_vim_interface_id = ingress_vim_interface_id
1078 if not ingress_vim_interface_id or not egress_vim_interface_id:
1079 error_text = "Error creating Service Function Instance, Ingress: {}, Egress: {}".format(
1080 ingress_vim_interface_id, egress_vim_interface_id)
1081 self.logger.error(error_text)
1082 task["error_msg"] = error_text
1083 task["status"] = "FAILED"
1084 task["vim_id"] = None
1085 return None
1086 # At the moment, every port associated with the VM will be used both as ingress and egress ports.
1087 # Bear in mind that different VIM connectors might support SFI differently. In the case of OpenStack,
1088 # only the first ingress and first egress ports will be used to create the SFI (Port Pair).
1089 ingress_port_id_list = [ingress_vim_interface_id]
1090 egress_port_id_list = [egress_vim_interface_id]
1091 name = "sfi-%s" % task["item_id"][:8]
1092 # By default no form of IETF SFC Encapsulation will be used
1093 vim_sfi_id = self.vim.new_sfi(name, ingress_port_id_list, egress_port_id_list, sfc_encap=False)
1094
1095 task["extra"]["created"] = True
1096 task["extra"]["vim_status"] = "ACTIVE"
1097 task["error_msg"] = None
1098 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1099 task["vim_id"] = vim_sfi_id
1100 instance_element_update = {"status": "ACTIVE", "vim_sfi_id": vim_sfi_id, "error_msg": None}
1101 return instance_element_update
1102
1103 except (vimconn.vimconnException, VimThreadException) as e:
1104 self.logger.error("Error creating Service Function Instance, task=%s: %s", task_id, str(e))
1105 error_text = self._format_vim_error_msg(str(e))
1106 task["error_msg"] = error_text
1107 task["status"] = "FAILED"
1108 task["vim_id"] = None
1109 instance_element_update = {"status": "VIM_ERROR", "vim_sfi_id": None, "error_msg": error_text}
1110 return instance_element_update
1111
1112 def del_sfi(self, task):
1113 sfi_vim_id = task["vim_id"]
1114 try:
1115 self.vim.delete_sfi(sfi_vim_id)
1116 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1117 task["error_msg"] = None
1118 return None
1119
1120 except vimconn.vimconnException as e:
1121 task["error_msg"] = self._format_vim_error_msg(str(e))
1122 if isinstance(e, vimconn.vimconnNotFoundException):
1123 # If not found mark as Done and fill error_msg
1124 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1125 return None
1126 task["status"] = "FAILED"
1127 return None
1128
1129 def new_sf(self, task):
1130 vim_sf_id = None
1131 try:
1132 task_id = task["instance_action_id"] + "." + str(task["task_index"])
1133 error_text = ""
1134 depending_tasks = ["TASK-" + str(dep_id) for dep_id in task["extra"]["depends_on"]]
1135 # sfis = task.get("depends").values()[0].get("extra").get("params")[5]
1136 sfis = [task.get("depends").get(dep_task) for dep_task in depending_tasks]
1137 sfi_id_list = []
1138 for sfi in sfis:
1139 sfi_id_list.append(sfi.get("vim_id"))
1140 name = "sf-%s" % task["item_id"][:8]
1141 # By default no form of IETF SFC Encapsulation will be used
1142 vim_sf_id = self.vim.new_sf(name, sfi_id_list, sfc_encap=False)
1143
1144 task["extra"]["created"] = True
1145 task["extra"]["vim_status"] = "ACTIVE"
1146 task["error_msg"] = None
1147 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1148 task["vim_id"] = vim_sf_id
1149 instance_element_update = {"status": "ACTIVE", "vim_sf_id": vim_sf_id, "error_msg": None}
1150 return instance_element_update
1151
1152 except (vimconn.vimconnException, VimThreadException) as e:
1153 self.logger.error("Error creating Service Function, task=%s: %s", task_id, str(e))
1154 error_text = self._format_vim_error_msg(str(e))
1155 task["error_msg"] = error_text
1156 task["status"] = "FAILED"
1157 task["vim_id"] = None
1158 instance_element_update = {"status": "VIM_ERROR", "vim_sf_id": None, "error_msg": error_text}
1159 return instance_element_update
1160
1161 def del_sf(self, task):
1162 sf_vim_id = task["vim_id"]
1163 try:
1164 self.vim.delete_sf(sf_vim_id)
1165 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1166 task["error_msg"] = None
1167 return None
1168
1169 except vimconn.vimconnException as e:
1170 task["error_msg"] = self._format_vim_error_msg(str(e))
1171 if isinstance(e, vimconn.vimconnNotFoundException):
1172 # If not found mark as Done and fill error_msg
1173 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1174 return None
1175 task["status"] = "FAILED"
1176 return None
1177
1178 def new_classification(self, task):
1179 vim_classification_id = None
1180 try:
1181 params = task["params"]
1182 task_id = task["instance_action_id"] + "." + str(task["task_index"])
1183 dep_id = "TASK-" + str(task["extra"]["depends_on"][0])
1184 error_text = ""
1185 interfaces = task.get("depends").get(dep_id).get("extra").get("interfaces").keys()
1186 # Bear in mind that different VIM connectors might support Classifications differently.
1187 # In the case of OpenStack, only the first VNF attached to the classifier will be used
1188 # to create the Classification(s) (the "logical source port" of the "Flow Classifier").
1189 # Since the VNFFG classifier match lacks the ethertype, classification defaults to
1190 # using the IPv4 flow classifier.
1191 name = "c-%s" % task["item_id"][:8]
1192 # if not CIDR is given for the IP addresses, add /32:
1193 ip_proto = int(params.get("ip_proto"))
1194 source_ip = params.get("source_ip")
1195 destination_ip = params.get("destination_ip")
1196 source_port = params.get("source_port")
1197 destination_port = params.get("destination_port")
1198 definition = {"logical_source_port": interfaces[0]}
1199 if ip_proto:
1200 if ip_proto == 1:
1201 ip_proto = 'icmp'
1202 elif ip_proto == 6:
1203 ip_proto = 'tcp'
1204 elif ip_proto == 17:
1205 ip_proto = 'udp'
1206 definition["protocol"] = ip_proto
1207 if source_ip:
1208 if '/' not in source_ip:
1209 source_ip += '/32'
1210 definition["source_ip_prefix"] = source_ip
1211 if source_port:
1212 definition["source_port_range_min"] = source_port
1213 definition["source_port_range_max"] = source_port
1214 if destination_port:
1215 definition["destination_port_range_min"] = destination_port
1216 definition["destination_port_range_max"] = destination_port
1217 if destination_ip:
1218 if '/' not in destination_ip:
1219 destination_ip += '/32'
1220 definition["destination_ip_prefix"] = destination_ip
1221
1222 vim_classification_id = self.vim.new_classification(
1223 name, 'legacy_flow_classifier', definition)
1224
1225 task["extra"]["created"] = True
1226 task["extra"]["vim_status"] = "ACTIVE"
1227 task["error_msg"] = None
1228 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1229 task["vim_id"] = vim_classification_id
1230 instance_element_update = {"status": "ACTIVE", "vim_classification_id": vim_classification_id,
1231 "error_msg": None}
1232 return instance_element_update
1233
1234 except (vimconn.vimconnException, VimThreadException) as e:
1235 self.logger.error("Error creating Classification, task=%s: %s", task_id, str(e))
1236 error_text = self._format_vim_error_msg(str(e))
1237 task["error_msg"] = error_text
1238 task["status"] = "FAILED"
1239 task["vim_id"] = None
1240 instance_element_update = {"status": "VIM_ERROR", "vim_classification_id": None, "error_msg": error_text}
1241 return instance_element_update
1242
1243 def del_classification(self, task):
1244 classification_vim_id = task["vim_id"]
1245 try:
1246 self.vim.delete_classification(classification_vim_id)
1247 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1248 task["error_msg"] = None
1249 return None
1250
1251 except vimconn.vimconnException as e:
1252 task["error_msg"] = self._format_vim_error_msg(str(e))
1253 if isinstance(e, vimconn.vimconnNotFoundException):
1254 # If not found mark as Done and fill error_msg
1255 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1256 return None
1257 task["status"] = "FAILED"
1258 return None
1259
1260 def new_sfp(self, task):
1261 vim_sfp_id = None
1262 try:
1263 task_id = task["instance_action_id"] + "." + str(task["task_index"])
1264 depending_tasks = [task.get("depends").get("TASK-" + str(tsk_id)) for tsk_id in
1265 task.get("extra").get("depends_on")]
1266 error_text = ""
1267 sf_id_list = []
1268 classification_id_list = []
1269 for dep in depending_tasks:
1270 vim_id = dep.get("vim_id")
1271 resource = dep.get("item")
1272 if resource == "instance_sfs":
1273 sf_id_list.append(vim_id)
1274 elif resource == "instance_classifications":
1275 classification_id_list.append(vim_id)
1276
1277 name = "sfp-%s" % task["item_id"][:8]
1278 # By default no form of IETF SFC Encapsulation will be used
1279 vim_sfp_id = self.vim.new_sfp(name, classification_id_list, sf_id_list, sfc_encap=False)
1280
1281 task["extra"]["created"] = True
1282 task["extra"]["vim_status"] = "ACTIVE"
1283 task["error_msg"] = None
1284 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1285 task["vim_id"] = vim_sfp_id
1286 instance_element_update = {"status": "ACTIVE", "vim_sfp_id": vim_sfp_id, "error_msg": None}
1287 return instance_element_update
1288
1289 except (vimconn.vimconnException, VimThreadException) as e:
1290 self.logger.error("Error creating Service Function, task=%s: %s", task_id, str(e))
1291 error_text = self._format_vim_error_msg(str(e))
1292 task["error_msg"] = error_text
1293 task["status"] = "FAILED"
1294 task["vim_id"] = None
1295 instance_element_update = {"status": "VIM_ERROR", "vim_sfp_id": None, "error_msg": error_text}
1296 return instance_element_update
1297
1298 def del_sfp(self, task):
1299 sfp_vim_id = task["vim_id"]
1300 try:
1301 self.vim.delete_sfp(sfp_vim_id)
1302 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1303 task["error_msg"] = None
1304 return None
1305
1306 except vimconn.vimconnException as e:
1307 task["error_msg"] = self._format_vim_error_msg(str(e))
1308 if isinstance(e, vimconn.vimconnNotFoundException):
1309 # If not found mark as Done and fill error_msg
1310 task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1311 return None
1312 task["status"] = "FAILED"
1313 return None