X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=src%2Femuvim%2Fapi%2Fsonata%2Fdummygatekeeper.py;h=f9ff50667ad29e586b0d7ed0727982f1a878023f;hb=08f91be07d9124986dacf05b7cd87eb8d3d8b338;hp=715432ebee9518e1219640b7558e766e3ddc3f04;hpb=6b5224d7f2f9c52802b37d5454ac02a8fe2c6634;p=osm%2Fvim-emu.git diff --git a/src/emuvim/api/sonata/dummygatekeeper.py b/src/emuvim/api/sonata/dummygatekeeper.py index 715432e..f9ff506 100755 --- a/src/emuvim/api/sonata/dummygatekeeper.py +++ b/src/emuvim/api/sonata/dummygatekeeper.py @@ -38,10 +38,13 @@ import uuid import hashlib import zipfile import yaml -from docker import Client as DockerClient +import threading +from docker import DockerClient, APIClient from flask import Flask, request import flask_restful as fr from collections import defaultdict +import pkg_resources +from subprocess import Popen logging.basicConfig() LOG = logging.getLogger("sonata-dummy-gatekeeper") @@ -61,11 +64,19 @@ GK_STANDALONE_MODE = False # should a new version of an image be pulled even if its available FORCE_PULL = False +# Automatically deploy SAPs (endpoints) of the service as new containers +# Attention: This is not a configuration switch but a global variable! Don't change its default value. +DEPLOY_SAP = False + +# flag to indicate if we use bidirectional forwarding rules in the automatic chaining process +BIDIRECTIONAL_CHAIN = False + class Gatekeeper(object): def __init__(self): self.services = dict() self.dcs = dict() + self.net = None self.vnf_counter = 0 # used to generate short names for VNFs (Mininet limitation) LOG.info("Create SONATA dummy gatekeeper.") @@ -106,6 +117,7 @@ class Service(object): self.remote_docker_image_urls = dict() self.instances = dict() self.vnf_name2docker_name = dict() + self.sap_identifiers = set() # lets generate a set of subnet configurations used for e-line chaining setup self.eline_subnets_src = generate_subnet_strings(50, start=200, subnet_size=24, ip=1) self.eline_subnets_dst = generate_subnet_strings(50, start=200, subnet_size=24, ip=2) @@ -121,6 +133,8 @@ class Service(object): self._load_package_descriptor() self._load_nsd() self._load_vnfd() + if DEPLOY_SAP: + self._load_saps() # 3. prepare container images (e.g. download or build Dockerfile) if BUILD_DOCKERFILE: self._load_docker_files() @@ -128,7 +142,7 @@ class Service(object): else: self._load_docker_urls() self._pull_predefined_dockerimages() - LOG.info("On-boarded service: %r" % self.manifest.get("package_name")) + LOG.info("On-boarded service: %r" % self.manifest.get("name")) def start_service(self): """ @@ -146,9 +160,16 @@ class Service(object): self.instances[instance_uuid] = dict() self.instances[instance_uuid]["vnf_instances"] = list() - # 2. compute placement of this service instance (adds DC names to VNFDs) + # 2. Configure the chaining of the network functions (currently only E-Line and E-LAN links supported) + vnf_id2vnf_name = defaultdict(lambda: "NotExistingNode", + reduce(lambda x, y: dict(x, **y), + map(lambda d: {d["vnf_id"]: d["vnf_name"]}, + self.nsd["network_functions"]))) + + # 3. compute placement of this service instance (adds DC names to VNFDs) if not GK_STANDALONE_MODE: - self._calculate_placement(FirstDcPlacement) + #self._calculate_placement(FirstDcPlacement) + self._calculate_placement(RoundRobinDcPlacement) # iterate over all vnfds that we have to start for vnfd in self.vnfds.itervalues(): vnfi = None @@ -156,53 +177,130 @@ class Service(object): vnfi = self._start_vnfd(vnfd) self.instances[instance_uuid]["vnf_instances"].append(vnfi) - # 3. Configure the chaining of the network functions (currently only E-Line links supported) - vnf_id2vnf_name = defaultdict(lambda: "NotExistingNode", - reduce(lambda x, y: dict(x, **y), - map(lambda d: {d["vnf_id"]: d["vnf_name"]}, - self.nsd["network_functions"]))) - - vlinks = self.nsd["virtual_links"] - fwd_links = self.nsd["forwarding_graphs"][0]["constituent_virtual_links"] - eline_fwd_links = [l for l in vlinks if (l["id"] in fwd_links) and (l["connectivity_type"] == "E-Line")] - - cookie = 1 # not clear why this is needed - to check with Steven - for link in eline_fwd_links: - src_id, src_if_name = link["connection_points_reference"][0].split(":") - dst_id, dst_if_name = link["connection_points_reference"][1].split(":") - - src_name = vnf_id2vnf_name[src_id] - dst_name = vnf_id2vnf_name[dst_id] - - LOG.debug( - "Setting up E-Line link. %s(%s:%s) -> %s(%s:%s)" % ( - src_name, src_id, src_if_name, dst_name, dst_id, dst_if_name)) - - if (src_name in self.vnfds) and (dst_name in self.vnfds): - network = self.vnfds[src_name].get("dc").net # there should be a cleaner way to find the DCNetwork - src_docker_name = self.vnf_name2docker_name[src_name] - dst_docker_name = self.vnf_name2docker_name[dst_name] - LOG.debug(src_docker_name) - ret = network.setChain( - src_docker_name, dst_docker_name, - vnf_src_interface=src_if_name, vnf_dst_interface=dst_if_name, - bidirectional=True, cmd="add-flow", cookie=cookie) - cookie += 1 - - # re-configure the VNFs IP assignment and ensure that a new subnet is used for each E-Link - src_vnfi = self._get_vnf_instance(instance_uuid, src_name) - if src_vnfi is not None: - self._vnf_reconfigure_network(src_vnfi, src_if_name, self.eline_subnets_src.pop(0)) - dst_vnfi = self._get_vnf_instance(instance_uuid, dst_name) - if dst_vnfi is not None: - self._vnf_reconfigure_network(dst_vnfi, dst_if_name, self.eline_subnets_dst.pop(0)) - - # 4. run the emulator specific entrypoint scripts in the VNFIs of this service instance + if "virtual_links" in self.nsd: + vlinks = self.nsd["virtual_links"] + fwd_links = self.nsd["forwarding_graphs"][0]["constituent_virtual_links"] + eline_fwd_links = [l for l in vlinks if (l["id"] in fwd_links) and (l["connectivity_type"] == "E-Line")] + elan_fwd_links = [l for l in vlinks if (l["id"] in fwd_links) and (l["connectivity_type"] == "E-LAN")] + + GK.net.deployed_elines.extend(eline_fwd_links) + GK.net.deployed_elans.extend(elan_fwd_links) + + # 4a. deploy E-Line links + # cookie is used as identifier for the flowrules installed by the dummygatekeeper + # eg. different services get a unique cookie for their flowrules + cookie = 1 + for link in eline_fwd_links: + src_id, src_if_name = link["connection_points_reference"][0].split(":") + dst_id, dst_if_name = link["connection_points_reference"][1].split(":") + + # check if there is a SAP in the link + if src_id in self.sap_identifiers: + src_docker_name = "{0}_{1}".format(src_id, src_if_name) + src_id = src_docker_name + else: + src_docker_name = src_id + + if dst_id in self.sap_identifiers: + dst_docker_name = "{0}_{1}".format(dst_id, dst_if_name) + dst_id = dst_docker_name + else: + dst_docker_name = dst_id + + src_name = vnf_id2vnf_name[src_id] + dst_name = vnf_id2vnf_name[dst_id] + + LOG.debug( + "Setting up E-Line link. %s(%s:%s) -> %s(%s:%s)" % ( + src_name, src_id, src_if_name, dst_name, dst_id, dst_if_name)) + + if (src_name in self.vnfds) and (dst_name in self.vnfds): + network = self.vnfds[src_name].get("dc").net # there should be a cleaner way to find the DCNetwork + LOG.debug(src_docker_name) + ret = network.setChain( + src_docker_name, dst_docker_name, + vnf_src_interface=src_if_name, vnf_dst_interface=dst_if_name, + bidirectional=BIDIRECTIONAL_CHAIN, cmd="add-flow", cookie=cookie, priority=10) + + # re-configure the VNFs IP assignment and ensure that a new subnet is used for each E-Link + src_vnfi = self._get_vnf_instance(instance_uuid, src_name) + if src_vnfi is not None: + self._vnf_reconfigure_network(src_vnfi, src_if_name, self.eline_subnets_src.pop(0)) + dst_vnfi = self._get_vnf_instance(instance_uuid, dst_name) + if dst_vnfi is not None: + self._vnf_reconfigure_network(dst_vnfi, dst_if_name, self.eline_subnets_dst.pop(0)) + + # 4b. deploy E-LAN links + base = 10 + for link in elan_fwd_links: + + elan_vnf_list=[] + + # generate lan ip address + ip = 1 + for intf in link["connection_points_reference"]: + ip_address = generate_lan_string("10.0", base, subnet_size=24, ip=ip) + vnf_id, intf_name = intf.split(":") + if vnf_id in self.sap_identifiers: + src_docker_name = "{0}_{1}".format(vnf_id, intf_name) + vnf_id = src_docker_name + else: + src_docker_name = vnf_id + vnf_name = vnf_id2vnf_name[vnf_id] + LOG.debug( + "Setting up E-LAN link. %s(%s:%s) -> %s" % ( + vnf_name, vnf_id, intf_name, ip_address)) + + if vnf_name in self.vnfds: + # re-configure the VNFs IP assignment and ensure that a new subnet is used for each E-LAN + # E-LAN relies on the learning switch capability of Ryu which has to be turned on in the topology + # (DCNetwork(controller=RemoteController, enable_learning=True)), so no explicit chaining is necessary. + vnfi = self._get_vnf_instance(instance_uuid, vnf_name) + if vnfi is not None: + self._vnf_reconfigure_network(vnfi, intf_name, ip_address) + # increase for the next ip address on this E-LAN + ip += 1 + + # add this vnf and interface to the E-LAN for tagging + network = self.vnfds[vnf_name].get("dc").net # there should be a cleaner way to find the DCNetwork + elan_vnf_list.append({'name':src_docker_name,'interface':intf_name}) + + + # install the VLAN tags for this E-LAN + network.setLAN(elan_vnf_list) + # increase the base ip address for the next E-LAN + base += 1 + + # 5. run the emulator specific entrypoint scripts in the VNFIs of this service instance self._trigger_emulator_start_scripts_in_vnfis(self.instances[instance_uuid]["vnf_instances"]) LOG.info("Service started. Instance id: %r" % instance_uuid) return instance_uuid + def stop_service(self, instance_uuid): + """ + This method stops a running service instance. + It iterates over all VNF instances, stopping them each + and removing them from their data center. + + :param instance_uuid: the uuid of the service instance to be stopped + """ + LOG.info("Stopping service %r" % self.uuid) + # get relevant information + # instance_uuid = str(self.uuid.uuid4()) + vnf_instances = self.instances[instance_uuid]["vnf_instances"] + + for v in vnf_instances: + self._stop_vnfi(v) + + if not GK_STANDALONE_MODE: + # remove placement? + # self._remove_placement(RoundRobinPlacement) + None + + # last step: remove the instance from the list of all instances + del self.instances[instance_uuid] + def _start_vnfd(self, vnfd): """ Start a single VNFD of this service @@ -222,15 +320,64 @@ class Service(object): assert(target_dc is not None) if not self._check_docker_image_exists(docker_name): raise Exception("Docker image %r not found. Abort." % docker_name) - # 3. do the dc.startCompute(name="foobar") call to run the container + + # 3. get the resource limits + res_req = u.get("resource_requirements") + cpu_list = res_req.get("cpu").get("cores") + if not cpu_list or len(cpu_list)==0: + cpu_list="1" + cpu_bw = res_req.get("cpu").get("cpu_bw") + if not cpu_bw: + cpu_bw=1 + mem_num = str(res_req.get("memory").get("size")) + if len(mem_num)==0: + mem_num="2" + mem_unit = str(res_req.get("memory").get("size_unit")) + if str(mem_unit)==0: + mem_unit="GB" + mem_limit = float(mem_num) + if mem_unit=="GB": + mem_limit=mem_limit*1024*1024*1024 + elif mem_unit=="MB": + mem_limit=mem_limit*1024*1024 + elif mem_unit=="KB": + mem_limit=mem_limit*1024 + mem_lim = int(mem_limit) + cpu_period, cpu_quota = self._calculate_cpu_cfs_values(float(cpu_bw)) + + # 4. do the dc.startCompute(name="foobar") call to run the container # TODO consider flavors, and other annotations intfs = vnfd.get("connection_points") - self.vnf_name2docker_name[vnf_name] = GK.get_next_vnf_name() + + # TODO: get all vnf id's from the nsd for this vnfd and use those as dockername + # use the vnf_id in the nsd as docker name + # so deployed containers can be easily mapped back to the nsd + vnf_name2id = defaultdict(lambda: "NotExistingNode", + reduce(lambda x, y: dict(x, **y), + map(lambda d: {d["vnf_name"]: d["vnf_id"]}, + self.nsd["network_functions"]))) + self.vnf_name2docker_name[vnf_name] = vnf_name2id[vnf_name] + # self.vnf_name2docker_name[vnf_name] = GK.get_next_vnf_name() + LOG.info("Starting %r as %r in DC %r" % (vnf_name, self.vnf_name2docker_name[vnf_name], vnfd.get("dc"))) LOG.debug("Interfaces for %r: %r" % (vnf_name, intfs)) - vnfi = target_dc.startCompute(self.vnf_name2docker_name[vnf_name], network=intfs, image=docker_name, flavor_name="small") + vnfi = target_dc.startCompute(self.vnf_name2docker_name[vnf_name], network=intfs, image=docker_name, flavor_name="small", + cpu_quota=cpu_quota, cpu_period=cpu_period, cpuset=cpu_list, mem_limit=mem_lim) return vnfi + def _stop_vnfi(self, vnfi): + """ + Stop a VNF instance. + + :param vnfi: vnf instance to be stopped + """ + # Find the correct datacenter + status = vnfi.getStatus() + dc = vnfi.datacenter + # stop the vnfi + LOG.info("Stopping the vnf instance contained in %r in DC %r" % (status["name"], dc)) + dc.stopCompute(status["name"]) + def _get_vnf_instance(self, instance_uuid, name): """ Returns the Docker object for the given VNF name (or Docker name). @@ -270,10 +417,14 @@ class Service(object): config = vnfi.dcinfo.get("Config", dict()) env = config.get("Env", list()) for env_var in env: - if "SON_EMU_CMD=" in env_var: - cmd = str(env_var.split("=")[1]) - LOG.info("Executing entrypoint script in %r: %r" % (vnfi.name, cmd)) - vnfi.cmdPrint(cmd) + var, cmd = map(str.strip, map(str, env_var.split('=', 1))) + LOG.debug("%r = %r" % (var , cmd)) + if var=="SON_EMU_CMD": + LOG.info("Executing entry point script in %r: %r" % (vnfi.name, cmd)) + # execute command in new thread to ensure that GK is not blocked by VNF + t = threading.Thread(target=vnfi.cmdPrint, args=(cmd,)) + t.daemon = True + t.start() def _unpack_service_package(self): """ @@ -303,6 +454,7 @@ class Service(object): self.package_content_path, make_relative_path(self.manifest.get("entry_service_template"))) self.nsd = load_yaml(nsd_path) + GK.net.deployed_nsds.append(self.nsd) LOG.debug("Loaded NSD: %r" % self.nsd.get("name")) def _load_vnfd(self): @@ -320,6 +472,28 @@ class Service(object): self.vnfds[vnfd.get("name")] = vnfd LOG.debug("Loaded VNFD: %r" % vnfd.get("name")) + def _load_saps(self): + # Each Service Access Point (connection_point) in the nsd is getting its own container + SAPs = [p["id"] for p in self.nsd["connection_points"] if p["type"] == "interface"] + for sap in SAPs: + # endpoints needed in this service + sap_vnf_id, sap_vnf_interface = sap.split(':') + # set of the connection_point ids found in the nsd (in the examples this is 'ns') + self.sap_identifiers.add(sap_vnf_id) + + sap_docker_name = "%s_%s" % (sap_vnf_id, sap_vnf_interface) + + # add SAP to self.vnfds + sapfile = pkg_resources.resource_filename(__name__, "sap_vnfd.yml") + sap_vnfd = load_yaml(sapfile) + sap_vnfd["connection_points"][0]["id"] = sap_vnf_interface + sap_vnfd["name"] = sap_docker_name + self.vnfds[sap_docker_name] = sap_vnfd + # add SAP vnf to list in the NSD so it is deployed later on + # each SAP get a unique VNFD and vnf_id in the NSD + self.nsd["network_functions"].append({"vnf_id": sap_docker_name, "vnf_name": sap_docker_name}) + LOG.debug("Loaded SAP: %r" % sap_vnfd.get("name")) + def _load_docker_files(self): """ Get all paths to Dockerfiles from VNFDs and store them in dict. @@ -369,12 +543,22 @@ class Service(object): dc = DockerClient() for url in self.remote_docker_image_urls.itervalues(): if not FORCE_PULL: # only pull if not present (speedup for development) - if len(dc.images(name=url)) > 0: + if len(dc.images.list(name=url)) > 0: LOG.debug("Image %r present. Skipping pull." % url) continue LOG.info("Pulling image: %r" % url) - dc.pull(url, - insecure_registry=True) + # this seems to fail with latest docker api version 2.0.2 + # dc.images.pull(url, + # insecure_registry=True) + #using docker cli instead + cmd = ["docker", + "pull", + url, + ] + Popen(cmd).wait() + + + def _check_docker_image_exists(self, image_name): """ @@ -382,7 +566,7 @@ class Service(object): :param image_name: name of the docker image :return: """ - return len(DockerClient().images(image_name)) > 0 + return len(DockerClient().images.list(name=image_name)) > 0 def _calculate_placement(self, algorithm): """ @@ -400,6 +584,29 @@ class Service(object): for name, vnfd in self.vnfds.iteritems(): LOG.info("Placed VNF %r on DC %r" % (name, str(vnfd.get("dc")))) + def _calculate_cpu_cfs_values(self, cpu_time_percentage): + """ + Calculate cpu period and quota for CFS + :param cpu_time_percentage: percentage of overall CPU to be used + :return: cpu_period, cpu_quota + """ + if cpu_time_percentage is None: + return -1, -1 + if cpu_time_percentage < 0: + return -1, -1 + # (see: https://www.kernel.org/doc/Documentation/scheduler/sched-bwc.txt) + # Attention minimum cpu_quota is 1ms (micro) + cpu_period = 1000000 # lets consider a fixed period of 1000000 microseconds for now + LOG.debug("cpu_period is %r, cpu_percentage is %r" % (cpu_period, cpu_time_percentage)) + cpu_quota = cpu_period * cpu_time_percentage # calculate the fraction of cpu time for this container + # ATTENTION >= 1000 to avoid a invalid argument system error ... no idea why + if cpu_quota < 1000: + LOG.debug("cpu_quota before correcting: %r" % cpu_quota) + cpu_quota = 1000 + LOG.warning("Increased CPU quota to avoid system error.") + LOG.debug("Calculated: cpu_period=%f / cpu_quota=%f" % (cpu_period, cpu_quota)) + return int(cpu_period), int(cpu_quota) + """ Some (simple) placement algorithms @@ -415,6 +622,20 @@ class FirstDcPlacement(object): vnfd["dc"] = list(dcs.itervalues())[0] +class RoundRobinDcPlacement(object): + """ + Placement: Distribute VNFs across all available DCs in a round robin fashion. + """ + def place(self, nsd, vnfds, dcs): + c = 0 + dcs_list = list(dcs.itervalues()) + for name, vnfd in vnfds.iteritems(): + vnfd["dc"] = dcs_list[c % len(dcs_list)] + c += 1 # inc. c to use next DC + + + + """ Resource definitions and API endpoints """ @@ -452,7 +673,7 @@ class Packages(fr.Resource): s = Service(service_uuid, file_hash, upload_path) GK.register_service_package(service_uuid, s) # generate the JSON result - return {"service_uuid": service_uuid, "size": size, "sha1": file_hash, "error": None} + return {"service_uuid": service_uuid, "size": size, "sha1": file_hash, "error": None}, 201 except Exception as ex: LOG.exception("Service package upload failed:") return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed"}, 500 @@ -462,6 +683,7 @@ class Packages(fr.Resource): Return a list of UUID's of uploaded service packages. :return: dict/list """ + LOG.info("GET /packages") return {"service_uuid_list": list(GK.services.iterkeys())} @@ -485,7 +707,7 @@ class Instantiations(fr.Resource): if service_uuid in GK.services: # ok, we have a service uuid, lets start the service service_instance_uuid = GK.services.get(service_uuid).start_service() - return {"service_instance_uuid": service_instance_uuid} + return {"service_instance_uuid": service_instance_uuid}, 201 return "Service not found", 404 def get(self): @@ -493,12 +715,50 @@ class Instantiations(fr.Resource): Returns a list of UUIDs containing all running services. :return: dict / list """ - return {"service_instance_list": [ + LOG.info("GET /instantiations") + return {"service_instantiations_list": [ list(s.instances.iterkeys()) for s in GK.services.itervalues()]} + def delete(self): + """ + Stops a running service specified by its service and instance UUID. + """ + # try to extract the service and instance UUID from the request + json_data = request.get_json(force=True) + service_uuid = json_data.get("service_uuid") + instance_uuid = json_data.get("service_instance_uuid") + + # try to be fuzzy + if service_uuid is None and len(GK.services) > 0: + #if we don't get a service uuid, we simply stop the last service in the list + service_uuid = list(GK.services.iterkeys())[0] + if instance_uuid is None and len(GK.services[service_uuid].instances) > 0: + instance_uuid = list(GK.services[service_uuid].instances.iterkeys())[0] + + if service_uuid in GK.services and instance_uuid in GK.services[service_uuid].instances: + # valid service and instance UUID, stop service + GK.services.get(service_uuid).stop_service(instance_uuid) + return "service instance with uuid %r stopped." % instance_uuid,200 + return "Service not found", 404 + +class Exit(fr.Resource): + + def put(self): + """ + Stop the running Containernet instance regardless of data transmitted + """ + list(GK.dcs.values())[0].net.stop() + + +def initialize_GK(): + global GK + GK = Gatekeeper() + + # create a single, global GK object -GK = Gatekeeper() +GK = None +initialize_GK() # setup Flask app = Flask(__name__) app.config['MAX_CONTENT_LENGTH'] = 512 * 1024 * 1024 # 512 MB max upload @@ -506,10 +766,17 @@ api = fr.Api(app) # define endpoints api.add_resource(Packages, '/packages') api.add_resource(Instantiations, '/instantiations') +api.add_resource(Exit, '/emulator/exit') + + +#def initialize_GK(): +# global GK +# GK = Gatekeeper() def start_rest_api(host, port, datacenters=dict()): GK.dcs = datacenters + GK.net = get_dc_network() # start the Flask server (not the best performance but ok for our use case) app.run(host=host, port=port, @@ -541,6 +808,14 @@ def make_relative_path(path): return path +def generate_lan_string(prefix, base, subnet_size=24, ip=0): + """ + Helper to generate different network configuration strings. + """ + r = "%s.%d.%d/%d" % (prefix, base, ip, subnet_size) + return r + + def generate_subnet_strings(n, start=1, subnet_size=24, ip=0): """ Helper to generate different network configuration strings. @@ -550,6 +825,14 @@ def generate_subnet_strings(n, start=1, subnet_size=24, ip=0): r.append("%d.0.0.%d/%d" % (i, ip, subnet_size)) return r +def get_dc_network(): + """ + retrieve the DCnetwork where this dummygatekeeper (GK) connects to. + Assume at least 1 datacenter is connected to this GK, and that all datacenters belong to the same DCNetwork + :return: + """ + assert (len(GK.dcs) > 0) + return GK.dcs.values()[0].net if __name__ == '__main__': """