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