1 # -*- coding: utf-8 -*-
4 # Copyright 2015 Telefonica Investigacion y Desarrollo, S.A.U.
5 # This file is part of openvim
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
12 # http://www.apache.org/licenses/LICENSE-2.0
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
20 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact with: nfvlabs@tid.es
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
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
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
52 depends_on: list with the 'task_index'es of tasks that must be completed before. e.g. a vm creation depends
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
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
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
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.
84 import vimconn_openvim
86 import vimconn_opennebula
87 import vimconn_openstack
91 from db_base
import db_base_Exception
92 from lib_osm_openvim
.ovim
import ovimException
93 from copy
import deepcopy
95 __author__
= "Alfonso Tierno, Pablo Montes"
96 __date__
= "$28-Sep-2017 12:07:15$"
99 "openvim": vimconn_openvim
,
101 "opennebula": vimconn_opennebula
,
102 "openstack": vimconn_openstack
,
103 "vmware": vimconn_vmware
,
108 def is_task_id(task_id
):
109 return task_id
.startswith("TASK-")
112 class VimThreadException(Exception):
116 class VimThreadExceptionNotFound(VimThreadException
):
120 class vim_thread(threading
.Thread
):
121 REFRESH_BUILD
= 5 # 5 seconds
122 REFRESH_ACTIVE
= 60 # 1 minute
124 REFRESH_DELETE
= 3600 * 10
126 def __init__(self
, task_lock
, name
=None, datacenter_name
=None, datacenter_tenant_id
=None,
127 db
=None, db_lock
=None, ovim
=None):
131 'name' name of thread
132 'host','user': host ip or name to manage and user
133 'db', 'db_lock': database class and lock to use it in exclusion
135 threading
.Thread
.__init
__(self
)
137 self
.error_status
= None
138 self
.datacenter_name
= datacenter_name
139 self
.datacenter_tenant_id
= datacenter_tenant_id
142 self
.name
= vimconn
["id"] + "." + vimconn
["config"]["datacenter_tenant_id"]
145 self
.vim_persistent_info
= {}
146 self
.my_id
= self
.name
[:64]
148 self
.logger
= logging
.getLogger('openmano.vim.' + self
.name
)
150 self
.db_lock
= db_lock
152 self
.task_lock
= task_lock
153 self
.task_queue
= Queue
.Queue(2000)
155 def get_vimconnector(self
):
157 from_
= "datacenter_tenants as dt join datacenters as d on dt.datacenter_id=d.uuid"
158 select_
= ('type', 'd.config as config', 'd.uuid as datacenter_id', 'vim_url', 'vim_url_admin',
159 'd.name as datacenter_name', 'dt.uuid as datacenter_tenant_id',
160 'dt.vim_tenant_name as vim_tenant_name', 'dt.vim_tenant_id as vim_tenant_id',
161 'user', 'passwd', 'dt.config as dt_config')
162 where_
= {"dt.uuid": self
.datacenter_tenant_id
}
163 vims
= self
.db
.get_rows(FROM
=from_
, SELECT
=select_
, WHERE
=where_
)
167 vim_config
.update(yaml
.load(vim
["config"]))
169 vim_config
.update(yaml
.load(vim
["dt_config"]))
170 vim_config
['datacenter_tenant_id'] = vim
.get('datacenter_tenant_id')
171 vim_config
['datacenter_id'] = vim
.get('datacenter_id')
175 vim_config
["wim_external_ports"] = self
.ovim
.get_of_port_mappings(
176 db_filter
={"region": vim_config
['datacenter_id'], "pci": None})
178 self
.vim
= vim_module
[vim
["type"]].vimconnector(
179 uuid
=vim
['datacenter_id'], name
=vim
['datacenter_name'],
180 tenant_id
=vim
['vim_tenant_id'], tenant_name
=vim
['vim_tenant_name'],
181 url
=vim
['vim_url'], url_admin
=vim
['vim_url_admin'],
182 user
=vim
['user'], passwd
=vim
['passwd'],
183 config
=vim_config
, persistent_info
=self
.vim_persistent_info
185 self
.error_status
= None
186 except Exception as e
:
187 self
.logger
.error("Cannot load vimconnector for vim_account {}: {}".format(self
.datacenter_tenant_id
, e
))
189 self
.error_status
= "Error loading vimconnector: {}".format(e
)
191 def _get_db_task(self
):
193 Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
201 # get 20 (database_limit) entries each time
202 vim_actions
= self
.db
.get_rows(FROM
="vim_wim_actions",
203 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
204 "status": ['SCHEDULED', 'BUILD', 'DONE'],
205 "worker": [None, self
.my_id
], "modified_at<=": now
207 ORDER_BY
=("modified_at", "created_at",),
208 LIMIT
=database_limit
)
211 # if vim_actions[0]["modified_at"] > now:
212 # return int(vim_actions[0] - now)
213 for task
in vim_actions
:
215 if task_related
== task
["related"]:
216 continue # ignore if a locking has already tried for these task set
217 task_related
= task
["related"]
219 self
.db
.update_rows("vim_wim_actions", UPDATE
={"worker": self
.my_id
}, modified_time
=0,
220 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
221 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
222 "worker": [None, self
.my_id
],
223 "related": task_related
,
224 "item": task
["item"],
226 # ... and read all related and check if locked
227 related_tasks
= self
.db
.get_rows(FROM
="vim_wim_actions",
228 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
229 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
230 "related": task_related
,
231 "item": task
["item"],
233 ORDER_BY
=("created_at",))
234 # check that all related tasks have been locked. If not release and try again. It can happen
235 # for race conditions if a new related task has been inserted by nfvo in the process
236 some_tasks_locked
= False
237 some_tasks_not_locked
= False
239 for relate_task
in related_tasks
:
240 if relate_task
["worker"] != self
.my_id
:
241 some_tasks_not_locked
= True
243 some_tasks_locked
= True
244 if not creation_task
and relate_task
["action"] in ("CREATE", "FIND"):
245 creation_task
= relate_task
246 if some_tasks_not_locked
:
247 if some_tasks_locked
: # unlock
248 self
.db
.update_rows("vim_wim_actions", UPDATE
={"worker": None}, modified_time
=0,
249 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
250 "worker": self
.my_id
,
251 "related": task_related
,
252 "item": task
["item"],
256 # task of creation must be the first in the list of related_task
257 assert(related_tasks
[0]["action"] in ("CREATE", "FIND"))
260 extra
= yaml
.load(task
["extra"])
263 task
["extra"] = extra
264 if extra
.get("depends_on"):
266 if extra
.get("params"):
267 task
["params"] = deepcopy(extra
["params"])
268 return task
, related_tasks
269 except Exception as e
:
270 self
.logger
.critical("Unexpected exception at _get_db_task: " + str(e
), exc_info
=True)
273 def _delete_task(self
, task
):
275 Determine if this task need to be done or superseded
279 def copy_extra_created(copy_to
, copy_from
):
280 copy_to
["created"] = copy_from
["created"]
281 if copy_from
.get("sdn_net_id"):
282 copy_to
["sdn_net_id"] = copy_from
["sdn_net_id"]
283 if copy_from
.get("interfaces"):
284 copy_to
["interfaces"] = copy_from
["interfaces"]
285 if copy_from
.get("created_items"):
286 if not copy_to
.get("created_items"):
287 copy_to
["created_items"] = {}
288 copy_to
["created_items"].update(copy_from
["created_items"])
291 dependency_task
= None
292 deletion_needed
= False
293 if task
["status"] == "FAILED":
294 return # TODO need to be retry??
296 # get all related tasks
297 related_tasks
= self
.db
.get_rows(FROM
="vim_wim_actions",
298 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
299 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
300 "action": ["FIND", "CREATE"],
301 "related": task
["related"],
303 ORDER_BY
=("created_at",),
305 for related_task
in related_tasks
:
306 if related_task
["item"] == task
["item"] and related_task
["item_id"] == task
["item_id"]:
307 task_create
= related_task
309 if related_task
["extra"]:
310 extra_created
= yaml
.load(related_task
["extra"])
311 if extra_created
.get("created"):
312 deletion_needed
= True
313 related_task
["extra"] = extra_created
314 elif not dependency_task
:
315 dependency_task
= related_task
316 if task_create
and dependency_task
:
319 # mark task_create as FINISHED
320 self
.db
.update_rows("vim_wim_actions", UPDATE
={"status": "FINISHED"},
321 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
322 "instance_action_id": task_create
["instance_action_id"],
323 "task_index": task_create
["task_index"]
325 if not deletion_needed
:
327 elif dependency_task
:
328 # move create information from task_create to relate_task
329 extra_new_created
= yaml
.load(dependency_task
["extra"]) or {}
330 extra_new_created
["created"] = extra_created
["created"]
331 copy_extra_created(copy_to
=extra_new_created
, copy_from
=extra_created
)
333 self
.db
.update_rows("vim_wim_actions",
334 UPDATE
={"extra": yaml
.safe_dump(extra_new_created
, default_flow_style
=True,
336 "vim_id": task_create
.get("vim_id")},
337 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
338 "instance_action_id": dependency_task
["instance_action_id"],
339 "task_index": dependency_task
["task_index"]
343 task
["vim_id"] = task_create
["vim_id"]
344 copy_extra_created(copy_to
=task
["extra"], copy_from
=task_create
["extra"])
347 except Exception as e
:
348 self
.logger
.critical("Unexpected exception at _delete_task: " + str(e
), exc_info
=True)
350 def _refres_vm(self
, task
):
351 """Call VIM to get VMs status"""
352 database_update
= None
354 vim_id
= task
["vim_id"]
355 vm_to_refresh_list
= [vim_id
]
357 vim_dict
= self
.vim
.refresh_vms_status(vm_to_refresh_list
)
358 vim_info
= vim_dict
[vim_id
]
359 except vimconn
.vimconnException
as e
:
360 # Mark all tasks at VIM_ERROR status
361 self
.logger
.error("task=several get-VM: vimconnException when trying to refresh vms " + str(e
))
362 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
364 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
365 self
.logger
.debug("task={} get-VM: vim_vm_id={} result={}".format(task_id
, task
["vim_id"], vim_info
))
367 # check and update interfaces
368 task_warning_msg
= ""
369 for interface
in vim_info
.get("interfaces", ()):
370 vim_interface_id
= interface
["vim_interface_id"]
371 if vim_interface_id
not in task
["extra"]["interfaces"]:
372 self
.logger
.critical("task={} get-VM: Interface not found {} on task info {}".format(
373 task_id
, vim_interface_id
, task
["extra"]["interfaces"]), exc_info
=True)
375 task_interface
= task
["extra"]["interfaces"][vim_interface_id
]
376 task_vim_interface
= task_interface
.get("vim_info")
377 if task_vim_interface
!= interface
:
379 if task_interface
.get("sdn_port_id"):
382 self
.ovim
.delete_port(task_interface
["sdn_port_id"], idempotent
=True)
383 task_interface
["sdn_port_id"] = None
384 except ovimException
as e
:
385 error_text
= "ovimException deleting external_port={}: {}".format(
386 task_interface
["sdn_port_id"], e
)
387 self
.logger
.error("task={} get-VM: {}".format(task_id
, error_text
), exc_info
=True)
388 task_warning_msg
+= error_text
389 # TODO Set error_msg at instance_nets instead of instance VMs
392 sdn_net_id
= task_interface
.get("sdn_net_id")
393 if sdn_net_id
and interface
.get("compute_node") and interface
.get("pci"):
394 sdn_port_name
= sdn_net_id
+ "." + task
["vim_id"]
395 sdn_port_name
= sdn_port_name
[:63]
398 sdn_port_id
= self
.ovim
.new_external_port(
399 {"compute_node": interface
["compute_node"],
400 "pci": interface
["pci"],
401 "vlan": interface
.get("vlan"),
402 "net_id": sdn_net_id
,
403 "region": self
.vim
["config"]["datacenter_id"],
404 "name": sdn_port_name
,
405 "mac": interface
.get("mac_address")})
406 task_interface
["sdn_port_id"] = sdn_port_id
407 except (ovimException
, Exception) as e
:
408 error_text
= "ovimException creating new_external_port compute_node={} pci={} vlan={} {}".\
409 format(interface
["compute_node"], interface
["pci"], interface
.get("vlan"), e
)
410 self
.logger
.error("task={} get-VM: {}".format(task_id
, error_text
), exc_info
=True)
411 task_warning_msg
+= error_text
412 # TODO Set error_msg at instance_nets instead of instance VMs
414 self
.db
.update_rows('instance_interfaces',
415 UPDATE
={"mac_address": interface
.get("mac_address"),
416 "ip_address": interface
.get("ip_address"),
417 "vim_interface_id": interface
.get("vim_interface_id"),
418 "vim_info": interface
.get("vim_info"),
419 "sdn_port_id": task_interface
.get("sdn_port_id"),
420 "compute_node": interface
.get("compute_node"),
421 "pci": interface
.get("pci"),
422 "vlan": interface
.get("vlan")},
423 WHERE
={'uuid': task_interface
["iface_id"]})
424 task_interface
["vim_info"] = interface
426 # check and update task and instance_vms database
427 vim_info_error_msg
= None
428 if vim_info
.get("error_msg"):
429 vim_info_error_msg
= self
._format
_vim
_error
_msg
(vim_info
["error_msg"] + task_warning_msg
)
430 elif task_warning_msg
:
431 vim_info_error_msg
= self
._format
_vim
_error
_msg
(task_warning_msg
)
432 task_vim_info
= task
["extra"].get("vim_info")
433 task_error_msg
= task
.get("error_msg")
434 task_vim_status
= task
["extra"].get("vim_status")
435 if task_vim_status
!= vim_info
["status"] or task_error_msg
!= vim_info_error_msg
or \
436 (vim_info
.get("vim_info") and task_vim_info
!= vim_info
["vim_info"]):
437 database_update
= {"status": vim_info
["status"], "error_msg": vim_info_error_msg
}
438 if vim_info
.get("vim_info"):
439 database_update
["vim_info"] = vim_info
["vim_info"]
441 task
["extra"]["vim_status"] = vim_info
["status"]
442 task
["error_msg"] = vim_info_error_msg
443 if vim_info
.get("vim_info"):
444 task
["extra"]["vim_info"] = vim_info
["vim_info"]
446 return database_update
448 def _refres_net(self
, task
):
449 """Call VIM to get network status"""
450 database_update
= None
452 vim_id
= task
["vim_id"]
453 net_to_refresh_list
= [vim_id
]
455 vim_dict
= self
.vim
.refresh_nets_status(net_to_refresh_list
)
456 vim_info
= vim_dict
[vim_id
]
457 except vimconn
.vimconnException
as e
:
458 # Mark all tasks at VIM_ERROR status
459 self
.logger
.error("task=several get-net: vimconnException when trying to refresh nets " + str(e
))
460 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
462 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
463 self
.logger
.debug("task={} get-net: vim_net_id={} result={}".format(task_id
, task
["vim_id"], vim_info
))
465 task_vim_info
= task
["extra"].get("vim_info")
466 task_vim_status
= task
["extra"].get("vim_status")
467 task_error_msg
= task
.get("error_msg")
468 task_sdn_net_id
= task
["extra"].get("sdn_net_id")
470 vim_info_status
= vim_info
["status"]
471 vim_info_error_msg
= vim_info
.get("error_msg")
476 sdn_net
= self
.ovim
.show_network(task_sdn_net_id
)
477 except (ovimException
, Exception) as e
:
478 text_error
= "ovimException getting network snd_net_id={}: {}".format(task_sdn_net_id
, e
)
479 self
.logger
.error("task={} get-net: {}".format(task_id
, text_error
), exc_info
=True)
480 sdn_net
= {"status": "ERROR", "last_error": text_error
}
481 if sdn_net
["status"] == "ERROR":
482 if not vim_info_error_msg
:
483 vim_info_error_msg
= str(sdn_net
.get("last_error"))
485 vim_info_error_msg
= "VIM_ERROR: {} && SDN_ERROR: {}".format(
486 self
._format
_vim
_error
_msg
(vim_info_error_msg
, 1024 // 2 - 14),
487 self
._format
_vim
_error
_msg
(sdn_net
["last_error"], 1024 // 2 - 14))
488 vim_info_status
= "ERROR"
489 elif sdn_net
["status"] == "BUILD":
490 if vim_info_status
== "ACTIVE":
491 vim_info_status
= "BUILD"
494 if vim_info_error_msg
:
495 vim_info_error_msg
= self
._format
_vim
_error
_msg
(vim_info_error_msg
)
496 if task_vim_status
!= vim_info_status
or task_error_msg
!= vim_info_error_msg
or \
497 (vim_info
.get("vim_info") and task_vim_info
!= vim_info
["vim_info"]):
498 task
["extra"]["vim_status"] = vim_info_status
499 task
["error_msg"] = vim_info_error_msg
500 if vim_info
.get("vim_info"):
501 task
["extra"]["vim_info"] = vim_info
["vim_info"]
502 database_update
= {"status": vim_info_status
, "error_msg": vim_info_error_msg
}
503 if vim_info
.get("vim_info"):
504 database_update
["vim_info"] = vim_info
["vim_info"]
505 return database_update
507 def _proccess_pending_tasks(self
, task
, related_tasks
):
508 old_task_status
= task
["status"]
509 create_or_find
= False # if as result of processing this task something is created or found
513 if task
["status"] == "SCHEDULED":
514 # check if tasks that this depends on have been completed
515 dependency_not_completed
= False
516 dependency_modified_at
= 0
517 for task_index
in task
["extra"].get("depends_on", ()):
518 task_dependency
= self
._look
_for
_task
(task
["instance_action_id"], task_index
)
519 if not task_dependency
:
520 raise VimThreadException(
521 "Cannot get depending net task trying to get depending task {}.{}".format(
522 task
["instance_action_id"], task_index
))
523 # task["depends"]["TASK-" + str(task_index)] = task_dependency #it references another object,so
524 # database must be look again
525 if task_dependency
["status"] == "SCHEDULED":
526 dependency_not_completed
= True
527 dependency_modified_at
= task_dependency
["modified_at"]
529 elif task_dependency
["status"] == "FAILED":
530 raise VimThreadException(
531 "Cannot {} {}, (task {}.{}) because depends on failed {}.{}, (task{}.{}): {}".format(
532 task
["action"], task
["item"],
533 task
["instance_action_id"], task
["task_index"],
534 task_dependency
["instance_action_id"], task_dependency
["task_index"],
535 task_dependency
["action"], task_dependency
["item"], task_dependency
.get("error_msg")))
537 task
["depends"]["TASK-"+str(task_index
)] = task_dependency
538 task
["depends"]["TASK-{}.{}".format(task
["instance_action_id"], task_index
)] = task_dependency
539 if dependency_not_completed
:
540 # Move this task to the time dependency is going to be modified plus 10 seconds.
541 self
.db
.update_rows("vim_wim_actions", modified_time
=dependency_modified_at
+ 10,
542 UPDATE
={"worker": None},
543 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
, "worker": self
.my_id
,
544 "related": task
["related"],
546 # task["extra"]["tries"] = task["extra"].get("tries", 0) + 1
547 # if task["extra"]["tries"] > 3:
548 # raise VimThreadException(
549 # "Cannot {} {}, (task {}.{}) because timeout waiting to complete {} {}, "
550 # "(task {}.{})".format(task["action"], task["item"],
551 # task["instance_action_id"], task["task_index"],
552 # task_dependency["instance_action_id"], task_dependency["task_index"]
553 # task_dependency["action"], task_dependency["item"]))
556 database_update
= None
557 if task
["action"] == "DELETE":
558 deleted_needed
= self
._delete
_task
(task
)
559 if not deleted_needed
:
560 task
["status"] = "SUPERSEDED" # with FINISHED instead of DONE it will not be refreshing
561 task
["error_msg"] = None
563 if task
["status"] == "SUPERSEDED":
564 # not needed to do anything but update database with the new status
565 database_update
= None
567 task
["status"] = "FAILED"
568 task
["error_msg"] = self
.error_status
569 database_update
= {"status": "VIM_ERROR", "error_msg": task
["error_msg"]}
570 elif task
["item_id"] != related_tasks
[0]["item_id"] and task
["action"] in ("FIND", "CREATE"):
571 # Do nothing, just copy values from one to another and updata database
572 task
["status"] = related_tasks
[0]["status"]
573 task
["error_msg"] = related_tasks
[0]["error_msg"]
574 task
["vim_id"] = related_tasks
[0]["vim_id"]
575 extra
= yaml
.load(related_tasks
[0]["extra"])
576 task
["extra"]["vim_status"] = extra
["vim_status"]
577 next_refresh
= related_tasks
[0]["modified_at"] + 0.001
578 database_update
= {"status": task
["extra"].get("vim_status", "VIM_ERROR"),
579 "error_msg": task
["error_msg"]}
580 if task
["item"] == 'instance_vms':
581 database_update
["vim_vm_id"] = task
["vim_id"]
582 elif task
["item"] == 'instance_nets':
583 database_update
["vim_net_id"] = task
["vim_id"]
584 elif task
["item"] == 'instance_vms':
585 if task
["status"] in ('BUILD', 'DONE') and task
["action"] in ("FIND", "CREATE"):
586 database_update
= self
._refres
_vm
(task
)
587 create_or_find
= True
588 elif task
["action"] == "CREATE":
589 create_or_find
= True
590 database_update
= self
.new_vm(task
)
591 elif task
["action"] == "DELETE":
594 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
595 elif task
["item"] == 'instance_nets':
596 if task
["status"] in ('BUILD', 'DONE') and task
["action"] in ("FIND", "CREATE"):
597 database_update
= self
._refres
_net
(task
)
598 create_or_find
= True
599 elif task
["action"] == "CREATE":
600 create_or_find
= True
601 database_update
= self
.new_net(task
)
602 elif task
["action"] == "DELETE":
604 elif task
["action"] == "FIND":
605 database_update
= self
.get_net(task
)
607 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
608 elif task
["item"] == 'instance_sfis':
609 if task
["action"] == "CREATE":
610 create_or_find
= True
611 database_update
= self
.new_sfi(task
)
612 elif task
["action"] == "DELETE":
615 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
616 elif task
["item"] == 'instance_sfs':
617 if task
["action"] == "CREATE":
618 create_or_find
= True
619 database_update
= self
.new_sf(task
)
620 elif task
["action"] == "DELETE":
623 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
624 elif task
["item"] == 'instance_classifications':
625 if task
["action"] == "CREATE":
626 create_or_find
= True
627 database_update
= self
.new_classification(task
)
628 elif task
["action"] == "DELETE":
629 self
.del_classification(task
)
631 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
632 elif task
["item"] == 'instance_sfps':
633 if task
["action"] == "CREATE":
634 create_or_find
= True
635 database_update
= self
.new_sfp(task
)
636 elif task
["action"] == "DELETE":
639 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
641 raise vimconn
.vimconnException(self
.name
+ "unknown task item {}".format(task
["item"]))
643 except VimThreadException
as e
:
644 task
["error_msg"] = str(e
)
645 task
["status"] = "FAILED"
646 database_update
= {"status": "VIM_ERROR", "error_msg": task
["error_msg"]}
647 if task
["item"] == 'instance_vms':
648 database_update
["vim_vm_id"] = None
649 elif task
["item"] == 'instance_nets':
650 database_update
["vim_net_id"] = None
652 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
653 self
.logger
.debug("task={} item={} action={} result={}:'{}' params={}".format(
654 task_id
, task
["item"], task
["action"], task
["status"],
655 task
["vim_id"] if task
["status"] == "DONE" else task
.get("error_msg"), task
["params"]))
658 if task
["status"] == "DONE":
659 next_refresh
= time
.time()
660 if task
["extra"].get("vim_status") == "BUILD":
661 next_refresh
+= self
.REFRESH_BUILD
662 elif task
["extra"].get("vim_status") in ("ERROR", "VIM_ERROR"):
663 next_refresh
+= self
.REFRESH_ERROR
664 elif task
["extra"].get("vim_status") == "DELETED":
665 next_refresh
+= self
.REFRESH_DELETE
667 next_refresh
+= self
.REFRESH_ACTIVE
668 elif task
["status"] == "FAILED":
669 next_refresh
= time
.time() + self
.REFRESH_DELETE
672 # modify all related task with action FIND/CREATED non SCHEDULED
674 table
="vim_wim_actions", modified_time
=next_refresh
+ 0.001,
675 UPDATE
={"status": task
["status"], "vim_id": task
.get("vim_id"),
676 "error_msg": task
["error_msg"],
679 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
680 "worker": self
.my_id
,
681 "action": ["FIND", "CREATE"],
682 "related": task
["related"],
683 "status<>": "SCHEDULED",
687 table
="vim_wim_actions", modified_time
=next_refresh
,
688 UPDATE
={"status": task
["status"], "vim_id": task
.get("vim_id"),
689 "error_msg": task
["error_msg"],
690 "extra": yaml
.safe_dump(task
["extra"], default_flow_style
=True, width
=256)},
691 WHERE
={"instance_action_id": task
["instance_action_id"], "task_index": task
["task_index"]})
694 table
="vim_wim_actions", modified_time
=0,
695 UPDATE
={"worker": None},
696 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
697 "worker": self
.my_id
,
698 "related": task
["related"],
701 # Update table instance_actions
702 if old_task_status
== "SCHEDULED" and task
["status"] != old_task_status
:
704 table
="instance_actions",
705 UPDATE
={("number_failed" if task
["status"] == "FAILED" else "number_done"): {"INCREMENT": 1}},
706 WHERE
={"uuid": task
["instance_action_id"]})
708 self
.db
.update_rows(table
=task
["item"],
709 UPDATE
=database_update
,
710 WHERE
={"related": task
["related"]})
711 except db_base_Exception
as e
:
712 self
.logger
.error("task={} Error updating database {}".format(task_id
, e
), exc_info
=True)
714 def insert_task(self
, task
):
716 self
.task_queue
.put(task
, False)
719 raise vimconn
.vimconnException(self
.name
+ ": timeout inserting a task")
721 def del_task(self
, task
):
723 if task
["status"] == "SCHEDULED":
724 task
["status"] = "SUPERSEDED"
726 else: # task["status"] == "processing"
727 self
.task_lock
.release()
731 self
.logger
.debug("Starting")
733 self
.get_vimconnector()
734 self
.logger
.debug("Vimconnector loaded")
735 reload_thread
= False
739 while not self
.task_queue
.empty():
740 task
= self
.task_queue
.get()
741 if isinstance(task
, list):
743 elif isinstance(task
, str):
746 elif task
== 'reload':
749 self
.task_queue
.task_done()
753 task
, related_tasks
= self
._get
_db
_task
()
755 self
._proccess
_pending
_tasks
(task
, related_tasks
)
759 except Exception as e
:
760 self
.logger
.critical("Unexpected exception at run: " + str(e
), exc_info
=True)
762 self
.logger
.debug("Finishing")
764 def _look_for_task(self
, instance_action_id
, task_id
):
766 Look for a concrete task at vim_actions database table
767 :param instance_action_id: The instance_action_id
768 :param task_id: Can have several formats:
769 <task index>: integer
770 TASK-<task index> :backward compatibility,
771 [TASK-]<instance_action_id>.<task index>: this instance_action_id overrides the one in the parameter
772 :return: Task dictionary or None if not found
774 if isinstance(task_id
, int):
777 if task_id
.startswith("TASK-"):
778 task_id
= task_id
[5:]
779 ins_action_id
, _
, task_index
= task_id
.rpartition(".")
781 instance_action_id
= ins_action_id
783 tasks
= self
.db
.get_rows(FROM
="vim_wim_actions", WHERE
={"instance_action_id": instance_action_id
,
784 "task_index": task_index
})
788 task
["params"] = None
791 extra
= yaml
.load(task
["extra"])
792 task
["extra"] = extra
793 task
["params"] = extra
.get("params")
799 def _format_vim_error_msg(error_text
, max_length
=1024):
800 if error_text
and len(error_text
) >= max_length
:
801 return error_text
[:max_length
// 2 - 3] + " ... " + error_text
[-max_length
// 2 + 3:]
804 def new_vm(self
, task
):
805 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
807 params
= task
["params"]
808 depends
= task
.get("depends")
811 if "net_id" in net
and is_task_id(net
["net_id"]): # change task_id into network_id
812 network_id
= task
["depends"][net
["net_id"]].get("vim_id")
814 raise VimThreadException(
815 "Cannot create VM because depends on a network not created or found: " +
816 str(depends
[net
["net_id"]]["error_msg"]))
817 net
["net_id"] = network_id
818 params_copy
= deepcopy(params
)
819 vim_vm_id
, created_items
= self
.vim
.new_vminstance(*params_copy
)
821 # fill task_interfaces. Look for snd_net_id at database for each interface
823 for iface
in params_copy
[5]:
824 task_interfaces
[iface
["vim_id"]] = {"iface_id": iface
["uuid"]}
825 result
= self
.db
.get_rows(
826 SELECT
=('sdn_net_id', 'interface_id'),
827 FROM
='instance_nets as ine join instance_interfaces as ii on ii.instance_net_id=ine.uuid',
828 WHERE
={'ii.uuid': iface
["uuid"]})
830 task_interfaces
[iface
["vim_id"]]["sdn_net_id"] = result
[0]['sdn_net_id']
831 task_interfaces
[iface
["vim_id"]]["interface_id"] = result
[0]['interface_id']
833 self
.logger
.critical("task={} new-VM: instance_nets uuid={} not found at DB".format(task_id
,
837 task
["vim_info"] = {}
838 task
["extra"]["interfaces"] = task_interfaces
839 task
["extra"]["created"] = True
840 task
["extra"]["created_items"] = created_items
841 task
["extra"]["vim_status"] = "BUILD"
842 task
["error_msg"] = None
843 task
["status"] = "DONE"
844 task
["vim_id"] = vim_vm_id
845 instance_element_update
= {"status": "BUILD", "vim_vm_id": vim_vm_id
, "error_msg": None}
846 return instance_element_update
848 except (vimconn
.vimconnException
, VimThreadException
) as e
:
849 self
.logger
.error("task={} new-VM: {}".format(task_id
, e
))
850 error_text
= self
._format
_vim
_error
_msg
(str(e
))
851 task
["error_msg"] = error_text
852 task
["status"] = "FAILED"
853 task
["vim_id"] = None
854 instance_element_update
= {"status": "VIM_ERROR", "vim_vm_id": None, "error_msg": error_text
}
855 return instance_element_update
857 def del_vm(self
, task
):
858 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
859 vm_vim_id
= task
["vim_id"]
860 interfaces
= task
["extra"].get("interfaces", ())
862 for iface
in interfaces
.values():
863 if iface
.get("sdn_port_id"):
866 self
.ovim
.delete_port(iface
["sdn_port_id"], idempotent
=True)
867 except ovimException
as e
:
868 self
.logger
.error("task={} del-VM: ovimException when deleting external_port={}: {} ".format(
869 task_id
, iface
["sdn_port_id"], e
), exc_info
=True)
870 # TODO Set error_msg at instance_nets
872 self
.vim
.delete_vminstance(vm_vim_id
, task
["extra"].get("created_items"))
873 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
874 task
["error_msg"] = None
877 except vimconn
.vimconnException
as e
:
878 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
879 if isinstance(e
, vimconn
.vimconnNotFoundException
):
880 # If not found mark as Done and fill error_msg
881 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
883 task
["status"] = "FAILED"
886 def _get_net_internal(self
, task
, filter_param
):
888 Common code for get_net and new_net. It looks for a network on VIM with the filter_params
889 :param task: task for this find or find-or-create action
890 :param filter_param: parameters to send to the vimconnector
891 :return: a dict with the content to update the instance_nets database table. Raises an exception on error, or
892 when network is not found or found more than one
894 vim_nets
= self
.vim
.get_network_list(filter_param
)
896 raise VimThreadExceptionNotFound("Network not found with this criteria: '{}'".format(filter_param
))
897 elif len(vim_nets
) > 1:
898 raise VimThreadException("More than one network found with this criteria: '{}'".format(filter_param
))
899 vim_net_id
= vim_nets
[0]["id"]
901 # Discover if this network is managed by a sdn controller
903 result
= self
.db
.get_rows(SELECT
=('sdn_net_id',), FROM
='instance_nets',
904 WHERE
={'vim_net_id': vim_net_id
, 'datacenter_tenant_id': self
.datacenter_tenant_id
},
905 ORDER
="instance_scenario_id")
907 sdn_net_id
= result
[0]['sdn_net_id']
909 task
["status"] = "DONE"
910 task
["extra"]["vim_info"] = {}
911 task
["extra"]["created"] = False
912 task
["extra"]["vim_status"] = "BUILD"
913 task
["extra"]["sdn_net_id"] = sdn_net_id
914 task
["error_msg"] = None
915 task
["vim_id"] = vim_net_id
916 instance_element_update
= {"vim_net_id": vim_net_id
, "created": False, "status": "BUILD",
917 "error_msg": None, "sdn_net_id": sdn_net_id
}
918 return instance_element_update
920 def get_net(self
, task
):
921 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
923 params
= task
["params"]
924 filter_param
= params
[0]
925 instance_element_update
= self
._get
_net
_internal
(task
, filter_param
)
926 return instance_element_update
928 except (vimconn
.vimconnException
, VimThreadException
) as e
:
929 self
.logger
.error("task={} get-net: {}".format(task_id
, e
))
930 task
["status"] = "FAILED"
931 task
["vim_id"] = None
932 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
933 instance_element_update
= {"vim_net_id": None, "status": "VIM_ERROR",
934 "error_msg": task
["error_msg"]}
935 return instance_element_update
937 def new_net(self
, task
):
940 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
944 if task
["extra"].get("find"):
945 action_text
= "finding"
946 filter_param
= task
["extra"]["find"][0]
948 instance_element_update
= self
._get
_net
_internal
(task
, filter_param
)
949 return instance_element_update
950 except VimThreadExceptionNotFound
:
953 params
= task
["params"]
954 action_text
= "creating VIM"
955 vim_net_id
, created_items
= self
.vim
.new_network(*params
[0:3])
959 wim_account_name
= None
961 wim_account_name
= params
[3]
963 sdn_controller
= self
.vim
.config
.get('sdn-controller')
964 if sdn_controller
and (net_type
== "data" or net_type
== "ptp"):
965 network
= {"name": net_name
, "type": net_type
, "region": self
.vim
["config"]["datacenter_id"]}
967 vim_net
= self
.vim
.get_network(vim_net_id
)
968 if vim_net
.get('encapsulation') != 'vlan':
969 raise vimconn
.vimconnException(
970 "net '{}' defined as type '{}' has not vlan encapsulation '{}'".format(
971 net_name
, net_type
, vim_net
['encapsulation']))
972 network
["vlan"] = vim_net
.get('segmentation_id')
973 action_text
= "creating SDN"
975 sdn_net_id
= self
.ovim
.new_network(network
)
977 if wim_account_name
and self
.vim
.config
["wim_external_ports"]:
978 # add external port to connect WIM. Try with compute node __WIM:wim_name and __WIM
979 action_text
= "attaching external port to ovim network"
980 sdn_port_name
= sdn_net_id
+ "." + task
["vim_id"]
981 sdn_port_name
= sdn_port_name
[:63]
983 "compute_node": "__WIM:" + wim_account_name
[0:58],
985 "vlan": network
["vlan"],
986 "net_id": sdn_net_id
,
987 "region": self
.vim
["config"]["datacenter_id"],
988 "name": sdn_port_name
,
992 sdn_external_port_id
= self
.ovim
.new_external_port(sdn_port_data
)
993 except ovimException
:
994 sdn_port_data
["compute_node"] = "__WIM"
996 sdn_external_port_id
= self
.ovim
.new_external_port(sdn_port_data
)
997 self
.logger
.debug("Added sdn_external_port {} to sdn_network {}".format(sdn_external_port_id
,
999 task
["status"] = "DONE"
1000 task
["extra"]["vim_info"] = {}
1001 task
["extra"]["sdn_net_id"] = sdn_net_id
1002 task
["extra"]["vim_status"] = "BUILD"
1003 task
["extra"]["created"] = True
1004 task
["extra"]["created_items"] = created_items
1005 task
["error_msg"] = None
1006 task
["vim_id"] = vim_net_id
1007 instance_element_update
= {"vim_net_id": vim_net_id
, "sdn_net_id": sdn_net_id
, "status": "BUILD",
1008 "created": True, "error_msg": None}
1009 return instance_element_update
1010 except (vimconn
.vimconnException
, ovimException
) as e
:
1011 self
.logger
.error("task={} new-net: Error {}: {}".format(task_id
, action_text
, e
))
1012 task
["status"] = "FAILED"
1013 task
["vim_id"] = vim_net_id
1014 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1015 task
["extra"]["sdn_net_id"] = sdn_net_id
1016 instance_element_update
= {"vim_net_id": vim_net_id
, "sdn_net_id": sdn_net_id
, "status": "VIM_ERROR",
1017 "error_msg": task
["error_msg"]}
1018 return instance_element_update
1020 def del_net(self
, task
):
1021 net_vim_id
= task
["vim_id"]
1022 sdn_net_id
= task
["extra"].get("sdn_net_id")
1025 # Delete any attached port to this sdn network. There can be ports associated to this network in case
1026 # it was manually done using 'openmano vim-net-sdn-attach'
1028 port_list
= self
.ovim
.get_ports(columns
={'uuid'},
1029 filter={'name': 'external_port', 'net_id': sdn_net_id
})
1030 for port
in port_list
:
1031 self
.ovim
.delete_port(port
['uuid'], idempotent
=True)
1032 self
.ovim
.delete_network(sdn_net_id
, idempotent
=True)
1034 self
.vim
.delete_network(net_vim_id
, task
["extra"].get("created_items"))
1035 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1036 task
["error_msg"] = None
1038 except ovimException
as e
:
1039 task
["error_msg"] = self
._format
_vim
_error
_msg
("ovimException obtaining and deleting external "
1040 "ports for net {}: {}".format(sdn_net_id
, str(e
)))
1041 except vimconn
.vimconnException
as e
:
1042 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1043 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1044 # If not found mark as Done and fill error_msg
1045 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1047 task
["status"] = "FAILED"
1050 # Service Function Instances
1051 def new_sfi(self
, task
):
1054 # Waits for interfaces to be ready (avoids failure)
1056 dep_id
= "TASK-" + str(task
["extra"]["depends_on"][0])
1057 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1059 interfaces
= task
["depends"][dep_id
]["extra"].get("interfaces")
1061 ingress_interface_id
= task
.get("extra").get("params").get("ingress_interface_id")
1062 egress_interface_id
= task
.get("extra").get("params").get("egress_interface_id")
1063 ingress_vim_interface_id
= None
1064 egress_vim_interface_id
= None
1065 for vim_interface
, interface_data
in interfaces
.iteritems():
1066 if interface_data
.get("interface_id") == ingress_interface_id
:
1067 ingress_vim_interface_id
= vim_interface
1069 if ingress_interface_id
!= egress_interface_id
:
1070 for vim_interface
, interface_data
in interfaces
.iteritems():
1071 if interface_data
.get("interface_id") == egress_interface_id
:
1072 egress_vim_interface_id
= vim_interface
1075 egress_vim_interface_id
= ingress_vim_interface_id
1076 if not ingress_vim_interface_id
or not egress_vim_interface_id
:
1077 error_text
= "Error creating Service Function Instance, Ingress: {}, Egress: {}".format(
1078 ingress_vim_interface_id
, egress_vim_interface_id
)
1079 self
.logger
.error(error_text
)
1080 task
["error_msg"] = error_text
1081 task
["status"] = "FAILED"
1082 task
["vim_id"] = None
1084 # At the moment, every port associated with the VM will be used both as ingress and egress ports.
1085 # Bear in mind that different VIM connectors might support SFI differently. In the case of OpenStack,
1086 # only the first ingress and first egress ports will be used to create the SFI (Port Pair).
1087 ingress_port_id_list
= [ingress_vim_interface_id
]
1088 egress_port_id_list
= [egress_vim_interface_id
]
1089 name
= "sfi-%s" % task
["item_id"][:8]
1090 # By default no form of IETF SFC Encapsulation will be used
1091 vim_sfi_id
= self
.vim
.new_sfi(name
, ingress_port_id_list
, egress_port_id_list
, sfc_encap
=False)
1093 task
["extra"]["created"] = True
1094 task
["extra"]["vim_status"] = "ACTIVE"
1095 task
["error_msg"] = None
1096 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1097 task
["vim_id"] = vim_sfi_id
1098 instance_element_update
= {"status": "ACTIVE", "vim_sfi_id": vim_sfi_id
, "error_msg": None}
1099 return instance_element_update
1101 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1102 self
.logger
.error("Error creating Service Function Instance, task=%s: %s", task_id
, str(e
))
1103 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1104 task
["error_msg"] = error_text
1105 task
["status"] = "FAILED"
1106 task
["vim_id"] = None
1107 instance_element_update
= {"status": "VIM_ERROR", "vim_sfi_id": None, "error_msg": error_text
}
1108 return instance_element_update
1110 def del_sfi(self
, task
):
1111 sfi_vim_id
= task
["vim_id"]
1113 self
.vim
.delete_sfi(sfi_vim_id
)
1114 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1115 task
["error_msg"] = None
1118 except vimconn
.vimconnException
as e
:
1119 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1120 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1121 # If not found mark as Done and fill error_msg
1122 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1124 task
["status"] = "FAILED"
1127 def new_sf(self
, task
):
1130 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1132 depending_tasks
= ["TASK-" + str(dep_id
) for dep_id
in task
["extra"]["depends_on"]]
1133 # sfis = task.get("depends").values()[0].get("extra").get("params")[5]
1134 sfis
= [task
.get("depends").get(dep_task
) for dep_task
in depending_tasks
]
1137 sfi_id_list
.append(sfi
.get("vim_id"))
1138 name
= "sf-%s" % task
["item_id"][:8]
1139 # By default no form of IETF SFC Encapsulation will be used
1140 vim_sf_id
= self
.vim
.new_sf(name
, sfi_id_list
, sfc_encap
=False)
1142 task
["extra"]["created"] = True
1143 task
["extra"]["vim_status"] = "ACTIVE"
1144 task
["error_msg"] = None
1145 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1146 task
["vim_id"] = vim_sf_id
1147 instance_element_update
= {"status": "ACTIVE", "vim_sf_id": vim_sf_id
, "error_msg": None}
1148 return instance_element_update
1150 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1151 self
.logger
.error("Error creating Service Function, task=%s: %s", task_id
, str(e
))
1152 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1153 task
["error_msg"] = error_text
1154 task
["status"] = "FAILED"
1155 task
["vim_id"] = None
1156 instance_element_update
= {"status": "VIM_ERROR", "vim_sf_id": None, "error_msg": error_text
}
1157 return instance_element_update
1159 def del_sf(self
, task
):
1160 sf_vim_id
= task
["vim_id"]
1162 self
.vim
.delete_sf(sf_vim_id
)
1163 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1164 task
["error_msg"] = None
1167 except vimconn
.vimconnException
as e
:
1168 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1169 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1170 # If not found mark as Done and fill error_msg
1171 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1173 task
["status"] = "FAILED"
1176 def new_classification(self
, task
):
1177 vim_classification_id
= None
1179 params
= task
["params"]
1180 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1181 dep_id
= "TASK-" + str(task
["extra"]["depends_on"][0])
1183 interfaces
= task
.get("depends").get(dep_id
).get("extra").get("interfaces").keys()
1184 # Bear in mind that different VIM connectors might support Classifications differently.
1185 # In the case of OpenStack, only the first VNF attached to the classifier will be used
1186 # to create the Classification(s) (the "logical source port" of the "Flow Classifier").
1187 # Since the VNFFG classifier match lacks the ethertype, classification defaults to
1188 # using the IPv4 flow classifier.
1189 name
= "c-%s" % task
["item_id"][:8]
1190 # if not CIDR is given for the IP addresses, add /32:
1191 ip_proto
= int(params
.get("ip_proto"))
1192 source_ip
= params
.get("source_ip")
1193 destination_ip
= params
.get("destination_ip")
1194 source_port
= params
.get("source_port")
1195 destination_port
= params
.get("destination_port")
1196 definition
= {"logical_source_port": interfaces
[0]}
1202 elif ip_proto
== 17:
1204 definition
["protocol"] = ip_proto
1206 if '/' not in source_ip
:
1208 definition
["source_ip_prefix"] = source_ip
1210 definition
["source_port_range_min"] = source_port
1211 definition
["source_port_range_max"] = source_port
1212 if destination_port
:
1213 definition
["destination_port_range_min"] = destination_port
1214 definition
["destination_port_range_max"] = destination_port
1216 if '/' not in destination_ip
:
1217 destination_ip
+= '/32'
1218 definition
["destination_ip_prefix"] = destination_ip
1220 vim_classification_id
= self
.vim
.new_classification(
1221 name
, 'legacy_flow_classifier', definition
)
1223 task
["extra"]["created"] = True
1224 task
["extra"]["vim_status"] = "ACTIVE"
1225 task
["error_msg"] = None
1226 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1227 task
["vim_id"] = vim_classification_id
1228 instance_element_update
= {"status": "ACTIVE", "vim_classification_id": vim_classification_id
,
1230 return instance_element_update
1232 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1233 self
.logger
.error("Error creating Classification, task=%s: %s", task_id
, str(e
))
1234 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1235 task
["error_msg"] = error_text
1236 task
["status"] = "FAILED"
1237 task
["vim_id"] = None
1238 instance_element_update
= {"status": "VIM_ERROR", "vim_classification_id": None, "error_msg": error_text
}
1239 return instance_element_update
1241 def del_classification(self
, task
):
1242 classification_vim_id
= task
["vim_id"]
1244 self
.vim
.delete_classification(classification_vim_id
)
1245 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1246 task
["error_msg"] = None
1249 except vimconn
.vimconnException
as e
:
1250 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1251 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1252 # If not found mark as Done and fill error_msg
1253 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1255 task
["status"] = "FAILED"
1258 def new_sfp(self
, task
):
1261 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1262 depending_tasks
= [task
.get("depends").get("TASK-" + str(tsk_id
)) for tsk_id
in
1263 task
.get("extra").get("depends_on")]
1266 classification_id_list
= []
1267 for dep
in depending_tasks
:
1268 vim_id
= dep
.get("vim_id")
1269 resource
= dep
.get("item")
1270 if resource
== "instance_sfs":
1271 sf_id_list
.append(vim_id
)
1272 elif resource
== "instance_classifications":
1273 classification_id_list
.append(vim_id
)
1275 name
= "sfp-%s" % task
["item_id"][:8]
1276 # By default no form of IETF SFC Encapsulation will be used
1277 vim_sfp_id
= self
.vim
.new_sfp(name
, classification_id_list
, sf_id_list
, sfc_encap
=False)
1279 task
["extra"]["created"] = True
1280 task
["extra"]["vim_status"] = "ACTIVE"
1281 task
["error_msg"] = None
1282 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1283 task
["vim_id"] = vim_sfp_id
1284 instance_element_update
= {"status": "ACTIVE", "vim_sfp_id": vim_sfp_id
, "error_msg": None}
1285 return instance_element_update
1287 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1288 self
.logger
.error("Error creating Service Function, task=%s: %s", task_id
, str(e
))
1289 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1290 task
["error_msg"] = error_text
1291 task
["status"] = "FAILED"
1292 task
["vim_id"] = None
1293 instance_element_update
= {"status": "VIM_ERROR", "vim_sfp_id": None, "error_msg": error_text
}
1294 return instance_element_update
1296 def del_sfp(self
, task
):
1297 sfp_vim_id
= task
["vim_id"]
1299 self
.vim
.delete_sfp(sfp_vim_id
)
1300 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1301 task
["error_msg"] = None
1304 except vimconn
.vimconnException
as e
:
1305 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1306 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1307 # If not found mark as Done and fill error_msg
1308 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1310 task
["status"] = "FAILED"