src/emuvim/api/sonata/dummygatekeeper.py

   1 """
   2 Copyright (c) 2015 SONATA-NFV and Paderborn University
   3 ALL RIGHTS RESERVED.
   4
   5 Licensed under the Apache License, Version 2.0 (the "License");
   6 you may not use this file except in compliance with the License.
   7 You may obtain a copy of the License at
   8
   9     http://www.apache.org/licenses/LICENSE-2.0
  10
  11 Unless required by applicable law or agreed to in writing, software
  12 distributed under the License is distributed on an "AS IS" BASIS,
  13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 See the License for the specific language governing permissions and
  15 limitations under the License.
  16
  17 Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
  18 nor the names of its contributors may be used to endorse or promote
  19 products derived from this software without specific prior written
  20 permission.
  21
  22 This work has been performed in the framework of the SONATA project,
  23 funded by the European Commission under Grant number 671517 through
  24 the Horizon 2020 and 5G-PPP programmes. The authors would like to
  25 acknowledge the contributions of their colleagues of the SONATA
  26 partner consortium (www.sonata-nfv.eu).
  27 """
  28 """
  29 This module implements a simple REST API that behaves like SONATA's gatekeeper.
  30
  31 It is only used to support the development of SONATA's SDK tools and to demonstrate
  32 the year 1 version of the emulator until the integration with WP4's orchestrator is done.
  33 """
  34
  35 import logging
  36 import os
  37 import uuid
  38 import hashlib
  39 import zipfile
  40 import yaml
  41 import threading
  42 from docker import DockerClient, APIClient
  43 from flask import Flask, request
  44 import flask_restful as fr
  45 from collections import defaultdict
  46 import pkg_resources
  47 from subprocess import Popen
  48 from random import randint
  49 import ipaddress
  50
  51 logging.basicConfig()
  52 LOG = logging.getLogger("sonata-dummy-gatekeeper")
  53 LOG.setLevel(logging.DEBUG)
  54 logging.getLogger("werkzeug").setLevel(logging.WARNING)
  55
  56 GK_STORAGE = "/tmp/son-dummy-gk/"
  57 UPLOAD_FOLDER = os.path.join(GK_STORAGE, "uploads/")
  58 CATALOG_FOLDER = os.path.join(GK_STORAGE, "catalog/")
  59
  60 # Enable Dockerfile build functionality
  61 BUILD_DOCKERFILE = False
  62
  63 # flag to indicate that we run without the emulator (only the bare API for integration testing)
  64 GK_STANDALONE_MODE = False
  65
  66 # should a new version of an image be pulled even if its available
  67 FORCE_PULL = False
  68
  69 # Automatically deploy SAPs (endpoints) of the service as new containers
  70 # Attention: This is not a configuration switch but a global variable! Don't change its default value.
  71 DEPLOY_SAP = False
  72
  73 # flag to indicate if we use bidirectional forwarding rules in the automatic chaining process
  74 BIDIRECTIONAL_CHAIN = False
  75
  76 # override the management interfaces in the descriptors with default docker0 interfaces in the containers
  77 USE_DOCKER_MGMT = True
  78
  79 def generate_subnets(prefix, base, subnet_size=50, mask=24):
  80     # Generate a list of ipaddress in subnets
  81     r = list()
  82     for net in range(base, base + subnet_size):
  83         subnet = "{0}.{1}.0/{2}".format(prefix, net, mask)
  84         r.append(ipaddress.ip_network(unicode(subnet)))
  85     return r
  86 # private subnet definitions for the generated interfaces
  87 # 10.10.xxx.0/24
  88 SAP_SUBNETS = generate_subnets('10.10', 0, subnet_size=50, mask=24)
  89 # 10.20.xxx.0/24
  90 ELAN_SUBNETS = generate_subnets('10.20', 0, subnet_size=50, mask=24)
  91 # 10.30.xxx.0/30
  92 ELINE_SUBNETS = generate_subnets('10.30', 0, subnet_size=50, mask=30)
  93
  94
  95 class Gatekeeper(object):
  96
  97     def __init__(self):
  98         self.services = dict()
  99         self.dcs = dict()
 100         self.net = None
 101         self.vnf_counter = 0  # used to generate short names for VNFs (Mininet limitation)
 102         LOG.info("Create SONATA dummy gatekeeper.")
 103
 104     def register_service_package(self, service_uuid, service):
 105         """
 106         register new service package
 107         :param service_uuid
 108         :param service object
 109         """
 110         self.services[service_uuid] = service
 111         # lets perform all steps needed to onboard the service
 112         service.onboard()
 113
 114     def get_next_vnf_name(self):
 115         self.vnf_counter += 1
 116         return "vnf%d" % self.vnf_counter
 117
 118
 119 class Service(object):
 120     """
 121     This class represents a NS uploaded as a *.son package to the
 122     dummy gatekeeper.
 123     Can have multiple running instances of this service.
 124     """
 125
 126     def __init__(self,
 127                  service_uuid,
 128                  package_file_hash,
 129                  package_file_path):
 130         self.uuid = service_uuid
 131         self.package_file_hash = package_file_hash
 132         self.package_file_path = package_file_path
 133         self.package_content_path = os.path.join(CATALOG_FOLDER, "services/%s" % self.uuid)
 134         self.manifest = None
 135         self.nsd = None
 136         self.vnfds = dict()
 137         self.saps = dict()
 138         self.saps_ext = list()
 139         self.saps_int = list()
 140         self.local_docker_files = dict()
 141         self.remote_docker_image_urls = dict()
 142         self.instances = dict()
 143         self.vnf_name2docker_name = dict()
 144         self.vnf_id2vnf_name = dict()
 145
 146     def onboard(self):
 147         """
 148         Do all steps to prepare this service to be instantiated
 149         :return:
 150         """
 151         # 1. extract the contents of the package and store them in our catalog
 152         self._unpack_service_package()
 153         # 2. read in all descriptor files
 154         self._load_package_descriptor()
 155         self._load_nsd()
 156         self._load_vnfd()
 157         if DEPLOY_SAP:
 158             self._load_saps()
 159         # create dict to translate vnf names
 160         self.vnf_id2vnf_name = defaultdict(lambda: "NotExistingNode",
 161                                            reduce(lambda x, y: dict(x, **y),
 162                                                   map(lambda d: {d["vnf_id"]: d["vnf_name"]},
 163                                                       self.nsd["network_functions"])))
 164         # 3. prepare container images (e.g. download or build Dockerfile)
 165         if BUILD_DOCKERFILE:
 166             self._load_docker_files()
 167             self._build_images_from_dockerfiles()
 168         else:
 169             self._load_docker_urls()
 170             self._pull_predefined_dockerimages()
 171         LOG.info("On-boarded service: %r" % self.manifest.get("name"))
 172
 173     def start_service(self):
 174         """
 175         This methods creates and starts a new service instance.
 176         It computes placements, iterates over all VNFDs, and starts
 177         each VNFD as a Docker container in the data center selected
 178         by the placement algorithm.
 179         :return:
 180         """
 181         LOG.info("Starting service %r" % self.uuid)
 182
 183         # 1. each service instance gets a new uuid to identify it
 184         instance_uuid = str(uuid.uuid4())
 185         # build a instances dict (a bit like a NSR :))
 186         self.instances[instance_uuid] = dict()
 187         self.instances[instance_uuid]["vnf_instances"] = list()
 188
 189         # 2. compute placement of this service instance (adds DC names to VNFDs)
 190         if not GK_STANDALONE_MODE:
 191             #self._calculate_placement(FirstDcPlacement)
 192             self._calculate_placement(RoundRobinDcPlacementWithSAPs)
 193
 194         # 3. start all vnfds that we have in the service (except SAPs)
 195         for vnfd in self.vnfds.itervalues():
 196             vnfi = None
 197             if not GK_STANDALONE_MODE:
 198                 vnfi = self._start_vnfd(vnfd)
 199             self.instances[instance_uuid]["vnf_instances"].append(vnfi)
 200
 201         # 4. start all SAPs in the service
 202         for sap in self.saps:
 203             self._start_sap(self.saps[sap], instance_uuid)
 204
 205         # 5. Deploy E-Line and E_LAN links
 206         if "virtual_links" in self.nsd:
 207             vlinks = self.nsd["virtual_links"]
 208             # constituent virtual links are not checked
 209             #fwd_links = self.nsd["forwarding_graphs"][0]["constituent_virtual_links"]
 210             eline_fwd_links = [l for l in vlinks if (l["connectivity_type"] == "E-Line")]
 211             elan_fwd_links = [l for l in vlinks if (l["connectivity_type"] == "E-LAN")]
 212
 213             GK.net.deployed_elines.extend(eline_fwd_links)
 214             GK.net.deployed_elans.extend(elan_fwd_links)
 215
 216             # 5a. deploy E-Line links
 217             self._connect_elines(eline_fwd_links, instance_uuid)
 218
 219             # 5b. deploy E-LAN links
 220             self._connect_elans(elan_fwd_links, instance_uuid)
 221
 222         # 6. run the emulator specific entrypoint scripts in the VNFIs of this service instance
 223         self._trigger_emulator_start_scripts_in_vnfis(self.instances[instance_uuid]["vnf_instances"])
 224
 225         LOG.info("Service started. Instance id: %r" % instance_uuid)
 226         return instance_uuid
 227
 228     def stop_service(self, instance_uuid):
 229         """
 230         This method stops a running service instance.
 231         It iterates over all VNF instances, stopping them each
 232         and removing them from their data center.
 233
 234         :param instance_uuid: the uuid of the service instance to be stopped
 235         """
 236         LOG.info("Stopping service %r" % self.uuid)
 237         # get relevant information
 238         # instance_uuid = str(self.uuid.uuid4())
 239         vnf_instances = self.instances[instance_uuid]["vnf_instances"]
 240
 241         for v in vnf_instances:
 242             self._stop_vnfi(v)
 243
 244         if not GK_STANDALONE_MODE:
 245             # remove placement?
 246             # self._remove_placement(RoundRobinPlacement)
 247             None
 248
 249         # last step: remove the instance from the list of all instances
 250         del self.instances[instance_uuid]
 251
 252     def _start_vnfd(self, vnfd):
 253         """
 254         Start a single VNFD of this service
 255         :param vnfd: vnfd descriptor dict
 256         :return:
 257         """
 258         # iterate over all deployment units within each VNFDs
 259         for u in vnfd.get("virtual_deployment_units"):
 260             # 1. get the name of the docker image to start and the assigned DC
 261             vnf_name = vnfd.get("name")
 262             if vnf_name not in self.remote_docker_image_urls:
 263                 raise Exception("No image name for %r found. Abort." % vnf_name)
 264             docker_name = self.remote_docker_image_urls.get(vnf_name)
 265             target_dc = vnfd.get("dc")
 266             # 2. perform some checks to ensure we can start the container
 267             assert(docker_name is not None)
 268             assert(target_dc is not None)
 269             if not self._check_docker_image_exists(docker_name):
 270                 raise Exception("Docker image %r not found. Abort." % docker_name)
 271
 272             # 3. get the resource limits
 273             res_req = u.get("resource_requirements")
 274             cpu_list = res_req.get("cpu").get("cores")
 275             if not cpu_list or len(cpu_list)==0:
 276                 cpu_list="1"
 277             cpu_bw = res_req.get("cpu").get("cpu_bw")
 278             if not cpu_bw:
 279                 cpu_bw=1
 280             mem_num = str(res_req.get("memory").get("size"))
 281             if len(mem_num)==0:
 282                 mem_num="2"
 283             mem_unit = str(res_req.get("memory").get("size_unit"))
 284             if str(mem_unit)==0:
 285                 mem_unit="GB"
 286             mem_limit = float(mem_num)
 287             if mem_unit=="GB":
 288                 mem_limit=mem_limit*1024*1024*1024
 289             elif mem_unit=="MB":
 290                 mem_limit=mem_limit*1024*1024
 291             elif mem_unit=="KB":
 292                 mem_limit=mem_limit*1024
 293             mem_lim = int(mem_limit)
 294             cpu_period, cpu_quota = self._calculate_cpu_cfs_values(float(cpu_bw))
 295
 296             vnf_name2id = defaultdict(lambda: "NotExistingNode",
 297                                       reduce(lambda x, y: dict(x, **y),
 298                                              map(lambda d: {d["vnf_name"]: d["vnf_id"]},
 299                                                  self.nsd["network_functions"])))
 300
 301             # check if we need to deploy the management ports (defined as type:management both on in the vnfd and nsd)
 302             intfs = vnfd.get("connection_points", [])
 303             mgmt_intf_names = []
 304             if USE_DOCKER_MGMT:
 305                 vnf_id = vnf_name2id[vnf_name]
 306                 mgmt_intfs = [vnf_id + ':' + intf['id'] for intf in intfs if intf.get('type') == 'management']
 307                 # check if any of these management interfaces are used in a management-type network in the nsd
 308                 for nsd_intf_name in mgmt_intfs:
 309                     vlinks = [ l["connection_points_reference"] for l in self.nsd.get("virtual_links", [])]
 310                     for link in vlinks:
 311                         if nsd_intf_name in link and self.check_mgmt_interface(link):
 312                             # this is indeed a management interface and can be skipped
 313                             vnf_id, vnf_interface, vnf_sap_docker_name = parse_interface(nsd_intf_name)
 314                             found_interfaces = [intf for intf in intfs if intf.get('id') == vnf_interface]
 315                             intfs.remove(found_interfaces[0])
 316                             mgmt_intf_names.append(vnf_interface)
 317
 318             # 4. do the dc.startCompute(name="foobar") call to run the container
 319             # TODO consider flavors, and other annotations
 320             # TODO: get all vnf id's from the nsd for this vnfd and use those as dockername
 321             # use the vnf_id in the nsd as docker name
 322             # so deployed containers can be easily mapped back to the nsd
 323
 324             self.vnf_name2docker_name[vnf_name] = vnf_name2id[vnf_name]
 325
 326             LOG.info("Starting %r as %r in DC %r" % (vnf_name, self.vnf_name2docker_name[vnf_name], vnfd.get("dc")))
 327             LOG.debug("Interfaces for %r: %r" % (vnf_name, intfs))
 328             vnfi = target_dc.startCompute(self.vnf_name2docker_name[vnf_name], network=intfs, image=docker_name, flavor_name="small",
 329                     cpu_quota=cpu_quota, cpu_period=cpu_period, cpuset=cpu_list, mem_limit=mem_lim)
 330
 331             # rename the docker0 interfaces (eth0) to the management port name defined in the VNFD
 332             if USE_DOCKER_MGMT:
 333                 for intf_name in mgmt_intf_names:
 334                     self._vnf_reconfigure_network(vnfi, 'eth0', new_name=intf_name)
 335
 336             return vnfi
 337
 338     def _stop_vnfi(self, vnfi):
 339         """
 340         Stop a VNF instance.
 341
 342         :param vnfi: vnf instance to be stopped
 343         """
 344         # Find the correct datacenter
 345         status = vnfi.getStatus()
 346         dc = vnfi.datacenter
 347         # stop the vnfi
 348         LOG.info("Stopping the vnf instance contained in %r in DC %r" % (status["name"], dc))
 349         dc.stopCompute(status["name"])
 350
 351     def _get_vnf_instance(self, instance_uuid, name):
 352         """
 353         Returns the Docker object for the given VNF name (or Docker name).
 354         :param instance_uuid: UUID of the service instance to search in.
 355         :param name: VNF name or Docker name. We are fuzzy here.
 356         :return:
 357         """
 358         dn = name
 359         if name in self.vnf_name2docker_name:
 360             dn = self.vnf_name2docker_name[name]
 361         for vnfi in self.instances[instance_uuid]["vnf_instances"]:
 362             if vnfi.name == dn:
 363                 return vnfi
 364         LOG.warning("No container with name: {0} found.".format(dn))
 365         return None
 366
 367     @staticmethod
 368     def _vnf_reconfigure_network(vnfi, if_name, net_str=None, new_name=None):
 369         """
 370         Reconfigure the network configuration of a specific interface
 371         of a running container.
 372         :param vnfi: container instance
 373         :param if_name: interface name
 374         :param net_str: network configuration string, e.g., 1.2.3.4/24
 375         :return:
 376         """
 377
 378         # assign new ip address
 379         if net_str is not None:
 380             intf = vnfi.intf(intf=if_name)
 381             if intf is not None:
 382                 intf.setIP(net_str)
 383                 LOG.debug("Reconfigured network of %s:%s to %r" % (vnfi.name, if_name, net_str))
 384             else:
 385                 LOG.warning("Interface not found: %s:%s. Network reconfiguration skipped." % (vnfi.name, if_name))
 386
 387         if new_name is not None:
 388             vnfi.cmd('ip link set', if_name, 'down')
 389             vnfi.cmd('ip link set', if_name, 'name', new_name)
 390             vnfi.cmd('ip link set', new_name, 'up')
 391             LOG.debug("Reconfigured interface name of %s:%s to %s" % (vnfi.name, if_name, new_name))
 392
 393
 394
 395     def _trigger_emulator_start_scripts_in_vnfis(self, vnfi_list):
 396         for vnfi in vnfi_list:
 397             config = vnfi.dcinfo.get("Config", dict())
 398             env = config.get("Env", list())
 399             for env_var in env:
 400                 var, cmd = map(str.strip, map(str, env_var.split('=', 1)))
 401                 LOG.debug("%r = %r" % (var , cmd))
 402                 if var=="SON_EMU_CMD":
 403                     LOG.info("Executing entry point script in %r: %r" % (vnfi.name, cmd))
 404                     # execute command in new thread to ensure that GK is not blocked by VNF
 405                     t = threading.Thread(target=vnfi.cmdPrint, args=(cmd,))
 406                     t.daemon = True
 407                     t.start()
 408
 409     def _unpack_service_package(self):
 410         """
 411         unzip *.son file and store contents in CATALOG_FOLDER/services/<service_uuid>/
 412         """
 413         LOG.info("Unzipping: %r" % self.package_file_path)
 414         with zipfile.ZipFile(self.package_file_path, "r") as z:
 415             z.extractall(self.package_content_path)
 416
 417
 418     def _load_package_descriptor(self):
 419         """
 420         Load the main package descriptor YAML and keep it as dict.
 421         :return:
 422         """
 423         self.manifest = load_yaml(
 424             os.path.join(
 425                 self.package_content_path, "META-INF/MANIFEST.MF"))
 426
 427     def _load_nsd(self):
 428         """
 429         Load the entry NSD YAML and keep it as dict.
 430         :return:
 431         """
 432         if "entry_service_template" in self.manifest:
 433             nsd_path = os.path.join(
 434                 self.package_content_path,
 435                 make_relative_path(self.manifest.get("entry_service_template")))
 436             self.nsd = load_yaml(nsd_path)
 437             GK.net.deployed_nsds.append(self.nsd)
 438
 439             LOG.debug("Loaded NSD: %r" % self.nsd.get("name"))
 440
 441     def _load_vnfd(self):
 442         """
 443         Load all VNFD YAML files referenced in MANIFEST.MF and keep them in dict.
 444         :return:
 445         """
 446         if "package_content" in self.manifest:
 447             for pc in self.manifest.get("package_content"):
 448                 if pc.get("content-type") == "application/sonata.function_descriptor":
 449                     vnfd_path = os.path.join(
 450                         self.package_content_path,
 451                         make_relative_path(pc.get("name")))
 452                     vnfd = load_yaml(vnfd_path)
 453                     self.vnfds[vnfd.get("name")] = vnfd
 454                     LOG.debug("Loaded VNFD: %r" % vnfd.get("name"))
 455
 456     def _load_saps(self):
 457         # create list of all SAPs
 458         # check if we need to deploy management ports
 459         if USE_DOCKER_MGMT:
 460             SAPs = [p for p in self.nsd["connection_points"] if 'management' not in p.get('type')]
 461         else:
 462             SAPs = [p for p in self.nsd["connection_points"]]
 463
 464         for sap in SAPs:
 465             # endpoint needed in this service
 466             sap_id, sap_interface, sap_docker_name = parse_interface(sap['id'])
 467             # make sure SAP has type set (default internal)
 468             sap["type"] = sap.get("type", 'internal')
 469
 470             # Each Service Access Point (connection_point) in the nsd is an IP address on the host
 471             if sap["type"] == "external":
 472                 # add to vnfds to calculate placement later on
 473                 sap_net = SAP_SUBNETS.pop(0)
 474                 self.saps[sap_docker_name] = {"name": sap_docker_name , "type": "external", "net": sap_net}
 475                 # add SAP vnf to list in the NSD so it is deployed later on
 476                 # each SAP get a unique VNFD and vnf_id in the NSD and custom type (only defined in the dummygatekeeper)
 477                 self.nsd["network_functions"].append(
 478                     {"vnf_id": sap_docker_name, "vnf_name": sap_docker_name, "vnf_type": "sap_ext"})
 479
 480             # Each Service Access Point (connection_point) in the nsd is getting its own container (default)
 481             elif sap["type"] == "internal" or sap["type"] == "management":
 482                 # add SAP to self.vnfds
 483                 sapfile = pkg_resources.resource_filename(__name__, "sap_vnfd.yml")
 484                 sap_vnfd = load_yaml(sapfile)
 485                 sap_vnfd["connection_points"][0]["id"] = sap_interface
 486                 sap_vnfd["name"] = sap_docker_name
 487                 sap_vnfd["type"] = "internal"
 488                 # add to vnfds to calculate placement later on and deploy
 489                 self.saps[sap_docker_name] = sap_vnfd
 490                 # add SAP vnf to list in the NSD so it is deployed later on
 491                 # each SAP get a unique VNFD and vnf_id in the NSD
 492                 self.nsd["network_functions"].append(
 493                     {"vnf_id": sap_docker_name, "vnf_name": sap_docker_name, "vnf_type": "sap_int"})
 494
 495             LOG.debug("Loaded SAP: name: {0}, type: {1}".format(sap_docker_name, sap['type']))
 496
 497         # create sap lists
 498         self.saps_ext = [self.saps[sap]['name'] for sap in self.saps if self.saps[sap]["type"] == "external"]
 499         self.saps_int = [self.saps[sap]['name'] for sap in self.saps if self.saps[sap]["type"] == "internal"]
 500
 501     def _start_sap(self, sap, instance_uuid):
 502         if not DEPLOY_SAP:
 503             return
 504
 505         LOG.info('start SAP: {0} ,type: {1}'.format(sap['name'],sap['type']))
 506         if sap["type"] == "internal":
 507             vnfi = None
 508             if not GK_STANDALONE_MODE:
 509                 vnfi = self._start_vnfd(sap)
 510             self.instances[instance_uuid]["vnf_instances"].append(vnfi)
 511
 512         elif sap["type"] == "external":
 513             target_dc = sap.get("dc")
 514             # add interface to dc switch
 515             target_dc.attachExternalSAP(sap['name'], str(sap['net']))
 516
 517     def _connect_elines(self, eline_fwd_links, instance_uuid):
 518         """
 519         Connect all E-LINE links in the NSD
 520         :param eline_fwd_links: list of E-LINE links in the NSD
 521         :param: instance_uuid of the service
 522         :return:
 523         """
 524         # cookie is used as identifier for the flowrules installed by the dummygatekeeper
 525         # eg. different services get a unique cookie for their flowrules
 526         cookie = 1
 527         for link in eline_fwd_links:
 528             # check if we need to deploy this link when its a management link:
 529             if USE_DOCKER_MGMT:
 530                 if self.check_mgmt_interface(link["connection_points_reference"]):
 531                     continue
 532
 533             src_id, src_if_name, src_sap_id = parse_interface(link["connection_points_reference"][0])
 534             dst_id, dst_if_name, dst_sap_id = parse_interface(link["connection_points_reference"][1])
 535
 536             setChaining = False
 537             # check if there is a SAP in the link and chain everything together
 538             if src_sap_id in self.saps and dst_sap_id in self.saps:
 539                 LOG.info('2 SAPs cannot be chained together : {0} - {1}'.format(src_sap_id, dst_sap_id))
 540                 continue
 541
 542             elif src_sap_id in self.saps_ext:
 543                 src_id = src_sap_id
 544                 src_if_name = src_sap_id
 545                 src_name = self.vnf_id2vnf_name[src_id]
 546                 dst_name = self.vnf_id2vnf_name[dst_id]
 547                 dst_vnfi = self._get_vnf_instance(instance_uuid, dst_name)
 548                 if dst_vnfi is not None:
 549                     # choose first ip address in sap subnet
 550                     sap_net = self.saps[src_sap_id]['net']
 551                     sap_ip = "{0}/{1}".format(str(sap_net[1]), sap_net.prefixlen)
 552                     self._vnf_reconfigure_network(dst_vnfi, dst_if_name, sap_ip)
 553                     setChaining = True
 554
 555             elif dst_sap_id in self.saps_ext:
 556                 dst_id = dst_sap_id
 557                 dst_if_name = dst_sap_id
 558                 src_name = self.vnf_id2vnf_name[src_id]
 559                 dst_name = self.vnf_id2vnf_name[dst_id]
 560                 src_vnfi = self._get_vnf_instance(instance_uuid, src_name)
 561                 if src_vnfi is not None:
 562                     sap_net = self.saps[dst_sap_id]['net']
 563                     sap_ip = "{0}/{1}".format(str(sap_net[1]), sap_net.prefixlen)
 564                     self._vnf_reconfigure_network(src_vnfi, src_if_name, sap_ip)
 565                     setChaining = True
 566
 567             # Link between 2 VNFs
 568             else:
 569                 # make sure we use the correct sap vnf name
 570                 if src_sap_id in self.saps_int:
 571                     src_id = src_sap_id
 572                 if dst_sap_id in self.saps_int:
 573                     dst_id = dst_sap_id
 574                 src_name = self.vnf_id2vnf_name[src_id]
 575                 dst_name = self.vnf_id2vnf_name[dst_id]
 576                 # re-configure the VNFs IP assignment and ensure that a new subnet is used for each E-Link
 577                 src_vnfi = self._get_vnf_instance(instance_uuid, src_name)
 578                 dst_vnfi = self._get_vnf_instance(instance_uuid, dst_name)
 579                 if src_vnfi is not None and dst_vnfi is not None:
 580                     eline_net = ELINE_SUBNETS.pop(0)
 581                     ip1 = "{0}/{1}".format(str(eline_net[1]), eline_net.prefixlen)
 582                     ip2 = "{0}/{1}".format(str(eline_net[2]), eline_net.prefixlen)
 583                     self._vnf_reconfigure_network(src_vnfi, src_if_name, ip1)
 584                     self._vnf_reconfigure_network(dst_vnfi, dst_if_name, ip2)
 585                     setChaining = True
 586
 587             # Set the chaining
 588             if setChaining:
 589                 ret = GK.net.setChain(
 590                     src_id, dst_id,
 591                     vnf_src_interface=src_if_name, vnf_dst_interface=dst_if_name,
 592                     bidirectional=BIDIRECTIONAL_CHAIN, cmd="add-flow", cookie=cookie, priority=10)
 593                 LOG.debug(
 594                     "Setting up E-Line link. %s(%s:%s) -> %s(%s:%s)" % (
 595                         src_name, src_id, src_if_name, dst_name, dst_id, dst_if_name))
 596
 597
 598     def _connect_elans(self, elan_fwd_links, instance_uuid):
 599         """
 600         Connect all E-LAN links in the NSD
 601         :param elan_fwd_links: list of E-LAN links in the NSD
 602         :param: instance_uuid of the service
 603         :return:
 604         """
 605         for link in elan_fwd_links:
 606             # check if we need to deploy this link when its a management link:
 607             if USE_DOCKER_MGMT:
 608                 if self.check_mgmt_interface(link["connection_points_reference"]):
 609                     continue
 610
 611             elan_vnf_list = []
 612             # check if an external SAP is in the E-LAN (then a subnet is already defined)
 613             intfs_elan = [intf for intf in link["connection_points_reference"]]
 614             lan_sap = self.check_ext_saps(intfs_elan)
 615             if lan_sap:
 616                 lan_net = self.saps[lan_sap]['net']
 617                 lan_hosts = list(lan_net.hosts())
 618                 sap_ip = str(lan_hosts.pop(0))
 619             else:
 620                 lan_net = ELAN_SUBNETS.pop(0)
 621                 lan_hosts = list(lan_net.hosts())
 622
 623             # generate lan ip address for all interfaces except external SAPs
 624             for intf in link["connection_points_reference"]:
 625
 626                 # skip external SAPs, they already have an ip
 627                 vnf_id, vnf_interface, vnf_sap_docker_name = parse_interface(intf)
 628                 if vnf_sap_docker_name in self.saps_ext:
 629                     elan_vnf_list.append({'name': vnf_sap_docker_name, 'interface': vnf_interface})
 630                     continue
 631
 632                 ip_address = "{0}/{1}".format(str(lan_hosts.pop(0)), lan_net.prefixlen)
 633                 vnf_id, intf_name, vnf_sap_id = parse_interface(intf)
 634
 635                 # make sure we use the correct sap vnf name
 636                 src_docker_name = vnf_id
 637                 if vnf_sap_id in self.saps_int:
 638                     src_docker_name = vnf_sap_id
 639                     vnf_id = vnf_sap_id
 640
 641                 vnf_name = self.vnf_id2vnf_name[vnf_id]
 642                 LOG.debug(
 643                     "Setting up E-LAN interface. %s(%s:%s) -> %s" % (
 644                         vnf_name, vnf_id, intf_name, ip_address))
 645
 646                 if vnf_name in self.vnfds:
 647                     # re-configure the VNFs IP assignment and ensure that a new subnet is used for each E-LAN
 648                     # E-LAN relies on the learning switch capability of Ryu which has to be turned on in the topology
 649                     # (DCNetwork(controller=RemoteController, enable_learning=True)), so no explicit chaining is necessary.
 650                     vnfi = self._get_vnf_instance(instance_uuid, vnf_name)
 651                     if vnfi is not None:
 652                         self._vnf_reconfigure_network(vnfi, intf_name, ip_address)
 653                         # add this vnf and interface to the E-LAN for tagging
 654                         elan_vnf_list.append({'name': src_docker_name, 'interface': intf_name})
 655
 656             # install the VLAN tags for this E-LAN
 657             GK.net.setLAN(elan_vnf_list)
 658
 659
 660     def _load_docker_files(self):
 661         """
 662         Get all paths to Dockerfiles from VNFDs and store them in dict.
 663         :return:
 664         """
 665         for k, v in self.vnfds.iteritems():
 666             for vu in v.get("virtual_deployment_units"):
 667                 if vu.get("vm_image_format") == "docker":
 668                     vm_image = vu.get("vm_image")
 669                     docker_path = os.path.join(
 670                         self.package_content_path,
 671                         make_relative_path(vm_image))
 672                     self.local_docker_files[k] = docker_path
 673                     LOG.debug("Found Dockerfile (%r): %r" % (k, docker_path))
 674
 675     def _load_docker_urls(self):
 676         """
 677         Get all URLs to pre-build docker images in some repo.
 678         :return:
 679         """
 680         # also merge sap dicts, because internal saps also need a docker container
 681         all_vnfs = self.vnfds.copy()
 682         all_vnfs.update(self.saps)
 683
 684         for k, v in all_vnfs.iteritems():
 685             for vu in v.get("virtual_deployment_units", {}):
 686                 if vu.get("vm_image_format") == "docker":
 687                     url = vu.get("vm_image")
 688                     if url is not None:
 689                         url = url.replace("http://", "")
 690                         self.remote_docker_image_urls[k] = url
 691                         LOG.debug("Found Docker image URL (%r): %r" % (k, self.remote_docker_image_urls[k]))
 692
 693     def _build_images_from_dockerfiles(self):
 694         """
 695         Build Docker images for each local Dockerfile found in the package: self.local_docker_files
 696         """
 697         if GK_STANDALONE_MODE:
 698             return  # do not build anything in standalone mode
 699         dc = DockerClient()
 700         LOG.info("Building %d Docker images (this may take several minutes) ..." % len(self.local_docker_files))
 701         for k, v in self.local_docker_files.iteritems():
 702             for line in dc.build(path=v.replace("Dockerfile", ""), tag=k, rm=False, nocache=False):
 703                 LOG.debug("DOCKER BUILD: %s" % line)
 704             LOG.info("Docker image created: %s" % k)
 705
 706     def _pull_predefined_dockerimages(self):
 707         """
 708         If the package contains URLs to pre-build Docker images, we download them with this method.
 709         """
 710         dc = DockerClient()
 711         for url in self.remote_docker_image_urls.itervalues():
 712             if not FORCE_PULL:  # only pull if not present (speedup for development)
 713                 if len(dc.images.list(name=url)) > 0:
 714                     LOG.debug("Image %r present. Skipping pull." % url)
 715                     continue
 716             LOG.info("Pulling image: %r" % url)
 717             # this seems to fail with latest docker api version 2.0.2
 718             # dc.images.pull(url,
 719             #        insecure_registry=True)
 720             #using docker cli instead
 721             cmd = ["docker",
 722                    "pull",
 723                    url,
 724                    ]
 725             Popen(cmd).wait()
 726
 727
 728
 729
 730     def _check_docker_image_exists(self, image_name):
 731         """
 732         Query the docker service and check if the given image exists
 733         :param image_name: name of the docker image
 734         :return:
 735         """
 736         return len(DockerClient().images.list(name=image_name)) > 0
 737
 738     def _calculate_placement(self, algorithm):
 739         """
 740         Do placement by adding the a field "dc" to
 741         each VNFD that points to one of our
 742         data center objects known to the gatekeeper.
 743         """
 744         assert(len(self.vnfds) > 0)
 745         assert(len(GK.dcs) > 0)
 746         # instantiate algorithm an place
 747         p = algorithm()
 748         p.place(self.nsd, self.vnfds, self.saps, GK.dcs)
 749         LOG.info("Using placement algorithm: %r" % p.__class__.__name__)
 750         # lets print the placement result
 751         for name, vnfd in self.vnfds.iteritems():
 752             LOG.info("Placed VNF %r on DC %r" % (name, str(vnfd.get("dc"))))
 753         for sap in self.saps:
 754             sap_dict = self.saps[sap]
 755             LOG.info("Placed SAP %r on DC %r" % (sap, str(sap_dict.get("dc"))))
 756
 757
 758     def _calculate_cpu_cfs_values(self, cpu_time_percentage):
 759         """
 760         Calculate cpu period and quota for CFS
 761         :param cpu_time_percentage: percentage of overall CPU to be used
 762         :return: cpu_period, cpu_quota
 763         """
 764         if cpu_time_percentage is None:
 765             return -1, -1
 766         if cpu_time_percentage < 0:
 767             return -1, -1
 768         # (see: https://www.kernel.org/doc/Documentation/scheduler/sched-bwc.txt)
 769         # Attention minimum cpu_quota is 1ms (micro)
 770         cpu_period = 1000000  # lets consider a fixed period of 1000000 microseconds for now
 771         LOG.debug("cpu_period is %r, cpu_percentage is %r" % (cpu_period, cpu_time_percentage))
 772         cpu_quota = cpu_period * cpu_time_percentage  # calculate the fraction of cpu time for this container
 773         # ATTENTION >= 1000 to avoid a invalid argument system error ... no idea why
 774         if cpu_quota < 1000:
 775             LOG.debug("cpu_quota before correcting: %r" % cpu_quota)
 776             cpu_quota = 1000
 777             LOG.warning("Increased CPU quota to avoid system error.")
 778         LOG.debug("Calculated: cpu_period=%f / cpu_quota=%f" % (cpu_period, cpu_quota))
 779         return int(cpu_period), int(cpu_quota)
 780
 781     def check_ext_saps(self, intf_list):
 782         # check if the list of interfacs contains an externl SAP
 783         saps_ext = [self.saps[sap]['name'] for sap in self.saps if self.saps[sap]["type"] == "external"]
 784         for intf_name in intf_list:
 785             vnf_id, vnf_interface, vnf_sap_docker_name = parse_interface(intf_name)
 786             if vnf_sap_docker_name in saps_ext:
 787                 return vnf_sap_docker_name
 788
 789     def check_mgmt_interface(self, intf_list):
 790         SAPs_mgmt = [p.get('id') for p in self.nsd["connection_points"] if 'management' in p.get('type')]
 791         for intf_name in intf_list:
 792             if intf_name in SAPs_mgmt:
 793                 return True
 794
 795 """
 796 Some (simple) placement algorithms
 797 """
 798
 799
 800 class FirstDcPlacement(object):
 801     """
 802     Placement: Always use one and the same data center from the GK.dcs dict.
 803     """
 804     def place(self, nsd, vnfds, saps, dcs):
 805         for name, vnfd in vnfds.iteritems():
 806             vnfd["dc"] = list(dcs.itervalues())[0]
 807
 808
 809 class RoundRobinDcPlacement(object):
 810     """
 811     Placement: Distribute VNFs across all available DCs in a round robin fashion.
 812     """
 813     def place(self, nsd, vnfds, saps, dcs):
 814         c = 0
 815         dcs_list = list(dcs.itervalues())
 816         for name, vnfd in vnfds.iteritems():
 817             vnfd["dc"] = dcs_list[c % len(dcs_list)]
 818             c += 1  # inc. c to use next DC
 819
 820 class RoundRobinDcPlacementWithSAPs(object):
 821     """
 822     Placement: Distribute VNFs across all available DCs in a round robin fashion,
 823     every SAP is instantiated on the same DC as the connected VNF.
 824     """
 825     def place(self, nsd, vnfds, saps, dcs):
 826
 827         # place vnfs
 828         c = 0
 829         dcs_list = list(dcs.itervalues())
 830         for name, vnfd in vnfds.iteritems():
 831             vnfd["dc"] = dcs_list[c % len(dcs_list)]
 832             c += 1  # inc. c to use next DC
 833
 834         # place SAPs
 835         vlinks = nsd.get("virtual_links", [])
 836         eline_fwd_links = [l for l in vlinks if (l["connectivity_type"] == "E-Line")]
 837         elan_fwd_links = [l for l in vlinks if  (l["connectivity_type"] == "E-LAN")]
 838
 839         vnf_id2vnf_name = defaultdict(lambda: "NotExistingNode",
 840                                       reduce(lambda x, y: dict(x, **y),
 841                                              map(lambda d: {d["vnf_id"]: d["vnf_name"]},
 842                                                  nsd["network_functions"])))
 843
 844         # SAPs on E-Line links are placed on the same DC as the VNF on the E-Line
 845         for link in eline_fwd_links:
 846             src_id, src_if_name, src_sap_id = parse_interface(link["connection_points_reference"][0])
 847             dst_id, dst_if_name, dst_sap_id = parse_interface(link["connection_points_reference"][1])
 848
 849             # check if there is a SAP in the link
 850             if src_sap_id in saps:
 851                 dst_vnf_name = vnf_id2vnf_name[dst_id]
 852                 # get dc where connected vnf is mapped to
 853                 dc = vnfds[dst_vnf_name]['dc']
 854                 saps[src_sap_id]['dc'] = dc
 855
 856             if dst_sap_id in saps:
 857                 src_vnf_name = vnf_id2vnf_name[src_id]
 858                 # get dc where connected vnf is mapped to
 859                 dc = vnfds[src_vnf_name]['dc']
 860                 saps[dst_sap_id]['dc'] = dc
 861
 862         # SAPs on E-LANs are placed on a random DC
 863         dcs_list = list(dcs.itervalues())
 864         dc_len = len(dcs_list)
 865         for link in elan_fwd_links:
 866             for intf in link["connection_points_reference"]:
 867                 # find SAP interfaces
 868                 intf_id, intf_name, intf_sap_id = parse_interface(intf)
 869                 if intf_sap_id in saps:
 870                     dc = dcs_list[randint(0, dc_len-1)]
 871                     saps[intf_sap_id]['dc'] = dc
 872
 873
 874
 875 """
 876 Resource definitions and API endpoints
 877 """
 878
 879
 880 class Packages(fr.Resource):
 881
 882     def post(self):
 883         """
 884         Upload a *.son service package to the dummy gatekeeper.
 885
 886         We expect request with a *.son file and store it in UPLOAD_FOLDER
 887         :return: UUID
 888         """
 889         try:
 890             # get file contents
 891             LOG.info("POST /packages called")
 892             # lets search for the package in the request
 893             is_file_object = False  # make API more robust: file can be in data or in files field
 894             if "package" in request.files:
 895                 son_file = request.files["package"]
 896                 is_file_object = True
 897             elif len(request.data) > 0:
 898                 son_file = request.data
 899             else:
 900                 return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed. file not found."}, 500
 901             # generate a uuid to reference this package
 902             service_uuid = str(uuid.uuid4())
 903             file_hash = hashlib.sha1(str(son_file)).hexdigest()
 904             # ensure that upload folder exists
 905             ensure_dir(UPLOAD_FOLDER)
 906             upload_path = os.path.join(UPLOAD_FOLDER, "%s.son" % service_uuid)
 907             # store *.son file to disk
 908             if is_file_object:
 909                 son_file.save(upload_path)
 910             else:
 911                 with open(upload_path, 'wb') as f:
 912                     f.write(son_file)
 913             size = os.path.getsize(upload_path)
 914             # create a service object and register it
 915             s = Service(service_uuid, file_hash, upload_path)
 916             GK.register_service_package(service_uuid, s)
 917             # generate the JSON result
 918             return {"service_uuid": service_uuid, "size": size, "sha1": file_hash, "error": None}, 201
 919         except Exception as ex:
 920             LOG.exception("Service package upload failed:")
 921             return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed"}, 500
 922
 923     def get(self):
 924         """
 925         Return a list of UUID's of uploaded service packages.
 926         :return: dict/list
 927         """
 928         LOG.info("GET /packages")
 929         return {"service_uuid_list": list(GK.services.iterkeys())}
 930
 931
 932 class Instantiations(fr.Resource):
 933
 934     def post(self):
 935         """
 936         Instantiate a service specified by its UUID.
 937         Will return a new UUID to identify the running service instance.
 938         :return: UUID
 939         """
 940         LOG.info("POST /instantiations (or /reqeusts) called")
 941         # try to extract the service uuid from the request
 942         json_data = request.get_json(force=True)
 943         service_uuid = json_data.get("service_uuid")
 944
 945         # lets be a bit fuzzy here to make testing easier
 946         if (service_uuid is None or service_uuid=="latest") and len(GK.services) > 0:
 947             # if we don't get a service uuid, we simple start the first service in the list
 948             service_uuid = list(GK.services.iterkeys())[0]
 949         if service_uuid in GK.services:
 950             # ok, we have a service uuid, lets start the service
 951             service_instance_uuid = GK.services.get(service_uuid).start_service()
 952             return {"service_instance_uuid": service_instance_uuid}, 201
 953         return "Service not found", 404
 954
 955     def get(self):
 956         """
 957         Returns a list of UUIDs containing all running services.
 958         :return: dict / list
 959         """
 960         LOG.info("GET /instantiations")
 961         return {"service_instantiations_list": [
 962             list(s.instances.iterkeys()) for s in GK.services.itervalues()]}
 963
 964     def delete(self):
 965         """
 966         Stops a running service specified by its service and instance UUID.
 967         """
 968         # try to extract the service  and instance UUID from the request
 969         json_data = request.get_json(force=True)
 970         service_uuid = json_data.get("service_uuid")
 971         instance_uuid = json_data.get("service_instance_uuid")
 972
 973         # try to be fuzzy
 974         if service_uuid is None and len(GK.services) > 0:
 975             #if we don't get a service uuid, we simply stop the last service in the list
 976             service_uuid = list(GK.services.iterkeys())[0]
 977         if instance_uuid is None and len(GK.services[service_uuid].instances) > 0:
 978             instance_uuid = list(GK.services[service_uuid].instances.iterkeys())[0]
 979
 980         if service_uuid in GK.services and instance_uuid in GK.services[service_uuid].instances:
 981             # valid service and instance UUID, stop service
 982             GK.services.get(service_uuid).stop_service(instance_uuid)
 983             return "service instance with uuid %r stopped." % instance_uuid,200
 984         return "Service not found", 404
 985
 986 class Exit(fr.Resource):
 987
 988     def put(self):
 989         """
 990         Stop the running Containernet instance regardless of data transmitted
 991         """
 992         list(GK.dcs.values())[0].net.stop()
 993
 994
 995 def initialize_GK():
 996     global GK
 997     GK = Gatekeeper()
 998
 999
