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
 229             # use the vnf_id in the nsd as docker name
 230             # so deployed containers can be easily mapped back to the nsd
 231             vnf_name2id = defaultdict(lambda: "NotExistingNode",
 232                                           reduce(lambda x, y: dict(x, **y),
 233                                                  map(lambda d: {d["vnf_name"]: d["vnf_id"]},
 234                                                      self.nsd["network_functions"])))
 235             self.vnf_name2docker_name[vnf_name] = vnf_name2id[vnf_name]
 236             # self.vnf_name2docker_name[vnf_name] = GK.get_next_vnf_name()
 237
 238             LOG.info("Starting %r as %r in DC %r" % (vnf_name, self.vnf_name2docker_name[vnf_name], vnfd.get("dc")))
 239             LOG.debug("Interfaces for %r: %r" % (vnf_name, intfs))
 240             vnfi = target_dc.startCompute(self.vnf_name2docker_name[vnf_name], network=intfs, image=docker_name, flavor_name="small")
 241             return vnfi
 242
 243     def _get_vnf_instance(self, instance_uuid, name):
 244         """
 245         Returns the Docker object for the given VNF name (or Docker name).
 246         :param instance_uuid: UUID of the service instance to search in.
 247         :param name: VNF name or Docker name. We are fuzzy here.
 248         :return:
 249         """
 250         dn = name
 251         if name in self.vnf_name2docker_name:
 252             dn = self.vnf_name2docker_name[name]
 253         for vnfi in self.instances[instance_uuid]["vnf_instances"]:
 254             if vnfi.name == dn:
 255                 return vnfi
 256         LOG.warning("No container with name: %r found.")
 257         return None
 258
 259     @staticmethod
 260     def _vnf_reconfigure_network(vnfi, if_name, net_str):
 261         """
 262         Reconfigure the network configuration of a specific interface
 263         of a running container.
 264         :param vnfi: container instacne
 265         :param if_name: interface name
 266         :param net_str: network configuration string, e.g., 1.2.3.4/24
 267         :return:
 268         """
 269         intf = vnfi.intf(intf=if_name)
 270         if intf is not None:
 271             intf.setIP(net_str)
 272             LOG.debug("Reconfigured network of %s:%s to %r" % (vnfi.name, if_name, net_str))
 273         else:
 274             LOG.warning("Interface not found: %s:%s. Network reconfiguration skipped." % (vnfi.name, if_name))
 275
 276
 277     def _trigger_emulator_start_scripts_in_vnfis(self, vnfi_list):
 278         for vnfi in vnfi_list:
 279             config = vnfi.dcinfo.get("Config", dict())
 280             env = config.get("Env", list())
 281             for env_var in env:
 282                 if "SON_EMU_CMD=" in env_var:
 283                     cmd = str(env_var.split("=")[1])
 284                     LOG.info("Executing entrypoint script in %r: %r" % (vnfi.name, cmd))
 285                     vnfi.cmdPrint(cmd)
 286
 287     def _unpack_service_package(self):
 288         """
 289         unzip *.son file and store contents in CATALOG_FOLDER/services/<service_uuid>/
 290         """
 291         LOG.info("Unzipping: %r" % self.package_file_path)
 292         with zipfile.ZipFile(self.package_file_path, "r") as z:
 293             z.extractall(self.package_content_path)
 294
 295
 296     def _load_package_descriptor(self):
 297         """
 298         Load the main package descriptor YAML and keep it as dict.
 299         :return:
 300         """
 301         self.manifest = load_yaml(
 302             os.path.join(
 303                 self.package_content_path, "META-INF/MANIFEST.MF"))
 304
 305     def _load_nsd(self):
 306         """
 307         Load the entry NSD YAML and keep it as dict.
 308         :return:
 309         """
 310         if "entry_service_template" in self.manifest:
 311             nsd_path = os.path.join(
 312                 self.package_content_path,
 313                 make_relative_path(self.manifest.get("entry_service_template")))
 314             self.nsd = load_yaml(nsd_path)
 315             LOG.debug("Loaded NSD: %r" % self.nsd.get("name"))
 316
 317     def _load_vnfd(self):
 318         """
 319         Load all VNFD YAML files referenced in MANIFEST.MF and keep them in dict.
 320         :return:
 321         """
 322         if "package_content" in self.manifest:
 323             for pc in self.manifest.get("package_content"):
 324                 if pc.get("content-type") == "application/sonata.function_descriptor":
 325                     vnfd_path = os.path.join(
 326                         self.package_content_path,
 327                         make_relative_path(pc.get("name")))
 328                     vnfd = load_yaml(vnfd_path)
 329                     self.vnfds[vnfd.get("name")] = vnfd
 330                     LOG.debug("Loaded VNFD: %r" % vnfd.get("name"))
 331
 332     def _load_docker_files(self):
 333         """
 334         Get all paths to Dockerfiles from VNFDs and store them in dict.
 335         :return:
 336         """
 337         for k, v in self.vnfds.iteritems():
 338             for vu in v.get("virtual_deployment_units"):
 339                 if vu.get("vm_image_format") == "docker":
 340                     vm_image = vu.get("vm_image")
 341                     docker_path = os.path.join(
 342                         self.package_content_path,
 343                         make_relative_path(vm_image))
 344                     self.local_docker_files[k] = docker_path
 345                     LOG.debug("Found Dockerfile (%r): %r" % (k, docker_path))
 346
 347     def _load_docker_urls(self):
 348         """
 349         Get all URLs to pre-build docker images in some repo.
 350         :return:
 351         """
 352         for k, v in self.vnfds.iteritems():
 353             for vu in v.get("virtual_deployment_units"):
 354                 if vu.get("vm_image_format") == "docker":
 355                     url = vu.get("vm_image")
 356                     if url is not None:
 357                         url = url.replace("http://", "")
 358                         self.remote_docker_image_urls[k] = url
 359                         LOG.debug("Found Docker image URL (%r): %r" % (k, self.remote_docker_image_urls[k]))
 360
 361     def _build_images_from_dockerfiles(self):
 362         """
 363         Build Docker images for each local Dockerfile found in the package: self.local_docker_files
 364         """
 365         if GK_STANDALONE_MODE:
 366             return  # do not build anything in standalone mode
 367         dc = DockerClient()
 368         LOG.info("Building %d Docker images (this may take several minutes) ..." % len(self.local_docker_files))
 369         for k, v in self.local_docker_files.iteritems():
 370             for line in dc.build(path=v.replace("Dockerfile", ""), tag=k, rm=False, nocache=False):
 371                 LOG.debug("DOCKER BUILD: %s" % line)
 372             LOG.info("Docker image created: %s" % k)
 373
 374     def _pull_predefined_dockerimages(self):
 375         """
 376         If the package contains URLs to pre-build Docker images, we download them with this method.
 377         """
 378         dc = DockerClient()
 379         for url in self.remote_docker_image_urls.itervalues():
 380             if not FORCE_PULL:  # only pull if not present (speedup for development)
 381                 if len(dc.images(name=url)) > 0:
 382                     LOG.debug("Image %r present. Skipping pull." % url)
 383                     continue
 384             LOG.info("Pulling image: %r" % url)
 385             dc.pull(url,
 386                     insecure_registry=True)
 387
 388     def _check_docker_image_exists(self, image_name):
 389         """
 390         Query the docker service and check if the given image exists
 391         :param image_name: name of the docker image
 392         :return:
 393         """
 394         return len(DockerClient().images(image_name)) > 0
 395
 396     def _calculate_placement(self, algorithm):
 397         """
 398         Do placement by adding the a field "dc" to
 399         each VNFD that points to one of our
 400         data center objects known to the gatekeeper.
 401         """
 402         assert(len(self.vnfds) > 0)
 403         assert(len(GK.dcs) > 0)
 404         # instantiate algorithm an place
 405         p = algorithm()
 406         p.place(self.nsd, self.vnfds, GK.dcs)
 407         LOG.info("Using placement algorithm: %r" % p.__class__.__name__)
 408         # lets print the placement result
 409         for name, vnfd in self.vnfds.iteritems():
 410             LOG.info("Placed VNF %r on DC %r" % (name, str(vnfd.get("dc"))))
 411
 412
 413 """
 414 Some (simple) placement algorithms
 415 """
 416
 417
 418 class FirstDcPlacement(object):
 419     """
 420     Placement: Always use one and the same data center from the GK.dcs dict.
 421     """
 422     def place(self, nsd, vnfds, dcs):
 423         for name, vnfd in vnfds.iteritems():
 424             vnfd["dc"] = list(dcs.itervalues())[0]
 425
 426
 427 """
 428 Resource definitions and API endpoints
 429 """
 430
 431
 432 class Packages(fr.Resource):
 433
 434     def post(self):
 435         """
 436         Upload a *.son service package to the dummy gatekeeper.
 437
 438         We expect request with a *.son file and store it in UPLOAD_FOLDER
 439         :return: UUID
 440         """
 441         try:
 442             # get file contents
 443             print(request.files)
 444             # lets search for the package in the request
 445             if "package" in request.files:
 446                 son_file = request.files["package"]
 447             # elif "file" in request.files:
 448             #     son_file = request.files["file"]
 449             else:
 450                 return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed. file not found."}, 500
 451             # generate a uuid to reference this package
 452             service_uuid = str(uuid.uuid4())
 453             file_hash = hashlib.sha1(str(son_file)).hexdigest()
 454             # ensure that upload folder exists
 455             ensure_dir(UPLOAD_FOLDER)
 456             upload_path = os.path.join(UPLOAD_FOLDER, "%s.son" % service_uuid)
 457             # store *.son file to disk
 458             son_file.save(upload_path)
 459             size = os.path.getsize(upload_path)
 460             # create a service object and register it
 461             s = Service(service_uuid, file_hash, upload_path)
 462             GK.register_service_package(service_uuid, s)
 463             # generate the JSON result
 464             return {"service_uuid": service_uuid, "size": size, "sha1": file_hash, "error": None}
 465         except Exception as ex:
 466             LOG.exception("Service package upload failed:")
 467             return {"service_uuid": None, "size": 0, "sha1": None, "error": "upload failed"}, 500
 468
 469     def get(self):
 470         """
 471         Return a list of UUID's of uploaded service packages.
 472         :return: dict/list
 473         """
 474         LOG.info("GET /packages")
 475         return {"service_uuid_list": list(GK.services.iterkeys())}
 476
 477
 478 class Instantiations(fr.Resource):
 479
 480     def post(self):
 481         """
 482         Instantiate a service specified by its UUID.
 483         Will return a new UUID to identify the running service instance.
 484         :return: UUID
 485         """
 486         # try to extract the service uuid from the request
 487         json_data = request.get_json(force=True)
 488         service_uuid = json_data.get("service_uuid")
 489
 490         # lets be a bit fuzzy here to make testing easier
 491         if service_uuid is None and len(GK.services) > 0:
 492             # if we don't get a service uuid, we simple start the first service in the list
 493             service_uuid = list(GK.services.iterkeys())[0]
 494
 495         if service_uuid in GK.services:
 496             # ok, we have a service uuid, lets start the service
 497             service_instance_uuid = GK.services.get(service_uuid).start_service()
 498             return {"service_instance_uuid": service_instance_uuid}
 499         return "Service not found", 404
 500
 501     def get(self):
 502         """
 503         Returns a list of UUIDs containing all running services.
 504         :return: dict / list
 505         """
 506         LOG.info("GET /instantiations")
 507         return {"service_instantiations_list": [
 508             list(s.instances.iterkeys()) for s in GK.services.itervalues()]}
 509
 510
 511 # create a single, global GK object
 512 GK = Gatekeeper()
 513 # setup Flask
 514 app = Flask(__name__)
 515 app.config['MAX_CONTENT_LENGTH'] = 512 * 1024 * 1024  # 512 MB max upload
 516 api = fr.Api(app)
 517 # define endpoints
 518 api.add_resource(Packages, '/packages')
 519 api.add_resource(Instantiations, '/instantiations')
 520
 521
 522 def start_rest_api(host, port, datacenters=dict()):
 523     GK.dcs = datacenters
 524     # start the Flask server (not the best performance but ok for our use case)
 525     app.run(host=host,
 526             port=port,
 527             debug=True,
 528             use_reloader=False  # this is needed to run Flask in a non-main thread
 529             )
 530
 531
 532 def ensure_dir(name):
 533     if not os.path.exists(name):
 534         os.makedirs(name)
 535
 536
 537 def load_yaml(path):
 538     with open(path, "r") as f:
 539         try:
 540             r = yaml.load(f)
 541         except yaml.YAMLError as exc:
 542             LOG.exception("YAML parse error")
 543             r = dict()
 544     return r
 545
 546
 547 def make_relative_path(path):
 548     if path.startswith("file://"):
 549         path = path.replace("file://", "", 1)
 550     if path.startswith("/"):
 551         path = path.replace("/", "", 1)
 552     return path
 553
 554
 555 def generate_subnet_strings(n, start=1, subnet_size=24, ip=0):
 556     """
 557     Helper to generate different network configuration strings.
 558     """
 559     r = list()
 560     for i in range(start, start + n):
 561         r.append("%d.0.0.%d/%d" % (i, ip, subnet_size))
 562     return r
 563
 564
 565 if __name__ == '__main__':
 566     """
 567     Lets allow to run the API in standalone mode.
 568     """
 569     GK_STANDALONE_MODE = True
 570     logging.getLogger("werkzeug").setLevel(logging.INFO)
 571     start_rest_api("0.0.0.0", 8000)
 572