rwlaunchpad/plugins/rwimagemgr/rift/tasklets/rwimagemgr/upload.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
  18 import asyncio
  19 import collections
  20 import itertools
  21 import os
  22 import time
  23 import threading
  24
  25 import rift.mano.cloud
  26
  27 import gi
  28 gi.require_version('RwImageMgmtYang', '1.0')
  29 from gi.repository import (
  30         RwImageMgmtYang,
  31         )
  32
  33
  34 class UploadJobError(Exception):
  35     pass
  36
  37
  38 class ImageUploadTaskError(Exception):
  39     pass
  40
  41
  42 class ImageUploadError(ImageUploadTaskError):
  43     pass
  44
  45
  46 class ImageListError(ImageUploadTaskError):
  47     pass
  48
  49
  50 class ImageUploadJobController(object):
  51     """ This class starts and manages ImageUploadJobs """
  52     MAX_COMPLETED_JOBS = 20
  53
  54     def __init__(self, log, loop, project, max_completed_jobs=MAX_COMPLETED_JOBS):
  55         self._log = log
  56         self._loop = loop
  57         self._project = project
  58         self._job_id_gen = itertools.count(1)
  59         self._max_completed_jobs = max_completed_jobs
  60
  61         self._jobs = {}
  62         self._completed_jobs = collections.deque(
  63                 maxlen=self._max_completed_jobs
  64                 )
  65
  66     @property
  67     def pb_msg(self):
  68         """ the UploadJobs protobuf message """
  69         upload_jobs_msg = RwImageMgmtYang.UploadJobs()
  70         for job in self._jobs.values():
  71             upload_jobs_msg.job.append(job.pb_msg)
  72
  73         return upload_jobs_msg
  74
  75     @property
  76     def jobs(self):
  77         """ the tracked list of ImageUploadJobs """
  78         return self._jobs.values()
  79
  80     @property
  81     def completed_jobs(self):
  82         """ completed jobs in the tracked list of ImageUploadJobs """
  83         return [job for job in self._jobs.values() if job in self._completed_jobs]
  84
  85     @property
  86     def active_jobs(self):
  87         """ in-progress jobs in the tracked list of ImageUploadJobs """
  88         return [job for job in self._jobs.values() if job not in self._completed_jobs]
  89
  90     def _add_job(self, job):
  91         self._jobs[job.id] = job
  92
  93     def _start_job(self, job, on_completed=None):
  94         def on_job_completed(_):
  95             self._log.debug("%s completed.  Adding to completed jobs list.", job)
  96
  97             # If adding a new completed job is going to overflow the
  98             # completed job list, find the first job that completed and
  99             # remove it from the tracked jobs.
 100             if len(self._completed_jobs) == self._completed_jobs.maxlen:
 101                 first_completed_job = self._completed_jobs[-1]
 102                 del self._jobs[first_completed_job.id]
 103
 104             self._completed_jobs.appendleft(job)
 105
 106         job_future = job.start()
 107         job_future.add_done_callback(on_job_completed)
 108
 109         if on_completed is not None:
 110             job_future.add_done_callback(on_completed)
 111
 112     def get_job(self, job_id):
 113         """ Get the UploadJob from the job id
 114
 115         Arguments:
 116             job_id - the job id that was previously added to the controller
 117
 118         Returns:
 119             The associated ImageUploadJob
 120
 121         Raises:
 122             LookupError - Could not find the job id
 123         """
 124         if job_id not in self._jobs:
 125             raise LookupError("Could not find job_id %s" % job_id)
 126
 127         return self._jobs[job_id]
 128
 129     def create_job(self, image_tasks, on_completed=None):
 130         """ Create and start a ImageUploadJob from a list of ImageUploadTasks
 131
 132         Arguments:
 133             image_tasks - a list of ImageUploadTasks
 134             on_completed - a callback which is added to the job future
 135
 136         Returns:
 137             A ImageUploadJob id
 138         """
 139         self._log.debug("Creating new job from %s image tasks", len(image_tasks))
 140         new_job = ImageUploadJob(
 141                 self._log,
 142                 self._loop,
 143                 image_tasks,
 144                 job_id=next(self._job_id_gen)
 145                 )
 146
 147         self._add_job(new_job)
 148         self._start_job(new_job, on_completed=on_completed)
 149
 150         return new_job.id
 151
 152
 153 class ImageUploadJob(object):
 154     """ This class manages a set of ImageUploadTasks
 155
 156     In order to push an image (or set of images) to many cloud accounts, and get a single
 157     status on that operation, we need a single status that represents all of those tasks.
 158
 159     The ImageUploadJob provides a single endpoint to control all the tasks and report
 160     when all images are successfully upload or when any one fails.
 161     """
 162     STATES = ("QUEUED", "IN_PROGRESS", "CANCELLING", "CANCELLED", "COMPLETED", "FAILED")
 163     TIMEOUT_JOB = 6 * 60 * 60  # 6 hours
 164     JOB_GEN = itertools.count(1)
 165
 166     def __init__(self, log, loop, upload_tasks, job_id=None, timeout_job=TIMEOUT_JOB):
 167         self._log = log
 168         self._loop = loop
 169         self._upload_tasks = upload_tasks
 170         self._job_id = next(ImageUploadJob.JOB_GEN) if job_id is None else job_id
 171         self._timeout_job = timeout_job
 172
 173         self._state = "QUEUED"
 174         self._state_stack = [self._state]
 175
 176         self._start_time = time.time()
 177         self._stop_time = 0
 178
 179         self._task_future_map = {}
 180         self._job_future = None
 181
 182     def __repr__(self):
 183         return "{}(job_id={}, state={})".format(
 184                 self.__class__.__name__, self._job_id, self._state
 185                 )
 186
 187     @property
 188     def id(self):
 189         return self._job_id
 190
 191     @property
 192     def state(self):
 193         """ The state of the ImageUploadJob """
 194         return self._state
 195
 196     @state.setter
 197     def state(self, new_state):
 198         """ Set the state of the ImageUploadJob """
 199         states = ImageUploadJob.STATES
 200         assert new_state in states
 201         assert states.index(new_state) >= states.index(self._state)
 202         self._state_stack.append(new_state)
 203
 204         self._state = new_state
 205
 206     @property
 207     def state_stack(self):
 208         """ The list of states that this job progressed through  """
 209         return self._state_stack
 210
 211     @property
 212     def pb_msg(self):
 213         """ The UploadJob protobuf message """
 214         task = RwImageMgmtYang.UploadJob.from_dict({
 215             "id": self._job_id,
 216             "status": self._state,
 217             "start_time": self._start_time,
 218             "upload_tasks": [task.pb_msg for task in self._upload_tasks]
 219         })
 220
 221         if self._stop_time:
 222             task.stop_time = self._stop_time
 223
 224         return task
 225
 226     def _start_upload_tasks(self):
 227         self._log.debug("Starting %s upload tasks", len(self._upload_tasks))
 228
 229         for upload_task in self._upload_tasks:
 230             upload_task.start()
 231
 232     @asyncio.coroutine
 233     def _wait_for_upload_tasks(self):
 234         self._log.debug("Waiting for upload tasks to complete")
 235
 236         wait_coroutines = [t.wait() for t in self._upload_tasks]
 237         if wait_coroutines:
 238             yield from asyncio.wait(
 239                     wait_coroutines,
 240                     timeout=self._timeout_job,
 241                     loop=self._loop
 242                     )
 243
 244         self._log.debug("All upload tasks completed")
 245
 246     def _set_final_job_state(self):
 247         failed_tasks = []
 248         for task in self._upload_tasks:
 249             if task.state != "COMPLETED":
 250                 failed_tasks.append(task)
 251
 252         if failed_tasks:
 253             self._log.error("%s had %s FAILED tasks.", self, len(failed_tasks))
 254             for ftask in failed_tasks:
 255                 self._log.error("%s : Failed to upload image : %s to cloud_account : %s", self, ftask.image_name, ftask.cloud_account)
 256             self.state = "FAILED"
 257         else:
 258             self._log.debug("%s tasks completed successfully", len(self._upload_tasks))
 259             self.state = "COMPLETED"
 260
 261     @asyncio.coroutine
 262     def _cancel_job(self):
 263         for task in self._upload_tasks:
 264             task.stop()
 265
 266         # TODO: Wait for all tasks to actually reach terminal
 267         # states.
 268
 269         self.state = "CANCELLED"
 270
 271     @asyncio.coroutine
 272     def _do_job(self):
 273         self.state = "IN_PROGRESS"
 274         self._start_upload_tasks()
 275         try:
 276             yield from self._wait_for_upload_tasks()
 277         except asyncio.CancelledError:
 278             self._log.debug("%s was cancelled.  Cancelling all tasks.",
 279                             self)
 280             self._loop.create_task(self._cancel_job())
 281             raise
 282         finally:
 283             self._stop_time = time.time()
 284             self._job_future = None
 285
 286         self._set_final_job_state()
 287
 288     @asyncio.coroutine
 289     def wait(self):
 290         """ Wait for the job to reach a terminal state """
 291         if self._job_future is None:
 292             raise UploadJobError("Job not started")
 293
 294         yield from asyncio.wait_for(
 295                 self._job_future,
 296                 self._timeout_job,
 297                 loop=self._loop
 298                 )
 299
 300     def start(self):
 301         """ Start the job and all child tasks """
 302         if self._state != "QUEUED":
 303             raise UploadJobError("Job already started")
 304
 305         self._job_future = self._loop.create_task(self._do_job())
 306         return self._job_future
 307
 308     def stop(self):
 309         """ Stop the job and all child tasks  """
 310         if self._job_future is not None:
 311             self.state = "CANCELLING"
 312             self._job_future.cancel()
 313
 314
 315 class ByteRateCalculator(object):
 316     """  This class produces a byte rate from inputted measurements"""
 317     def __init__(self, rate_time_constant):
 318         self._rate = 0
 319         self._time_constant = rate_time_constant
 320
 321     @property
 322     def rate(self):
 323         return self._rate
 324
 325     def add_measurement(self, num_bytes, time_delta):
 326         rate = num_bytes / time_delta
 327         if self._rate == 0:
 328             self._rate = rate
 329         else:
 330             self._rate += ((rate - self._rate) / self._time_constant)
 331
 332         return self._rate
 333
 334
 335 class UploadProgressWriteProxy(object):
 336     """ This class implements a write proxy with produces various progress stats
 337
 338     In order to keep the complexity of the UploadTask down, this class acts as a
 339     proxy for a file write.  By providing the original handle to be written to
 340     and having the client class call write() on this object, we can produce the
 341     various statistics to be consumed.
 342     """
 343     RATE_TIME_CONSTANT = 5
 344
 345     def __init__(self, log, loop, bytes_total, write_hdl):
 346         self._log = log
 347         self._loop = loop
 348         self._bytes_total = bytes_total
 349         self._write_hdl = write_hdl
 350
 351         self._bytes_written = 0
 352         self._byte_rate = 0
 353
 354         self._rate_calc = ByteRateCalculator(UploadProgressWriteProxy.RATE_TIME_CONSTANT)
 355         self._rate_task = None
 356
 357     def write(self, data):
 358         self._write_hdl.write(data)
 359         self._bytes_written += len(data)
 360
 361     def close(self):
 362         self._write_hdl.close()
 363         if self._rate_task is not None:
 364             self._log.debug("stopping rate monitoring task")
 365             self._rate_task.cancel()
 366
 367     def start_rate_monitoring(self):
 368         """ Start the rate monitoring task """
 369         @asyncio.coroutine
 370         def periodic_rate_task():
 371             while True:
 372                 start_time = time.time()
 373                 start_bytes = self._bytes_written
 374                 yield from asyncio.sleep(1, loop=self._loop)
 375                 time_period = time.time() - start_time
 376                 num_bytes = self._bytes_written - start_bytes
 377
 378                 self._byte_rate = self._rate_calc.add_measurement(num_bytes, time_period)
 379
 380         self._log.debug("starting rate monitoring task")
 381         self._rate_task = self._loop.create_task(periodic_rate_task())
 382
 383     @property
 384     def progress_percent(self):
 385         if self._bytes_total == 0:
 386             return 0
 387
 388         return int(self._bytes_written / self._bytes_total * 100)
 389
 390     @property
 391     def bytes_written(self):
 392         return self._bytes_written
 393
 394     @property
 395     def bytes_total(self):
 396         return self._bytes_total
 397
 398     @property
 399     def bytes_rate(self):
 400         return self._byte_rate
 401
 402
 403 class GlanceImagePipeGen(object):
 404     """ This class produces a read file handle from a generator that produces bytes
 405
 406     The CAL API takes a file handle as an input.  The Glance API creates a generator
 407     that produces byte strings.  This class acts as the mediator by creating a pipe
 408     and pumping the bytestring from the generator into the write side of the pipe.
 409
 410     A pipe has the useful feature here that it will block at the buffer size until
 411     the reader has consumed.  This allows us to only pull from glance and push at the
 412     pace of the reader preventing us from having to store the images locally on disk.
 413     """
 414     def __init__(self, log, loop, data_gen):
 415         self._log = log
 416         self._loop = loop
 417         self._data_gen = data_gen
 418
 419         read_fd, write_fd = os.pipe()
 420
 421         self._read_hdl = os.fdopen(read_fd, 'rb')
 422         self._write_hdl = os.fdopen(write_fd, 'wb')
 423         self._close_hdl = self._write_hdl
 424
 425     @property
 426     def write_hdl(self):
 427         return self._write_hdl
 428
 429     @write_hdl.setter
 430     def write_hdl(self, new_write_hdl):
 431         self._write_hdl = new_write_hdl
 432
 433     @property
 434     def read_hdl(self):
 435         return self._read_hdl
 436
 437     def _gen_writer(self):
 438         self._log.debug("starting image data write to pipe")
 439         try:
 440             for data in self._data_gen:
 441                 try:
 442                     self._write_hdl.write(data)
 443                 except (BrokenPipeError, ValueError) as e:
 444                     self._log.warning("write pipe closed: %s", str(e))
 445                     return
 446
 447         except Exception as e:
 448             self._log.exception("error when writing data to pipe: %s", str(e))
 449
 450         finally:
 451             self._log.debug("closing write side of pipe")
 452             try:
 453                 self._write_hdl.close()
 454             except OSError:
 455                 pass
 456
 457     def start(self):
 458         t = threading.Thread(target=self._gen_writer)
 459         t.daemon = True
 460         t.start()
 461
 462     def stop(self):
 463         self._log.debug("stop requested, closing write side of pipe")
 464         self._write_hdl.close()
 465
 466
 467 class AccountImageUploadTask(object):
 468     """ This class manages an create_image task from an image info and file handle
 469
 470     Manage the upload of a image to a configured cloud account.
 471     """
 472     STATES = ("QUEUED", "CHECK_IMAGE_EXISTS", "UPLOADING", "CANCELLING", "CANCELLED", "COMPLETED", "FAILED")
 473
 474     TIMEOUT_CHECK_EXISTS = 10
 475     TIMEOUT_IMAGE_UPLOAD = 6 * 60 * 60  # 6 hours
 476
 477     def __init__(self, log, loop, account, image_info, image_hdl,
 478                  timeout_exists=TIMEOUT_CHECK_EXISTS, timeout_upload=TIMEOUT_IMAGE_UPLOAD,
 479                  progress_info=None, write_canceller=None
 480                  ):
 481         self._log = log
 482         self._loop = loop
 483         self._account = account
 484         self._image_info = image_info.deep_copy()
 485         self._image_hdl = image_hdl
 486
 487         self._timeout_exists = timeout_exists
 488         self._timeout_upload = timeout_upload
 489
 490         self._progress_info = progress_info
 491         self._write_canceller = write_canceller
 492
 493         self._state = "QUEUED"
 494         self._state_stack = [self._state]
 495
 496         self._detail = "Task is waiting to be started"
 497         self._start_time = time.time()
 498         self._stop_time = 0
 499         self._upload_future = None
 500
 501         if not image_info.has_field("name"):
 502             raise ValueError("image info must have name field")
 503
 504     @property
 505     def state(self):
 506         return self._state
 507
 508     @state.setter
 509     def state(self, new_state):
 510         states = AccountImageUploadTask.STATES
 511         assert new_state in states
 512         assert states.index(new_state) >= states.index(self._state)
 513         self._state_stack.append(new_state)
 514
 515         self._state = new_state
 516
 517     @property
 518     def state_stack(self):
 519         return self._state_stack
 520
 521     @property
 522     def image_id(self):
 523         """ The image name being uploaded """
 524         return self._image_info.id
 525
 526     @property
 527     def image_name(self):
 528         """ The image name being uploaded """
 529         return self._image_info.name
 530
 531     @property
 532     def image_checksum(self):
 533         """ The image checksum being uploaded """
 534         if self._image_info.has_field("checksum"):
 535             return self._image_info.checksum
 536
 537         return None
 538
 539     @property
 540     def cloud_account(self):
 541         """ The cloud account name which the image is being uploaded to """
 542         return self._account.name
 543
 544     @property
 545     def pb_msg(self):
 546         """ The UploadTask protobuf message """
 547         task = RwImageMgmtYang.UploadTask.from_dict({
 548             "cloud_account": self.cloud_account,
 549             "image_id": self.image_id,
 550             "image_name": self.image_name,
 551             "status": self.state,
 552             "detail": self._detail,
 553             "start_time": self._start_time,
 554         })
 555
 556         if self.image_checksum is not None:
 557             task.image_checksum = self.image_checksum
 558
 559         if self._stop_time:
 560             task.stop_time = self._stop_time
 561
 562         if self._progress_info:
 563             task.bytes_written = self._progress_info.bytes_written
 564             task.bytes_total = self._progress_info.bytes_total
 565             task.progress_percent = self._progress_info.progress_percent
 566             task.bytes_per_second = self._progress_info.bytes_rate
 567
 568         if self.state == "COMPLETED":
 569             task.progress_percent = 100
 570
 571         return task
 572
 573     def _get_account_images(self):
 574         account_images = []
 575         self._log.debug("getting image list for account {}".format(self._account.name))
 576         try:
 577             account_images = self._account.get_image_list()
 578         except rift.mano.cloud.CloudAccountCalError as e:
 579             msg = "could not get image list for account {}".format(self._account.name)
 580             self._log.error(msg)
 581             raise ImageListError(msg) from e
 582
 583         return account_images
 584
 585     def _has_existing_image(self):
 586         account = self._account
 587
 588         account_images = self._get_account_images()
 589
 590         matching_images = [i for i in account_images if i.name == self.image_name]
 591
 592         if self.image_checksum is not None:
 593             matching_images = [i for i in matching_images if i.checksum == self.image_checksum]
 594
 595         if matching_images:
 596             self._log.debug("found matching image with checksum in account %s",
 597                             account.name)
 598             return True
 599
 600         self._log.debug("did not find matching image with checksum in account %s",
 601                         account.name)
 602         return False
 603
 604     def _upload_image(self):
 605         image = self._image_info
 606         account = self._account
 607
 608         image.fileno = self._image_hdl.fileno()
 609
 610         self._log.debug("uploading to account {}: {}".format(account.name, image))
 611         try:
 612             image.id = account.create_image(image)
 613         except rift.mano.cloud.CloudAccountCalError as e:
 614             msg = "error when uploading image {} to cloud account: {}".format(image.name, str(e))
 615             self._log.error(msg)
 616             raise ImageUploadError(msg) from e
 617
 618         self._log.debug('uploaded image (id: {}) to account{}: {}'.format(
 619                         image.id, account.name, image.name))
 620
 621         return image.id
 622
 623     @asyncio.coroutine
 624     def _do_upload(self):
 625         try:
 626             self.state = "CHECK_IMAGE_EXISTS"
 627             has_image = yield from asyncio.wait_for(
 628                     self._loop.run_in_executor(None, self._has_existing_image),
 629                     timeout=self._timeout_exists,
 630                     loop=self._loop
 631                     )
 632             if has_image:
 633                 self.state = "COMPLETED"
 634                 self._detail = "Image already exists on destination"
 635                 return
 636
 637             self.state = "UPLOADING"
 638             self._detail = "Uploading image"
 639
 640             # Note that if the upload times out, the upload thread may still
 641             # stick around.  We'll need another method of cancelling the task
 642             # through the VALA interface.
 643             image_id = yield from asyncio.wait_for(
 644                     self._loop.run_in_executor(None, self._upload_image),
 645                     timeout=self._timeout_upload,
 646                     loop=self._loop
 647                     )
 648
 649         except asyncio.CancelledError as e:
 650             self.state = "CANCELLED"
 651             self._detail = "Image upload cancelled"
 652
 653         except ImageUploadTaskError as e:
 654             self.state = "FAILED"
 655             self._detail = str(e)
 656
 657         except asyncio.TimeoutError as e:
 658             self.state = "FAILED"
 659             self._detail = "Timed out during upload task: %s" % str(e)
 660
 661         else:
 662             # If the user does not provide a checksum and performs a URL source
 663             # upload with an incorrect URL, then Glance does not indicate a failure
 664             # and the CAL cannot detect an incorrect upload.  In this case, use
 665             # the bytes_written to detect a bad upload and mark the task as failed.
 666             if self._progress_info and self._progress_info.bytes_written == 0:
 667                 self.state = "FAILED"
 668                 self._detail = "No bytes written.  Possible bad image source."
 669                 return
 670
 671             self.state = "COMPLETED"
 672             self._detail = "Image successfully uploaded.  Image id: %s" % image_id
 673
 674         finally:
 675             self._stop_time = time.time()
 676             self._upload_future = None
 677
 678     @asyncio.coroutine
 679     def wait(self):
 680         """ Wait for the upload task to complete """
 681         if self._upload_future is None:
 682             raise ImageUploadError("Task not started")
 683
 684         yield from asyncio.wait_for(
 685                 self._upload_future,
 686                 self._timeout_upload, loop=self._loop
 687                 )
 688
 689     def start(self):
 690         """ Start the upload task """
 691         if self._state != "QUEUED":
 692             raise ImageUploadError("Task already started")
 693
 694         self._log.info("Starting %s", self)
 695
 696         self._upload_future = self._loop.create_task(self._do_upload())
 697
 698         return self._upload_future
 699
 700     def stop(self):
 701         """ Stop the upload task in progress """
 702         if self._upload_future is None:
 703             self._log.warning("Cannot cancel %s.  Not in progress.", self)
 704             return
 705
 706         self.state = "CANCELLING"
 707         self._detail = "Cancellation has been requested"
 708
 709         self._log.info("Cancelling %s", self)
 710         self._upload_future.cancel()
 711         if self._write_canceller is not None:
 712             self._write_canceller.stop()