Vipin Balachandran
5ffe3c73ba
Currently we create an instance of FixedIntervalLoopingCall (to update lease progress) even if the write handle is not of type VmdkWriteHandle. Also, we do not create any progress updater for VmdkReadHandle. Refactoring _start_transfer to create updater only if the handle is of type VmdkHandle. Change-Id: I6e306a5ff453f16e9bdfbd97996206db7db8447e
343 lines
14 KiB
Python
343 lines
14 KiB
Python
# 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 logging
|
|
import tarfile
|
|
|
|
from eventlet import timeout
|
|
import six
|
|
|
|
from oslo_utils import units
|
|
from oslo_vmware._i18n import _
|
|
from oslo_vmware.common import loopingcall
|
|
from oslo_vmware import constants
|
|
from oslo_vmware import exceptions
|
|
from oslo_vmware import image_util
|
|
from oslo_vmware.objects import datastore as ds_obj
|
|
from oslo_vmware import rw_handles
|
|
from oslo_vmware import vim_util
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
NFC_LEASE_UPDATE_PERIOD = 60 # update NFC lease every 60sec.
|
|
CHUNK_SIZE = 64 * units.Ki # default chunk size for image transfer
|
|
|
|
|
|
def _create_progress_updater(handle):
|
|
if isinstance(handle, rw_handles.VmdkHandle):
|
|
updater = loopingcall.FixedIntervalLoopingCall(handle.update_progress)
|
|
updater.start(interval=NFC_LEASE_UPDATE_PERIOD)
|
|
return updater
|
|
|
|
|
|
def _start_transfer(read_handle, write_handle, timeout_secs):
|
|
# read_handle/write_handle could be an NFC lease, so we need to
|
|
# periodically update its progress
|
|
read_updater = _create_progress_updater(read_handle)
|
|
write_updater = _create_progress_updater(write_handle)
|
|
|
|
timer = timeout.Timeout(timeout_secs)
|
|
try:
|
|
while True:
|
|
data = read_handle.read(CHUNK_SIZE)
|
|
if not data:
|
|
break
|
|
write_handle.write(data)
|
|
except timeout.Timeout as excep:
|
|
msg = (_('Timeout, read_handle: "%(src)s", write_handle: "%(dest)s"') %
|
|
{'src': read_handle,
|
|
'dest': write_handle})
|
|
LOG.exception(msg)
|
|
raise exceptions.ImageTransferException(msg, excep)
|
|
except Exception as excep:
|
|
msg = (_('Error, read_handle: "%(src)s", write_handle: "%(dest)s"') %
|
|
{'src': read_handle,
|
|
'dest': write_handle})
|
|
LOG.exception(msg)
|
|
raise exceptions.ImageTransferException(msg, excep)
|
|
finally:
|
|
timer.cancel()
|
|
if read_updater:
|
|
read_updater.stop()
|
|
if write_updater:
|
|
write_updater.stop()
|
|
read_handle.close()
|
|
write_handle.close()
|
|
|
|
|
|
def download_image(image, image_meta, session, datastore, rel_path,
|
|
bypass=True, timeout_secs=7200):
|
|
"""Transfer an image to a datastore.
|
|
|
|
:param image: file-like iterator
|
|
:param image_meta: image metadata
|
|
:param session: VMwareAPISession object
|
|
:param datastore: Datastore object
|
|
:param rel_path: path where the file will be stored in the datastore
|
|
:param bypass: if set to True, bypass vCenter to download the image
|
|
:param timeout_secs: time in seconds to wait for the xfer to complete
|
|
"""
|
|
image_size = int(image_meta['size'])
|
|
method = 'PUT'
|
|
if bypass:
|
|
hosts = datastore.get_connected_hosts(session)
|
|
host = ds_obj.Datastore.choose_host(hosts)
|
|
host_name = session.invoke_api(vim_util, 'get_object_property',
|
|
session.vim, host, 'name')
|
|
ds_url = datastore.build_url(session._scheme, host_name, rel_path,
|
|
constants.ESX_DATACENTER_PATH)
|
|
cookie = ds_url.get_transfer_ticket(session, method)
|
|
conn = ds_url.connect(method, image_size, cookie)
|
|
else:
|
|
ds_url = datastore.build_url(session._scheme, session._host, rel_path)
|
|
cookie = '%s=%s' % (constants.SOAP_COOKIE_KEY,
|
|
session.vim.get_http_cookie().strip("\""))
|
|
conn = ds_url.connect(method, image_size, cookie)
|
|
conn.write = conn.send
|
|
|
|
read_handle = rw_handles.ImageReadHandle(image)
|
|
_start_transfer(read_handle, conn, timeout_secs)
|
|
|
|
|
|
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,
|
|
cacerts=kwargs.get('cacerts'))
|
|
_start_transfer(read_handle, write_handle, timeout_secs)
|
|
LOG.debug("Downloaded image: %s from image service as a flat file.",
|
|
image_id)
|
|
|
|
|
|
def download_file(
|
|
read_handle, host, port, dc_name, ds_name, cookies,
|
|
upload_file_path, file_size, cacerts, timeout_secs):
|
|
"""Download file to VMware server.
|
|
|
|
:param read_handle: file read handle
|
|
:param host: VMware server host name or IP address
|
|
:param port: VMware server port number
|
|
:param dc_name: name of the datacenter which contains the destination
|
|
datastore
|
|
:param ds_name: name of the destination datastore
|
|
:param cookies: cookies to build the cookie header while establishing
|
|
http connection with VMware server
|
|
:param upload_file_path: destination datastore file path
|
|
:param file_size: source file size
|
|
:param cacerts: CA bundle file to use for SSL verification
|
|
:param timeout_secs: timeout in seconds to wait for the download to
|
|
complete
|
|
"""
|
|
write_handle = rw_handles.FileWriteHandle(host,
|
|
port,
|
|
dc_name,
|
|
ds_name,
|
|
cookies,
|
|
upload_file_path,
|
|
file_size,
|
|
cacerts=cacerts)
|
|
_start_transfer(read_handle, write_handle, timeout_secs)
|
|
|
|
|
|
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,
|
|
kwargs.get('http_method', 'PUT'))
|
|
_start_transfer(read_handle, write_handle, timeout_secs)
|
|
return write_handle.get_imported_vm()
|
|
|
|
|
|
def _get_vmdk_handle(ova_handle):
|
|
|
|
with tarfile.open(mode="r|", fileobj=ova_handle) as tar:
|
|
vmdk_name = None
|
|
for tar_info in tar:
|
|
if tar_info and tar_info.name.endswith(".ovf"):
|
|
vmdk_name = image_util.get_vmdk_name_from_ovf(
|
|
tar.extractfile(tar_info))
|
|
elif vmdk_name and tar_info.name.startswith(vmdk_name):
|
|
# Actual file name is <vmdk_name>.XXXXXXX
|
|
return tar.extractfile(tar_info)
|
|
|
|
|
|
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
|
|
"""
|
|
metadata = image_service.show(context, image_id)
|
|
container_format = metadata.get('container_format')
|
|
|
|
LOG.debug("Downloading image: %(id)s (container: %(container)s) from image"
|
|
" service as a stream optimized file.",
|
|
{'id': image_id,
|
|
'container': container_format})
|
|
|
|
# TODO(vbala) catch specific exceptions raised by download call
|
|
read_iter = image_service.download(context, image_id)
|
|
read_handle = rw_handles.ImageReadHandle(read_iter)
|
|
|
|
if container_format == 'ova':
|
|
read_handle = _get_vmdk_handle(read_handle)
|
|
if read_handle is None:
|
|
raise exceptions.ImageTransferException(
|
|
_("No vmdk found in the OVA image %s.") % image_id)
|
|
|
|
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(read_handle, write_handle, timeout_secs)
|
|
LOG.debug("Downloaded virtual disk: %s.", vmdk_file_path)
|
|
|
|
|
|
# TODO(vbala) Remove dependency on image service provided by the client.
|
|
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)
|
|
|
|
# TODO(vbala) Remove this after we delete the keyword argument 'is_public'
|
|
# from all client code.
|
|
if 'is_public' in kwargs:
|
|
LOG.debug("Ignoring keyword argument 'is_public'.")
|
|
|
|
if 'image_version' in kwargs:
|
|
LOG.warning("The keyword argument 'image_version' is deprecated "
|
|
"and will be ignored in the next release.")
|
|
|
|
image_ver = six.text_type(kwargs.get('image_version'))
|
|
image_metadata = {'disk_format': 'vmdk',
|
|
'name': kwargs.get('image_name'),
|
|
'properties': {'vmware_image_version': image_ver,
|
|
'vmware_disktype': 'streamOptimized',
|
|
'owner_id': owner_id}}
|
|
|
|
updater = loopingcall.FixedIntervalLoopingCall(read_handle.update_progress)
|
|
try:
|
|
updater.start(interval=NFC_LEASE_UPDATE_PERIOD)
|
|
image_service.update(context, image_id, image_metadata,
|
|
data=read_handle)
|
|
finally:
|
|
updater.stop()
|
|
read_handle.close()
|
|
LOG.debug("Uploaded image: %s.", image_id)
|