# -*- coding: utf-8 -*-
##
-# Copyright 2015 Telefónica Investigación y Desarrollo, S.A.U.
+# Copyright 2015 Telefonica Investigacion y Desarrollo, S.A.U.
# This file is part of openvim
# All Rights Reserved.
#
import Queue
import logging
import vimconn
+import vimconn_openvim
+import vimconn_aws
+import vimconn_opennebula
+import vimconn_openstack
+import vimconn_vmware
import yaml
from db_base import db_base_Exception
from lib_osm_openvim.ovim import ovimException
__author__ = "Alfonso Tierno, Pablo Montes"
__date__ = "$28-Sep-2017 12:07:15$"
+vim_module = {
+ "openvim": vimconn_openvim,
+ "aws": vimconn_aws,
+ "opennebula": vimconn_opennebula,
+ "openstack": vimconn_openstack,
+ "vmware": vimconn_vmware,
+}
def is_task_id(task_id):
return task_id.startswith("TASK-")
REFRESH_BUILD = 5 # 5 seconds
REFRESH_ACTIVE = 60 # 1 minute
- def __init__(self, myvimconn, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None,
+ def __init__(self, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None,
db=None, db_lock=None, ovim=None):
"""Init a thread.
Arguments:
'db', 'db_lock': database class and lock to use it in exclusion
"""
threading.Thread.__init__(self)
- if isinstance(myvimconn, vimconn.vimconnException):
- self.vim = None
- self.error_status = "Error accesing to VIM: {}".format(myvimconn)
- else:
- self.vim = myvimconn
- self.error_status = None
+ self.vim = None
+ self.error_status = None
self.datacenter_name = datacenter_name
self.datacenter_tenant_id = datacenter_tenant_id
self.ovim = ovim
self.name = vimconn["id"] + "." + vimconn["config"]["datacenter_tenant_id"]
else:
self.name = name
+ self.vim_persistent_info = {}
self.logger = logging.getLogger('openmano.vim.'+self.name)
self.db = db
<task2> # e.g. DELETE task
"""
+ def get_vimconnector(self):
+ try:
+ from_= "datacenter_tenants as dt join datacenters as d on dt.datacenter_id=d.uuid"
+ select_ = ('type', 'd.config as config', 'd.uuid as datacenter_id', 'vim_url', 'vim_url_admin',
+ 'd.name as datacenter_name', 'dt.uuid as datacenter_tenant_id',
+ 'dt.vim_tenant_name as vim_tenant_name', 'dt.vim_tenant_id as vim_tenant_id',
+ 'user', 'passwd', 'dt.config as dt_config')
+ where_ = {"dt.uuid": self.datacenter_tenant_id}
+ with self.db_lock:
+ vims = self.db.get_rows(FROM=from_, SELECT=select_, WHERE=where_)
+ vim = vims[0]
+ vim_config = {}
+ if vim["config"]:
+ vim_config.update(yaml.load(vim["config"]))
+ if vim["dt_config"]:
+ vim_config.update(yaml.load(vim["dt_config"]))
+ vim_config['datacenter_tenant_id'] = vim.get('datacenter_tenant_id')
+ vim_config['datacenter_id'] = vim.get('datacenter_id')
+
+ self.vim = vim_module[vim["type"]].vimconnector(
+ uuid=vim['datacenter_id'], name=vim['datacenter_name'],
+ tenant_id=vim['vim_tenant_id'], tenant_name=vim['vim_tenant_name'],
+ url=vim['vim_url'], url_admin=vim['vim_url_admin'],
+ user=vim['user'], passwd=vim['passwd'],
+ config=vim_config, persistent_info=self.vim_persistent_info
+ )
+ self.error_status = None
+ except Exception as e:
+ self.logger.error("Cannot load vimconnector for vim_account {}: {}".format(self.datacenter_tenant_id, e))
+ self.vim = None
+ self.error_status = "Error loading vimconnector: {}".format(e)
+
def _reload_vim_actions(self):
"""
Read actions from database and reload them at memory. Fill self.refresh_list, pending_list, vim_actions
def run(self):
self.logger.debug("Starting")
while True:
+ self.get_vimconnector()
+ self.logger.debug("Vimconnector loaded")
self._reload_vim_actions()
reload_thread = False
sdn_net_id = None
with self.db_lock:
result = self.db.get_rows(SELECT=('sdn_net_id',), FROM='instance_nets',
- WHERE={'vim_net_id': vim_net_id, 'instance_scenario_id': None,
- 'datacenter_tenant_id': self.datacenter_tenant_id})
+ WHERE={'vim_net_id': vim_net_id,
+ 'datacenter_tenant_id': self.datacenter_tenant_id},
+ ORDER="instance_scenario_id")
if result:
sdn_net_id = result[0]['sdn_net_id']
def new_sfi(self, task):
vim_sfi_id = None
try:
- params = task["params"]
+ dep_id = "TASK-" + str(task["extra"]["depends_on"][0])
task_id = task["instance_action_id"] + "." + str(task["task_index"])
- depends = task.get("depends")
error_text = ""
- interfaces = task.get("depends").values()[0].get("extra").get("params")[5]
+ interfaces = task.get("depends").get(dep_id).get("extra").get("interfaces").keys()
# At the moment, every port associated with the VM will be used both as ingress and egress ports.
# Bear in mind that different VIM connectors might support SFI differently. In the case of OpenStack, only the
# first ingress and first egress ports will be used to create the SFI (Port Pair).
- port_id_list = [interfaces[0].get("vim_id")]
+ port_id_list = [interfaces[0]]
name = "sfi-%s" % task["item_id"][:8]
# By default no form of IETF SFC Encapsulation will be used
vim_sfi_id = self.vim.new_sfi(name, port_id_list, port_id_list, sfc_encap=False)
def new_sf(self, task):
vim_sf_id = None
try:
- params = task["params"]
task_id = task["instance_action_id"] + "." + str(task["task_index"])
- depends = task.get("depends")
error_text = ""
+ depending_tasks = [ "TASK-" + str(dep_id) for dep_id in task["extra"]["depends_on"]]
#sfis = task.get("depends").values()[0].get("extra").get("params")[5]
- sfis = task.get("depends").values()
+ sfis = [task.get("depends").get(dep_task) for dep_task in depending_tasks]
sfi_id_list = []
for sfi in sfis:
sfi_id_list.append(sfi.get("vim_id"))
try:
params = task["params"]
task_id = task["instance_action_id"] + "." + str(task["task_index"])
- depends = task.get("depends")
+ depending_task = "TASK-" + str(task.get("extra").get("depends_on")[0])
error_text = ""
- interfaces = task.get("depends").values()[0].get("extra").get("params")[5]
+ interfaces = task.get("depends").get(depending_task).get("vim_interfaces").keys()
# Bear in mind that different VIM connectors might support Classifications differently.
# In the case of OpenStack, only the first VNF attached to the classifier will be used
# to create the Classification(s) (the "logical source port" of the "Flow Classifier").
if '/' not in destination_ip:
destination_ip += '/32'
definition = {
- "logical_source_port": interfaces[0].get("vim_id"),
+ "logical_source_port": interfaces[0],
"protocol": ip_proto,
"source_ip_prefix": source_ip,
"destination_ip_prefix": destination_ip,
try:
params = task["params"]
task_id = task["instance_action_id"] + "." + str(task["task_index"])
- depends = task.get("depends")
+ depending_tasks = [task.get("depends").get("TASK-" + str(tsk_id)) for tsk_id in task.get("extra").get("depends_on")]
error_text = ""
- deps = task.get("depends").values()
sf_id_list = []
classification_id_list = []
- for dep in deps:
+ for dep in depending_tasks:
vim_id = dep.get("vim_id")
resource = dep.get("item")
if resource == "instance_sfs":
projects / osm / RO.git / blobdiff