# Copyright (c) 2014 VMware, Inc. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. """ Functions and classes for image transfer between ESX/VC & image service. """ import errno import logging from eventlet import event from eventlet import greenthread from eventlet import queue from eventlet import timeout from oslo.vmware._i18n import _ from oslo.vmware import exceptions from oslo.vmware import rw_handles LOG = logging.getLogger(__name__) IMAGE_SERVICE_POLL_INTERVAL = 5 FILE_READ_WRITE_TASK_SLEEP_TIME = 0.01 BLOCKING_QUEUE_SIZE = 10 class BlockingQueue(queue.LightQueue): """Producer-Consumer queue to share data between reader/writer threads.""" def __init__(self, max_size, max_transfer_size): """Initializes the queue with the given parameters. :param max_size: maximum queue size; if max_size is less than zero or None, the queue size is infinite. :param max_transfer_size: maximum amount of data that can be _transferred using this queue """ queue.LightQueue.__init__(self, max_size) self._max_transfer_size = max_transfer_size self._transferred = 0 def read(self, chunk_size): """Read data from the queue. This method blocks until data is available. The input chunk size is ignored since we have ensured that the data chunks written to the pipe by the image reader thread is the same as the chunks asked for by the image writer thread. """ if (self._max_transfer_size is 0 or self._transferred < self._max_transfer_size): data_item = self.get() self._transferred += len(data_item) return data_item else: LOG.debug("Completed transfer of size %s.", self._transferred) return "" def write(self, data): """Write data into the queue. :param data: data to be written """ self.put(data) # Below methods are provided in order to enable treating the queue # as a file handle. def seek(self, offset, whence=0): """Set the file's current position at the offset. This method throws IOError since seek cannot be supported for a pipe. """ raise IOError(errno.ESPIPE, "Illegal seek") def tell(self): """Get the current file position.""" return self._transferred def close(self): pass def __str__(self): return "blocking queue" class ImageWriter(object): """Class to write the image to the image service from an input file.""" def __init__(self, context, input_file, image_service, image_id, image_meta=None): """Initializes the image writer instance with given parameters. :param context: write context needed by the image service :param input_file: file to read the image data from :param image_service: handle to image service :param image_id: ID of the image in the image service :param image_meta: image meta-data """ if not image_meta: image_meta = {} self._context = context self._input_file = input_file self._image_service = image_service self._image_id = image_id self._image_meta = image_meta self._running = False def start(self): """Start the image write task. :returns: the event indicating the status of the write task """ self._done = event.Event() def _inner(): """Task performing the image write operation. This method performs image data transfer through an update call. After the update, it waits until the image state becomes 'active', 'killed' or unknown. If the final state is not 'active' an instance of ImageTransferException is thrown. :raises: ImageTransferException """ LOG.debug("Calling image service update on image: %(image)s " "with meta: %(meta)s", {'image': self._image_id, 'meta': self._image_meta}) try: self._image_service.update(self._context, self._image_id, self._image_meta, data=self._input_file) self._running = True while self._running: LOG.debug("Retrieving status of image: %s.", self._image_id) image_meta = self._image_service.show(self._context, self._image_id) image_status = image_meta.get('status') if image_status == 'active': self.stop() LOG.debug("Image: %s is now active.", self._image_id) self._done.send(True) elif image_status == 'killed': self.stop() excep_msg = (_("Image: %s is in killed state.") % self._image_id) LOG.error(excep_msg) excep = exceptions.ImageTransferException(excep_msg) self._done.send_exception(excep) elif image_status in ['saving', 'queued']: LOG.debug("Image: %(image)s is in %(state)s state; " "sleeping for %(sleep)d seconds.", {'image': self._image_id, 'state': image_status, 'sleep': IMAGE_SERVICE_POLL_INTERVAL}) greenthread.sleep(IMAGE_SERVICE_POLL_INTERVAL) else: self.stop() excep_msg = (_("Image: %(image)s is in unknown " "state: %(state)s.") % {'image': self._image_id, 'state': image_status}) LOG.error(excep_msg) excep = exceptions.ImageTransferException(excep_msg) self._done.send_exception(excep) except Exception as excep: self.stop() excep_msg = (_("Error occurred while writing image: %s") % self._image_id) LOG.exception(excep_msg) excep = exceptions.ImageTransferException(excep_msg, excep) self._done.send_exception(excep) LOG.debug("Starting image write task for image: %(image)s with" " source: %(source)s.", {'source': self._input_file, 'image': self._image_id}) greenthread.spawn(_inner) return self._done def stop(self): """Stop the image writing task.""" LOG.debug("Stopping the writing task for image: %s.", self._image_id) self._running = False def wait(self): """Wait for the image writer task to complete. This method returns True if the writer thread completes successfully. In case of error, it raises ImageTransferException. :raises ImageTransferException """ return self._done.wait() def close(self): """This is a NOP.""" pass def __str__(self): string = "Image Writer " % (self._input_file, self._image_id) return string class FileReadWriteTask(object): """Task which reads data from the input file and writes to the output file. This class defines the task which copies the given input file to the given output file. The copy operation involves reading chunks of data from the input file and writing the same to the output file. """ def __init__(self, input_file, output_file): """Initializes the read-write task with the given input parameters. :param input_file: the input file handle :param output_file: the output file handle """ self._input_file = input_file self._output_file = output_file self._running = False def start(self): """Start the file read - file write task. :returns: the event indicating the status of the read-write task """ self._done = event.Event() def _inner(): """Task performing the file read-write operation.""" self._running = True while self._running: try: data = self._input_file.read(rw_handles.READ_CHUNKSIZE) if not data: LOG.debug("File read-write task is done.") self.stop() self._done.send(True) self._output_file.write(data) # update lease progress if applicable if hasattr(self._input_file, "update_progress"): self._input_file.update_progress() if hasattr(self._output_file, "update_progress"): self._output_file.update_progress() greenthread.sleep(FILE_READ_WRITE_TASK_SLEEP_TIME) except Exception as excep: self.stop() excep_msg = _("Error occurred during file read-write " "task.") LOG.exception(excep_msg) excep = exceptions.ImageTransferException(excep_msg, excep) self._done.send_exception(excep) LOG.debug("Starting file read-write task with source: %(source)s " "and destination: %(dest)s.", {'source': self._input_file, 'dest': self._output_file}) greenthread.spawn(_inner) return self._done def stop(self): """Stop the read-write task.""" LOG.debug("Stopping the file read-write task.") self._running = False def wait(self): """Wait for the file read-write task to complete. This method returns True if the read-write thread completes successfully. In case of error, it raises ImageTransferException. :raises: ImageTransferException """ return self._done.wait() def __str__(self): string = ("File Read-Write Task " % (self._input_file, self._output_file)) return string # Functions to perform image transfer between VMware servers and image service. def _start_transfer(context, timeout_secs, read_file_handle, max_data_size, write_file_handle=None, image_service=None, image_id=None, image_meta=None): """Start the image transfer. The image reader reads the data from the image source and writes to the blocking queue. The image source is always a file handle (VmdkReadHandle or ImageReadHandle); therefore, a FileReadWriteTask is created for this transfer. The image writer reads the data from the blocking queue and writes it to the image destination. The image destination is either a file or VMDK in VMware datastore or an image in the image service. If the destination is a file or VMDK in VMware datastore, the method creates a FileReadWriteTask which reads from the blocking queue and writes to either FileWriteHandle or VmdkWriteHandle. In the case of image service as the destination, an instance of ImageWriter task is created which reads from the blocking queue and writes to the image service. :param context: write context needed for the image service :param timeout_secs: time in seconds to wait for the transfer to complete :param read_file_handle: handle to read data from :param max_data_size: maximum transfer size :param write_file_handle: handle to write data to; if this is None, then param image_service and param image_id should be set. :param image_service: image service handle :param image_id: ID of the image in the image service :param image_meta: image meta-data :raises: ImageTransferException, ValueError """ # Create the blocking queue blocking_queue = BlockingQueue(BLOCKING_QUEUE_SIZE, max_data_size) # Create the image reader reader = FileReadWriteTask(read_file_handle, blocking_queue) # Create the image writer if write_file_handle: # File or VMDK in VMware datastore is the image destination writer = FileReadWriteTask(blocking_queue, write_file_handle) elif image_service and image_id: # Image service image is the destination writer = ImageWriter(context, blocking_queue, image_service, image_id, image_meta) else: excep_msg = _("No image destination given.") LOG.error(excep_msg) raise ValueError(excep_msg) # Start the reader and writer LOG.debug("Starting image transfer with reader: %(reader)s and writer: " "%(writer)s", {'reader': reader, 'writer': writer}) reader.start() writer.start() timer = timeout.Timeout(timeout_secs) try: # Wait for the reader and writer to complete reader.wait() writer.wait() except (timeout.Timeout, exceptions.ImageTransferException) as excep: excep_msg = (_("Error occurred during image transfer with reader: " "%(reader)s and writer: %(writer)s") % {'reader': reader, 'writer': writer}) LOG.exception(excep_msg) reader.stop() writer.stop() if isinstance(excep, exceptions.ImageTransferException): raise raise exceptions.ImageTransferException(excep_msg, excep) finally: timer.cancel() read_file_handle.close() if write_file_handle: write_file_handle.close() def download_flat_image(context, timeout_secs, image_service, image_id, **kwargs): """Download flat image from the image service to VMware server. :param context: image service write context :param timeout_secs: time in seconds to wait for the download to complete :param image_service: image service handle :param image_id: ID of the image to be downloaded :param kwargs: keyword arguments to configure the destination file write handle :raises: VimConnectionException, ImageTransferException, ValueError """ LOG.debug("Downloading image: %s from image service as a flat file.", image_id) # TODO(vbala) catch specific exceptions raised by download call read_iter = image_service.download(context, image_id) read_handle = rw_handles.ImageReadHandle(read_iter) file_size = int(kwargs.get('image_size')) write_handle = rw_handles.FileWriteHandle(kwargs.get('host'), kwargs.get('port'), kwargs.get('data_center_name'), kwargs.get('datastore_name'), kwargs.get('cookies'), kwargs.get('file_path'), file_size) _start_transfer(context, timeout_secs, read_handle, file_size, write_file_handle=write_handle) LOG.debug("Downloaded image: %s from image service as a flat file.", image_id) def download_stream_optimized_data(context, timeout_secs, read_handle, **kwargs): """Download stream optimized data to VMware server. :param context: image service write context :param timeout_secs: time in seconds to wait for the download to complete :param read_handle: handle from which to read the image data :param kwargs: keyword arguments to configure the destination VMDK write handle :returns: managed object reference of the VM created for import to VMware server :raises: VimException, VimFaultException, VimAttributeException, VimSessionOverLoadException, VimConnectionException, ImageTransferException, ValueError """ file_size = int(kwargs.get('image_size')) write_handle = rw_handles.VmdkWriteHandle(kwargs.get('session'), kwargs.get('host'), kwargs.get('port'), kwargs.get('resource_pool'), kwargs.get('vm_folder'), kwargs.get('vm_import_spec'), file_size) _start_transfer(context, timeout_secs, read_handle, file_size, write_file_handle=write_handle) return write_handle.get_imported_vm() def download_stream_optimized_image(context, timeout_secs, image_service, image_id, **kwargs): """Download stream optimized image from image service to VMware server. :param context: image service write context :param timeout_secs: time in seconds to wait for the download to complete :param image_service: image service handle :param image_id: ID of the image to be downloaded :param kwargs: keyword arguments to configure the destination VMDK write handle :returns: managed object reference of the VM created for import to VMware server :raises: VimException, VimFaultException, VimAttributeException, VimSessionOverLoadException, VimConnectionException, ImageTransferException, ValueError """ LOG.debug("Downloading image: %s from image service as a stream " "optimized file.", image_id) # TODO(vbala) catch specific exceptions raised by download call read_iter = image_service.download(context, image_id) read_handle = rw_handles.ImageReadHandle(read_iter) imported_vm = download_stream_optimized_data(context, timeout_secs, read_handle, **kwargs) LOG.debug("Downloaded image: %s from image service as a stream " "optimized file.", image_id) return imported_vm def copy_stream_optimized_disk( context, timeout_secs, write_handle, **kwargs): """Copy virtual disk from VMware server to the given write handle. :param context: context :param timeout_secs: time in seconds to wait for the copy to complete :param write_handle: copy destination :param kwargs: keyword arguments to configure the source VMDK read handle :raises: VimException, VimFaultException, VimAttributeException, VimSessionOverLoadException, VimConnectionException, ImageTransferException, ValueError """ vmdk_file_path = kwargs.get('vmdk_file_path') LOG.debug("Copying virtual disk: %(vmdk_path)s to %(dest)s.", {'vmdk_path': vmdk_file_path, 'dest': write_handle.name}) file_size = kwargs.get('vmdk_size') read_handle = rw_handles.VmdkReadHandle(kwargs.get('session'), kwargs.get('host'), kwargs.get('port'), kwargs.get('vm'), kwargs.get('vmdk_file_path'), file_size) _start_transfer(context, timeout_secs, read_handle, file_size, write_file_handle=write_handle) LOG.debug("Downloaded virtual disk: %s.", vmdk_file_path) def upload_image(context, timeout_secs, image_service, image_id, owner_id, **kwargs): """Upload the VM's disk file to image service. :param context: image service write context :param timeout_secs: time in seconds to wait for the upload to complete :param image_service: image service handle :param image_id: upload destination image ID :param kwargs: keyword arguments to configure the source VMDK read handle :raises: VimException, VimFaultException, VimAttributeException, VimSessionOverLoadException, VimConnectionException, ImageTransferException, ValueError """ LOG.debug("Uploading to image: %s.", image_id) file_size = kwargs.get('vmdk_size') read_handle = rw_handles.VmdkReadHandle(kwargs.get('session'), kwargs.get('host'), kwargs.get('port'), kwargs.get('vm'), kwargs.get('vmdk_file_path'), file_size) # Set the image properties. It is important to set the 'size' to 0. # Otherwise, the image service client will use the VM's disk capacity # which will not be the image size after upload, since it is converted # to a stream-optimized sparse disk. image_metadata = {'disk_format': 'vmdk', 'is_public': kwargs.get('is_public'), 'name': kwargs.get('image_name'), 'status': 'active', 'container_format': 'bare', 'size': 0, 'properties': {'vmware_image_version': kwargs.get('image_version'), 'vmware_disktype': 'streamOptimized', 'owner_id': owner_id}} # Passing 0 as the file size since data size to be transferred cannot be # predetermined. _start_transfer(context, timeout_secs, read_handle, 0, image_service=image_service, image_id=image_id, image_meta=image_metadata) LOG.debug("Uploaded image: %s.", image_id)