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
92 from db_base
import db_base_Exception
93 from lib_osm_openvim
.ovim
import ovimException
94 from copy
import deepcopy
96 __author__
= "Alfonso Tierno, Pablo Montes"
97 __date__
= "$28-Sep-2017 12:07:15$"
100 "openvim": vimconn_openvim
,
102 "opennebula": vimconn_opennebula
,
103 "openstack": vimconn_openstack
,
104 "vmware": vimconn_vmware
,
106 "azure": vimconn_azure
,
110 def is_task_id(task_id
):
111 return task_id
.startswith("TASK-")
114 class VimThreadException(Exception):
118 class VimThreadExceptionNotFound(VimThreadException
):
122 class vim_thread(threading
.Thread
):
123 REFRESH_BUILD
= 5 # 5 seconds
124 REFRESH_ACTIVE
= 60 # 1 minute
126 REFRESH_DELETE
= 3600 * 10
128 def __init__(self
, task_lock
, name
=None, datacenter_name
=None, datacenter_tenant_id
=None,
129 db
=None, db_lock
=None, ovim
=None):
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
137 threading
.Thread
.__init
__(self
)
139 self
.error_status
= None
140 self
.datacenter_name
= datacenter_name
141 self
.datacenter_tenant_id
= datacenter_tenant_id
144 self
.name
= vimconn
["id"] + "." + vimconn
["config"]["datacenter_tenant_id"]
147 self
.vim_persistent_info
= {}
148 self
.my_id
= self
.name
[:64]
150 self
.logger
= logging
.getLogger('openmano.vim.' + self
.name
)
152 self
.db_lock
= db_lock
154 self
.task_lock
= task_lock
155 self
.task_queue
= Queue
.Queue(2000)
157 def get_vimconnector(self
):
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_
)
169 vim_config
.update(yaml
.load(vim
["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')
177 vim_config
["wim_external_ports"] = self
.ovim
.get_of_port_mappings(
178 db_filter
={"region": vim_config
['datacenter_id'], "pci": None})
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
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
))
191 self
.error_status
= "Error loading vimconnector: {}".format(e
)
193 def _get_db_task(self
):
195 Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
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
209 ORDER_BY
=("modified_at", "created_at",),
210 LIMIT
=database_limit
)
213 # if vim_actions[0]["modified_at"] > now:
214 # return int(vim_actions[0] - now)
215 for task
in vim_actions
:
217 if task_related
== task
["related"]:
218 continue # ignore if a locking has already tried for these task set
219 task_related
= task
["related"]
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"],
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"],
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
241 for relate_task
in related_tasks
:
242 if relate_task
["worker"] != self
.my_id
:
243 some_tasks_not_locked
= True
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"],
258 # task of creation must be the first in the list of related_task
259 assert(related_tasks
[0]["action"] in ("CREATE", "FIND"))
261 task
["params"] = None
263 extra
= yaml
.load(task
["extra"])
266 task
["extra"] = extra
267 if extra
.get("depends_on"):
269 if extra
.get("params"):
270 task
["params"] = deepcopy(extra
["params"])
271 return task
, related_tasks
272 except Exception as e
:
273 self
.logger
.critical("Unexpected exception at _get_db_task: " + str(e
), exc_info
=True)
276 def _delete_task(self
, task
):
278 Determine if this task need to be done or superseded
282 def copy_extra_created(copy_to
, copy_from
):
283 copy_to
["created"] = copy_from
["created"]
284 if copy_from
.get("sdn_net_id"):
285 copy_to
["sdn_net_id"] = copy_from
["sdn_net_id"]
286 if copy_from
.get("interfaces"):
287 copy_to
["interfaces"] = copy_from
["interfaces"]
288 if copy_from
.get("created_items"):
289 if not copy_to
.get("created_items"):
290 copy_to
["created_items"] = {}
291 copy_to
["created_items"].update(copy_from
["created_items"])
294 dependency_task
= None
295 deletion_needed
= False
296 if task
["status"] == "FAILED":
297 return # TODO need to be retry??
299 # get all related tasks
300 related_tasks
= self
.db
.get_rows(FROM
="vim_wim_actions",
301 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
302 "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'],
303 "action": ["FIND", "CREATE"],
304 "related": task
["related"],
306 ORDER_BY
=("created_at",),
308 for related_task
in related_tasks
:
309 if related_task
["item"] == task
["item"] and related_task
["item_id"] == task
["item_id"]:
310 task_create
= related_task
312 if related_task
["extra"]:
313 extra_created
= yaml
.load(related_task
["extra"])
314 if extra_created
.get("created"):
315 deletion_needed
= True
316 related_task
["extra"] = extra_created
317 elif not dependency_task
:
318 dependency_task
= related_task
319 if task_create
and dependency_task
:
322 # mark task_create as FINISHED
323 self
.db
.update_rows("vim_wim_actions", UPDATE
={"status": "FINISHED"},
324 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
325 "instance_action_id": task_create
["instance_action_id"],
326 "task_index": task_create
["task_index"]
328 if not deletion_needed
:
330 elif dependency_task
:
331 # move create information from task_create to relate_task
332 extra_new_created
= yaml
.load(dependency_task
["extra"]) or {}
333 extra_new_created
["created"] = extra_created
["created"]
334 copy_extra_created(copy_to
=extra_new_created
, copy_from
=extra_created
)
336 self
.db
.update_rows("vim_wim_actions",
337 UPDATE
={"extra": yaml
.safe_dump(extra_new_created
, default_flow_style
=True,
339 "vim_id": task_create
.get("vim_id")},
340 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
341 "instance_action_id": dependency_task
["instance_action_id"],
342 "task_index": dependency_task
["task_index"]
346 task
["vim_id"] = task_create
["vim_id"]
347 copy_extra_created(copy_to
=task
["extra"], copy_from
=task_create
["extra"])
350 except Exception as e
:
351 self
.logger
.critical("Unexpected exception at _delete_task: " + str(e
), exc_info
=True)
353 def _refres_vm(self
, task
):
354 """Call VIM to get VMs status"""
355 database_update
= None
357 vim_id
= task
["vim_id"]
358 vm_to_refresh_list
= [vim_id
]
360 vim_dict
= self
.vim
.refresh_vms_status(vm_to_refresh_list
)
361 vim_info
= vim_dict
[vim_id
]
362 except vimconn
.vimconnException
as e
:
363 # Mark all tasks at VIM_ERROR status
364 self
.logger
.error("task=several get-VM: vimconnException when trying to refresh vms " + str(e
))
365 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
367 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
368 self
.logger
.debug("task={} get-VM: vim_vm_id={} result={}".format(task_id
, task
["vim_id"], vim_info
))
370 # check and update interfaces
371 task_warning_msg
= ""
372 for interface
in vim_info
.get("interfaces", ()):
373 vim_interface_id
= interface
["vim_interface_id"]
374 if vim_interface_id
not in task
["extra"]["interfaces"]:
375 self
.logger
.critical("task={} get-VM: Interface not found {} on task info {}".format(
376 task_id
, vim_interface_id
, task
["extra"]["interfaces"]), exc_info
=True)
378 task_interface
= task
["extra"]["interfaces"][vim_interface_id
]
379 task_vim_interface
= task_interface
.get("vim_info")
380 if task_vim_interface
!= interface
:
382 if task_interface
.get("sdn_port_id"):
385 self
.ovim
.delete_port(task_interface
["sdn_port_id"], idempotent
=True)
386 task_interface
["sdn_port_id"] = None
387 except ovimException
as e
:
388 error_text
= "ovimException deleting external_port={}: {}".format(
389 task_interface
["sdn_port_id"], e
)
390 self
.logger
.error("task={} get-VM: {}".format(task_id
, error_text
), exc_info
=True)
391 task_warning_msg
+= error_text
392 # TODO Set error_msg at instance_nets instead of instance VMs
395 sdn_net_id
= task_interface
.get("sdn_net_id")
396 if sdn_net_id
and interface
.get("compute_node") and interface
.get("pci"):
397 sdn_port_name
= sdn_net_id
+ "." + task
["vim_id"]
398 sdn_port_name
= sdn_port_name
[:63]
401 sdn_port_id
= self
.ovim
.new_external_port(
402 {"compute_node": interface
["compute_node"],
403 "pci": interface
["pci"],
404 "vlan": interface
.get("vlan"),
405 "net_id": sdn_net_id
,
406 "region": self
.vim
["config"]["datacenter_id"],
407 "name": sdn_port_name
,
408 "mac": interface
.get("mac_address")})
409 task_interface
["sdn_port_id"] = sdn_port_id
410 except (ovimException
, Exception) as e
:
411 error_text
= "ovimException creating new_external_port compute_node={} pci={} vlan={} {}".\
412 format(interface
["compute_node"], interface
["pci"], interface
.get("vlan"), e
)
413 self
.logger
.error("task={} get-VM: {}".format(task_id
, error_text
), exc_info
=True)
414 task_warning_msg
+= error_text
415 # TODO Set error_msg at instance_nets instead of instance VMs
417 self
.db
.update_rows('instance_interfaces',
418 UPDATE
={"mac_address": interface
.get("mac_address"),
419 "ip_address": interface
.get("ip_address"),
420 "vim_interface_id": interface
.get("vim_interface_id"),
421 "vim_info": interface
.get("vim_info"),
422 "sdn_port_id": task_interface
.get("sdn_port_id"),
423 "compute_node": interface
.get("compute_node"),
424 "pci": interface
.get("pci"),
425 "vlan": interface
.get("vlan")},
426 WHERE
={'uuid': task_interface
["iface_id"]})
427 task_interface
["vim_info"] = interface
429 # check and update task and instance_vms database
430 vim_info_error_msg
= None
431 if vim_info
.get("error_msg"):
432 vim_info_error_msg
= self
._format
_vim
_error
_msg
(vim_info
["error_msg"] + task_warning_msg
)
433 elif task_warning_msg
:
434 vim_info_error_msg
= self
._format
_vim
_error
_msg
(task_warning_msg
)
435 task_vim_info
= task
["extra"].get("vim_info")
436 task_error_msg
= task
.get("error_msg")
437 task_vim_status
= task
["extra"].get("vim_status")
438 if task_vim_status
!= vim_info
["status"] or task_error_msg
!= vim_info_error_msg
or \
439 (vim_info
.get("vim_info") and task_vim_info
!= vim_info
["vim_info"]):
440 database_update
= {"status": vim_info
["status"], "error_msg": vim_info_error_msg
}
441 if vim_info
.get("vim_info"):
442 database_update
["vim_info"] = vim_info
["vim_info"]
444 task
["extra"]["vim_status"] = vim_info
["status"]
445 task
["error_msg"] = vim_info_error_msg
446 if vim_info
.get("vim_info"):
447 task
["extra"]["vim_info"] = vim_info
["vim_info"]
449 return database_update
451 def _refres_net(self
, task
):
452 """Call VIM to get network status"""
453 database_update
= None
455 vim_id
= task
["vim_id"]
456 net_to_refresh_list
= [vim_id
]
458 vim_dict
= self
.vim
.refresh_nets_status(net_to_refresh_list
)
459 vim_info
= vim_dict
[vim_id
]
460 except vimconn
.vimconnException
as e
:
461 # Mark all tasks at VIM_ERROR status
462 self
.logger
.error("task=several get-net: vimconnException when trying to refresh nets " + str(e
))
463 vim_info
= {"status": "VIM_ERROR", "error_msg": str(e
)}
465 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
466 self
.logger
.debug("task={} get-net: vim_net_id={} result={}".format(task_id
, task
["vim_id"], vim_info
))
468 task_vim_info
= task
["extra"].get("vim_info")
469 task_vim_status
= task
["extra"].get("vim_status")
470 task_error_msg
= task
.get("error_msg")
471 task_sdn_net_id
= task
["extra"].get("sdn_net_id")
473 vim_info_status
= vim_info
["status"]
474 vim_info_error_msg
= vim_info
.get("error_msg")
479 sdn_net
= self
.ovim
.show_network(task_sdn_net_id
)
480 except (ovimException
, Exception) as e
:
481 text_error
= "ovimException getting network snd_net_id={}: {}".format(task_sdn_net_id
, e
)
482 self
.logger
.error("task={} get-net: {}".format(task_id
, text_error
), exc_info
=True)
483 sdn_net
= {"status": "ERROR", "last_error": text_error
}
484 if sdn_net
["status"] == "ERROR":
485 if not vim_info_error_msg
:
486 vim_info_error_msg
= str(sdn_net
.get("last_error"))
488 vim_info_error_msg
= "VIM_ERROR: {} && SDN_ERROR: {}".format(
489 self
._format
_vim
_error
_msg
(vim_info_error_msg
, 1024 // 2 - 14),
490 self
._format
_vim
_error
_msg
(sdn_net
["last_error"], 1024 // 2 - 14))
491 vim_info_status
= "ERROR"
492 elif sdn_net
["status"] == "BUILD":
493 if vim_info_status
== "ACTIVE":
494 vim_info_status
= "BUILD"
497 if vim_info_error_msg
:
498 vim_info_error_msg
= self
._format
_vim
_error
_msg
(vim_info_error_msg
)
499 if task_vim_status
!= vim_info_status
or task_error_msg
!= vim_info_error_msg
or \
500 (vim_info
.get("vim_info") and task_vim_info
!= vim_info
["vim_info"]):
501 task
["extra"]["vim_status"] = vim_info_status
502 task
["error_msg"] = vim_info_error_msg
503 if vim_info
.get("vim_info"):
504 task
["extra"]["vim_info"] = vim_info
["vim_info"]
505 database_update
= {"status": vim_info_status
, "error_msg": vim_info_error_msg
}
506 if vim_info
.get("vim_info"):
507 database_update
["vim_info"] = vim_info
["vim_info"]
508 return database_update
510 def _proccess_pending_tasks(self
, task
, related_tasks
):
511 old_task_status
= task
["status"]
512 create_or_find
= False # if as result of processing this task something is created or found
516 if task
["status"] == "SCHEDULED":
517 # check if tasks that this depends on have been completed
518 dependency_not_completed
= False
519 dependency_modified_at
= 0
520 for task_index
in task
["extra"].get("depends_on", ()):
521 task_dependency
= self
._look
_for
_task
(task
["instance_action_id"], task_index
)
522 if not task_dependency
:
523 raise VimThreadException(
524 "Cannot get depending net task trying to get depending task {}.{}".format(
525 task
["instance_action_id"], task_index
))
526 # task["depends"]["TASK-" + str(task_index)] = task_dependency #it references another object,so
527 # database must be look again
528 if task_dependency
["status"] == "SCHEDULED":
529 dependency_not_completed
= True
530 dependency_modified_at
= task_dependency
["modified_at"]
532 elif task_dependency
["status"] == "FAILED":
533 raise VimThreadException(
534 "Cannot {} {}, (task {}.{}) because depends on failed {}.{}, (task{}.{}): {}".format(
535 task
["action"], task
["item"],
536 task
["instance_action_id"], task
["task_index"],
537 task_dependency
["instance_action_id"], task_dependency
["task_index"],
538 task_dependency
["action"], task_dependency
["item"], task_dependency
.get("error_msg")))
540 task
["depends"]["TASK-"+str(task_index
)] = task_dependency
541 task
["depends"]["TASK-{}.{}".format(task
["instance_action_id"], task_index
)] = task_dependency
542 if dependency_not_completed
:
543 # Move this task to the time dependency is going to be modified plus 10 seconds.
544 self
.db
.update_rows("vim_wim_actions", modified_time
=dependency_modified_at
+ 10,
545 UPDATE
={"worker": None},
546 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
, "worker": self
.my_id
,
547 "related": task
["related"],
549 # task["extra"]["tries"] = task["extra"].get("tries", 0) + 1
550 # if task["extra"]["tries"] > 3:
551 # raise VimThreadException(
552 # "Cannot {} {}, (task {}.{}) because timeout waiting to complete {} {}, "
553 # "(task {}.{})".format(task["action"], task["item"],
554 # task["instance_action_id"], task["task_index"],
555 # task_dependency["instance_action_id"], task_dependency["task_index"]
556 # task_dependency["action"], task_dependency["item"]))
559 database_update
= None
560 if task
["action"] == "DELETE":
561 deleted_needed
= self
._delete
_task
(task
)
562 if not deleted_needed
:
563 task
["status"] = "SUPERSEDED" # with FINISHED instead of DONE it will not be refreshing
564 task
["error_msg"] = None
566 if task
["status"] == "SUPERSEDED":
567 # not needed to do anything but update database with the new status
568 database_update
= None
570 task
["status"] = "FAILED"
571 task
["error_msg"] = self
.error_status
572 database_update
= {"status": "VIM_ERROR", "error_msg": task
["error_msg"]}
573 elif task
["item_id"] != related_tasks
[0]["item_id"] and task
["action"] in ("FIND", "CREATE"):
574 # Do nothing, just copy values from one to another and updata database
575 task
["status"] = related_tasks
[0]["status"]
576 task
["error_msg"] = related_tasks
[0]["error_msg"]
577 task
["vim_id"] = related_tasks
[0]["vim_id"]
578 extra
= yaml
.load(related_tasks
[0]["extra"])
579 task
["extra"]["vim_status"] = extra
.get("vim_status")
580 next_refresh
= related_tasks
[0]["modified_at"] + 0.001
581 database_update
= {"status": task
["extra"].get("vim_status", "VIM_ERROR"),
582 "error_msg": task
["error_msg"]}
583 if task
["item"] == 'instance_vms':
584 database_update
["vim_vm_id"] = task
["vim_id"]
585 elif task
["item"] == 'instance_nets':
586 database_update
["vim_net_id"] = task
["vim_id"]
587 elif task
["item"] == 'instance_vms':
588 if task
["status"] in ('BUILD', 'DONE') and task
["action"] in ("FIND", "CREATE"):
589 database_update
= self
._refres
_vm
(task
)
590 create_or_find
= True
591 elif task
["action"] == "CREATE":
592 create_or_find
= True
593 database_update
= self
.new_vm(task
)
594 elif task
["action"] == "DELETE":
597 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
598 elif task
["item"] == 'instance_nets':
599 if task
["status"] in ('BUILD', 'DONE') and task
["action"] in ("FIND", "CREATE"):
600 database_update
= self
._refres
_net
(task
)
601 create_or_find
= True
602 elif task
["action"] == "CREATE":
603 create_or_find
= True
604 database_update
= self
.new_net(task
)
605 elif task
["action"] == "DELETE":
607 elif task
["action"] == "FIND":
608 database_update
= self
.get_net(task
)
610 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
611 elif task
["item"] == 'instance_sfis':
612 if task
["action"] == "CREATE":
613 create_or_find
= True
614 database_update
= self
.new_sfi(task
)
615 elif task
["action"] == "DELETE":
618 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
619 elif task
["item"] == 'instance_sfs':
620 if task
["action"] == "CREATE":
621 create_or_find
= True
622 database_update
= self
.new_sf(task
)
623 elif task
["action"] == "DELETE":
626 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
627 elif task
["item"] == 'instance_classifications':
628 if task
["action"] == "CREATE":
629 create_or_find
= True
630 database_update
= self
.new_classification(task
)
631 elif task
["action"] == "DELETE":
632 self
.del_classification(task
)
634 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
635 elif task
["item"] == 'instance_sfps':
636 if task
["action"] == "CREATE":
637 create_or_find
= True
638 database_update
= self
.new_sfp(task
)
639 elif task
["action"] == "DELETE":
642 raise vimconn
.vimconnException(self
.name
+ "unknown task action {}".format(task
["action"]))
644 raise vimconn
.vimconnException(self
.name
+ "unknown task item {}".format(task
["item"]))
646 except VimThreadException
as e
:
647 task
["error_msg"] = str(e
)
648 task
["status"] = "FAILED"
649 database_update
= {"status": "VIM_ERROR", "error_msg": task
["error_msg"]}
650 if task
["item"] == 'instance_vms':
651 database_update
["vim_vm_id"] = None
652 elif task
["item"] == 'instance_nets':
653 database_update
["vim_net_id"] = None
655 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
656 self
.logger
.debug("task={} item={} action={} result={}:'{}' params={}".format(
657 task_id
, task
["item"], task
["action"], task
["status"],
658 task
["vim_id"] if task
["status"] == "DONE" else task
.get("error_msg"), task
["params"]))
661 if task
["status"] == "DONE":
662 next_refresh
= time
.time()
663 if task
["extra"].get("vim_status") == "BUILD":
664 next_refresh
+= self
.REFRESH_BUILD
665 elif task
["extra"].get("vim_status") in ("ERROR", "VIM_ERROR"):
666 next_refresh
+= self
.REFRESH_ERROR
667 elif task
["extra"].get("vim_status") == "DELETED":
668 next_refresh
+= self
.REFRESH_DELETE
670 next_refresh
+= self
.REFRESH_ACTIVE
671 elif task
["status"] == "FAILED":
672 next_refresh
= time
.time() + self
.REFRESH_DELETE
675 # modify all related task with action FIND/CREATED non SCHEDULED
677 table
="vim_wim_actions", modified_time
=next_refresh
+ 0.001,
678 UPDATE
={"status": task
["status"], "vim_id": task
.get("vim_id"),
679 "error_msg": task
["error_msg"],
682 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
683 "worker": self
.my_id
,
684 "action": ["FIND", "CREATE"],
685 "related": task
["related"],
686 "status<>": "SCHEDULED",
690 table
="vim_wim_actions", modified_time
=next_refresh
,
691 UPDATE
={"status": task
["status"], "vim_id": task
.get("vim_id"),
692 "error_msg": task
["error_msg"],
693 "extra": yaml
.safe_dump(task
["extra"], default_flow_style
=True, width
=256)},
694 WHERE
={"instance_action_id": task
["instance_action_id"], "task_index": task
["task_index"]})
697 table
="vim_wim_actions", modified_time
=0,
698 UPDATE
={"worker": None},
699 WHERE
={"datacenter_vim_id": self
.datacenter_tenant_id
,
700 "worker": self
.my_id
,
701 "related": task
["related"],
704 # Update table instance_actions
705 if old_task_status
== "SCHEDULED" and task
["status"] != old_task_status
:
707 table
="instance_actions",
708 UPDATE
={("number_failed" if task
["status"] == "FAILED" else "number_done"): {"INCREMENT": 1}},
709 WHERE
={"uuid": task
["instance_action_id"]})
711 where_filter
= {"related": task
["related"]}
712 if task
["item"] == "instance_nets" and task
["datacenter_vim_id"]:
713 where_filter
["datacenter_tenant_id"] = task
["datacenter_vim_id"]
714 self
.db
.update_rows(table
=task
["item"],
715 UPDATE
=database_update
,
717 except db_base_Exception
as e
:
718 self
.logger
.error("task={} Error updating database {}".format(task_id
, e
), exc_info
=True)
720 def insert_task(self
, task
):
722 self
.task_queue
.put(task
, False)
725 raise vimconn
.vimconnException(self
.name
+ ": timeout inserting a task")
727 def del_task(self
, task
):
729 if task
["status"] == "SCHEDULED":
730 task
["status"] = "SUPERSEDED"
732 else: # task["status"] == "processing"
733 self
.task_lock
.release()
737 self
.logger
.debug("Starting")
739 self
.get_vimconnector()
740 self
.logger
.debug("Vimconnector loaded")
741 reload_thread
= False
745 while not self
.task_queue
.empty():
746 task
= self
.task_queue
.get()
747 if isinstance(task
, list):
749 elif isinstance(task
, str):
752 elif task
== 'reload':
755 self
.task_queue
.task_done()
759 task
, related_tasks
= self
._get
_db
_task
()
761 self
._proccess
_pending
_tasks
(task
, related_tasks
)
765 except Exception as e
:
766 self
.logger
.critical("Unexpected exception at run: " + str(e
), exc_info
=True)
768 self
.logger
.debug("Finishing")
770 def _look_for_task(self
, instance_action_id
, task_id
):
772 Look for a concrete task at vim_actions database table
773 :param instance_action_id: The instance_action_id
774 :param task_id: Can have several formats:
775 <task index>: integer
776 TASK-<task index> :backward compatibility,
777 [TASK-]<instance_action_id>.<task index>: this instance_action_id overrides the one in the parameter
778 :return: Task dictionary or None if not found
780 if isinstance(task_id
, int):
783 if task_id
.startswith("TASK-"):
784 task_id
= task_id
[5:]
785 ins_action_id
, _
, task_index
= task_id
.rpartition(".")
787 instance_action_id
= ins_action_id
789 tasks
= self
.db
.get_rows(FROM
="vim_wim_actions", WHERE
={"instance_action_id": instance_action_id
,
790 "task_index": task_index
})
794 task
["params"] = None
797 extra
= yaml
.load(task
["extra"])
798 task
["extra"] = extra
799 task
["params"] = extra
.get("params")
805 def _format_vim_error_msg(error_text
, max_length
=1024):
806 if error_text
and len(error_text
) >= max_length
:
807 return error_text
[:max_length
// 2 - 3] + " ... " + error_text
[-max_length
// 2 + 3:]
810 def new_vm(self
, task
):
811 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
813 params
= task
["params"]
814 depends
= task
.get("depends")
817 if "net_id" in net
and is_task_id(net
["net_id"]): # change task_id into network_id
818 network_id
= task
["depends"][net
["net_id"]].get("vim_id")
820 raise VimThreadException(
821 "Cannot create VM because depends on a network not created or found: " +
822 str(depends
[net
["net_id"]]["error_msg"]))
823 net
["net_id"] = network_id
824 params_copy
= deepcopy(params
)
825 vim_vm_id
, created_items
= self
.vim
.new_vminstance(*params_copy
)
827 # fill task_interfaces. Look for snd_net_id at database for each interface
829 for iface
in params_copy
[5]:
830 task_interfaces
[iface
["vim_id"]] = {"iface_id": iface
["uuid"]}
831 result
= self
.db
.get_rows(
832 SELECT
=('sdn_net_id', 'interface_id'),
833 FROM
='instance_nets as ine join instance_interfaces as ii on ii.instance_net_id=ine.uuid',
834 WHERE
={'ii.uuid': iface
["uuid"]})
836 task_interfaces
[iface
["vim_id"]]["sdn_net_id"] = result
[0]['sdn_net_id']
837 task_interfaces
[iface
["vim_id"]]["interface_id"] = result
[0]['interface_id']
839 self
.logger
.critical("task={} new-VM: instance_nets uuid={} not found at DB".format(task_id
,
843 task
["vim_info"] = {}
844 task
["extra"]["interfaces"] = task_interfaces
845 task
["extra"]["created"] = True
846 task
["extra"]["created_items"] = created_items
847 task
["extra"]["vim_status"] = "BUILD"
848 task
["error_msg"] = None
849 task
["status"] = "DONE"
850 task
["vim_id"] = vim_vm_id
851 instance_element_update
= {"status": "BUILD", "vim_vm_id": vim_vm_id
, "error_msg": None}
852 return instance_element_update
854 except (vimconn
.vimconnException
, VimThreadException
) as e
:
855 self
.logger
.error("task={} new-VM: {}".format(task_id
, e
))
856 error_text
= self
._format
_vim
_error
_msg
(str(e
))
857 task
["error_msg"] = error_text
858 task
["status"] = "FAILED"
859 task
["vim_id"] = None
860 instance_element_update
= {"status": "VIM_ERROR", "vim_vm_id": None, "error_msg": error_text
}
861 return instance_element_update
863 def del_vm(self
, task
):
864 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
865 vm_vim_id
= task
["vim_id"]
866 interfaces
= task
["extra"].get("interfaces", ())
868 for iface
in interfaces
.values():
869 if iface
.get("sdn_port_id"):
872 self
.ovim
.delete_port(iface
["sdn_port_id"], idempotent
=True)
873 except ovimException
as e
:
874 self
.logger
.error("task={} del-VM: ovimException when deleting external_port={}: {} ".format(
875 task_id
, iface
["sdn_port_id"], e
), exc_info
=True)
876 # TODO Set error_msg at instance_nets
878 self
.vim
.delete_vminstance(vm_vim_id
, task
["extra"].get("created_items"))
879 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
880 task
["error_msg"] = None
883 except vimconn
.vimconnException
as e
:
884 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
885 if isinstance(e
, vimconn
.vimconnNotFoundException
):
886 # If not found mark as Done and fill error_msg
887 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
889 task
["status"] = "FAILED"
892 def _get_net_internal(self
, task
, filter_param
):
894 Common code for get_net and new_net. It looks for a network on VIM with the filter_params
895 :param task: task for this find or find-or-create action
896 :param filter_param: parameters to send to the vimconnector
897 :return: a dict with the content to update the instance_nets database table. Raises an exception on error, or
898 when network is not found or found more than one
900 vim_nets
= self
.vim
.get_network_list(filter_param
)
902 raise VimThreadExceptionNotFound("Network not found with this criteria: '{}'".format(filter_param
))
903 elif len(vim_nets
) > 1:
904 raise VimThreadException("More than one network found with this criteria: '{}'".format(filter_param
))
905 vim_net_id
= vim_nets
[0]["id"]
907 # Discover if this network is managed by a sdn controller
909 result
= self
.db
.get_rows(SELECT
=('sdn_net_id',), FROM
='instance_nets',
910 WHERE
={'vim_net_id': vim_net_id
, 'datacenter_tenant_id': self
.datacenter_tenant_id
},
911 ORDER
="instance_scenario_id")
913 sdn_net_id
= result
[0]['sdn_net_id']
915 task
["status"] = "DONE"
916 task
["extra"]["vim_info"] = {}
917 task
["extra"]["created"] = False
918 task
["extra"]["vim_status"] = "BUILD"
919 task
["extra"]["sdn_net_id"] = sdn_net_id
920 task
["error_msg"] = None
921 task
["vim_id"] = vim_net_id
922 instance_element_update
= {"vim_net_id": vim_net_id
, "created": False, "status": "BUILD",
923 "error_msg": None, "sdn_net_id": sdn_net_id
}
924 return instance_element_update
926 def get_net(self
, task
):
927 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
929 params
= task
["params"]
930 filter_param
= params
[0]
931 instance_element_update
= self
._get
_net
_internal
(task
, filter_param
)
932 return instance_element_update
934 except (vimconn
.vimconnException
, VimThreadException
) as e
:
935 self
.logger
.error("task={} get-net: {}".format(task_id
, e
))
936 task
["status"] = "FAILED"
937 task
["vim_id"] = None
938 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
939 instance_element_update
= {"vim_net_id": None, "status": "VIM_ERROR",
940 "error_msg": task
["error_msg"]}
941 return instance_element_update
943 def new_net(self
, task
):
946 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
950 if task
["extra"].get("find"):
951 action_text
= "finding"
952 filter_param
= task
["extra"]["find"][0]
954 instance_element_update
= self
._get
_net
_internal
(task
, filter_param
)
955 return instance_element_update
956 except VimThreadExceptionNotFound
:
959 params
= task
["params"]
960 action_text
= "creating VIM"
961 vim_net_id
, created_items
= self
.vim
.new_network(*params
[0:3])
965 wim_account_name
= None
967 wim_account_name
= params
[3]
969 sdn_controller
= self
.vim
.config
.get('sdn-controller')
970 if sdn_controller
and (net_type
== "data" or net_type
== "ptp"):
971 network
= {"name": net_name
, "type": net_type
, "region": self
.vim
["config"]["datacenter_id"]}
973 vim_net
= self
.vim
.get_network(vim_net_id
)
974 if vim_net
.get('encapsulation') != 'vlan':
975 raise vimconn
.vimconnException(
976 "net '{}' defined as type '{}' has not vlan encapsulation '{}'".format(
977 net_name
, net_type
, vim_net
['encapsulation']))
978 network
["vlan"] = vim_net
.get('segmentation_id')
979 action_text
= "creating SDN"
981 sdn_net_id
= self
.ovim
.new_network(network
)
983 if wim_account_name
and self
.vim
.config
["wim_external_ports"]:
984 # add external port to connect WIM. Try with compute node __WIM:wim_name and __WIM
985 action_text
= "attaching external port to ovim network"
986 sdn_port_name
= "external_port"
988 "compute_node": "__WIM:" + wim_account_name
[0:58],
990 "vlan": network
["vlan"],
991 "net_id": sdn_net_id
,
992 "region": self
.vim
["config"]["datacenter_id"],
993 "name": sdn_port_name
,
997 sdn_external_port_id
= self
.ovim
.new_external_port(sdn_port_data
)
998 except ovimException
:
999 sdn_port_data
["compute_node"] = "__WIM"
1001 sdn_external_port_id
= self
.ovim
.new_external_port(sdn_port_data
)
1002 self
.logger
.debug("Added sdn_external_port {} to sdn_network {}".format(sdn_external_port_id
,
1004 task
["status"] = "DONE"
1005 task
["extra"]["vim_info"] = {}
1006 task
["extra"]["sdn_net_id"] = sdn_net_id
1007 task
["extra"]["vim_status"] = "BUILD"
1008 task
["extra"]["created"] = True
1009 task
["extra"]["created_items"] = created_items
1010 task
["error_msg"] = None
1011 task
["vim_id"] = vim_net_id
1012 instance_element_update
= {"vim_net_id": vim_net_id
, "sdn_net_id": sdn_net_id
, "status": "BUILD",
1013 "created": True, "error_msg": None}
1014 return instance_element_update
1015 except (vimconn
.vimconnException
, ovimException
) as e
:
1016 self
.logger
.error("task={} new-net: Error {}: {}".format(task_id
, action_text
, e
))
1017 task
["status"] = "FAILED"
1018 task
["vim_id"] = vim_net_id
1019 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1020 task
["extra"]["sdn_net_id"] = sdn_net_id
1021 instance_element_update
= {"vim_net_id": vim_net_id
, "sdn_net_id": sdn_net_id
, "status": "VIM_ERROR",
1022 "error_msg": task
["error_msg"]}
1023 return instance_element_update
1025 def del_net(self
, task
):
1026 net_vim_id
= task
["vim_id"]
1027 sdn_net_id
= task
["extra"].get("sdn_net_id")
1030 self
.vim
.delete_network(net_vim_id
, task
["extra"].get("created_items"))
1032 # Delete any attached port to this sdn network. There can be ports associated to this network in case
1033 # it was manually done using 'openmano vim-net-sdn-attach'
1035 port_list
= self
.ovim
.get_ports(columns
={'uuid'},
1036 filter={'name': 'external_port', 'net_id': sdn_net_id
})
1037 for port
in port_list
:
1038 self
.ovim
.delete_port(port
['uuid'], idempotent
=True)
1039 self
.ovim
.delete_network(sdn_net_id
, idempotent
=True)
1040 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1041 task
["error_msg"] = None
1043 except ovimException
as e
:
1044 task
["error_msg"] = self
._format
_vim
_error
_msg
("ovimException obtaining and deleting external "
1045 "ports for net {}: {}".format(sdn_net_id
, str(e
)))
1046 except vimconn
.vimconnException
as e
:
1047 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1048 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1049 # If not found mark as Done and fill error_msg
1050 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1052 task
["status"] = "FAILED"
1055 # Service Function Instances
1056 def new_sfi(self
, task
):
1059 # Waits for interfaces to be ready (avoids failure)
1061 dep_id
= "TASK-" + str(task
["extra"]["depends_on"][0])
1062 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1064 interfaces
= task
["depends"][dep_id
]["extra"].get("interfaces")
1066 ingress_interface_id
= task
.get("extra").get("params").get("ingress_interface_id")
1067 egress_interface_id
= task
.get("extra").get("params").get("egress_interface_id")
1068 ingress_vim_interface_id
= None
1069 egress_vim_interface_id
= None
1070 for vim_interface
, interface_data
in interfaces
.iteritems():
1071 if interface_data
.get("interface_id") == ingress_interface_id
:
1072 ingress_vim_interface_id
= vim_interface
1074 if ingress_interface_id
!= egress_interface_id
:
1075 for vim_interface
, interface_data
in interfaces
.iteritems():
1076 if interface_data
.get("interface_id") == egress_interface_id
:
1077 egress_vim_interface_id
= vim_interface
1080 egress_vim_interface_id
= ingress_vim_interface_id
1081 if not ingress_vim_interface_id
or not egress_vim_interface_id
:
1082 error_text
= "Error creating Service Function Instance, Ingress: {}, Egress: {}".format(
1083 ingress_vim_interface_id
, egress_vim_interface_id
)
1084 self
.logger
.error(error_text
)
1085 task
["error_msg"] = error_text
1086 task
["status"] = "FAILED"
1087 task
["vim_id"] = None
1089 # At the moment, every port associated with the VM will be used both as ingress and egress ports.
1090 # Bear in mind that different VIM connectors might support SFI differently. In the case of OpenStack,
1091 # only the first ingress and first egress ports will be used to create the SFI (Port Pair).
1092 ingress_port_id_list
= [ingress_vim_interface_id
]
1093 egress_port_id_list
= [egress_vim_interface_id
]
1094 name
= "sfi-%s" % task
["item_id"][:8]
1095 # By default no form of IETF SFC Encapsulation will be used
1096 vim_sfi_id
= self
.vim
.new_sfi(name
, ingress_port_id_list
, egress_port_id_list
, sfc_encap
=False)
1098 task
["extra"]["created"] = True
1099 task
["extra"]["vim_status"] = "ACTIVE"
1100 task
["error_msg"] = None
1101 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1102 task
["vim_id"] = vim_sfi_id
1103 instance_element_update
= {"status": "ACTIVE", "vim_sfi_id": vim_sfi_id
, "error_msg": None}
1104 return instance_element_update
1106 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1107 self
.logger
.error("Error creating Service Function Instance, task=%s: %s", task_id
, str(e
))
1108 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1109 task
["error_msg"] = error_text
1110 task
["status"] = "FAILED"
1111 task
["vim_id"] = None
1112 instance_element_update
= {"status": "VIM_ERROR", "vim_sfi_id": None, "error_msg": error_text
}
1113 return instance_element_update
1115 def del_sfi(self
, task
):
1116 sfi_vim_id
= task
["vim_id"]
1118 self
.vim
.delete_sfi(sfi_vim_id
)
1119 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1120 task
["error_msg"] = None
1123 except vimconn
.vimconnException
as e
:
1124 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1125 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1126 # If not found mark as Done and fill error_msg
1127 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1129 task
["status"] = "FAILED"
1132 def new_sf(self
, task
):
1135 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1137 depending_tasks
= ["TASK-" + str(dep_id
) for dep_id
in task
["extra"]["depends_on"]]
1138 # sfis = task.get("depends").values()[0].get("extra").get("params")[5]
1139 sfis
= [task
.get("depends").get(dep_task
) for dep_task
in depending_tasks
]
1142 sfi_id_list
.append(sfi
.get("vim_id"))
1143 name
= "sf-%s" % task
["item_id"][:8]
1144 # By default no form of IETF SFC Encapsulation will be used
1145 vim_sf_id
= self
.vim
.new_sf(name
, sfi_id_list
, sfc_encap
=False)
1147 task
["extra"]["created"] = True
1148 task
["extra"]["vim_status"] = "ACTIVE"
1149 task
["error_msg"] = None
1150 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1151 task
["vim_id"] = vim_sf_id
1152 instance_element_update
= {"status": "ACTIVE", "vim_sf_id": vim_sf_id
, "error_msg": None}
1153 return instance_element_update
1155 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1156 self
.logger
.error("Error creating Service Function, task=%s: %s", task_id
, str(e
))
1157 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1158 task
["error_msg"] = error_text
1159 task
["status"] = "FAILED"
1160 task
["vim_id"] = None
1161 instance_element_update
= {"status": "VIM_ERROR", "vim_sf_id": None, "error_msg": error_text
}
1162 return instance_element_update
1164 def del_sf(self
, task
):
1165 sf_vim_id
= task
["vim_id"]
1167 self
.vim
.delete_sf(sf_vim_id
)
1168 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1169 task
["error_msg"] = None
1172 except vimconn
.vimconnException
as e
:
1173 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1174 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1175 # If not found mark as Done and fill error_msg
1176 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1178 task
["status"] = "FAILED"
1181 def new_classification(self
, task
):
1182 vim_classification_id
= None
1184 params
= task
["params"]
1185 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1186 dep_id
= "TASK-" + str(task
["extra"]["depends_on"][0])
1188 interfaces
= task
.get("depends").get(dep_id
).get("extra").get("interfaces").keys()
1189 # Bear in mind that different VIM connectors might support Classifications differently.
1190 # In the case of OpenStack, only the first VNF attached to the classifier will be used
1191 # to create the Classification(s) (the "logical source port" of the "Flow Classifier").
1192 # Since the VNFFG classifier match lacks the ethertype, classification defaults to
1193 # using the IPv4 flow classifier.
1194 name
= "c-%s" % task
["item_id"][:8]
1195 # if not CIDR is given for the IP addresses, add /32:
1196 ip_proto
= int(params
.get("ip_proto"))
1197 source_ip
= params
.get("source_ip")
1198 destination_ip
= params
.get("destination_ip")
1199 source_port
= params
.get("source_port")
1200 destination_port
= params
.get("destination_port")
1201 definition
= {"logical_source_port": interfaces
[0]}
1207 elif ip_proto
== 17:
1209 definition
["protocol"] = ip_proto
1211 if '/' not in source_ip
:
1213 definition
["source_ip_prefix"] = source_ip
1215 definition
["source_port_range_min"] = source_port
1216 definition
["source_port_range_max"] = source_port
1217 if destination_port
:
1218 definition
["destination_port_range_min"] = destination_port
1219 definition
["destination_port_range_max"] = destination_port
1221 if '/' not in destination_ip
:
1222 destination_ip
+= '/32'
1223 definition
["destination_ip_prefix"] = destination_ip
1225 vim_classification_id
= self
.vim
.new_classification(
1226 name
, 'legacy_flow_classifier', definition
)
1228 task
["extra"]["created"] = True
1229 task
["extra"]["vim_status"] = "ACTIVE"
1230 task
["error_msg"] = None
1231 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1232 task
["vim_id"] = vim_classification_id
1233 instance_element_update
= {"status": "ACTIVE", "vim_classification_id": vim_classification_id
,
1235 return instance_element_update
1237 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1238 self
.logger
.error("Error creating Classification, task=%s: %s", task_id
, str(e
))
1239 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1240 task
["error_msg"] = error_text
1241 task
["status"] = "FAILED"
1242 task
["vim_id"] = None
1243 instance_element_update
= {"status": "VIM_ERROR", "vim_classification_id": None, "error_msg": error_text
}
1244 return instance_element_update
1246 def del_classification(self
, task
):
1247 classification_vim_id
= task
["vim_id"]
1249 self
.vim
.delete_classification(classification_vim_id
)
1250 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1251 task
["error_msg"] = None
1254 except vimconn
.vimconnException
as e
:
1255 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1256 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1257 # If not found mark as Done and fill error_msg
1258 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1260 task
["status"] = "FAILED"
1263 def new_sfp(self
, task
):
1266 task_id
= task
["instance_action_id"] + "." + str(task
["task_index"])
1267 depending_tasks
= [task
.get("depends").get("TASK-" + str(tsk_id
)) for tsk_id
in
1268 task
.get("extra").get("depends_on")]
1271 classification_id_list
= []
1272 for dep
in depending_tasks
:
1273 vim_id
= dep
.get("vim_id")
1274 resource
= dep
.get("item")
1275 if resource
== "instance_sfs":
1276 sf_id_list
.append(vim_id
)
1277 elif resource
== "instance_classifications":
1278 classification_id_list
.append(vim_id
)
1280 name
= "sfp-%s" % task
["item_id"][:8]
1281 # By default no form of IETF SFC Encapsulation will be used
1282 vim_sfp_id
= self
.vim
.new_sfp(name
, classification_id_list
, sf_id_list
, sfc_encap
=False)
1284 task
["extra"]["created"] = True
1285 task
["extra"]["vim_status"] = "ACTIVE"
1286 task
["error_msg"] = None
1287 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1288 task
["vim_id"] = vim_sfp_id
1289 instance_element_update
= {"status": "ACTIVE", "vim_sfp_id": vim_sfp_id
, "error_msg": None}
1290 return instance_element_update
1292 except (vimconn
.vimconnException
, VimThreadException
) as e
:
1293 self
.logger
.error("Error creating Service Function, task=%s: %s", task_id
, str(e
))
1294 error_text
= self
._format
_vim
_error
_msg
(str(e
))
1295 task
["error_msg"] = error_text
1296 task
["status"] = "FAILED"
1297 task
["vim_id"] = None
1298 instance_element_update
= {"status": "VIM_ERROR", "vim_sfp_id": None, "error_msg": error_text
}
1299 return instance_element_update
1301 def del_sfp(self
, task
):
1302 sfp_vim_id
= task
["vim_id"]
1304 self
.vim
.delete_sfp(sfp_vim_id
)
1305 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1306 task
["error_msg"] = None
1309 except vimconn
.vimconnException
as e
:
1310 task
["error_msg"] = self
._format
_vim
_error
_msg
(str(e
))
1311 if isinstance(e
, vimconn
.vimconnNotFoundException
):
1312 # If not found mark as Done and fill error_msg
1313 task
["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing
1315 task
["status"] = "FAILED"