Clean up orphans boot volumes (dettached) in all compartments and subscribed region of a tenancy

oci_cleanup_block_storage.py

#!/usr/bin/env python3
# coding: utf-8
# Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
# This script clean-up all detached block storage (boot and block volumes)
#
#
#
#
import oci




##############################################################
def cleanup_block_storage(oci_config_file, tenancy_id, list_target_compartment_id, dry_run ):
    if (dry_run):
        print('***** dry-run, volumes not deleted ******')
    home_region_config = oci_config_file
    identityc = oci.identity.IdentityClient(home_region_config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY)
    region_subscriptions = identityc.list_region_subscriptions(tenancy_id).data
    region_config = home_region_config
    compartments = list_of_compartments(identityc, tenancy_id, list_target_compartment_id)

    for compartment in compartments:
        if (compartment.lifecycle_state != 'ACTIVE'):
            continue

        print('compartment: '+compartment.name+' ['+compartment.id+']')
        compartment_id = compartment.id

        for region in region_subscriptions:
            if (region.status != 'READY'):
                continue
            for key in home_region_config:
                if key == 'region':
                    region_config[key] = region.region_name
            cleanup_block_storage_on_region(region_config, region.region_name, tenancy_id, compartment_id, dry_run)
        print('\n')


def list_of_compartments(identity_client, tenancy_id, list_target_compartment_id):
    if (list_target_compartment_id is None):
        #list_compartments do not return the root compartment, need additional call to retrieve it.
        compartments = identity_client.list_compartments(tenancy_id, compartment_id_in_subtree=True, access_level="ANY").data
        compartments.append(identity_client.get_compartment(tenancy_id).data)
    else:
        compartments = list()
        for target_compartment_id in list_target_compartment_id:
            compartments.append(identity_client.get_compartment(target_compartment_id).data)
    
    return compartments


##############################################################
def cleanup_block_storage_on_region(config, region_name, tenancy_id, compartment_id, dry_run):

    identityc = oci.identity.IdentityClient(config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY)
    blockc = oci.core.BlockstorageClient(config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY)
    computec = oci.core.ComputeClient(config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY)
    
    availability_domains = identityc.list_availability_domains(tenancy_id).data    

    boot_volume_list = list()
    block_volume_list = list()
    boot_volume_attachment_list = list()

    #iterate over list of objects that are AD scope
    for ad in availability_domains:            
        # print('> '+ad.name)
        
        boot_volume_list.extend(blockc.list_boot_volumes(ad.name, compartment_id).data)
        block_volume_list.extend(blockc.list_volumes(compartment_id, availability_domain=ad).data)
        boot_volume_attachment_list.extend(computec.list_boot_volume_attachments(ad, compartment_id).data)
        
        
    block_volume_attachment_list = computec.list_volume_attachments(compartment_id).data

    orphan_boot_volume = list_orphan_boot_volume_id(region_name, boot_volume_attachment_list, boot_volume_list)    
    orphan_block_volume = list_orphan_block_volume_id(region_name, block_volume_attachment_list, block_volume_list)

    if (not dry_run):
        if (len(orphan_block_volume) > 0 or len(orphan_boot_volume) > 0):
            print('>>> deleting volume storage in '+region_name)
            delete_volumes(blockc, orphan_boot_volume, orphan_block_volume)
        else:
            print('>>> nothing to delete in '+region_name)

                        
##############################################################
def delete_volumes(blockc, list_boot_volume_id, list_block_volume_id):
    for boot_volume_id in list_boot_volume_id:
        blockc.delete_boot_volume(boot_volume_id)
    
    for block_volume_id in list_block_volume_id:
        blockc.delete_volume(block_volume_id)



##############################################################
def list_orphan_boot_volume_id(region_name, boot_volume_attachment_list, list_all_boot_volumes):
    orphan_boot_volume_ids = set()
    all_boot_volume_ids = set()
    attached_boot_volume_ids = set()

    for boot_volume_attachment in boot_volume_attachment_list:
        if boot_volume_attachment.lifecycle_state == "ATTACHED":
            attached_boot_volume_ids.add(boot_volume_attachment.boot_volume_id)
        

    for boot_volume in list_all_boot_volumes:
        if boot_volume.lifecycle_state == "AVAILABLE":
            all_boot_volume_ids.add(boot_volume.id)
        
    orphan_boot_volume_ids = all_boot_volume_ids.difference(attached_boot_volume_ids)

    if (len(all_boot_volume_ids) > 0 or len(attached_boot_volume_ids) > 0 or len(orphan_boot_volume_ids)):
        print('>'+region_name)
        print('>> total boot volumes='+str(len(all_boot_volume_ids)))
        print('>> boot volume attachments='+str(len(attached_boot_volume_ids)))
        print('>> boot volume orphans='+str(len(orphan_boot_volume_ids)))


    return orphan_boot_volume_ids


def list_orphan_block_volume_id(region_name, block_volume_attachment_list, list_all_block_volumes):
    orphan_block_volume_ids = set()
    all_block_volume_ids = set()
    attached_block_volume_ids = set()

    for block_volume_attachment in block_volume_attachment_list:
        if block_volume_attachment.lifecycle_state == "ATTACHED":
            attached_block_volume_ids.add(block_volume_attachment.volume_id)
        

    for block_volume in list_all_block_volumes:
        if block_volume.lifecycle_state == "AVAILABLE":
            all_block_volume_ids.add(block_volume.id)
        
    orphan_block_volume_ids = all_block_volume_ids.difference(attached_block_volume_ids)

    if (len(all_block_volume_ids) > 0 or len(attached_block_volume_ids) > 0 or len(orphan_block_volume_ids)):
        print('>'+region_name)
        print('>> total block volumes='+str(len(all_block_volume_ids)))
        print('>> block volume attachments='+str(len(attached_block_volume_ids)))
        print('>> block volume orphans='+str(len(orphan_block_volume_ids)))


    return orphan_block_volume_ids

##############################################################
##### main
##############################################################

oci_config_file = oci.config.from_file('./oci_config') 

# list_target_compartment_id = ["ocid1.tenancy.oc1.", "ocid1.compartment.oc1.."]

# check all compartments 
list_target_compartment_id = None
dry_run = True
# dry_run = False

cleanup_block_storage(oci_config_file, oci_config_file["tenancy"], list_target_compartment_id, dry_run)