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