openstack
/
cinder
Code Issues Proposed changes

VMware: Backend driver for VStorageObject 

Volume driver based on VMware VStorageObject aka First Class
Disk (FCD). This driver requires a minimum vCenter version
of 6.5.

Implements: blueprint vmware-fcd-driver
Change-Id: I983e0b7c650a2358e4af9862365d29dfa107210a
This commit is contained in:
Vipin Balachandran 2017-11-09 17:25:12 -08:00
parent 4cbc03a0bd
commit 377549c67c
5 changed files with 1231 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

505
cinder/tests/unit/volume/drivers/vmware/test_fcd.py Normal file
View File

@ -0,0 +1,505 @@
# Copyright (c) 2017 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.
"""
Test suite for VMware vCenter FCD driver.
"""
import ddt
import mock
from oslo_utils import units
from oslo_vmware import image_transfer
from oslo_vmware.objects import datastore
from oslo_vmware import vim_util
from cinder import context
from cinder import exception as cinder_exceptions
from cinder import test
from cinder.tests.unit import fake_snapshot
from cinder.tests.unit import fake_volume
from cinder.volume import configuration
from cinder.volume.drivers.vmware import datastore as hub
from cinder.volume.drivers.vmware import fcd
from cinder.volume.drivers.vmware import vmdk
from cinder.volume.drivers.vmware import volumeops
@ddt.ddt
class VMwareVStorageObjectDriverTestCase(test.TestCase):
IP = 'localhost'
PORT = 2321
IMG_TX_TIMEOUT = 10
VMDK_DRIVER = vmdk.VMwareVcVmdkDriver
FCD_DRIVER = fcd.VMwareVStorageObjectDriver
VOL_ID = 'abcdefab-cdef-abcd-efab-cdefabcdefab'
DISPLAY_NAME = 'foo'
VOL_TYPE_ID = 'd61b8cb3-aa1b-4c9b-b79e-abcdbda8b58a'
VOL_SIZE = 2
PROJECT_ID = 'd45beabe-f5de-47b7-b462-0d9ea02889bc'
IMAGE_ID = 'eb87f4b0-d625-47f8-bb45-71c43b486d3a'
IMAGE_NAME = 'image-1'
def setUp(self):
super(VMwareVStorageObjectDriverTestCase, self).setUp()
self._config = mock.Mock(spec=configuration.Configuration)
self._config.vmware_host_ip = self.IP
self._config.vmware_host_port = self.PORT
self._config.vmware_image_transfer_timeout_secs = self.IMG_TX_TIMEOUT
self._driver = fcd.VMwareVStorageObjectDriver(
configuration=self._config)
self._context = context.get_admin_context()
@mock.patch.object(VMDK_DRIVER, 'do_setup')
@mock.patch.object(FCD_DRIVER, 'volumeops')
def test_do_setup(self, vops, vmdk_do_setup):
self._driver.do_setup(self._context)
vmdk_do_setup.assert_called_once_with(self._context)
self.assertFalse(self._driver._storage_policy_enabled)
vops.set_vmx_version.assert_called_once_with('vmx-13')
def test_get_volume_stats(self):
stats = self._driver.get_volume_stats()
self.assertEqual('VMware', stats['vendor_name'])
self.assertEqual(self._driver.VERSION, stats['driver_version'])
self.assertEqual(self._driver.STORAGE_TYPE, stats['storage_protocol'])
self.assertEqual(0, stats['reserved_percentage'])
self.assertEqual('unknown', stats['total_capacity_gb'])
self.assertEqual('unknown', stats['free_capacity_gb'])
def _create_volume_dict(self,
vol_id=VOL_ID,
display_name=DISPLAY_NAME,
volume_type_id=VOL_TYPE_ID,
status='available',
size=VOL_SIZE,
attachment=None,
project_id=PROJECT_ID):
return {'id': vol_id,
'display_name': display_name,
'name': 'volume-%s' % vol_id,
'volume_type_id': volume_type_id,
'status': status,
'size': size,
'volume_attachment': attachment,
'project_id': project_id,
}
def _create_volume_obj(self,
vol_id=VOL_ID,
display_name=DISPLAY_NAME,
volume_type_id=VOL_TYPE_ID,
status='available',
size=VOL_SIZE,
attachment=None,
project_id=PROJECT_ID):
vol = self._create_volume_dict(
vol_id, display_name, volume_type_id, status, size, attachment,
project_id)
return fake_volume.fake_volume_obj(self._context, **vol)
@mock.patch.object(FCD_DRIVER, '_select_datastore')
def test_select_ds_fcd(self, select_datastore):
datastore = mock.sentinel.datastore
summary = mock.Mock(datastore=datastore)
select_datastore.return_value = (mock.ANY, mock.ANY, summary)
volume = self._create_volume_obj()
ret = self._driver._select_ds_fcd(volume)
self.assertEqual(datastore, ret)
exp_req = {hub.DatastoreSelector.SIZE_BYTES: volume.size * units.Gi}
select_datastore.assert_called_once_with(exp_req)
@mock.patch.object(FCD_DRIVER, '_select_datastore')
@mock.patch.object(FCD_DRIVER, 'volumeops')
def _test_get_temp_image_folder(
self, vops, select_datastore, preallocated=False):
host = mock.sentinel.host
summary = mock.Mock()
summary.name = 'ds-1'
select_datastore.return_value = (host, mock.ANY, summary)
dc_ref = mock.sentinel.dc_ref
vops.get_dc.return_value = dc_ref
size_bytes = units.Gi
ret = self._driver._get_temp_image_folder(size_bytes, preallocated)
self.assertEqual(
(dc_ref, summary, vmdk.TMP_IMAGES_DATASTORE_FOLDER_PATH), ret)
exp_req = {hub.DatastoreSelector.SIZE_BYTES: size_bytes}
if preallocated:
exp_req[hub.DatastoreSelector.HARD_AFFINITY_DS_TYPE] = (
{hub.DatastoreType.NFS, hub.DatastoreType.VMFS})
select_datastore.assert_called_once_with(exp_req)
vops.get_dc.assert_called_once_with(host)
vops.create_datastore_folder.assert_called_once_with(
summary.name, vmdk.TMP_IMAGES_DATASTORE_FOLDER_PATH, dc_ref)
def test_get_temp_image_folder(self):
self._test_get_temp_image_folder()
def test_get_temp_image_folder_preallocated(self):
self._test_get_temp_image_folder(preallocated=True)
@mock.patch.object(VMDK_DRIVER, '_get_disk_type')
@ddt.data(('eagerZeroedThick', 'eagerZeroedThick'),
('thick', 'preallocated'),
('thin', 'thin'))
@ddt.unpack
def test_get_disk_type(
self, extra_spec_disk_type, exp_ret_val, vmdk_get_disk_type):
vmdk_get_disk_type.return_value = extra_spec_disk_type
volume = mock.sentinel.volume
ret = self._driver._get_disk_type(volume)
self.assertEqual(exp_ret_val, ret)
@mock.patch.object(FCD_DRIVER, '_select_ds_fcd')
@mock.patch.object(FCD_DRIVER, '_get_disk_type')
@mock.patch.object(FCD_DRIVER, 'volumeops')
def test_create_volume(self, vops, get_disk_type, select_ds_fcd):
ds_ref = mock.sentinel.ds_ref
select_ds_fcd.return_value = ds_ref
disk_type = mock.sentinel.disk_type
get_disk_type.return_value = disk_type
fcd_loc = mock.Mock()
provider_loc = mock.sentinel.provider_loc
fcd_loc.provider_location.return_value = provider_loc
vops.create_fcd.return_value = fcd_loc
volume = self._create_volume_obj()
ret = self._driver.create_volume(volume)
self.assertEqual({'provider_location': provider_loc}, ret)
select_ds_fcd.assert_called_once_with(volume)
get_disk_type.assert_called_once_with(volume)
vops.create_fcd.assert_called_once_with(
volume.name, volume.size * units.Ki, ds_ref, disk_type)
@mock.patch.object(volumeops.FcdLocation, 'from_provider_location')
@mock.patch.object(FCD_DRIVER, 'volumeops')
def test_delete_fcd(self, vops, from_provider_loc):
fcd_loc = mock.sentinel.fcd_loc
from_provider_loc.return_value = fcd_loc
provider_loc = mock.sentinel.provider_loc
self._driver._delete_fcd(provider_loc)
from_provider_loc.test_assert_called_once_with(provider_loc)
vops.delete_fcd.assert_called_once_with(fcd_loc)
@mock.patch.object(FCD_DRIVER, '_delete_fcd')
def test_delete_volume(self, delete_fcd):
volume = self._create_volume_obj()
self._driver.delete_volume(volume)
delete_fcd.assert_called_once_with(volume.provider_location)
@mock.patch.object(volumeops.FcdLocation, 'from_provider_location')
@mock.patch.object(FCD_DRIVER, '_get_adapter_type')
def test_initialize_connection(
self, get_adapter_type, from_provider_location):
fcd_loc = mock.Mock(
fcd_id=mock.sentinel.fcd_id, ds_ref_val=mock.sentinel.ds_ref_val)
from_provider_location.return_value = fcd_loc
adapter_type = mock.sentinel.adapter_type
get_adapter_type.return_value = adapter_type
volume = self._create_volume_obj()
connector = mock.sentinel.connector
ret = self._driver.initialize_connection(volume, connector)
self.assertEqual(self._driver.STORAGE_TYPE, ret['driver_volume_type'])
self.assertEqual(fcd_loc.fcd_id, ret['data']['id'])
self.assertEqual(fcd_loc.ds_ref_val, ret['data']['ds_ref_val'])
self.assertEqual(adapter_type, ret['data']['adapter_type'])
def test_container_format(self):
self._driver._validate_container_format('bare', mock.sentinel.image_id)
def test_container_format_invalid(self):
self.assertRaises(cinder_exceptions.ImageUnacceptable,
self._driver._validate_container_format,
'ova',
mock.sentinel.image_id)
def _create_image_meta(self,
_id=IMAGE_ID,
name=IMAGE_NAME,
disk_format='vmdk',
size=1 * units.Gi,
container_format='bare',
vmware_disktype='streamOptimized',
vmware_adaptertype='lsiLogic',
is_public=True):
return {'id': _id,
'name': name,
'disk_format': disk_format,
'size': size,
'container_format': container_format,
'properties': {'vmware_disktype': vmware_disktype,
'vmware_adaptertype': vmware_adaptertype,
},
'is_public': is_public,
}
@mock.patch.object(FCD_DRIVER, '_get_temp_image_folder')
@mock.patch.object(FCD_DRIVER, '_create_virtual_disk_from_sparse_image')
@mock.patch.object(FCD_DRIVER,
'_create_virtual_disk_from_preallocated_image')
@mock.patch.object(FCD_DRIVER, 'volumeops')
@mock.patch.object(datastore, 'DatastoreURL')
@ddt.data(vmdk.ImageDiskType.PREALLOCATED, vmdk.ImageDiskType.SPARSE,
vmdk.ImageDiskType.STREAM_OPTIMIZED)
def test_copy_image_to_volume(self,
disk_type,
datastore_url_cls,
vops,
create_disk_from_preallocated_image,
create_disk_from_sparse_image,
get_temp_image_folder):
image_meta = self._create_image_meta(vmware_disktype=disk_type)
image_service = mock.Mock()
image_service.show.return_value = image_meta
dc_ref = mock.sentinel.dc_ref
datastore = mock.sentinel.datastore
summary = mock.Mock(datastore=datastore)
summary.name = 'ds1'
folder_path = mock.sentinel.folder_path
get_temp_image_folder.return_value = (dc_ref, summary, folder_path)
vmdk_path = mock.Mock()
vmdk_path.get_descriptor_ds_file_path.return_value = (
"[ds1] cinder_vol/foo.vmdk")
if disk_type == vmdk.ImageDiskType.PREALLOCATED:
create_disk_from_preallocated_image.return_value = vmdk_path
else:
create_disk_from_sparse_image.return_value = vmdk_path
dc_path = '/test-dc'
vops.get_inventory_path.return_value = dc_path
ds_url = mock.sentinel.ds_url
datastore_url_cls.return_value = ds_url
fcd_loc = mock.Mock()
provider_location = mock.sentinel.provider_location
fcd_loc.provider_location.return_value = provider_location
vops.register_disk.return_value = fcd_loc
volume = self._create_volume_obj()
image_id = self.IMAGE_ID
ret = self._driver.copy_image_to_volume(
self._context, volume, image_service, image_id)
self.assertEqual({'provider_location': provider_location}, ret)
get_temp_image_folder.assert_called_once_with(volume.size * units.Gi)
if disk_type == vmdk.ImageDiskType.PREALLOCATED:
create_disk_from_preallocated_image.assert_called_once_with(
self._context, image_service, image_id, image_meta['size'],
dc_ref, summary.name, folder_path, volume.id,
volumeops.VirtualDiskAdapterType.LSI_LOGIC)
else:
create_disk_from_sparse_image.assert_called_once_with(
self._context, image_service, image_id, image_meta['size'],
dc_ref, summary.name, folder_path, volume.id)
datastore_url_cls.assert_called_once_with(
'https', self._driver.configuration.vmware_host_ip,
'cinder_vol/foo.vmdk', '/test-dc', 'ds1')
vops.register_disk.assert_called_once_with(
str(ds_url),
volume.name,
summary.datastore)
@mock.patch.object(volumeops.FcdLocation, 'from_provider_location')
@mock.patch.object(FCD_DRIVER, 'volumeops')
@mock.patch.object(vim_util, 'get_moref')
@mock.patch.object(FCD_DRIVER, '_create_backing')
@mock.patch.object(image_transfer, 'upload_image')
@mock.patch.object(VMDK_DRIVER, 'session')
@mock.patch.object(FCD_DRIVER, '_delete_temp_backing')
def test_copy_volume_to_image(
self, delete_temp_backing, session, upload_image, create_backing,
get_moref, vops, from_provider_loc):
fcd_loc = mock.Mock()
ds_ref = mock.sentinel.ds_ref
fcd_loc.ds_ref.return_value = ds_ref
from_provider_loc.return_value = fcd_loc
host_ref_val = mock.sentinel.host_ref_val
vops.get_connected_hosts.return_value = [host_ref_val]
host = mock.sentinel.host
get_moref.return_value = host
backing = mock.sentinel.backing
create_backing.return_value = backing
vmdk_file_path = mock.sentinel.vmdk_file_path
vops.get_vmdk_path.return_value = vmdk_file_path
vops.get_backing_by_uuid.return_value = backing
volume = self._create_volume_obj()
image_service = mock.sentinel.image_service
image_meta = self._create_image_meta()
self._driver.copy_volume_to_image(
self._context, volume, image_service, image_meta)
from_provider_loc.assert_called_once_with(volume.provider_location)
vops.get_connected_hosts.assert_called_once_with(ds_ref)
create_backing.assert_called_once_with(
volume, host, {vmdk.CREATE_PARAM_DISK_LESS: True})
vops.attach_fcd.assert_called_once_with(backing, fcd_loc)
vops.get_vmdk_path.assert_called_once_with(backing)
conf = self._driver.configuration
upload_image.assert_called_once_with(
self._context,
conf.vmware_image_transfer_timeout_secs,
image_service,
image_meta['id'],
volume.project_id,
session=session,
host=conf.vmware_host_ip,
port=conf.vmware_host_port,
vm=backing,
vmdk_file_path=vmdk_file_path,
vmdk_size=volume.size * units.Gi,
image_name=image_meta['name'])
vops.detach_fcd.assert_called_once_with(backing, fcd_loc)
delete_temp_backing.assert_called_once_with(backing)
@mock.patch.object(volumeops.FcdLocation, 'from_provider_location')
@mock.patch.object(FCD_DRIVER, 'volumeops')
def test_extend_volume(self, vops, from_provider_loc):
fcd_loc = mock.sentinel.fcd_loc
from_provider_loc.return_value = fcd_loc
volume = self._create_volume_obj()
new_size = 3
self._driver.extend_volume(volume, new_size)
from_provider_loc.assert_called_once_with(volume.provider_location)
vops.extend_fcd.assert_called_once_with(
fcd_loc, new_size * units.Ki)
@mock.patch.object(volumeops.FcdLocation, 'from_provider_location')
@mock.patch.object(FCD_DRIVER, 'volumeops')
def test_clone_fcd(self, vops, from_provider_loc):
fcd_loc = mock.sentinel.fcd_loc
from_provider_loc.return_value = fcd_loc
dest_fcd_loc = mock.sentinel.dest_fcd_loc
vops.clone_fcd.return_value = dest_fcd_loc
provider_loc = mock.sentinel.provider_loc
name = mock.sentinel.name
dest_ds_ref = mock.sentinel.dest_ds_ref
disk_type = mock.sentinel.disk_type
ret = self._driver._clone_fcd(
provider_loc, name, dest_ds_ref, disk_type)
self.assertEqual(dest_fcd_loc, ret)
from_provider_loc.assert_called_once_with(provider_loc)
vops.clone_fcd.assert_called_once_with(
name, fcd_loc, dest_ds_ref, disk_type)
@mock.patch.object(FCD_DRIVER, '_select_ds_fcd')
@mock.patch.object(FCD_DRIVER, '_clone_fcd')
def test_create_snapshot(self, clone_fcd, select_ds_fcd):
ds_ref = mock.sentinel.ds_ref
select_ds_fcd.return_value = ds_ref
dest_fcd_loc = mock.Mock()
provider_location = mock.sentinel.provider_location
dest_fcd_loc.provider_location.return_value = provider_location
clone_fcd.return_value = dest_fcd_loc
volume = self._create_volume_obj()
snapshot = fake_snapshot.fake_snapshot_obj(
self._context, volume=volume)
ret = self._driver.create_snapshot(snapshot)
self.assertEqual({'provider_location': provider_location}, ret)
select_ds_fcd.assert_called_once_with(snapshot.volume)
clone_fcd.assert_called_once_with(
volume.provider_location, snapshot.name, ds_ref)
@mock.patch.object(FCD_DRIVER, '_delete_fcd')
def test_delete_snapshot(self, delete_fcd):
volume = self._create_volume_obj()
snapshot = fake_snapshot.fake_snapshot_obj(
self._context, volume=volume)
self._driver.delete_snapshot(snapshot)
delete_fcd.assert_called_once_with(snapshot.provider_location)
@mock.patch.object(FCD_DRIVER, 'volumeops')
@ddt.data((1, 1), (1, 2))
@ddt.unpack
def test_extend_if_needed(self, cur_size, new_size, vops):
fcd_loc = mock.sentinel.fcd_loc
self._driver._extend_if_needed(fcd_loc, cur_size, new_size)
if new_size > cur_size:
vops.extend_fcd.assert_called_once_with(
fcd_loc, new_size * units.Ki)
else:
vops.extend_fcd.assert_not_called()
@mock.patch.object(FCD_DRIVER, '_select_ds_fcd')
@mock.patch.object(FCD_DRIVER, '_get_disk_type')
@mock.patch.object(FCD_DRIVER, '_clone_fcd')
@mock.patch.object(FCD_DRIVER, '_extend_if_needed')
def test_create_volume_from_fcd(
self, extend_if_needed, clone_fcd, get_disk_type, select_ds_fcd):
ds_ref = mock.sentinel.ds_ref
select_ds_fcd.return_value = ds_ref
disk_type = mock.sentinel.disk_type
get_disk_type.return_value = disk_type
cloned_fcd_loc = mock.Mock()
dest_provider_loc = mock.sentinel.dest_provider_loc
cloned_fcd_loc.provider_location.return_value = dest_provider_loc
clone_fcd.return_value = cloned_fcd_loc
provider_loc = mock.sentinel.provider_loc
cur_size = 1
volume = self._create_volume_obj()
ret = self._driver._create_volume_from_fcd(
provider_loc, cur_size, volume)
self.assertEqual({'provider_location': dest_provider_loc}, ret)
select_ds_fcd.test_assert_called_once_with(volume)
get_disk_type.test_assert_called_once_with(volume)
clone_fcd.assert_called_once_with(
provider_loc, volume.name, ds_ref, disk_type=disk_type)
extend_if_needed.assert_called_once_with(
cloned_fcd_loc, cur_size, volume.size)
@mock.patch.object(FCD_DRIVER, '_create_volume_from_fcd')
def test_create_volume_from_snapshot(self, create_volume_from_fcd):
src_volume = self._create_volume_obj()
snapshot = fake_snapshot.fake_snapshot_obj(
self._context, volume=src_volume)
volume = mock.sentinel.volume
self._driver.create_volume_from_snapshot(volume, snapshot)
create_volume_from_fcd.assert_called_once_with(
snapshot.provider_location, snapshot.volume.size, volume)
@mock.patch.object(FCD_DRIVER, '_create_volume_from_fcd')
def test_create_cloned_volume(self, create_volume_from_fcd):
src_volume = self._create_volume_obj()
volume = mock.sentinel.volume
self._driver.create_cloned_volume(volume, src_volume)
create_volume_from_fcd.assert_called_once_with(
src_volume.provider_location, src_volume.size, volume)

