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 from docker import Client as DockerClient
  42 from flask import Flask, request
  43 import flask_restful as fr
  44 from collections import defaultdict
  45
  46 logging.basicConfig()
  47 LOG = logging.getLogger("sonata-dummy-gatekeeper")
  48 LOG.setLevel(logging.DEBUG)
  49 logging.getLogger("werkzeug").setLevel(logging.WARNING)
  50
  51 GK_STORAGE = "/tmp/son-dummy-gk/"
  52 UPLOAD_FOLDER = os.path.join(GK_STORAGE, "uploads/")
  53 CATALOG_FOLDER = os.path.join(GK_STORAGE, "catalog/")
  54
  55 # Enable Dockerfile build functionality
  56 BUILD_DOCKERFILE = False
  57
  58 # flag to indicate that we run without the emulator (only the bare API for integration testing)
  59 GK_STANDALONE_MODE = False
  60
  61 # should a new version of an image be pulled even if its available
  62 FORCE_PULL = False
  63
  64 class Gatekeeper(object):
  65
  66     def __init__(self):
  67         self.services = dict()
  68         self.dcs = dict()
  69         self.vnf_counter = 0  # used to generate short names for VNFs (Mininet limitation)
  70         LOG.info("Create SONATA dummy gatekeeper.")
  71
  72     def register_service_package(self, service_uuid, service):
  73         """
  74         register new service package
  75         :param service_uuid
  76         :param service object
  77         """
  78         self.services[service_uuid] = service
  79         # lets perform all steps needed to onboard the service
  80         service.onboard()
  81
  82     def get_next_vnf_name(self):
  83         self.vnf_counter += 1
  84         return "vnf%d" % self.vnf_counter
  85
  86
  87 class Service(object):
  88     """
  89     This class represents a NS uploaded as a *.son package to the
  90     dummy gatekeeper.
  91     Can have multiple running instances of this service.
  92     """
  93
  94     def __init__(self,
  95                  service_uuid,
  96                  package_file_hash,
  97                  package_file_path):
  98         self.uuid = service_uuid
  99         self.package_file_hash = package_file_hash
 100         self.package_file_path = package_file_path
 101         self.package_content_path = os.path.join(CATALOG_FOLDER, "services/%s" % self.uuid)
 102         self.manifest = None
 103         self.nsd = None
 104         self.vnfds = dict()
 105         self.local_docker_files = dict()
 106         self.remote_docker_image_urls = dict()
 107         self.instances = dict()
 108         self.vnf_name2docker_name = dict()
 109         # lets generate a set of subnet configurations used for e-line chaining setup
 110         self.eline_subnets_src = generate_subnet_strings(50, start=200, subnet_size=24, ip=1)
 111         self.eline_subnets_dst = generate_subnet_strings(50, start=200, subnet_size=24, ip=2)
 112
 113     def onboard(self):
 114         """
 115         Do all steps to prepare this service to be instantiated
 116         :return:
 117         """
 118         # 1. extract the contents of the package and store them in our catalog
 119         self._unpack_service_package()
 120         # 2. read in all descriptor files
 121         self._load_package_descriptor()
 122         self._load_nsd()
 123         self._load_vnfd()
 124         # 3. prepare container images (e.g. download or build Dockerfile)
 125         if BUILD_DOCKERFILE:
 126             self._load_docker_files()
 127             self._build_images_from_dockerfiles()
 128         else:
 129             self._load_docker_urls()
 130             self._pull_predefined_dockerimages()
 131         LOG.info("On-boarded service: %r" % self.manifest.get("package_name"))
 132
 133     def start_service(self):
 134         """
 135         This methods creates and starts a new service instance.
 136         It computes placements, iterates over all VNFDs, and starts
 137         each VNFD as a Docker container in the data center selected
 138         by the placement algorithm.
 139         :return:
 140         """
 141         LOG.info("Starting service %r" % self.uuid)
 142
 143         # 1. each service instance gets a new uuid to identify it
 144         instance_uuid = str(uuid.uuid4())
 145         # build a instances dict (a bit like a NSR :))
 146         self.instances[instance_uuid] = dict()
 147         self.instances[instance_uuid]["vnf_instances"] = list()
 148
 149         # 2. compute placement of this service instance (adds DC names to VNFDs)
 150         if not GK_STANDALONE_MODE:
 151             self._calculate_placement(FirstDcPlacement)
 152         # iterate over all vnfds that we have to start
 153         for vnfd in self.vnfds.itervalues():
 154             vnfi = None
 155             if not GK_STANDALONE_MODE:
 156                 vnfi = self._start_vnfd(vnfd)
 157             self.instances[instance_uuid]["vnf_instances"].append(vnfi)
 158
 159         # 3. Configure the chaining of the network functions (currently only E-Line links supported)
 160         vnf_id2vnf_name = defaultdict(lambda: "NotExistingNode",
 161                                 reduce(lambda x, y: dict(x, **y),
 162                                        map(lambda d: {d["vnf_id"]: d["vnf_name"]},
 163                                            self.nsd["network_functions"])))
 164
 165         vlinks = self.nsd["virtual_links"]
 166         fwd_links = self.nsd["forwarding_graphs"][0]["constituent_virtual_links"]
 167         eline_fwd_links = [l for l in vlinks if (l["id"] in fwd_links) and (l["connectivity_type"] == "E-Line")]
 168
 169         cookie = 1  # not clear why this is needed - to check with Steven
 170         for link in eline_fwd_links:
 171             src_id, src_if_name = link["connection_points_reference"][0].split(":")
 172             dst_id, dst_if_name = link["connection_points_reference"][1].split(":")
 173
 174             src_name = vnf_id2vnf_name[src_id]
 175             dst_name = vnf_id2vnf_name[dst_id]
 176
 177             LOG.debug(
 178                 "Setting up E-Line link. %s(%s:%s) -> %s(%s:%s)" % (
 179                     src_name, src_id, src_if_name, dst_name, dst_id, dst_if_name))
 180
 181             if (src_name in self.vnfds) and (dst_name in self.vnfds):
 182                 network = self.vnfds[src_name].get("dc").net  # there should be a cleaner way to find the DCNetwork
 183                 src_docker_name = self.vnf_name2docker_name[src_name]
 184                 dst_docker_name = self.vnf_name2docker_name[dst_name]
 185                 LOG.debug(src_docker_name)
 186                 ret = network.setChain(
 187                     src_docker_name, dst_docker_name,
 188                     vnf_src_interface=src_if_name, vnf_dst_interface=dst_if_name,
 189                     bidirectional=True, cmd="add-flow", cookie=cookie)
 190                 cookie += 1
 191
 192                 # re-configure the VNFs IP assignment and ensure that a new subnet is used for each E-Link
 193                 src_vnfi = self._get_vnf_instance(instance_uuid, src_name)
 194                 if src_vnfi is not None:
 195                     self._vnf_reconfigure_network(src_vnfi, src_if_name, self.eline_subnets_src.pop(0))
 196                 dst_vnfi = self._get_vnf_instance(instance_uuid, dst_name)
 197                 if dst_vnfi is not None:
 198                     self._vnf_reconfigure_network(dst_vnfi, dst_if_name, self.eline_subnets_dst.pop(0))
 199
 200         # 4. run the emulator specific entrypoint scripts in the VNFIs of this service instance
 201         self._trigger_emulator_start_scripts_in_vnfis(self.instances[instance_uuid]["vnf_instances"])
 202
 203         LOG.info("Service started. Instance id: %r" % instance_uuid)
 204         return instance_uuid
 205
 206     def _start_vnfd(self, vnfd):
 207         """
 208         Start a single VNFD of this service
 209         :param vnfd: vnfd descriptor dict
 210         :return:
 211         """
 212         # iterate over all deployment units within each VNFDs
 213         for u in vnfd.get("virtual_deployment_units"):
 214             # 1. get the name of the docker image to start and the assigned DC
 215             vnf_name = vnfd.get("name")
 216             if vnf_name not in self.remote_docker_image_urls:
 217                 raise Exception("No image name for %r found. Abort." % vnf_name)
 218             docker_name = self.remote_docker_image_urls.get(vnf_name)
 219             target_dc = vnfd.get("dc")
 220             # 2. perform some checks to ensure we can start the container
 221             assert(docker_name is not None)
 222             assert(target_dc is not None)
 223             if not self._check_docker_image_exists(docker_name):
 224                 raise Exception("Docker image %r not found. Abort." % docker_name)
 225             # 3. do the dc.startCompute(name="foobar") call to run the container
 226             # TODO consider flavors, and other annotations
 227             intfs = vnfd.get("connection_points")
 228             self.vnf_name2docker_name[vnf_name] = GK.get_next_vnf_name()
 229             LOG.info("Starting %r as %r in DC %r" % (vnf_name, self.vnf_name2docker_name[vnf_name], vnfd.get("dc")))
 230             LOG.debug("Interfaces for %r: %r" % (vnf_name, intfs))
 231             vnfi = target_dc.startCompute(self.vnf_name2docker_name[vnf_name], network=intfs, image=docker_name, flavor_name="small")
 232             return vnfi
 233
 234     def _get_vnf_instance(self, instance_uuid, name):
 235         """
 236         Returns the Docker object for the given VNF name (or Docker name).
 237         :param instance_uuid: UUID of the service instance to search in.
 238         :param name: VNF name or Docker name. We are fuzzy here.
 239         :return:
 240         """
 241         dn = name
 242         if name in self.vnf_name2docker_name:
 243             dn = self.vnf_name2docker_name[name]
 244         for vnfi in self.instances[instance_uuid]["vnf_instances"]:
 245             if vnfi.name == dn:
 246                 return vnfi
 247         LOG.warning("No container with name: %r found.")
 248         return None
 249
 250     @staticmethod
 251     def _vnf_reconfigure_network(vnfi, if_name, net_str):
 252         """
 253         Reconfigure the network configuration of a specific interface
 254         of a running container.
 255         :param vnfi: container instacne
 256         :param if_name: interface name
 257         :param net_str: network configuration string, e.g., 1.2.3.4/24
 258         :return:
 259         """
 260         intf = vnfi.intf(intf=if_name)
 261         if intf is not None:
 262             intf.setIP(net_str)
 263             LOG.debug("Reconfigured network of %s:%s to %r" % (vnfi.name, if_name, net_str))
 264         else:
 265             LOG.warning("Interface not found: %s:%s. Network reconfiguration skipped." % (vnfi.name, if_name))
 266
 267
 268     def _trigger_emulator_start_scripts_in_vnfis(self, vnfi_list):
 269         for vnfi in vnfi_list:
 270             config = vnfi.dcinfo.get("Config", dict())
 271             env = config.get("Env", list())
 272             for env_var in env:
 273                 if "SON_EMU_CMD=" in env_var:
 274                     cmd = str(env_var.split("=")[1])
 275                     LOG.info("Executing entrypoint script in %r: %r" % (vnfi.name, cmd))
 276                     vnfi.cmdPrint(cmd)
 277
 278     def _unpack_service_package(self):
 279         """
 280         unzip *.son file and store contents in CATALOG_FOLDER/services/<service_uuid>/
 281         """
 282         LOG.info("Unzipping: %r" % self.package_file_path)
 283         with zipfile.ZipFile(self.package_file_path, "r") as z:
 284             z.extractall(self.package_content_path)
 285
 286
 287     def _load_package_descriptor(self):
 288         """
 289         Load the main package descriptor YAML and keep it as dict.
 290         :return:
 291         """
 292         self.manifest = load_yaml(
 293             os.path.join(
 294                 self.package_content_path, "META-INF/MANIFEST.MF"))
 295
 296     def _load_nsd(self):
 297         """
 298         Load the entry NSD YAML and keep it as dict.
 299         :return:
 300         """
 301         if "entry_service_template" in self.manifest:
 302             nsd_path = os.path.join(
 303                 self.package_content_path,
 304                 make_relative_path(self.manifest.get("entry_service_template")))
 305             self.nsd = load_yaml(nsd_path)
 306             LOG.debug("Loaded NSD: %r" % self.nsd.get("name"))
 307
 308     def _load_vnfd(self):
 309         """
 310         Load all VNFD YAML files referenced in MANIFEST.MF and keep them in dict.
 311         :return:
 312         """
 313         if "package_content" in self.manifest:
 314             for pc in self.manifest.get("package_content"):
 315                 if pc.get("content-type") == "application/sonata.function_descriptor":
 316                     vnfd_path = os.path.join(
 317                         self.package_content_path,
 318                         make_relative_path(pc.get("name")))
 319                     vnfd = load_yaml(vnfd_path)
 320                     self.vnfds[vnfd.get("name")] = vnfd
 321                     LOG.debug("Loaded VNFD: %r" % vnfd.get("name"))
 322
 323     def _load_docker_files(self):
 324         """
 325         Get all paths to Dockerfiles from VNFDs and store them in dict.
 326         :return:
 327         """
 328         for k, v in self.vnfds.iteritems():
 329             for vu in v.get("virtual_deployment_units"):
 330                 if vu.get("vm_image_format") == "docker":
 331                     vm_image = vu.get("vm_image")
 332                     docker_path = os.path.join(
 333                         self.package_content_path,
 334                         make_relative_path(vm_image))
 335                     self.local_docker_files[k] = docker_path
 336                     LOG.debug("Found Dockerfile (%r): %r" % (k, docker_path))
 337
 338     def _load_docker_urls(self):
 339         """
 340         Get all URLs to pre-build docker images in some repo.
 341         :return:
 342         """
 343         for k, v in self.vnfds.iteritems():
 344             for vu in v.get("virtual_deployment_units"):
 345                 if vu.get("vm_image_format") == "docker":
 346                     url = vu.get("vm_image")
 347                     if url is not None:
 348                         url = url.replace("http://", "")
 349                         self.remote_docker_image_urls[k] = url
 350                         LOG.debug("Found Docker image URL (%r): %r" % (k, self.remote_docker_image_urls[k]))
 351
 352     def _build_images_from_dockerfiles(self):
 353         """
 354         Build Docker images for each local Dockerfile found in the package: self.local_docker_files
 355         """
 356         if GK_STANDALONE_MODE:
 357             return  # do not build anything in standalone mode
 358         dc = DockerClient()
 359         LOG.info("Building %d Docker images (this may take several minutes) ..." % len(self.local_docker_files))
 360         for k, v in self.local_docker_files.iteritems():
 361             for line in dc.build(path=v.replace("Dockerfile", ""), tag=k, rm=False, nocache=False):
 362                 LOG.debug("DOCKER BUILD: %s" % line)
 363             LOG.info("Docker image created: %s" % k)
 364
 365     def _pull_predefined_dockerimages(self):
 366         """
 367         If the package contains URLs to pre-build Docker images, we download them with this method.
 368         """
 369         dc = DockerClient()
 370         for url in self.remote_docker_image_urls.itervalues():
 371             if not FORCE_PULL:  # only pull if not present (speedup for development)
 372                 if len(dc.images(name=url)) > 0:
 373                     LOG.debug("Image %r present. Skipping pull." % url)
 374                     continue
 375             LOG.info("Pulling image: %r" % url)
 376             dc.pull(url,
 377                     insecure_registry=True)
 378
 379     def _check_docker_image_exists(self, image_name):
 380         """
 381         Query the docker service and check if the given image exists
 382         :param image_name: name of the docker image
 383         :return:
 384         """
 385         return len(DockerClient().images(image_name)) > 0
 386
 387     def _calculate_placement(self, algorithm):
 388         """
 389         Do placement by adding the a field "dc" to
 390         each VNFD that points to one of our
 391         data center objects known to the gatekeeper.
 392         """
 393         assert(len(self.vnfds) > 0)
 394         assert(len(GK.dcs) > 0)
 395         # instantiate algorithm an place
 396         p = algorithm()
 397         p.place(self.nsd, self.vnfds, GK.dcs)
 398         LOG.info("Using placement algorithm: %r" % p.__class__.__name__)
 399         # lets print the placement result
 400         for name, vnfd in self.vnfds.iteritems():
 401             LOG.info("Placed VNF %r on DC %r" % (name, str(vnfd.get("dc"))))
 402
 403
 404 """
 405 Some (simple) placement algorithms
 406 """
 407
 408
 409 class FirstDcPlacement(object):
 410     """
 411     Placement: Always use one and the same data center from the GK.dcs dict.
 412     """
 413     def place(self, nsd, vnfds, dcs):
 414         for name, vnfd in vnfds.iteritems():
 415             vnfd["dc"] = list(dcs.itervalues())[0]
 416
 417
 418 """
 419 Resource definitions and API endpoints
 420 """
 421
 422
 423 class Packages(fr.Resource):
 424
 425     def post(self):
 426         """
 427         Upload a *.son service package to the dummy gatekeeper.
 428
 429         We expect request with a *.son file and store it in UPLOAD_FOLDER
 430         :return: UUID
 431         """
 432         try:
 433             # get file contents
 434             print(request.files)
 435             # lets search for the package in the request
 436             if "package" in request.files:
 437                 son_file = request.files["package"]
 438             # elif "file" in request.files:
 439             #     son_file = request.files["file"]
 440             else:
 441                 return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed. file not found."}, 500
 442             # generate a uuid to reference this package
 443             service_uuid = str(uuid.uuid4())
 444             file_hash = hashlib.sha1(str(son_file)).hexdigest()
 445             # ensure that upload folder exists
 446             ensure_dir(UPLOAD_FOLDER)
 447             upload_path = os.path.join(UPLOAD_FOLDER, "%s.son" % service_uuid)
 448             # store *.son file to disk
 449             son_file.save(upload_path)
 450             size = os.path.getsize(upload_path)
 451             # create a service object and register it
 452             s = Service(service_uuid, file_hash, upload_path)
 453             GK.register_service_package(service_uuid, s)
 454             # generate the JSON result
 455             return {"service_uuid": service_uuid, "size": size, "sha1": file_hash, "error": None}
 456         except Exception as ex:
 457             LOG.exception("Service package upload failed:")
 458             return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed"}, 500
 459
 460     def get(self):
 461         """
 462         Return a list of UUID's of uploaded service packages.
 463         :return: dict/list
 464         """
 465         LOG.info("GET /packages")
 466         return {"service_uuid_list": list(GK.services.iterkeys())}
 467
 468
 469 class Instantiations(fr.Resource):
 470
 471     def post(self):
 472         """
 473         Instantiate a service specified by its UUID.
 474         Will return a new UUID to identify the running service instance.
 475         :return: UUID
 476         """
 477         # try to extract the service uuid from the request
 478         json_data = request.get_json(force=True)
 479         service_uuid = json_data.get("service_uuid")
 480
 481         # lets be a bit fuzzy here to make testing easier
 482         if service_uuid is None and len(GK.services) > 0:
 483             # if we don't get a service uuid, we simple start the first service in the list
 484             service_uuid = list(GK.services.iterkeys())[0]
 485
 486         if service_uuid in GK.services:
 487             # ok, we have a service uuid, lets start the service
 488             service_instance_uuid = GK.services.get(service_uuid).start_service()
 489             return {"service_instance_uuid": service_instance_uuid}
 490         return "Service not found", 404
 491
 492     def get(self):
 493         """
 494         Returns a list of UUIDs containing all running services.
 495         :return: dict / list
 496         """
 497         LOG.info("GET /instantiations")
 498         return {"service_instantiations_list": [
 499             list(s.instances.iterkeys()) for s in GK.services.itervalues()]}
 500
 501
 502 # create a single, global GK object
 503 GK = Gatekeeper()
 504 # setup Flask
 505 app = Flask(__name__)
 506 app.config['MAX_CONTENT_LENGTH'] = 512 * 1024 * 1024  # 512 MB max upload
 507 api = fr.Api(app)
 508 # define endpoints
 509 api.add_resource(Packages, '/packages')
 510 api.add_resource(Instantiations, '/instantiations')
 511
 512
 513 def start_rest_api(host, port, datacenters=dict()):
 514     GK.dcs = datacenters
 515     # start the Flask server (not the best performance but ok for our use case)
 516     app.run(host=host,
 517             port=port,
 518             debug=True,
 519             use_reloader=False  # this is needed to run Flask in a non-main thread
 520             )
 521
 522
 523 def ensure_dir(name):
 524     if not os.path.exists(name):
 525         os.makedirs(name)
 526
 527
 528 def load_yaml(path):
 529     with open(path, "r") as f:
 530         try:
 531             r = yaml.load(f)
 532         except yaml.YAMLError as exc:
 533             LOG.exception("YAML parse error")
 534             r = dict()
 535     return r
 536
 537
 538 def make_relative_path(path):
 539     if path.startswith("file://"):
 540         path = path.replace("file://", "", 1)
 541     if path.startswith("/"):
 542         path = path.replace("/", "", 1)
 543     return path
 544
 545
 546 def generate_subnet_strings(n, start=1, subnet_size=24, ip=0):
 547     """
 548     Helper to generate different network configuration strings.
 549     """
 550     r = list()
 551     for i in range(start, start + n):
 552         r.append("%d.0.0.%d/%d" % (i, ip, subnet_size))
 553     return r
 554
 555
 556 if __name__ == '__main__':
 557     """
 558     Lets allow to run the API in standalone mode.
 559     """
 560     GK_STANDALONE_MODE = True
 561     logging.getLogger("werkzeug").setLevel(logging.INFO)
 562     start_rest_api("0.0.0.0", 8000)
 563