KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use retype_volume method in tempest

Best Python code snippet using tempest_python

migration_planner.py

Source:migration_planner.py

...14    print("openstack server add volume %s %s" % (srvid, volid))15def remove_snapshot(snapid):16    print("openstack snapshot delete %s" % snapid)17    18def retype_volume(volid, newtype):19    print("cinder retype --migration-policy on-demand %s %s" % (volid, newtype))20    info("Monitor migration process using: openstack volume show %s | grep migration_status" % volid)21def patch_volume_boot_index(vol, enable):22    if enable:23        warn("patching boot volume info for volume %s" % vol)24        query = "update block_device_mapping set boot_index=999 where deleted=0 and volume_id='%s' and boot_index=0"25    else:26        query = "update block_device_mapping set boot_index=0 where deleted=0 and volume_id='%s' and boot_index is NULL"27    print("echo \""+(query % vol)+"\" | sudo mysql nova")28def instance_power(srv, power):29    pcmd = 'start' if power else 'stop'30    if not power:31        warn("instance %s is running: shutting down" % srv)32    print("openstack server %s %s" % (pcmd, srv))33loader = loading.get_plugin_loader('password')34auth = loader.load_from_options(35    auth_url             = os.getenv('OS_AUTH_URL'),36    username             = os.getenv('OS_USERNAME'),37    user_domain_name     = os.getenv('OS_USER_DOMAIN_NAME'),38    user_id              = os.getenv('OS_USER_ID'),39    project_name         = os.getenv('OS_PROJECT_NAME'),40    project_domain_name  = os.getenv('OS_PROJECT_DOMAIN_NAME'),41    project_id           = os.getenv('OS_PROJECT_ID'),42    password             = os.getenv('OS_PASSWORD'),43)44sess = session.Session(auth=auth, verify=os.getenv('OS_CACERT', True))45cinder = c_client.Client(os.getenv('OS_VOLUME_API_VERSION', '2'),46                         session=sess,47                         endpoint_type=os.getenv('OS_ENDPOINT_TYPE', 'public'))48nova = n_client.Client(os.getenv('OS_COMPUTE_API_VERSION', '2'),49                       session=sess,50                       endpoint_type=os.getenv('OS_ENDPOINT_TYPE', 'public'))51NO_ROLLING_FLAG = '--no-rolling'52if len(sys.argv) < 2:53    print("Syntax: %s [%s] <fromtype>=<totype>..." % (sys.argv[0], NO_ROLLING_FLAG))54    sys.exit(0)55no_rolling_migration = False56restart_instances = []57available_volume_types = set([t.name for t in cinder.volume_types.list()])58voltype_map = {}59for arg in sys.argv[1:]:60    if arg == NO_ROLLING_FLAG:61        no_rolling_migration = True62    elif '=' in arg:63        vtfrom, vtto = (x.strip() for x in arg.split('=', 1))64        for t in (vtfrom, vtto):65            if t not in available_volume_types:66                warn("unknown volume type: %s" % t)67                continue68        voltype_map[vtfrom] = vtto69backlog = {}70srvs_with_volumes = set()71info("Migration plan for the following volume types:")72for f, t in sorted(voltype_map.items()):73    info("%s -> %s" % (f.rjust(15), t))74snapshots = {x.id: x.volume_id75    for x in cinder.volume_snapshots.list(search_opts=dict(all_tenants=1))}76vols_with_snapshots = {77    x: set([k for k, v in snapshots.items() if v == x])78        for x in snapshots.values()79}80vol_count = vol_total_size = 081# First migrate all detached volumes82info("Migrating detached volumes")83for v in cinder.volumes.list(search_opts=dict(all_tenants=1)):84    if v.volume_type in voltype_map:85        if v.id in vols_with_snapshots:86            warn("volume %s has snapshot(s): removing" % v.id)87            for snapid in vols_with_snapshots[v.id]:88                remove_snapshot(snapid)89        srvs = [x['server_id'] for x in v.attachments]90        if len(srvs) > 1:91            # This should not happen since multi-attach is not supported yet92            warn("Volume %s is attached to multiple instances: ignoring" % v.id)93        else:94            vol_count += 195            vol_total_size += v.size96            if srvs:97                srvs_with_volumes.add(srvs[0])98                backlog[v.id] = v.volume_type99            else:100                retype_volume(v.id, voltype_map[v.volume_type])101# Then, for each idle server, detach all volumes, migrate if needed and re-attach102info("Migrating attached volumes")103for srv in srvs_with_volumes:104    s = nova.servers.get(srv)105    info("Migrating volumes attached to instance %s (%s)" % (srv, s.name))106    if s.status not in ('ACTIVE', 'SHUTOFF'):107        warn("Instance %s is in an invalid state (%s): skipping" % (srv, s.status))108        continue109    srv_running = s.status == 'ACTIVE'110    if srv_running:111        instance_power(srv, False)112        if no_rolling_migration:113            restart_instances.append(srv)114    vols = [x['id'] for x in s.to_dict().get('os-extended-volumes:volumes_attached')]115    workaround_required = not s.image and vols[0] in backlog116    if workaround_required:117        # This instance boots from a volume hosted on a legacy volume type:118        # apply workaround119        patch_volume_boot_index(vols[0], True)120    for v in vols:121        detach_volume(srv, v)122    for v in vols:123        if v in backlog:124            # We only want to retype if the volume has one of the legacy125            # volume types but we still want to detach and reattach the126            # volume to keep the right attachment order127            retype_volume(v, voltype_map[backlog[v]])128    for v in vols:129        attach_volume(srv, v)130    if workaround_required:131        # Undo workaround132        patch_volume_boot_index(vols[0], False)133    if srv_running and not no_rolling_migration:134        instance_power(srv, True)135# In case we are not doing a rolling migration, power back on the instances136# at the end of the migration137if restart_instances:138    info("Before proceding, make sure nova is correctly configured to connect to the new backend")139    for srv in restart_instances:140            instance_power(srv, True)141    ...

