rwlaunchpad/plugins/rwlaunchpadtasklet/rift/tasklets/rwlaunchpad/uploader.py

   1
   2 #
   3 #   Copyright 2016 RIFT.IO Inc
   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 import abc
  18 import asyncio
  19 import collections
  20 import os
  21 import tempfile
  22 import threading
  23 import uuid
  24 import zlib
  25 import re
  26
  27 import tornado
  28 import tornado.escape
  29 import tornado.ioloop
  30 import tornado.web
  31 import tornado.httputil
  32 import tornadostreamform.multipart_streamer as multipart_streamer
  33
  34 import requests
  35
  36 # disable unsigned certificate warning
  37 from requests.packages.urllib3.exceptions import InsecureRequestWarning
  38 requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
  39
  40 import gi
  41 gi.require_version('RwLaunchpadYang', '1.0')
  42 gi.require_version('ProjectNsdYang', '1.0')
  43 gi.require_version('ProjectVnfdYang', '1.0')
  44
  45 from gi.repository import (
  46         ProjectNsdYang as NsdYang,
  47         ProjectVnfdYang as VnfdYang,
  48         )
  49 import rift.mano.cloud
  50
  51 import rift.package.checksums
  52 import rift.package.convert
  53 import rift.package.handler as pkg_handler
  54 import rift.package.icon
  55 import rift.package.package
  56 import rift.package.script
  57 import rift.package.store
  58
  59 from gi.repository import (
  60    RwDts as rwdts,
  61    RwPkgMgmtYang
  62    )
  63 import rift.downloader as downloader
  64 import rift.mano.dts as mano_dts
  65 import rift.tasklets
  66
  67 from . import (
  68         export,
  69         extract,
  70         image,
  71         message,
  72         onboard,
  73         state,
  74         )
  75
  76 from .message import (
  77         MessageException,
  78
  79         # Onboard Error Messages
  80         OnboardChecksumMismatch,
  81         OnboardDescriptorError,
  82         OnboardDescriptorExistsError,
  83         OnboardDescriptorFormatError,
  84         OnboardError,
  85         OnboardExtractionError,
  86         OnboardImageUploadError,
  87         OnboardMissingContentBoundary,
  88         OnboardMissingContentType,
  89         OnboardMissingTerminalBoundary,
  90         OnboardUnreadableHeaders,
  91         OnboardUnreadablePackage,
  92         OnboardUnsupportedMediaType,
  93
  94         # Onboard Status Messages
  95         OnboardDescriptorOnboard,
  96         OnboardFailure,
  97         OnboardImageUpload,
  98         OnboardPackageUpload,
  99         OnboardPackageValidation,
 100         OnboardStart,
 101         OnboardSuccess,
 102
 103         DownloadError,
 104         DownloadSuccess,
 105
 106         # Update Error Messages
 107         UpdateChecksumMismatch,
 108         UpdateDescriptorError,
 109         UpdateDescriptorFormatError,
 110         UpdateError,
 111         UpdateExtractionError,
 112         UpdateImageUploadError,
 113         UpdateMissingContentBoundary,
 114         UpdateMissingContentType,
 115         UpdatePackageNotFoundError,
 116         UpdateUnreadableHeaders,
 117         UpdateUnreadablePackage,
 118         UpdateUnsupportedMediaType,
 119
 120         # Update Status Messages
 121         UpdateDescriptorUpdate,
 122         UpdateDescriptorUpdated,
 123         UpdatePackageUpload,
 124         UpdateStart,
 125         UpdateSuccess,
 126         UpdateFailure,
 127         )
 128
 129 from .tosca import ExportTosca
 130
 131 from .onboard import OnboardError as OnboardException
 132
 133 MB = 1024 * 1024
 134 GB = 1024 * MB
 135
 136 MAX_STREAMED_SIZE = 5 * GB
 137
 138 # Shortcuts
 139 RPC_PACKAGE_CREATE_ENDPOINT = RwPkgMgmtYang.YangOutput_RwPkgMgmt_PackageCreate
 140 RPC_PACKAGE_UPDATE_ENDPOINT = RwPkgMgmtYang.YangOutput_RwPkgMgmt_PackageUpdate
 141
 142 class HttpMessageError(Exception):
 143     def __init__(self, code, msg):
 144         self.code = code
 145         self.msg = msg
 146
 147
 148 class UploadRpcHandler(mano_dts.AbstractRpcHandler):
 149     def __init__(self, application):
 150         """
 151         Args:
 152             application: UploaderApplication
 153         """
 154         super().__init__(application.log, application.dts, application.loop)
 155         self.application = application
 156
 157     @property
 158     def xpath(self):
 159         return "/rw-pkg-mgmt:package-create"
 160
 161     @asyncio.coroutine
 162     def callback(self, ks_path, msg):
 163         transaction_id = str(uuid.uuid4())
 164         log = self.application.get_logger(transaction_id)
 165         log.message(OnboardStart())
 166
 167         self.log.debug("Package create RPC: {}".format(msg))
 168
 169         auth = None
 170         if msg.username is not None:
 171             auth = (msg.username, msg.password)
 172
 173         try:
 174             project = msg.project_name
 175         except AttributeError as e:
 176             self._log.warning("Did not get project name in RPC: {}".
 177                               format(msg.as_dict()))
 178             project = rift.mano.utils.project.DEFAULT_PROJECT
 179
 180         self.application.onboard(
 181                 msg.external_url,
 182                 transaction_id,
 183                 auth=auth,
 184                 project=project,
 185                 )
 186
 187         rpc_op = RPC_PACKAGE_CREATE_ENDPOINT.from_dict({
 188             "transaction_id": transaction_id,
 189             "project_name": project,
 190         })
 191
 192         return rpc_op
 193
 194
 195 class UpdateRpcHandler(mano_dts.AbstractRpcHandler):
 196     def __init__(self, application):
 197         """
 198         Args:
 199             application: UploaderApplication
 200         """
 201         super().__init__(application.log, application.dts, application.loop)
 202         self.application = application
 203
 204     @property
 205     def xpath(self):
 206         return "/rw-pkg-mgmt:package-update"
 207
 208     @asyncio.coroutine
 209     def callback(self, ks_path, msg):
 210
 211         transaction_id = str(uuid.uuid4())
 212         log = self.application.get_logger(transaction_id)
 213         log.message(UpdateStart())
 214
 215         auth = None
 216         if msg.username is not None:
 217             auth = (msg.username, msg.password)
 218
 219         self.application.update(
 220                 msg.external_url,
 221                 transaction_id,
 222                 auth=auth,
 223                 project=msg.project_name,
 224                 )
 225
 226         rpc_op = RPC_PACKAGE_UPDATE_ENDPOINT.from_dict({
 227             "transaction_id": transaction_id,
 228             "project_name": msg.project_name,
 229         })
 230
 231         return rpc_op
 232
 233
 234 class UploadStateHandler(state.StateHandler):
 235     STARTED = OnboardStart
 236     SUCCESS = OnboardSuccess
 237     FAILURE = OnboardFailure
 238
 239
 240 class UpdateStateHandler(state.StateHandler):
 241     STARTED = UpdateStart
 242     SUCCESS = UpdateSuccess
 243     FAILURE = UpdateFailure
 244
 245
 246 class UpdatePackage(downloader.DownloaderProtocol):
 247
 248     def __init__(self, log, loop, project, url, auth,
 249                  onboarder, uploader, package_store_map, transaction_id):
 250         super().__init__()
 251         self.log = log
 252         self.loop = loop
 253         self.project = project
 254         self.url = url
 255         self.auth = auth
 256         self.onboarder = onboarder
 257         self.uploader = uploader
 258         self.package_store_map = package_store_map
 259         self.transaction_id = transaction_id
 260
 261
 262     def _update_package(self, packages):
 263
 264         # Extract package could return multiple packages if
 265         # the package is converted
 266         for pkg in packages:
 267             with pkg as temp_package:
 268                 package_checksums = self.validate_package(temp_package)
 269                 stored_package = self.update_package(temp_package)
 270                 self.validate_descriptor_fields(temp_package)
 271
 272                 try:
 273                     self.extract_icons(temp_package)
 274                     self.update_descriptors(temp_package)
 275
 276                 except Exception:
 277                     self.delete_stored_package(stored_package)
 278                     raise
 279
 280                 else:
 281                     self.upload_images(temp_package, package_checksums)
 282
 283     def extract(self, packages):
 284         try:
 285             self._update_package(packages)
 286             self.log.message(UpdateSuccess())
 287
 288         except MessageException as e:
 289             self.log.message(e.msg)
 290             self.log.message(UpdateFailure())
 291             raise UpdateFailure(str(e))
 292
 293         except Exception as e:
 294             self.log.exception(e)
 295             if str(e):
 296                 self.log.message(UpdateError(str(e)))
 297             self.log.message(UpdateFailure())
 298
 299     def on_download_succeeded(self, job):
 300         self.log.message(DownloadSuccess("Package downloaded."))
 301
 302         extractor = extract.UploadPackageExtractor(self.log)
 303         file_backed_packages = extractor.create_packages_from_upload(
 304                 job.filename, job.filepath
 305                 )
 306         try:
 307             self.extract(file_backed_packages)
 308         except Exception as e:
 309             raise Exception("Error in Package Update")
 310
 311     def on_download_finished(self, job):
 312         self.log.debug("*** Download completed")
 313         if hasattr(self.project, 'update_status_handler'):
 314             self.project.update_status_handler.update_status(job, self.transaction_id)
 315
 316     def on_download_progress(self, job):
 317         self.log.debug("*** Download in progress")
 318         if hasattr(self.project, 'update_status_handler'):
 319             self.project.update_status_handler.update_status(job, self.transaction_id)
 320
 321     def on_download_failed(self, job):
 322         self.log.error(job.detail)
 323         self.log.message(DownloadError("Package download failed. {}".format(job.detail)))
 324         self.log.message(UpdateFailure())
 325
 326     def download_package(self):
 327
 328         _, filename = tempfile.mkstemp()
 329         url_downloader = downloader.UrlDownloader(
 330                 self.url,
 331                 auth=self.auth,
 332                 file_obj=filename,
 333                 decompress_on_fly=True,
 334                 log=self.log)
 335         url_downloader.delegate = self
 336         url_downloader.download()
 337
 338     def get_package_store(self, package):
 339         return self.package_store_map[package.descriptor_type]
 340
 341     def update_package(self, package):
 342         store = self.get_package_store(package)
 343
 344         try:
 345             store.update_package(package)
 346         except rift.package.store.PackageNotFoundError as e:
 347             # If the package doesn't exist, then it is possible the descriptor was onboarded
 348             # out of band.  In that case, just store the package as is
 349             self.log.warning("Package not found, storing new package instead.")
 350             store.store_package(package)
 351
 352         stored_package = store.get_package(package.descriptor_id)
 353
 354         return stored_package
 355
 356     def delete_stored_package(self, package):
 357         self.log.info("Deleting stored package: %s", package)
 358         store = self.get_package_store(package)
 359         try:
 360             store.delete_package(package.descriptor_id)
 361         except Exception as e:
 362             self.log.warning("Failed to delete package from store: %s", str(e))
 363
 364     def upload_images(self, package, package_checksums):
 365         image_file_map = rift.package.image.get_package_image_files(package)
 366         name_hdl_map = {name: package.open(image_file_map[name]) for name in image_file_map}
 367         if not image_file_map:
 368             return
 369
 370         try:
 371             for image_name, image_hdl in name_hdl_map.items():
 372                 image_file = image_file_map[image_name]
 373                 if image_file in package_checksums:
 374                     image_checksum = package_checksums[image_file]
 375                 else:
 376                     self.log.warning("checksum not provided for image %s.  Calculating checksum",
 377                                      image_file)
 378                     image_checksum = rift.package.checksums.checksum(
 379                             package.open(image_file_map[image_name])
 380                             )
 381                 try:
 382                     self.uploader.upload_image(image_name, image_checksum, image_hdl)
 383                     self.uploader.upload_image_to_cloud_accounts(image_name, image_checksum, self.project)
 384
 385                 except image.ImageUploadError as e:
 386                     self.log.exception("Failed to upload image: %s", image_name)
 387                     raise MessageException(OnboardImageUploadError(str(e))) from e
 388
 389         finally:
 390             _ = [image_hdl.close() for image_hdl in name_hdl_map.values()]
 391
 392     def extract_icons(self, package):
 393         try:
 394             icon_extractor = rift.package.icon.PackageIconExtractor(self.log)
 395             icon_extractor.extract_icons(package)
 396         except rift.package.icon.IconExtractionError as e:
 397             raise MessageException(UpdateExtractionError()) from e
 398
 399     def validate_descriptor_fields(self, package):
 400         # We can add more VNFD validations here. Currently we are validating only cloud-init
 401         if package.descriptor_msg is not None:
 402             self.validate_cloud_init_file(package)
 403
 404     def validate_cloud_init_file(self, package):
 405         """ This validation is for VNFDs with associated VDUs. """
 406         if 'vdu' in package.descriptor_msg.as_dict():
 407             for vdu in package.descriptor_msg.as_dict()['vdu']:
 408                 if 'cloud_init_file' in vdu:
 409                     cloud_init_file = vdu['cloud_init_file']
 410                     for file in package.files:
 411                         if file.endswith('/' + cloud_init_file) is True:
 412                             return
 413                     raise MessageException(
 414                         OnboardError("Cloud-Init file reference in VNFD does not match with cloud-init file"))
 415
 416     def validate_package(self, package):
 417         checksum_validator = rift.package.package.PackageChecksumValidator(self.log)
 418
 419         try:
 420             checksum_validator.validate(package)
 421         except rift.package.package.PackageFileChecksumError as e:
 422             raise MessageException(UpdateChecksumMismatch(e.filename)) from e
 423         except rift.package.package.PackageValidationError as e:
 424             raise MessageException(UpdateUnreadablePackage()) from e
 425
 426         return checksum_validator.checksums
 427
 428     def update_descriptors(self, package):
 429         descriptor_msg = package.descriptor_msg
 430
 431         self.log.message(UpdateDescriptorUpdate())
 432
 433         try:
 434             self.onboarder.update(descriptor_msg, project=self.project.name)
 435         except onboard.UpdateError as e:
 436             raise MessageException(UpdateDescriptorError(package.descriptor_file)) from e
 437
 438
 439 class OnboardPackage(downloader.DownloaderProtocol):
 440
 441     def __init__(self, log, loop, project, url, auth,
 442                  onboarder, uploader, package_store_map, transaction_id):
 443         self.log = log
 444         self.loop = loop
 445         self.project = project
 446         self.url = url
 447         self.auth = auth
 448         self.onboarder = onboarder
 449         self.uploader = uploader
 450         self.package_store_map = package_store_map
 451         self.transaction_id = transaction_id
 452
 453     def _onboard_package(self, packages):
 454         # Extract package could return multiple packages if
 455         # the package is converted
 456         for pkg in packages:
 457             with pkg as temp_package:
 458                 package_checksums = self.validate_package(temp_package)
 459                 stored_package = self.store_package(temp_package)
 460                 try:
 461                     self.validate_descriptor_fields(temp_package)
 462                 except Exception as e:
 463                     self.log.exception("Descriptor validation Failed")
 464                     raise
 465                 try:
 466                     self.extract_icons(temp_package)
 467                     self.onboard_descriptors(temp_package)
 468
 469                 except Exception as e:
 470                     if "data-exists" not in e.msg.text:
 471                         self.delete_stored_package(stored_package)
 472                     raise
 473                 else:
 474                     self.upload_images(temp_package, package_checksums)
 475
 476     def extract(self, packages):
 477         try:
 478             self._onboard_package(packages)
 479             self.log.message(OnboardSuccess())
 480
 481         except MessageException as e:
 482             self.log.message(e.msg)
 483             self.log.message(OnboardFailure())
 484             raise OnboardException(OnboardFailure())
 485
 486
 487         except Exception as e:
 488             self.log.exception(e)
 489             if str(e):
 490                 self.log.message(OnboardError(str(e)))
 491             self.log.message(OnboardFailure())
 492
 493     def on_download_succeeded(self, job):
 494         self.log.message(DownloadSuccess("Package downloaded."))
 495
 496         extractor = extract.UploadPackageExtractor(self.log)
 497         file_backed_packages = extractor.create_packages_from_upload(
 498                 job.filename, job.filepath
 499                 )
 500         try:
 501             self.extract(file_backed_packages)
 502         except Exception as e:
 503             raise Exception("Error in Onboarding Package")
 504
 505     def on_download_finished(self, job):
 506         self.log.debug("*** Download completed")
 507         if hasattr(self.project, 'upload_status_handler'):
 508             self.project.upload_status_handler.upload_status(job, self.transaction_id)
 509
 510     def on_download_progress(self, job):
 511         self.log.debug("*** Download in progress")
 512         if hasattr(self.project, 'upload_status_handler'):
 513             self.project.upload_status_handler.upload_status(job, self.transaction_id)
 514
 515     def on_download_failed(self, job):
 516         self.log.error(job.detail)
 517         self.log.message(DownloadError("Package download failed. {}".format(job.detail)))
 518         self.log.message(OnboardFailure())
 519
 520     def download_package(self):
 521
 522         self.log.debug("Before pkg download, project = {}".format(self.project.name))
 523         _, filename = tempfile.mkstemp()
 524         url_downloader = downloader.UrlDownloader(
 525                 self.url,
 526                 auth=self.auth,
 527                 file_obj=filename,
 528                 decompress_on_fly=True,
 529                 log=self.log)
 530         url_downloader.delegate = self
 531         url_downloader.download()
 532
 533     def get_package_store(self, package):
 534         return self.package_store_map[package.descriptor_type]
 535
 536     def store_package(self, package):
 537         store = self.get_package_store(package)
 538
 539         try:
 540             store.store_package(package)
 541         except rift.package.store.PackageExistsError as e:
 542             store.update_package(package)
 543
 544         stored_package = store.get_package(package.descriptor_id)
 545
 546         return stored_package
 547
 548     def delete_stored_package(self, package):
 549         self.log.info("Deleting stored package: %s", package)
 550         store = self.get_package_store(package)
 551         try:
 552             store.delete_package(package.descriptor_id)
 553         except Exception as e:
 554             self.log.warning("Failed to delete package from store: %s", str(e))
 555
 556     def upload_images(self, package, package_checksums):
 557         image_file_map = rift.package.image.get_package_image_files(package)
 558         if not image_file_map:
 559             return
 560
 561         name_hdl_map = {name: package.open(image_file_map[name]) for name in image_file_map}
 562         try:
 563             for image_name, image_hdl in name_hdl_map.items():
 564                 image_file = image_file_map[image_name]
 565                 if image_file in package_checksums:
 566                     image_checksum = package_checksums[image_file]
 567                 else:
 568                     self.log.warning("checksum not provided for image %s.  Calculating checksum",
 569                                      image_file)
 570                     image_checksum = rift.package.checksums.checksum(
 571                             package.open(image_file_map[image_name])
 572                             )
 573                 try:
 574                     set_image_property = {}
 575                     self.uploader.upload_image(image_name, image_checksum, image_hdl, set_image_property)
 576                     self.uploader.upload_image_to_cloud_accounts(image_name, image_checksum, self.project.name)
 577
 578                 except image.ImageUploadError as e:
 579                     raise MessageException(OnboardImageUploadError(str(e))) from e
 580
 581         finally:
 582             _ = [image_hdl.close() for image_hdl in name_hdl_map.values()]
 583
 584     def extract_icons(self, package):
 585         try:
 586             icon_extractor = rift.package.icon.PackageIconExtractor(self.log)
 587             icon_extractor.extract_icons(package)
 588         except rift.package.icon.IconExtractionError as e:
 589             raise MessageException(OnboardExtractionError()) from e
 590
 591     def validate_descriptor_fields(self, package):
 592         # We can add more VNFD/NSD validations here.
 593         if package.descriptor_msg is not None:
 594             try:
 595                 self.validate_cloud_init_file(package)
 596                 self.validate_vld_mgmt_network(package)
 597             except Exception as e:
 598                 raise
 599
 600     def validate_vld_mgmt_network(self, package):
 601         """ This is validation at onboarding of NSD for atleast one of the VL's to have mgmt network true
 602             and have minimum one connection point"""
 603         if package.descriptor_type == 'nsd':
 604             for  vld in package.descriptor_msg.as_dict().get('vld',[]):
 605                 if vld.get('mgmt_network', False) is True and \
 606                     len(vld.get('vnfd_connection_point_ref',[])) > 0 :
 607                     break
 608             else:
 609                 self.log.error(("AtLeast One of the VL's should have Management Network as True "
 610                                  "and have minimum one connection point"))
 611                 raise Exception("Management Network not defined.")
 612
 613     def validate_cloud_init_file(self, package):
 614         """ This validation is for VNFDs with associated VDUs. """
 615         if 'vdu' in package.descriptor_msg.as_dict():
 616             for vdu in package.descriptor_msg.as_dict()['vdu']:
 617                 if 'cloud_init_file' in vdu:
 618                     cloud_init_file = vdu['cloud_init_file']
 619                     for file in package.files:
 620                         if file.endswith('/' + cloud_init_file) is True:
 621                             return
 622                     raise MessageException(
 623                         OnboardError("Cloud-Init file reference in VNFD does not match with cloud-init file"))
 624
 625     def validate_package(self, package):
 626         validators = (
 627                 rift.package.package.PackageChecksumValidator(self.log),
 628                 rift.package.package.PackageConstructValidator(self.log),
 629                 )
 630
 631         # Run the validators for checksum and package construction for imported pkgs
 632         for validator in validators:
 633             try:
 634                 validator.validate(package)
 635
 636             except rift.package.package.PackageFileChecksumError as e:
 637                 raise MessageException(OnboardChecksumMismatch(e.filename)) from e
 638             except rift.package.package.PackageValidationError as e:
 639                 raise MessageException(OnboardUnreadablePackage()) from e
 640
 641         return validators[0].checksums
 642
 643     def onboard_descriptors(self, package):
 644         def process_error_messsage(exception, package):
 645             """
 646             This method captures error reason. This needs to be enhanced with time.
 647             """
 648             exception_msg = str(exception)
 649             match_duplicate = re.findall('<error-tag>(.*?)</error-tag>', exception_msg, re.DOTALL)
 650
 651             if len(match_duplicate) > 0:
 652                 error_message = str(match_duplicate[0])
 653                 return error_message
 654
 655             match = re.findall('<tailf:missing-element>(.*?)</tailf:missing-element>', exception_msg, re.DOTALL)
 656             error_message = ""
 657             if len(match) > 0:
 658                 for element in match:
 659                     element_message = "Missing element : {}".format(element)
 660                     error_message += element_message
 661             else:
 662                 error_message = package.descriptor_file
 663             return error_message
 664
 665         def process_exception(exception, package):
 666             return OnboardDescriptorError(process_error_messsage(exception, package))
 667
 668         descriptor_msg = package.descriptor_msg
 669         self.log.message(OnboardDescriptorOnboard())
 670
 671         try:
 672             self.onboarder.onboard(descriptor_msg, project=self.project.name)
 673         except onboard.OnboardError as e:
 674             raise MessageException(process_exception(e, package)) from e
 675
 676
 677 class UploaderApplication(tornado.web.Application):
 678
 679     @classmethod
 680     def from_tasklet(cls, tasklet):
 681         manifest = tasklet.tasklet_info.get_pb_manifest()
 682         use_ssl = manifest.bootstrap_phase.rwsecurity.use_ssl
 683         ssl_cert = manifest.bootstrap_phase.rwsecurity.cert
 684         ssl_key = manifest.bootstrap_phase.rwsecurity.key
 685         return cls(
 686             tasklet,
 687             ssl=(ssl_cert, ssl_key))
 688
 689     def __init__(
 690             self,
 691             tasklet,
 692             ssl=None,
 693             vnfd_store=None,
 694             nsd_store=None):
 695
 696         self.log = tasklet.log
 697         self.loop = tasklet.loop
 698         self.dts = tasklet.dts
 699
 700         self.accounts = {}
 701         self.ro_accounts = {}
 702
 703         self.use_ssl = False
 704         self.ssl_cert, self.ssl_key = None, None
 705         if ssl:
 706             self.use_ssl = True
 707             self.ssl_cert, self.ssl_key = ssl
 708
 709         self.messages = collections.defaultdict(list)
 710         self.export_dir = os.path.join(os.environ['RIFT_VAR_ROOT'], 'launchpad/exports')
 711
 712         self.uploader = image.ImageUploader(self.log, self.loop, self.dts)
 713         self.onboarder = onboard.DescriptorOnboarder(
 714                 self.log, "127.0.0.1", 8008, self.use_ssl, self.ssl_cert, self.ssl_key
 715                 )
 716
 717         self.exporter = export.DescriptorPackageArchiveExporter(self.log)
 718         self.loop.create_task(export.periodic_export_cleanup(self.log, self.loop, self.export_dir))
 719
 720         self.tasklet = tasklet
 721         self.get_vnfd_catalog = tasklet.get_vnfd_catalog
 722         self.get_nsd_catalog = tasklet.get_nsd_catalog
 723         catalog_map = {
 724                  "vnfd": self.get_vnfd_catalog,
 725                  "nsd": self.get_nsd_catalog
 726                  }
 727
 728         self.upload_handler = UploadRpcHandler(self)
 729         self.update_handler = UpdateRpcHandler(self)
 730         self.export_handler = export.ExportRpcHandler(self, catalog_map)
 731
 732         attrs = dict(log=self.log, loop=self.loop)
 733
 734         super(UploaderApplication, self).__init__([
 735             (r"/api/package/vnfd/(.*)", pkg_handler.FileRestApiHandler, {
 736                 'path': rift.package.store.VnfdPackageFilesystemStore.DEFAULT_ROOT_DIR}),
 737             (r"/api/package/nsd/(.*)", pkg_handler.FileRestApiHandler, {
 738                 'path': rift.package.store.NsdPackageFilesystemStore.DEFAULT_ROOT_DIR}),
 739
 740             (r"/api/upload/([^/]+)/state", UploadStateHandler, attrs),
 741             (r"/api/update/([^/]+)/state", UpdateStateHandler, attrs),
 742             (r"/api/export/([^/]+)/state", export.ExportStateHandler, attrs),
 743
 744             (r"/api/export/([^/]+.tar.gz)", tornado.web.StaticFileHandler, {
 745                 "path": self.export_dir,
 746                 }),
 747             (r"/api/export/([^/]+.zip)", tornado.web.StaticFileHandler, {
 748                 "path": self.export_dir,
 749                 }),
 750             ])
 751
 752     @asyncio.coroutine
 753     def register(self):
 754         yield from self.upload_handler.register()
 755         yield from self.update_handler.register()
 756         yield from self.export_handler.register()
 757
 758     def get_logger(self, transaction_id):
 759         return message.Logger(self.log, self.messages[transaction_id])
 760
 761     def build_store_map(self, project=None):
 762         ''' Use project information to build vnfd/nsd filesystem stores with appropriate
 763         package directory root.
 764         '''
 765         vnfd_store = rift.package.store.VnfdPackageFilesystemStore(self.log) if not \
 766             project else rift.package.store.VnfdPackageFilesystemStore(self.log, project=project)
 767         nsd_store = rift.package.store.NsdPackageFilesystemStore(self.log) if not \
 768             project else rift.package.store.NsdPackageFilesystemStore(self.log, project=project)
 769
 770         return dict(vnfd = vnfd_store, nsd = nsd_store)
 771
 772     def onboard(self, url, transaction_id, auth=None, project=None):
 773         log = message.Logger(self.log, self.messages[transaction_id])
 774
 775         try:
 776             self.project = self.tasklet._get_project(project)
 777         except Exception as e:
 778             self.log.error("Exception raised ...%s" % (str(e)))
 779             self.log.exception(e)
 780
 781         self.package_store_map = self.build_store_map(project)
 782         onboard_package = OnboardPackage(
 783                 log,
 784                 self.loop,
 785                 self.project,
 786                 url,
 787                 auth,
 788                 self.onboarder,
 789                 self.uploader,
 790                 self.package_store_map,
 791                 transaction_id
 792                 )
 793
 794         self.loop.run_in_executor(None, onboard_package.download_package)
 795
 796     def update(self, url, transaction_id, auth=None, project=None):
 797         log = message.Logger(self.log, self.messages[transaction_id])
 798
 799         try:
 800             self.project = self.tasklet._get_project(project)
 801         except Exception as e:
 802             self.log.error("Exception raised ...%s" % (str(e)))
 803             self.log.exception(e)
 804
 805         self.package_store_map = self.build_store_map(project)
 806         update_package = UpdatePackage(
 807                 log,
 808                 self.loop,
 809                 self.project,
 810                 url,
 811                 auth,
 812                 self.onboarder,
 813                 self.uploader,
 814                 self.package_store_map,
 815                 transaction_id
 816                 )
 817
 818         self.loop.run_in_executor(None, update_package.download_package)