1000
1001 # create a single, global GK object
1002 GK = None
1003 initialize_GK()
1004 # setup Flask
1005 app = Flask(__name__)
1006 app.config['MAX_CONTENT_LENGTH'] = 512 * 1024 * 1024  # 512 MB max upload
1007 api = fr.Api(app)
1008 # define endpoints
1009 api.add_resource(Packages, '/packages', '/api/v2/packages')
1010 api.add_resource(Instantiations, '/instantiations', '/api/v2/instantiations', '/api/v2/requests')
1011 api.add_resource(Exit, '/emulator/exit')
1012
1013
1014 #def initialize_GK():
1015 #    global GK
1016 #    GK = Gatekeeper()
1017
1018
1019 def start_rest_api(host, port, datacenters=dict()):
1020     GK.dcs = datacenters
1021     GK.net = get_dc_network()
1022     # start the Flask server (not the best performance but ok for our use case)
1023     app.run(host=host,
1024             port=port,
1025             debug=True,
1026             use_reloader=False  # this is needed to run Flask in a non-main thread
1027             )
1028
1029
1030 def ensure_dir(name):
1031     if not os.path.exists(name):
1032         os.makedirs(name)
1033
1034
1035 def load_yaml(path):
1036     with open(path, "r") as f:
1037         try:
1038             r = yaml.load(f)
1039         except yaml.YAMLError as exc:
1040             LOG.exception("YAML parse error")
1041             r = dict()
1042     return r
1043
1044
1045 def make_relative_path(path):
1046     if path.startswith("file://"):
1047         path = path.replace("file://", "", 1)
1048     if path.startswith("/"):
1049         path = path.replace("/", "", 1)
1050     return path
1051
1052
1053 def get_dc_network():
1054     """
1055     retrieve the DCnetwork where this dummygatekeeper (GK) connects to.
1056     Assume at least 1 datacenter is connected to this GK, and that all datacenters belong to the same DCNetwork
1057     :return:
1058     """
1059     assert (len(GK.dcs) > 0)
1060     return GK.dcs.values()[0].net
1061
1062
1063 def parse_interface(interface_name):
1064     """
1065     convert the interface name in the nsd to the according vnf_id, vnf_interface names
1066     :param interface_name:
1067     :return:
1068     """
1069
1070     if ':' in interface_name:
1071         vnf_id, vnf_interface = interface_name.split(':')
1072         vnf_sap_docker_name = interface_name.replace(':', '_')
1073     else:
1074         vnf_id = interface_name
1075         vnf_interface = interface_name
1076         vnf_sap_docker_name = interface_name
1077
1078     return vnf_id, vnf_interface, vnf_sap_docker_name
1079
1080 if __name__ == '__main__':
1081     """
1082     Lets allow to run the API in standalone mode.
1083     """
1084     GK_STANDALONE_MODE = True
1085     logging.getLogger("werkzeug").setLevel(logging.INFO)
1086     start_rest_api("0.0.0.0", 8000)
1087