257
cinder/tests/unit/volume/drivers/vmware/test_vmware_volumeops.py
View File

@ -1843,6 +1843,222 @@ class VolumeOpsTestCase(test.TestCase):
destinationDatacenter=mock.sentinel.dest_dc_ref)
self.session.wait_for_task.assert_called_once_with(mock.sentinel.task)
@ddt.data(volumeops.VirtualDiskType.EAGER_ZEROED_THICK,
volumeops.VirtualDiskType.PREALLOCATED,
volumeops.VirtualDiskType.THIN)
def test_create_fcd_backing_spec(self, disk_type):
spec = mock.Mock()
self.session.vim.client.factory.create.return_value = spec
ds_ref = mock.sentinel.ds_ref
ret = self.vops._create_fcd_backing_spec(disk_type, ds_ref)
if disk_type == volumeops.VirtualDiskType.PREALLOCATED:
prov_type = 'lazyZeroedThick'
else:
prov_type = disk_type
self.assertEqual(prov_type, ret.provisioningType)
self.assertEqual(ds_ref, ret.datastore)
self.session.vim.client.factory.create.assert_called_once_with(
'ns0:VslmCreateSpecDiskFileBackingSpec')
@mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
'_create_fcd_backing_spec')
def test_create_fcd(self, create_fcd_backing_spec):
spec = mock.Mock()
self.session.vim.client.factory.create.return_value = spec
backing_spec = mock.sentinel.backing_spec
create_fcd_backing_spec.return_value = backing_spec
task = mock.sentinel.task
self.session.invoke_api.return_value = task
task_info = mock.Mock()
fcd_id = mock.sentinel.fcd_id
task_info.result.config.id.id = fcd_id
self.session.wait_for_task.return_value = task_info
name = mock.sentinel.name
size_mb = 1024
ds_ref_val = mock.sentinel.ds_ref_val
ds_ref = mock.Mock(value=ds_ref_val)
disk_type = mock.sentinel.disk_type
ret = self.vops.create_fcd(name, size_mb, ds_ref, disk_type)
self.assertEqual(fcd_id, ret.fcd_id)
self.assertEqual(ds_ref_val, ret.ds_ref_val)
self.session.vim.client.factory.create.assert_called_once_with(
'ns0:VslmCreateSpec')
create_fcd_backing_spec.assert_called_once_with(disk_type, ds_ref)
self.assertEqual(1024, spec.capacityInMB)
self.assertEqual(name, spec.name)
self.assertEqual(backing_spec, spec.backingSpec)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'CreateDisk_Task',
self.session.vim.service_content.vStorageObjectManager,
spec=spec)
self.session.wait_for_task.assert_called_once_with(task)
def test_delete_fcd(self):
task = mock.sentinel.task
self.session.invoke_api.return_value = task
fcd_location = mock.Mock()
fcd_id = mock.sentinel.fcd_id
fcd_location.id.return_value = fcd_id
ds_ref = mock.sentinel.ds_ref
fcd_location.ds_ref.return_value = ds_ref
self.vops.delete_fcd(fcd_location)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'DeleteVStorageObject_Task',
self.session.vim.service_content.vStorageObjectManager,
id=fcd_id,
datastore=ds_ref)
self.session.wait_for_task(task)
@mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
'_create_fcd_backing_spec')
def test_clone_fcd(self, create_fcd_backing_spec):
spec = mock.Mock()
self.session.vim.client.factory.create.return_value = spec
backing_spec = mock.sentinel.backing_spec
create_fcd_backing_spec.return_value = backing_spec
task = mock.sentinel.task
self.session.invoke_api.return_value = task
task_info = mock.Mock()
fcd_id = mock.sentinel.fcd_id
task_info.result.config.id.id = fcd_id
self.session.wait_for_task.return_value = task_info
fcd_location = mock.Mock()
fcd_id = mock.sentinel.fcd_id
fcd_location.id.return_value = fcd_id
ds_ref = mock.sentinel.ds_ref
fcd_location.ds_ref.return_value = ds_ref
name = mock.sentinel.name
dest_ds_ref_val = mock.sentinel.dest_ds_ref_val
dest_ds_ref = mock.Mock(value=dest_ds_ref_val)
disk_type = mock.sentinel.disk_type
ret = self.vops.clone_fcd(name, fcd_location, dest_ds_ref, disk_type)
self.assertEqual(fcd_id, ret.fcd_id)
self.assertEqual(dest_ds_ref_val, ret.ds_ref_val)
self.session.vim.client.factory.create.assert_called_once_with(
'ns0:VslmCloneSpec')
create_fcd_backing_spec.assert_called_once_with(disk_type, dest_ds_ref)
self.assertEqual(name, spec.name)
self.assertEqual(backing_spec, spec.backingSpec)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'CloneVStorageObject_Task',
self.session.vim.service_content.vStorageObjectManager,
id=fcd_id,
datastore=ds_ref,
spec=spec)
self.session.wait_for_task.assert_called_once_with(task)
def test_extend_fcd(self):
task = mock.sentinel.task
self.session.invoke_api.return_value = task
fcd_location = mock.Mock()
fcd_id = mock.sentinel.fcd_id
fcd_location.id.return_value = fcd_id
ds_ref = mock.sentinel.ds_ref
fcd_location.ds_ref.return_value = ds_ref
new_size_mb = 1024
self.vops.extend_fcd(fcd_location, new_size_mb)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'ExtendDisk_Task',
self.session.vim.service_content.vStorageObjectManager,
id=fcd_id,
datastore=ds_ref,
newCapacityInMB=new_size_mb)
self.session.wait_for_task(task)
def test_register_disk(self):
fcd = mock.Mock()
fcd_id = mock.sentinel.fcd_id
fcd.config.id = mock.Mock(id=fcd_id)
self.session.invoke_api.return_value = fcd
vmdk_url = mock.sentinel.vmdk_url
name = mock.sentinel.name
ds_ref_val = mock.sentinel.ds_ref_val
ds_ref = mock.Mock(value=ds_ref_val)
ret = self.vops.register_disk(vmdk_url, name, ds_ref)
self.assertEqual(fcd_id, ret.fcd_id)
self.assertEqual(ds_ref_val, ret.ds_ref_val)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'RegisterDisk',
self.session.vim.service_content.vStorageObjectManager,
path=vmdk_url,
name=name)
@mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
'_create_controller_config_spec')
@mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
'_reconfigure_backing')
def test_attach_fcd(self, reconfigure_backing, create_controller_spec):
reconfig_spec = mock.Mock()
self.session.vim.client.factory.create.return_value = reconfig_spec
spec = mock.Mock()
create_controller_spec.return_value = spec
task = mock.sentinel.task
self.session.invoke_api.return_value = task
backing = mock.sentinel.backing
fcd_location = mock.Mock()
fcd_id = mock.sentinel.fcd_id
fcd_location.id.return_value = fcd_id
ds_ref = mock.sentinel.ds_ref
fcd_location.ds_ref.return_value = ds_ref
self.vops.attach_fcd(backing, fcd_location)
self.session.vim.client.factory.create.assert_called_once_with(
'ns0:VirtualMachineConfigSpec')
create_controller_spec.assert_called_once_with(
volumeops.VirtualDiskAdapterType.LSI_LOGIC)
self.assertEqual([spec], reconfig_spec.deviceChange)
reconfigure_backing.assert_called_once_with(backing, reconfig_spec)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'AttachDisk_Task',
backing,
diskId=fcd_id,
datastore=ds_ref)
self.session.wait_for_task.assert_called_once_with(task)
def test_detach_fcd(self):
task = mock.sentinel.task
self.session.invoke_api.return_value = task
backing = mock.sentinel.backing
fcd_location = mock.Mock()
fcd_id = mock.sentinel.fcd_id
fcd_location.id.return_value = fcd_id
self.vops.detach_fcd(backing, fcd_location)
self.session.invoke_api.assert_called_once_with(
self.session.vim,
'DetachDisk_Task',
backing,
diskId=fcd_id)
self.session.wait_for_task.assert_called_once_with(task)
class VirtualDiskPathTest(test.TestCase):
"""Unit tests for VirtualDiskPath."""
@ -1991,3 +2207,44 @@ class ControllerTypeTest(test.TestCase):
volumeops.ControllerType.PARA_VIRTUAL))
self.assertFalse(volumeops.ControllerType.is_scsi_controller(
volumeops.ControllerType.IDE))
class FcdLocationTest(test.TestCase):
"""Unit tests for FcdLocation."""
def test_create(self):
fcd_id = mock.sentinel.fcd_id
fcd_id_obj = mock.Mock(id=fcd_id)
ds_ref_val = mock.sentinel.ds_ref_val
ds_ref = mock.Mock(value=ds_ref_val)
fcd_loc = volumeops.FcdLocation.create(fcd_id_obj, ds_ref)
self.assertEqual(fcd_id, fcd_loc.fcd_id)
self.assertEqual(ds_ref_val, fcd_loc.ds_ref_val)
def test_provider_location(self):
fcd_loc = volumeops.FcdLocation('123', 'ds1')
self.assertEqual('123@ds1', fcd_loc.provider_location())
def test_ds_ref(self):
fcd_loc = volumeops.FcdLocation('123', 'ds1')
ds_ref = fcd_loc.ds_ref()
self.assertEqual('ds1', ds_ref.value)
def test_id(self):
id_obj = mock.Mock()
cf = mock.Mock()
cf.create.return_value = id_obj
fcd_loc = volumeops.FcdLocation('123', 'ds1')
fcd_id = fcd_loc.id(cf)
self.assertEqual('123', fcd_id.id)
cf.create.assert_called_once_with('ns0:ID')
def test_from_provider_location(self):
fcd_loc = volumeops.FcdLocation.from_provider_location('123@ds1')
self.assertEqual('123', fcd_loc.fcd_id)
self.assertEqual('ds1', fcd_loc.ds_ref_val)
def test_str(self):
fcd_loc = volumeops.FcdLocation('123', 'ds1')
self.assertEqual('123@ds1', str(fcd_loc))

