3 # Copyright 2016 RIFT.IO Inc
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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.
25 import rift
.mano
.cloud
28 gi
.require_version('RwImageMgmtYang', '1.0')
29 from gi
.repository
import (
34 class UploadJobError(Exception):
38 class ImageUploadTaskError(Exception):
42 class ImageUploadError(ImageUploadTaskError
):
46 class ImageListError(ImageUploadTaskError
):
50 class ImageUploadJobController(object):
51 """ This class starts and manages ImageUploadJobs """
52 MAX_COMPLETED_JOBS
= 20
54 def __init__(self
, log
, loop
, project
, max_completed_jobs
=MAX_COMPLETED_JOBS
):
57 self
._project
= project
58 self
._job
_id
_gen
= itertools
.count(1)
59 self
._max
_completed
_jobs
= max_completed_jobs
62 self
._completed
_jobs
= collections
.deque(
63 maxlen
=self
._max
_completed
_jobs
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
)
73 return upload_jobs_msg
77 """ the tracked list of ImageUploadJobs """
78 return self
._jobs
.values()
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
]
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
]
90 def _add_job(self
, job
):
91 self
._jobs
[job
.id] = job
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
)
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]
104 self
._completed
_jobs
.appendleft(job
)
106 job_future
= job
.start()
107 job_future
.add_done_callback(on_job_completed
)
109 if on_completed
is not None:
110 job_future
.add_done_callback(on_completed
)
112 def get_job(self
, job_id
):
113 """ Get the UploadJob from the job id
116 job_id - the job id that was previously added to the controller
119 The associated ImageUploadJob
122 LookupError - Could not find the job id
124 if job_id
not in self
._jobs
:
125 raise LookupError("Could not find job_id %s" % job_id
)
127 return self
._jobs
[job_id
]
129 def create_job(self
, image_tasks
, on_completed
=None):
130 """ Create and start a ImageUploadJob from a list of ImageUploadTasks
133 image_tasks - a list of ImageUploadTasks
134 on_completed - a callback which is added to the job future
139 self
._log
.debug("Creating new job from %s image tasks", len(image_tasks
))
140 new_job
= ImageUploadJob(
144 job_id
=next(self
._job
_id
_gen
)
147 self
._add
_job
(new_job
)
148 self
._start
_job
(new_job
, on_completed
=on_completed
)
153 class ImageUploadJob(object):
154 """ This class manages a set of ImageUploadTasks
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.
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.
162 STATES
= ("QUEUED", "IN_PROGRESS", "CANCELLING", "CANCELLED", "COMPLETED", "FAILED")
163 TIMEOUT_JOB
= 6 * 60 * 60 # 6 hours
164 JOB_GEN
= itertools
.count(1)
166 def __init__(self
, log
, loop
, upload_tasks
, job_id
=None, timeout_job
=TIMEOUT_JOB
):
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
173 self
._state
= "QUEUED"
174 self
._state
_stack
= [self
._state
]
176 self
._start
_time
= time
.time()
179 self
._task
_future
_map
= {}
180 self
._job
_future
= None
183 return "{}(job_id={}, state={})".format(
184 self
.__class
__.__name
__, self
._job
_id
, self
._state
193 """ The state of the ImageUploadJob """
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
)
204 self
._state
= new_state
207 def state_stack(self
):
208 """ The list of states that this job progressed through """
209 return self
._state
_stack
213 """ The UploadJob protobuf message """
214 task
= RwImageMgmtYang
.UploadJob
.from_dict({
216 "status": self
._state
,
217 "start_time": self
._start
_time
,
218 "upload_tasks": [task
.pb_msg
for task
in self
._upload
_tasks
]
222 task
.stop_time
= self
._stop
_time
226 def _start_upload_tasks(self
):
227 self
._log
.debug("Starting %s upload tasks", len(self
._upload
_tasks
))
229 for upload_task
in self
._upload
_tasks
:
233 def _wait_for_upload_tasks(self
):
234 self
._log
.debug("Waiting for upload tasks to complete")
236 wait_coroutines
= [t
.wait() for t
in self
._upload
_tasks
]
238 yield from asyncio
.wait(
240 timeout
=self
._timeout
_job
,
244 self
._log
.debug("All upload tasks completed")
246 def _set_final_job_state(self
):
248 for task
in self
._upload
_tasks
:
249 if task
.state
!= "COMPLETED":
250 failed_tasks
.append(task
)
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"
258 self
._log
.debug("%s tasks completed successfully", len(self
._upload
_tasks
))
259 self
.state
= "COMPLETED"
262 def _cancel_job(self
):
263 for task
in self
._upload
_tasks
:
266 # TODO: Wait for all tasks to actually reach terminal
269 self
.state
= "CANCELLED"
273 self
.state
= "IN_PROGRESS"
274 self
._start
_upload
_tasks
()
276 yield from self
._wait
_for
_upload
_tasks
()
277 except asyncio
.CancelledError
:
278 self
._log
.debug("%s was cancelled. Cancelling all tasks.",
280 self
._loop
.create_task(self
._cancel
_job
())
283 self
._stop
_time
= time
.time()
284 self
._job
_future
= None
286 self
._set
_final
_job
_state
()
290 """ Wait for the job to reach a terminal state """
291 if self
._job
_future
is None:
292 raise UploadJobError("Job not started")
294 yield from asyncio
.wait_for(
301 """ Start the job and all child tasks """
302 if self
._state
!= "QUEUED":
303 raise UploadJobError("Job already started")
305 self
._job
_future
= self
._loop
.create_task(self
._do
_job
())
306 return self
._job
_future
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()
315 class ByteRateCalculator(object):
316 """ This class produces a byte rate from inputted measurements"""
317 def __init__(self
, rate_time_constant
):
319 self
._time
_constant
= rate_time_constant
325 def add_measurement(self
, num_bytes
, time_delta
):
326 rate
= num_bytes
/ time_delta
330 self
._rate
+= ((rate
- self
._rate
) / self
._time
_constant
)
335 class UploadProgressWriteProxy(object):
336 """ This class implements a write proxy with produces various progress stats
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.
343 RATE_TIME_CONSTANT
= 5
345 def __init__(self
, log
, loop
, bytes_total
, write_hdl
):
348 self
._bytes
_total
= bytes_total
349 self
._write
_hdl
= write_hdl
351 self
._bytes
_written
= 0
354 self
._rate
_calc
= ByteRateCalculator(UploadProgressWriteProxy
.RATE_TIME_CONSTANT
)
355 self
._rate
_task
= None
357 def write(self
, data
):
358 self
._write
_hdl
.write(data
)
359 self
._bytes
_written
+= len(data
)
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()
367 def start_rate_monitoring(self
):
368 """ Start the rate monitoring task """
370 def periodic_rate_task():
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
378 self
._byte
_rate
= self
._rate
_calc
.add_measurement(num_bytes
, time_period
)
380 self
._log
.debug("starting rate monitoring task")
381 self
._rate
_task
= self
._loop
.create_task(periodic_rate_task())
384 def progress_percent(self
):
385 if self
._bytes
_total
== 0:
388 return int(self
._bytes
_written
/ self
._bytes
_total
* 100)
391 def bytes_written(self
):
392 return self
._bytes
_written
395 def bytes_total(self
):
396 return self
._bytes
_total
399 def bytes_rate(self
):
400 return self
._byte
_rate
403 class GlanceImagePipeGen(object):
404 """ This class produces a read file handle from a generator that produces bytes
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.
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.
414 def __init__(self
, log
, loop
, data_gen
):
417 self
._data
_gen
= data_gen
419 read_fd
, write_fd
= os
.pipe()
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
427 return self
._write
_hdl
430 def write_hdl(self
, new_write_hdl
):
431 self
._write
_hdl
= new_write_hdl
435 return self
._read
_hdl
437 def _gen_writer(self
):
438 self
._log
.debug("starting image data write to pipe")
440 for data
in self
._data
_gen
:
442 self
._write
_hdl
.write(data
)
443 except (BrokenPipeError
, ValueError) as e
:
444 self
._log
.warning("write pipe closed: %s", str(e
))
447 except Exception as e
:
448 self
._log
.exception("error when writing data to pipe: %s", str(e
))
451 self
._log
.debug("closing write side of pipe")
453 self
._write
_hdl
.close()
458 t
= threading
.Thread(target
=self
._gen
_writer
)
463 self
._log
.debug("stop requested, closing write side of pipe")
464 self
._write
_hdl
.close()
467 class AccountImageUploadTask(object):
468 """ This class manages an create_image task from an image info and file handle
470 Manage the upload of a image to a configured cloud account.
472 STATES
= ("QUEUED", "CHECK_IMAGE_EXISTS", "UPLOADING", "CANCELLING", "CANCELLED", "COMPLETED", "FAILED")
474 TIMEOUT_CHECK_EXISTS
= 10
475 TIMEOUT_IMAGE_UPLOAD
= 6 * 60 * 60 # 6 hours
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
483 self
._account
= account
484 self
._image
_info
= image_info
.deep_copy()
485 self
._image
_hdl
= image_hdl
487 self
._timeout
_exists
= timeout_exists
488 self
._timeout
_upload
= timeout_upload
490 self
._progress
_info
= progress_info
491 self
._write
_canceller
= write_canceller
493 self
._state
= "QUEUED"
494 self
._state
_stack
= [self
._state
]
496 self
._detail
= "Task is waiting to be started"
497 self
._start
_time
= time
.time()
499 self
._upload
_future
= None
501 if not image_info
.has_field("name"):
502 raise ValueError("image info must have name field")
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
)
515 self
._state
= new_state
518 def state_stack(self
):
519 return self
._state
_stack
523 """ The image name being uploaded """
524 return self
._image
_info
.id
527 def image_name(self
):
528 """ The image name being uploaded """
529 return self
._image
_info
.name
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
540 def cloud_account(self
):
541 """ The cloud account name which the image is being uploaded to """
542 return self
._account
.name
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
,
556 if self
.image_checksum
is not None:
557 task
.image_checksum
= self
.image_checksum
560 task
.stop_time
= self
._stop
_time
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
568 if self
.state
== "COMPLETED":
569 task
.progress_percent
= 100
573 def _get_account_images(self
):
575 self
._log
.debug("getting image list for account {}".format(self
._account
.name
))
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
)
581 raise ImageListError(msg
) from e
583 return account_images
585 def _has_existing_image(self
):
586 account
= self
._account
588 account_images
= self
._get
_account
_images
()
590 matching_images
= [i
for i
in account_images
if i
.name
== self
.image_name
]
592 if self
.image_checksum
is not None:
593 matching_images
= [i
for i
in matching_images
if i
.checksum
== self
.image_checksum
]
596 self
._log
.debug("found matching image with checksum in account %s",
600 self
._log
.debug("did not find matching image with checksum in account %s",
604 def _upload_image(self
):
605 image
= self
._image
_info
606 account
= self
._account
608 image
.fileno
= self
._image
_hdl
.fileno()
610 self
._log
.debug("uploading to account {}: {}".format(account
.name
, image
))
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
))
616 raise ImageUploadError(msg
) from e
618 self
._log
.debug('uploaded image (id: {}) to account{}: {}'.format(
619 image
.id, account
.name
, image
.name
))
624 def _do_upload(self
):
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
,
633 self
.state
= "COMPLETED"
634 self
._detail
= "Image already exists on destination"
637 self
.state
= "UPLOADING"
638 self
._detail
= "Uploading image"
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
,
649 except asyncio
.CancelledError
as e
:
650 self
.state
= "CANCELLED"
651 self
._detail
= "Image upload cancelled"
653 except ImageUploadTaskError
as e
:
654 self
.state
= "FAILED"
655 self
._detail
= str(e
)
657 except asyncio
.TimeoutError
as e
:
658 self
.state
= "FAILED"
659 self
._detail
= "Timed out during upload task: %s" % str(e
)
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."
671 self
.state
= "COMPLETED"
672 self
._detail
= "Image successfully uploaded. Image id: %s" % image_id
675 self
._stop
_time
= time
.time()
676 self
._upload
_future
= None
680 """ Wait for the upload task to complete """
681 if self
._upload
_future
is None:
682 raise ImageUploadError("Task not started")
684 yield from asyncio
.wait_for(
686 self
._timeout
_upload
, loop
=self
._loop
690 """ Start the upload task """
691 if self
._state
!= "QUEUED":
692 raise ImageUploadError("Task already started")
694 self
._log
.info("Starting %s", self
)
696 self
._upload
_future
= self
._loop
.create_task(self
._do
_upload
())
698 return self
._upload
_future
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
)
706 self
.state
= "CANCELLING"
707 self
._detail
= "Cancellation has been requested"
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()