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