313
cinder/volume/drivers/vmware/fcd.py Normal file
View File

@ -0,0 +1,313 @@
# Copyright (c) 2017 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.
"""
VMware VStorageObject driver
Volume driver based on VMware VStorageObject aka First Class Disk (FCD). This
driver requires a minimum vCenter version of 6.5.
"""
from oslo_log import log as logging
from oslo_utils import units
from oslo_vmware import image_transfer
from oslo_vmware.objects import datastore
from oslo_vmware import vim_util
from cinder import exception
from cinder.i18n import _
from cinder import interface
from cinder.volume.drivers.vmware import datastore as hub
from cinder.volume.drivers.vmware import vmdk
from cinder.volume.drivers.vmware import volumeops as vops
LOG = logging.getLogger(__name__)
@interface.volumedriver
class VMwareVStorageObjectDriver(vmdk.VMwareVcVmdkDriver):
"""Volume driver based on VMware VStorageObject"""
# 1.0 - initial version based on vSphere 6.5 vStorageObject APIs
VERSION = '1.0.0'
# ThirdPartySystems wiki page
CI_WIKI_NAME = "VMware_CI"
# minimum supported vCenter version
MIN_SUPPORTED_VC_VERSION = '6.5'
STORAGE_TYPE = 'vstorageobject'
def do_setup(self, context):
"""Any initialization the volume driver needs to do while starting.
:param context: The admin context.
"""
super(VMwareVStorageObjectDriver, self).do_setup(context)
self._storage_policy_enabled = False
self.volumeops.set_vmx_version('vmx-13')
def get_volume_stats(self, refresh=False):
"""Collects volume backend stats.
:param refresh: Whether to discard any cached values and force a full
refresh of stats.
:returns: dict of appropriate values.
"""
stats = super(VMwareVStorageObjectDriver, self).get_volume_stats(
refresh=refresh)
stats['storage_protocol'] = self.STORAGE_TYPE
return stats
def _select_ds_fcd(self, volume):
req = {}
req[hub.DatastoreSelector.SIZE_BYTES] = volume.size * units.Gi
(_host_ref, _resource_pool, summary) = self._select_datastore(req)
return summary.datastore
def _get_temp_image_folder(self, size_bytes, preallocated=False):
req = {}
req[hub.DatastoreSelector.SIZE_BYTES] = size_bytes
if preallocated:
req[hub.DatastoreSelector.HARD_AFFINITY_DS_TYPE] = (
hub.DatastoreType.get_all_types() -
{hub.DatastoreType.VSAN, hub.DatastoreType.VVOL})
(host_ref, _resource_pool, summary) = self._select_datastore(req)
folder_path = vmdk.TMP_IMAGES_DATASTORE_FOLDER_PATH
dc_ref = self.volumeops.get_dc(host_ref)
self.volumeops.create_datastore_folder(
summary.name, folder_path, dc_ref)
return (dc_ref, summary, folder_path)
def _get_disk_type(self, volume):
extra_spec_disk_type = super(
VMwareVStorageObjectDriver, self)._get_disk_type(volume)
return vops.VirtualDiskType.get_virtual_disk_type(extra_spec_disk_type)
def create_volume(self, volume):
"""Create a new volume on the backend.
:param volume: Volume object containing specifics to create.
:returns: (Optional) dict of database updates for the new volume.
"""
disk_type = self._get_disk_type(volume)
ds_ref = self._select_ds_fcd(volume)
fcd_loc = self.volumeops.create_fcd(
volume.name, volume.size * units.Ki, ds_ref, disk_type)
return {'provider_location': fcd_loc.provider_location()}
def _delete_fcd(self, provider_loc):
fcd_loc = vops.FcdLocation.from_provider_location(provider_loc)
self.volumeops.delete_fcd(fcd_loc)
def delete_volume(self, volume):
"""Delete a volume from the backend.
:param volume: The volume to delete.
"""
self._delete_fcd(volume.provider_location)
def initialize_connection(self, volume, connector, initiator_data=None):
"""Allow connection to connector and return connection info.
:param volume: The volume to be attached.
:param connector: Dictionary containing information about what is being
connected to.
:param initiator_data: (Optional) A dictionary of driver_initiator_data
objects with key-value pairs that have been
saved for this initiator by a driver in previous
initialize_connection calls.
:returns: A dictionary of connection information.
"""
fcd_loc = vops.FcdLocation.from_provider_location(
volume.provider_location)
connection_info = {'driver_volume_type': self.STORAGE_TYPE}
connection_info['data'] = {
'id': fcd_loc.fcd_id,
'ds_ref_val': fcd_loc.ds_ref_val,
'adapter_type': self._get_adapter_type(volume)
}
LOG.debug("Connection info for volume %(name)s: %(connection_info)s.",
{'name': volume.name, 'connection_info': connection_info})
return connection_info
def _validate_container_format(self, container_format, image_id):
if container_format and container_format != 'bare':
msg = _("Container format: %s is unsupported, only 'bare' "
"is supported.") % container_format
LOG.error(msg)
raise exception.ImageUnacceptable(image_id=image_id, reason=msg)
def copy_image_to_volume(self, context, volume, image_service, image_id):
"""Fetch the image from image_service and write it to the volume.
:param context: Security/policy info for the request.
:param volume: The volume to create.
:param image_service: The image service to use.
:param image_id: The image identifier.
:returns: Model updates.
"""
metadata = image_service.show(context, image_id)
self._validate_disk_format(metadata['disk_format'])
self._validate_container_format(
metadata.get('container_format'), image_id)
properties = metadata['properties'] or {}
disk_type = properties.get('vmware_disktype',
vmdk.ImageDiskType.PREALLOCATED)
vmdk.ImageDiskType.validate(disk_type)
size_bytes = metadata['size']
dc_ref, summary, folder_path = self._get_temp_image_folder(
volume.size * units.Gi)
disk_name = volume.id
if disk_type in [vmdk.ImageDiskType.SPARSE,
vmdk.ImageDiskType.STREAM_OPTIMIZED]:
vmdk_path = self._create_virtual_disk_from_sparse_image(
context, image_service, image_id, size_bytes, dc_ref,
summary.name, folder_path, disk_name)
else:
vmdk_path = self._create_virtual_disk_from_preallocated_image(
context, image_service, image_id, size_bytes, dc_ref,
summary.name, folder_path, disk_name,
vops.VirtualDiskAdapterType.LSI_LOGIC)
ds_path = datastore.DatastorePath.parse(
vmdk_path.get_descriptor_ds_file_path())
dc_path = self.volumeops.get_inventory_path(dc_ref)
vmdk_url = datastore.DatastoreURL(
'https', self.configuration.vmware_host_ip, ds_path.rel_path,
dc_path, ds_path.datastore)
fcd_loc = self.volumeops.register_disk(
str(vmdk_url), volume.name, summary.datastore)
return {'provider_location': fcd_loc.provider_location()}
def copy_volume_to_image(self, context, volume, image_service, image_meta):
"""Copy the volume to the specified image.
:param context: Security/policy info for the request.
:param volume: The volume to copy.
:param image_service: The image service to use.
:param image_meta: Information about the image.
:returns: Model updates.
"""
self._validate_disk_format(image_meta['disk_format'])
fcd_loc = vops.FcdLocation.from_provider_location(
volume.provider_location)
hosts = self.volumeops.get_connected_hosts(fcd_loc.ds_ref())
host = vim_util.get_moref(hosts[0], 'HostSystem')
LOG.debug("Selected host: %(host)s for downloading fcd: %(fcd_loc)s.",
{'host': host, 'fcd_loc': fcd_loc})
attached = False
try:
create_params = {vmdk.CREATE_PARAM_DISK_LESS: True}
backing = self._create_backing(volume, host, create_params)
self.volumeops.attach_fcd(backing, fcd_loc)
attached = True
vmdk_file_path = self.volumeops.get_vmdk_path(backing)
conf = self.configuration
image_transfer.upload_image(
context,
conf.vmware_image_transfer_timeout_secs,
image_service,
image_meta['id'],
volume.project_id,
session=self.session,
host=conf.vmware_host_ip,
port=conf.vmware_host_port,
vm=backing,
vmdk_file_path=vmdk_file_path,
vmdk_size=volume.size * units.Gi,
image_name=image_meta['name'])
finally:
if attached:
self.volumeops.detach_fcd(backing, fcd_loc)
backing = self.volumeops.get_backing_by_uuid(volume.id)
if backing:
self._delete_temp_backing(backing)
def extend_volume(self, volume, new_size):
"""Extend the size of a volume.
:param volume: The volume to extend.
:param new_size: The new desired size of the volume.
"""
fcd_loc = vops.FcdLocation.from_provider_location(
volume.provider_location)
self.volumeops.extend_fcd(fcd_loc, new_size * units.Ki)
def _clone_fcd(self, provider_loc, name, dest_ds_ref,
disk_type=vops.VirtualDiskType.THIN):
fcd_loc = vops.FcdLocation.from_provider_location(provider_loc)
return self.volumeops.clone_fcd(name, fcd_loc, dest_ds_ref, disk_type)
def create_snapshot(self, snapshot):
"""Creates a snapshot.
:param snapshot: Information for the snapshot to be created.
"""
ds_ref = self._select_ds_fcd(snapshot.volume)
cloned_fcd_loc = self._clone_fcd(
snapshot.volume.provider_location, snapshot.name, ds_ref)
return {'provider_location': cloned_fcd_loc.provider_location()}
def delete_snapshot(self, snapshot):
"""Deletes a snapshot.
:param snapshot: The snapshot to delete.
"""
self._delete_fcd(snapshot.provider_location)
def _extend_if_needed(self, fcd_loc, cur_size, new_size):
if new_size > cur_size:
self.volumeops.extend_fcd(fcd_loc, new_size * units.Ki)
def _create_volume_from_fcd(self, provider_loc, cur_size, volume):
ds_ref = self._select_ds_fcd(volume)
disk_type = self._get_disk_type(volume)
cloned_fcd_loc = self._clone_fcd(
provider_loc, volume.name, ds_ref, disk_type=disk_type)
self._extend_if_needed(cloned_fcd_loc, cur_size, volume.size)
return {'provider_location': cloned_fcd_loc.provider_location()}
def create_volume_from_snapshot(self, volume, snapshot):
"""Creates a volume from a snapshot.
:param volume: The volume to be created.
:param snapshot: The snapshot from which to create the volume.
:returns: A dict of database updates for the new volume.
"""
return self._create_volume_from_fcd(
snapshot.provider_location, snapshot.volume.size, volume)
def create_cloned_volume(self, volume, src_vref):
"""Creates a clone of the specified volume.
:param volume: New Volume object
:param src_vref: Source Volume object
"""
return self._create_volume_from_fcd(
src_vref.provider_location, src_vref.size, volume)