volumes.py

Source:volumes.py

1# Copyright 2022 Red Hat, Inc.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15import copy16from tempest.lib.api_schema.response.volume.v3_65 import volumes17# Volume microversion 3.69:18# The 'shared_targets' attribute is now a tristate boolean.19common_show_volume = copy.deepcopy(volumes.common_show_volume)20common_show_volume['properties'].update(21    {'shared_targets': {'type': ['boolean', 'null']}})22create_volume = copy.deepcopy(volumes.create_volume)23create_volume['response_body']['properties']['volume']['properties'].update(24    {'shared_targets': {'type': ['boolean', 'null']}})25# copy unchanged volumes schema26attachments = copy.deepcopy(volumes.attachments)27list_volumes_no_detail = copy.deepcopy(volumes.list_volumes_no_detail)28# show_volume refers to common_show_volume29show_volume = copy.deepcopy(volumes.show_volume)30show_volume['response_body']['properties']['volume'] = common_show_volume31# list_volumes_detail refers to latest common_show_volume32list_volumes_detail = copy.deepcopy(common_show_volume)33list_volumes_with_detail = copy.deepcopy(volumes.list_volumes_with_detail)34list_volumes_with_detail['response_body']['properties']['volumes']['items'] \35    = list_volumes_detail36update_volume = copy.deepcopy(volumes.update_volume)37delete_volume = copy.deepcopy(volumes.delete_volume)38show_volume_summary = copy.deepcopy(volumes.show_volume_summary)39attach_volume = copy.deepcopy(volumes.attach_volume)40set_bootable_volume = copy.deepcopy(volumes.set_bootable_volume)41detach_volume = copy.deepcopy(volumes.detach_volume)42reserve_volume = copy.deepcopy(volumes.reserve_volume)43unreserve_volume = copy.deepcopy(volumes.unreserve_volume)44extend_volume = copy.deepcopy(volumes.extend_volume)45reset_volume_status = copy.deepcopy(volumes.reset_volume_status)46update_volume_readonly = copy.deepcopy(volumes.update_volume_readonly)47force_delete_volume = copy.deepcopy(volumes.force_delete_volume)48retype_volume = copy.deepcopy(volumes.retype_volume)49force_detach_volume = copy.deepcopy(volumes.force_detach_volume)50create_volume_metadata = copy.deepcopy(volumes.create_volume_metadata)51show_volume_metadata = copy.deepcopy(volumes.show_volume_metadata)52update_volume_metadata = copy.deepcopy(volumes.update_volume_metadata)53update_volume_metadata_item = copy.deepcopy(54    volumes.update_volume_metadata_item)55update_volume_image_metadata = copy.deepcopy(56    volumes.update_volume_image_metadata)57delete_volume_image_metadata = copy.deepcopy(58    volumes.delete_volume_image_metadata)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.