154
cinder/volume/drivers/vmware/volumeops.py
View File

@ -286,6 +286,10 @@ class VMwareVolumeOps(object):
self._extension_type = extension_type
self._folder_cache = {}
self._backing_ref_cache = {}
self._vmx_version = None
def set_vmx_version(self, vmx_version):
self._vmx_version = vmx_version
def get_backing(self, name, backing_uuid):
"""Get the backing based on name or uuid.
@ -751,11 +755,11 @@ class VMwareVolumeOps(object):
create_spec.numCPUs = 1
create_spec.memoryMB = 128
create_spec.files = vm_file_info
# Set the hardware version to a compatible version supported by
# Set the default hardware version to a compatible version supported by
# vSphere 5.0. This will ensure that the backing VM can be migrated
# without any incompatibility issues in a mixed cluster of ESX hosts
# with versions 5.0 or above.
create_spec.version = "vmx-08"
create_spec.version = self._vmx_version or "vmx-08"
if profileId:
vmProfile = cf.create('ns0:VirtualMachineDefinedProfileSpec')
@ -1744,3 +1748,149 @@ class VMwareVolumeOps(object):
def mark_backing_as_template(self, backing):
LOG.debug("Marking backing: %s as template.", backing)
self._session.invoke_api(self._session.vim, 'MarkAsTemplate', backing)
def _create_fcd_backing_spec(self, disk_type, ds_ref):
backing_spec = self._session.vim.client.factory.create(
'ns0:VslmCreateSpecDiskFileBackingSpec')
if disk_type == VirtualDiskType.PREALLOCATED:
disk_type = 'lazyZeroedThick'
backing_spec.provisioningType = disk_type
backing_spec.datastore = ds_ref
return backing_spec
def create_fcd(self, name, size_mb, ds_ref, disk_type):
spec = self._session.vim.client.factory.create('ns0:VslmCreateSpec')
spec.capacityInMB = size_mb
spec.name = name
spec.backingSpec = self._create_fcd_backing_spec(disk_type, ds_ref)
LOG.debug("Creating fcd with spec: %(spec)s on datastore: %(ds_ref)s.",
{'spec': spec, 'ds_ref': ds_ref})
vstorage_mgr = self._session.vim.service_content.vStorageObjectManager
task = self._session.invoke_api(self._session.vim,
'CreateDisk_Task',
vstorage_mgr,
spec=spec)
task_info = self._session.wait_for_task(task)
fcd_loc = FcdLocation.create(task_info.result.config.id, ds_ref)
LOG.debug("Created fcd: %s.", fcd_loc)
return fcd_loc
def delete_fcd(self, fcd_location):
cf = self._session.vim.client.factory
vstorage_mgr = self._session.vim.service_content.vStorageObjectManager
LOG.debug("Deleting fcd: %s.", fcd_location)
task = self._session.invoke_api(self._session.vim,
'DeleteVStorageObject_Task',
vstorage_mgr,
id=fcd_location.id(cf),
datastore=fcd_location.ds_ref())
self._session.wait_for_task(task)
def clone_fcd(self, name, fcd_location, dest_ds_ref, disk_type):
cf = self._session.vim.client.factory
spec = cf.create('ns0:VslmCloneSpec')
spec.name = name
spec.backingSpec = self._create_fcd_backing_spec(disk_type,
dest_ds_ref)
LOG.debug("Copying fcd: %(fcd_loc)s to datastore: %(ds_ref)s with "
"spec: %(spec)s.",
{'fcd_loc': fcd_location,
'spec': spec,
'ds_ref': dest_ds_ref})
vstorage_mgr = self._session.vim.service_content.vStorageObjectManager
task = self._session.invoke_api(self._session.vim,
'CloneVStorageObject_Task',
vstorage_mgr,
id=fcd_location.id(cf),
datastore=fcd_location.ds_ref(),
spec=spec)
task_info = self._session.wait_for_task(task)
dest_fcd_loc = FcdLocation.create(task_info.result.config.id,
dest_ds_ref)
LOG.debug("Clone fcd: %s.", dest_fcd_loc)
return dest_fcd_loc
def extend_fcd(self, fcd_location, new_size_mb):
cf = self._session.vim.client.factory
vstorage_mgr = self._session.vim.service_content.vStorageObjectManager
LOG.debug("Extending fcd: %(fcd_loc)s to %(size)s.",
{'fcd_loc': fcd_location, 'size': new_size_mb})
task = self._session.invoke_api(self._session.vim,
'ExtendDisk_Task',
vstorage_mgr,
id=fcd_location.id(cf),
datastore=fcd_location.ds_ref(),
newCapacityInMB=new_size_mb)
self._session.wait_for_task(task)
def register_disk(self, vmdk_url, name, ds_ref):
vstorage_mgr = self._session.vim.service_content.vStorageObjectManager
LOG.debug("Registering disk: %s as fcd.", vmdk_url)
fcd = self._session.invoke_api(self._session.vim,
'RegisterDisk',
vstorage_mgr,
path=vmdk_url,
name=name)
fcd_loc = FcdLocation.create(fcd.config.id, ds_ref)
LOG.debug("Created fcd: %s.", fcd_loc)
return fcd_loc
def attach_fcd(self, backing, fcd_location):
cf = self._session.vim.client.factory
reconfig_spec = cf.create('ns0:VirtualMachineConfigSpec')
spec = self._create_controller_config_spec(
VirtualDiskAdapterType.LSI_LOGIC)
reconfig_spec.deviceChange = [spec]
self._reconfigure_backing(backing, reconfig_spec)
LOG.debug("Attaching fcd: %(fcd_loc)s to %(backing)s.",
{'fcd_loc': fcd_location, 'backing': backing})
task = self._session.invoke_api(self._session.vim,
"AttachDisk_Task",
backing,
diskId=fcd_location.id(cf),
datastore=fcd_location.ds_ref())
self._session.wait_for_task(task)
def detach_fcd(self, backing, fcd_location):
cf = self._session.vim.client.factory
LOG.debug("Detaching fcd: %(fcd_loc)s from %(backing)s.",
{'fcd_loc': fcd_location, 'backing': backing})
task = self._session.invoke_api(self._session.vim,
"DetachDisk_Task",
backing,
diskId=fcd_location.id(cf))
self._session.wait_for_task(task)
class FcdLocation(object):
def __init__(self, fcd_id, ds_ref_val):
self.fcd_id = fcd_id
self.ds_ref_val = ds_ref_val
@classmethod
def create(cls, fcd_id_obj, ds_ref):
return cls(fcd_id_obj.id, ds_ref.value)
def provider_location(self):
return "%s@%s" % (self.fcd_id, self.ds_ref_val)
def ds_ref(self):
return vim_util.get_moref(self.ds_ref_val, 'Datastore')
def id(self, cf):
id_obj = cf.create('ns0:ID')
id_obj.id = self.fcd_id
return id_obj
@classmethod
def from_provider_location(cls, provider_location):
fcd_id, ds_ref_val = provider_location.split('@')
return cls(fcd_id, ds_ref_val)
def __str__(self):
return self.provider_location()

4
releasenotes/notes/bp-vmware-fcd-fbe19ee577d2e9e4.yaml Normal file
View File

@ -0,0 +1,4 @@
---
features:
- |
Added backend driver for VMware VStorageObject (First Class Disk).