Script File to Delete Resources

You must create a script file to remove the autoscaling resources before you destroy the stack in Oracle WebLogic Server for OCI.

You can copy the following scripts to the script file, remove_resources.py file in Cloud Shell.

""" # # Copyright (c) 2021, 2022, Oracle Corporation and/or its affiliates. # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. """ import os import sys import oci from oci.events import models as event_models from oci.functions import models as fn_models from oci.monitoring import models as monitoring_models from oci.artifacts import models as artifacts_models from oci.ons import models as ons_models from oci import pagination import argparse import textwrap REGION = '' def get_config(): if REGION != '': return {"region": REGION} return {} """ Lists and deletes the resources like instances, policies, volumes, VCN related resources, logs and tags etc.. """ class CleanUpResources: def __init__(self, service_name_prefix): self.service_name_prefix = service_name_prefix # delegate token should be present at /etc/oci/delegation_token in cloud shell if os.path.exists('/etc/oci/delegation_token'): with open('/etc/oci/delegation_token', 'r') as file: delegation_token = file.read() self.signer = oci.auth.signers.InstancePrincipalsDelegationTokenSigner(delegation_token=delegation_token) else: print("ERROR: In the Cloud shell the delegation token does not exist at location /etc/oci/delegation_token." "Run the script from the Cloud shell, where you need to delete the resources.") sys.exit(1) self.vcn_client = oci.core.VirtualNetworkClient(config=get_config(), signer=self.signer) self.virtual_network_composite_operations = oci.core.VirtualNetworkClientCompositeOperations(self.vcn_client) self.log_client = oci.logging.LoggingManagementClient(config=get_config(), signer=self.signer) self.log_composite_operations = oci.logging.LoggingManagementClientCompositeOperations(self.log_client) self.identity_client = oci.identity.IdentityClient(config={}, signer=self.signer) self.identity_client_composite_operations = oci.identity.IdentityClientCompositeOperations(self.identity_client) self.events_client = oci.events.EventsClient(config=get_config(), signer=self.signer) self.events_client_composite_operations = oci.events.EventsClientCompositeOperations(self.events_client) self.fn_client = oci.functions.FunctionsManagementClient(config=get_config(), signer=self.signer) self.fn_composite_operations = oci.functions.FunctionsManagementClientCompositeOperations(client=self.fn_client) self.monitoring_client = oci.monitoring.MonitoringClient(config=get_config(), signer=self.signer) self.monitoring_composite_operations = oci.monitoring.MonitoringClientCompositeOperations( client=self.monitoring_client) self.artifacts_client = oci.artifacts.ArtifactsClient(config=get_config(), signer=self.signer) self.artifacts_composite_operations = oci.artifacts.ArtifactsClientCompositeOperations( client=self.artifacts_client) self.ons_control_plane_client = oci.ons.NotificationControlPlaneClient(config=get_config(), signer=self.signer) # Lists all the resources based on the service name prefix def list_all_resources(self): search_client = oci.resource_search.ResourceSearchClient(config=get_config(), signer=self.signer) running_resources = ["RUNNING", "Running", "AVAILABLE", "STOPPED", "Stopped", "ACTIVE", "CREATED", "INACTIVE"] resource_not_required = ["PrivateIp", "Vnic"] # https://docs.oracle.com/en-us/iaas/Content/Search/Concepts/queryoverview.htm#resourcetypes structured_search = oci.resource_search.models.StructuredSearchDetails( query="query all resources where displayname =~ '{}'".format(self.service_name_prefix), type='Structured', matching_context_type=oci.resource_search.models.SearchDetails.MATCHING_CONTEXT_TYPE_NONE) resources = search_client.search_resources(structured_search) resources_details = [] no_of_resources = 0 # Tags and default route table doesn't start with service prefix tagname_resource = "wlsoci-" + self.service_name_prefix default_rt = "Default Route Table for " + self.service_name_prefix # Logs and unified agent config resources use service_prefix with underscore instead of hyphen log_resource = self.service_name_prefix[:-1] + "_" print( "Resource Name Resource Type Resource Lifecycle State OCID DOC") print( "=================================================================================================================================================") for resource in resources.data.items: resource_name = resource.display_name if (resource_name.startswith( self.service_name_prefix) or tagname_resource in resource_name or default_rt in resource_name or log_resource in resource_name) and ( resource.lifecycle_state in running_resources) and ( resource.resource_type not in resource_not_required): resources_details.append(resource) no_of_resources = no_of_resources + 1 print("{} {} {} {} {}".format(resource.display_name, resource.resource_type, resource.lifecycle_state, resource.identifier, resource.time_created)) print( "================================================================================================================================================") print("Total number of resources {}".format(len(resources_details))) return resources_details # Removes all resources based on the service name prefix def cleanup_resources(self, delete_list): print("Deleting the resources") self.delete_all_autoscaling_resources(delete_list) self.delete_policies(delete_list) self.delete_instance(delete_list) self.delete_block_volumes(delete_list) self.delete_load_balancer(delete_list) self.delete_subnet(delete_list) self.delete_sec_list(delete_list) self.delete_route_table(delete_list) self.delete_dhcp_options(delete_list) self.delete_internet_gateway(delete_list) self.delete_service_gateway(delete_list) self.delete_local_peering_gateway(delete_list) self.delete_nat_gateway(delete_list) self.delete_vcn_resources(delete_list) self.delete_unified_agent_configuration(delete_list) self.delete_log(delete_list) self.delete_log_group(delete_list) self.delete_mount_targets(delete_list) self.delete_fss(delete_list) self.delete_tag_namespace(delete_list) self.delete_boot_volumes(delete_list) # Delete Policies def delete_policies(self, delete_list): for resource in delete_list: if resource.resource_type == "Policy": policy_ocid = resource.identifier print("Deleting policy: {0}, with ocid: {1}".format(resource.display_name, policy_ocid)) try: self.identity_client_composite_operations.delete_policy_and_wait_for_state( policy_ocid, wait_for_states=[oci.identity.models.Policy.LIFECYCLE_STATE_DELETED]) print("Deleted policy successfully!") except Exception as e: print("Error while deleting the policy {0}, policy id {1}, Error message {2}".format( resource.display_name, policy_ocid, str(e))) # Delete Dynamic Group def delete_dynamic_group(self): tenancy = os.environ['OCI_TENANCY'] dynamic_group_list = self.identity_client.list_dynamic_groups(tenancy).data for d_group in dynamic_group_list: if self.service_name_prefix in d_group.name: print("Deleting the dynamic group: {0}, with ocid: {1}".format(d_group.name, d_group.id)) try: self.identity_client_composite_operations.delete_dynamic_group_and_wait_for_state( d_group.id, wait_for_states=[oci.identity.models.DynamicGroup.LIFECYCLE_STATE_DELETED]) print("Deleted the dynamic group successfully!") except Exception as e: print("Error while deleting the dynamic group name {}, ocid {}, Error message {}".format( d_group.name, d_group.id, str(e))) # Delete Block Volumes def delete_block_volumes(self, delete_list): bv_client = oci.core.BlockstorageClient(config=get_config(), signer=self.signer) bv_composite_operations = oci.core.BlockstorageClientCompositeOperations(bv_client) for resource in delete_list: if resource.resource_type == "Volume": bv_ocid = resource.identifier try: print( "Deleting the block volume: {0}, with ocid {1}".format(resource.display_name, bv_ocid)) bv_composite_operations.delete_volume_and_wait_for_state( bv_ocid, wait_for_states=[oci.core.models.Volume.LIFECYCLE_STATE_TERMINATED]) print("Deleted the block volume successfully!") except Exception as e: print( "Error while deleting the block volume {0}, ocid {1}, Error message {2}".format( resource.display_name, bv_ocid, str(e))) # Delete all compute instances def delete_instance(self, delete_list): compute_client = oci.core.ComputeClient(config=get_config(), signer=self.signer) compute_composite_operations = oci.core.ComputeClientCompositeOperations(compute_client) for resource in delete_list: if resource.resource_type == "Instance": instance_ocid = resource.identifier instance_name = resource.display_name print("Deleting the compute instance: {0}, with ocid {1}".format(instance_name, instance_ocid)) try: compute_composite_operations.terminate_instance_and_wait_for_state( instance_ocid, wait_for_states=[oci.core.models.Instance.LIFECYCLE_STATE_TERMINATED]) print("Deleted the compute instance successfully!") except Exception as e: print( "Error while deleting the instance {0}, ocid {1}, Error message {2}".format( instance_name, instance_ocid, str(e))) # Delete all Subnets in the VCN def delete_subnet(self, delete_list): for resource in delete_list: if resource.resource_type == "Subnet": subnet_ocid = resource.identifier print( "Deleting subnet: {0}, with ocid {1}".format(resource.display_name, resource.identifier)) try: self.virtual_network_composite_operations.delete_subnet_and_wait_for_state( subnet_ocid, wait_for_states=[oci.core.models.Subnet.LIFECYCLE_STATE_TERMINATED]) print("Deleted subnet successfully!") except Exception as e: print( "Error while deleting the subnet {0}, ocid {1}, Error message {2}".format(resource.display_name, subnet_ocid, str(e))) # Delete Security lists def delete_sec_list(self, delete_list): for resource in delete_list: if resource.resource_type == "SecurityList": sec_list_name = resource.display_name sec_list_ocid = resource.identifier if not ("Default" in sec_list_name): print( "Deleting the security list: {0}, with ocid {1}".format(resource.display_name, resource.identifier)) try: self.virtual_network_composite_operations.delete_security_list_and_wait_for_state( sec_list_ocid, wait_for_states=[oci.core.models.SecurityList.LIFECYCLE_STATE_TERMINATED]) print("Deleted the security list successfully!") except Exception as e: print( "Error while deleting the security list {0}, ocid {1}, Error message {2}".format( resource.display_name, sec_list_ocid, str(e))) # Delete Load balancers def delete_load_balancer(self, delete_list): lb_client = oci.load_balancer.LoadBalancerClient(config=get_config(), signer=self.signer) lb_composite_operations = oci.load_balancer.LoadBalancerClientCompositeOperations(lb_client) for resource in delete_list: if resource.resource_type == "LoadBalancer": lb_name = resource.display_name lb_ocid = resource.identifier print("Deleting Load balancer {0} with ocid {1}".format(lb_name, lb_ocid)) try: lb_composite_operations.delete_load_balancer_and_wait_for_state( lb_ocid, wait_for_states=[oci.load_balancer.models.WorkRequest.LIFECYCLE_STATE_SUCCEEDED]) print("Load balancer deleted successfully!") except Exception as e: print( "Error while deleting the loadbalancer {0}, ocid {1}, Error message {2}".format( lb_name, lb_ocid, str(e))) # Delete Route tables def delete_route_table(self, delete_list): for resource in delete_list: if resource.resource_type == "RouteTable": route_table_name = resource.display_name route_table_ocid = resource.identifier # Removing the route rules from the tables rt_details = oci.core.models.UpdateRouteTableDetails() rt_details.route_rules = [] self.virtual_network_composite_operations.update_route_table_and_wait_for_state( route_table_ocid, rt_details, wait_for_states=[oci.core.models.RouteTable.LIFECYCLE_STATE_AVAILABLE]) self.vcn_client.update_route_table(route_table_ocid, rt_details) # Default route table can't be deleted from VCN if not ("Default" in route_table_name): print( "Deleting the route table: {0}, with ocid {1}".format(resource.display_name, resource.identifier)) try: self.virtual_network_composite_operations.delete_route_table_and_wait_for_state( route_table_ocid, wait_for_states=[oci.core.models.RouteTable.LIFECYCLE_STATE_TERMINATED]) print("Deleted the route table successfully!") except Exception as e: print("Error while deleting the route table {0}, ocid {1}, Error message {2}".format( resource.display_name, route_table_ocid, str(e))) if "associated with Subnet" in str(e): try: self.delete_subnet_route_table_association(route_table_ocid) # After removing the association again retrying the removal of route table # This is for Db subnet route table self.virtual_network_composite_operations.delete_route_table_and_wait_for_state( route_table_ocid, wait_for_states=[oci.core.models.RouteTable.LIFECYCLE_STATE_TERMINATED]) print("Deleted the route table successfully!") except Exception as e: print("Error while deleting the route table after removing the association " "{0}, ocid {1}, Error message {2}".format (resource.display_name, route_table_ocid, str(e))) # Delete Subnet and route table association to remove route table def delete_subnet_route_table_association(self, route_table_ocid): default_rt_id_in_vcn = "" print("Route table is associated with a subnet. Removing the association between the subnet and route table") rt_res = self.vcn_client.get_route_table(route_table_ocid).data vcn_id = rt_res.vcn_id compartment_id = rt_res.compartment_id list_route_rables_vcn = self.vcn_client.list_route_tables(compartment_id=compartment_id, vcn_id=vcn_id).data for rt in list_route_rables_vcn: if "Default Route" in rt.display_name: default_rt_id_in_vcn = rt.id list_subnets = self.vcn_client.list_subnets(compartment_id=compartment_id, vcn_id=vcn_id).data for subnet in list_subnets: subnet_ocid = subnet.id if subnet.route_table_id == route_table_ocid: subnet_details = oci.core.models.UpdateSubnetDetails() subnet_details.route_table_id = default_rt_id_in_vcn try: self.virtual_network_composite_operations.update_subnet_and_wait_for_state( subnet_ocid, subnet_details, wait_for_states=[oci.core.models.Subnet.LIFECYCLE_STATE_AVAILABLE]) print("Removed the association between the subnet and route table.") except Exception as e: print("Error while removing the association between the subnet and route table {}".format(str(e))) # Delete DHCP Options def delete_dhcp_options(self, delete_list): for resource in delete_list: if resource.resource_type == "DHCPOptions": dhcp_name = resource.display_name dhcp_ocid = resource.identifier if not ("Default" in dhcp_name): print( "Deleting the DHCP options: {0}, with ocid {1}".format(resource.display_name, dhcp_ocid)) try: self.virtual_network_composite_operations.delete_dhcp_options_and_wait_for_state(dhcp_ocid, wait_for_states=[ oci.core.models.DhcpOptions.LIFECYCLE_STATE_TERMINATED]) print("Deleted the DHCP options successfully!") except Exception as e: print( "Error while deleting the DHCP options {0}, ocid {1}, Error message {2} ".format( resource.display_name, dhcp_ocid, str(e))) # Delete Internet Gateway def delete_internet_gateway(self, delete_list): for resource in delete_list: if resource.resource_type == "InternetGateway": ig_ocid = resource.identifier print("Deleting the Internet Gateway: {0}, with ocid {1}".format(resource.display_name, ig_ocid)) try: self.virtual_network_composite_operations.delete_internet_gateway_and_wait_for_state(ig_ocid, wait_for_states=[ oci.core.models.InternetGateway.LIFECYCLE_STATE_TERMINATED]) print("Deleted the Internet Gateway successfully!") except Exception as e: print("Error while deleting the Internet Gateway {0}, ocid {1}, Error message {2}".format( resource.display_name, ig_ocid, str(e))) # Delete Service Gateway def delete_service_gateway(self, delete_list): for resource in delete_list: if resource.resource_type == "ServiceGateway": svc_gateway_ocid = resource.identifier print("Deleting the service gateway: {0}, with ocid {1}".format(resource.display_name, svc_gateway_ocid)) try: self.virtual_network_composite_operations.delete_service_gateway_and_wait_for_state( svc_gateway_ocid, wait_for_states=[oci.core.models.ServiceGateway.LIFECYCLE_STATE_TERMINATED]) print("Deleted the service gateway successfully!") except Exception as e: print("Error while deleting the service gateway {0}, ocid {1}, Error message {2}".format( resource.display_name, svc_gateway_ocid, str(e))) # Delete Local Peering Gateway def delete_local_peering_gateway(self, delete_list): for resource in delete_list: if resource.resource_type == "LocalPeeringGateway": lpg_ocid = resource.identifier print("Deleting the local peering gateway: {0}, with ocid {1}".format(resource.display_name, lpg_ocid)) try: self.virtual_network_composite_operations.delete_local_peering_gateway_and_wait_for_state( lpg_ocid, wait_for_states=[oci.core.models.LocalPeeringGateway.LIFECYCLE_STATE_TERMINATED]) print("Deleted local peering gateway successfully!") except Exception as e: print("Error while deleting the local peering gateway {0}, ocid {1}, Error message {2}".format( resource.display_name, lpg_ocid, str(e))) # Delete Nat Gateway def delete_nat_gateway(self, delete_list): for resource in delete_list: if resource.resource_type == "NatGateway": nat_ocid = resource.identifier print("Deleting the NAT gateway: {0}, with ocid {1}".format(resource.display_name, nat_ocid)) try: self.virtual_network_composite_operations.delete_nat_gateway_and_wait_for_state( nat_gateway_id=nat_ocid, wait_for_states=[oci.core.models.NatGateway.LIFECYCLE_STATE_TERMINATED] ) print("Deleted the NAT gateway successfully!") except Exception as e: print("Error while deleting the NAT gateway {0}, ocid {1}, Error message {2}".format( resource.display_name, nat_ocid, str(e))) # Delete VCN def delete_vcn_resources(self, delete_list): for resource in delete_list: if resource.resource_type == "Vcn": vcn_ocid = resource.identifier vcn_name = resource.display_name print("Deleting the VCN: {0}, with ocid {1}".format(vcn_name, vcn_ocid)) try: self.virtual_network_composite_operations.delete_vcn_and_wait_for_state(vcn_ocid, oci.core.models.Vcn.LIFECYCLE_STATE_TERMINATED) print("Deleted the VCN successfully!") except Exception as e: print("Error while deleting the VCN {0}, VCN id {1}, Error message {2}".format(vcn_name, vcn_ocid, str(e))) # Deleting the Unified Agent Configuration def delete_unified_agent_configuration(self, delete_list): for resource in delete_list: if resource.resource_type == "UnifiedAgentConfiguration": uac_ocid = resource.identifier print("Deleting the unified agent configuration: {}, with ocid {}".format(resource.display_name, uac_ocid)) try: self.log_composite_operations.delete_unified_agent_configuration_and_wait_for_state( uac_ocid, wait_for_states=[oci.logging.models.WorkRequest.STATUS_SUCCEEDED]) print("Deleted the unified agent configuration successfully!") except Exception as e: print( "Error while deleting the unified agent configuration name {0}, ocid {1} - Error message {2}".format( resource.display_name, uac_ocid, str(e))) # Delete logs in a Log groups def delete_log(self, delete_list, name=None): """ Delete log resources for the service. :param delete_list: :param name: if name is set, only delete the named log resource, otherwise delete all log resources in the delete_list :return: """ for resource in delete_list: if resource.resource_type == "LogGroup": log_group_ocid = resource.identifier list_logs = self.log_client.list_logs(log_group_ocid).data for log in list_logs: to_delete = True if name is not None: if log.display_name != name: to_delete = False if to_delete: print("Deleting the log name {0}, with log ocid {1}".format(log.display_name, log.id)) try: self.log_composite_operations.delete_log_and_wait_for_state( log_group_ocid, log.id, wait_for_states=[oci.logging.models.WorkRequest.STATUS_SUCCEEDED]) print("Deleted the log {} successfully!".format(log.display_name)) except Exception as e: print("Error while deleting the log name {}, log ocid {}, Error message {}".format( log.display_name, log.id, str(e))) # Delete Log Group def delete_log_group(self, delete_list): for resource in delete_list: if resource.resource_type == "LogGroup": log_group_ocid = resource.identifier print("Deleting the log group: {0}, with ocid {1}".format(resource.display_name, log_group_ocid)) try: self.log_composite_operations.delete_log_group_and_wait_for_state( log_group_ocid, wait_for_states=[oci.logging.models.WorkRequest.STATUS_SUCCEEDED]) print("Deleted log group successfully!") except Exception as e: print("Error while deleting the log group {0}, ocid {1}, Error message {2}".format( resource.display_name, log_group_ocid, str(e))) # Delete the Mount targets def delete_mount_targets(self, delete_list): mt_client = oci.file_storage.FileStorageClient(config=get_config(), signer=self.signer) mt_composite_operations = oci.file_storage.FileStorageClientCompositeOperations(mt_client) for resource in delete_list: if resource.resource_type == "MountTarget": mt_ocid = resource.identifier print("Deleting the mount target {0}, with ocid {1}".format(resource.display_name, mt_ocid)) try: mt_composite_operations.delete_mount_target_and_wait_for_state( mt_ocid, wait_for_states=[oci.file_storage.models.MountTarget.LIFECYCLE_STATE_DELETED]) print("Deleted the mount target successfully!") except Exception as e: print("Error while deleting the mount target {0}, ocid {1}, Error message {2}".format( resource.display_name, mt_ocid, str(e))) # Delete FSS def delete_fss(self, delete_list): fss_client = oci.file_storage.FileStorageClient(config=get_config(), signer=self.signer) fss_composite_operations = oci.file_storage.FileStorageClientCompositeOperations(fss_client) for resource in delete_list: if resource.resource_type == "FileSystem": fss_ocid = resource.identifier try: # Get the list of exports to delete list_exports = fss_client.list_exports(file_system_id=fss_ocid).data for export in list_exports: export_ocid = export.id print("Deleting the export id {}".format(export_ocid)) fss_composite_operations.delete_export_and_wait_for_state( export_id=export_ocid, wait_for_states=[oci.file_storage.models.Export.LIFECYCLE_STATE_DELETED]) print("Deleted the exports successfully!") except Exception as e: print("Error while deleting the export, Error message {}".format(str(e))) try: print("Deleting the FSS: {0}, with ocid {1}".format(resource.display_name, fss_ocid)) fss_composite_operations.delete_file_system_and_wait_for_state( fss_ocid, wait_for_states=[oci.file_storage.models.FileSystem.LIFECYCLE_STATE_DELETED]) print("Deleted the FSS successfully!") except Exception as e: print("Error while deleting the FSS name {0}, ocid {1}, Error message {2}".format( resource.display_name, fss_ocid, str(e))) # Deletion of TagNamespace def delete_tag_namespace(self, delete_list): for resource in delete_list: if resource.resource_type == "TagNamespace": tag_ns_name = resource.display_name tag_ns_ocid = resource.identifier print("Deleting the tag namespace {0}, with ocid {1}".format(tag_ns_name, tag_ns_ocid)) try: # Retiring the tag namespace tag_status = self.identity_client.get_tag_namespace(tag_namespace_id=tag_ns_ocid).data print("Tag namespace: {} and isRetired: {}".format(tag_ns_name, tag_status.is_retired)) if not tag_status.is_retired: print("Retiring the tag namespace {}".format(tag_ns_name)) tag_ns_details = oci.identity.models.UpdateTagNamespaceDetails() tag_ns_details.is_retired = True self.identity_client_composite_operations.update_tag_namespace_and_wait_for_state( tag_namespace_id=tag_ns_ocid, update_tag_namespace_details=tag_ns_details, wait_for_states=[ oci.identity.models.TagNamespace.LIFECYCLE_STATE_INACTIVE]) tag_status = self.identity_client.get_tag_namespace(tag_namespace_id=tag_ns_ocid).data print("Tag status before deleting {}".format(tag_status.is_retired)) print("Deleting the tag namespace {}".format(tag_ns_name)) # Tag namespace deletion is taking too long time. So not waiting for the completion. self.identity_client.cascade_delete_tag_namespace(tag_namespace_id=tag_ns_ocid) print("Asynchronous deletion of Tag namespaces is enabled." "Check the deletion status manually. Tag name {0} with ocid {1}".format(tag_ns_name, tag_ns_ocid)) except Exception as e: print("Error while deleting the Tag namespace {0}, ocid {1}, Error message {2} " .format(tag_ns_name, tag_ns_ocid, str(e))) # Deleting the unattached boot volumes def delete_boot_volumes(self, delete_list): bv_client = oci.core.BlockstorageClient(config=get_config(), signer=self.signer) bv_composite_operations = oci.core.BlockstorageClientCompositeOperations(bv_client) for resource in delete_list: if resource.resource_type == "BootVolume" and resource.lifecycle_state == "AVAILABLE": bv_ocid = resource.identifier bv_name = resource.display_name print("Deleting the boot volume {}, with ocid {} ".format(bv_name, bv_ocid)) try: bv_composite_operations.delete_boot_volume_and_wait_for_state( boot_volume_id=bv_ocid, wait_for_states=[oci.core.models.BootVolume.LIFECYCLE_STATE_TERMINATED]) print("Deleted the boot volume successfully!") except Exception as e: print("Error while deleting the boot volume name {}, ocid {}, Error message {}".format(bv_name, bv_ocid, str(e))) def delete_functions(self, delete_list): """ Deletes OCI functions and function application for the service. :param delete_list: :return: """ for resource in delete_list: if resource.resource_type == "FunctionsApplication": fn_app_id = resource.identifier fn_app_name = resource.display_name try: result = pagination.list_call_get_all_results( self.fn_client.list_functions, fn_app_id ) # Delete all functions within function application print("Deleting functions within function application {}".format(fn_app_name)) for fn in result.data: self.fn_composite_operations.delete_function_and_wait_for_state( function_id=fn.id, wait_for_states=[fn_models.Function.LIFECYCLE_STATE_DELETED] ) print("Deleted the function {} successfully!".format(fn.display_name)) except Exception as e: print("Error while deleting the functions within application id {}, Error message {}".format(fn_app_id, str(e))) def delete_functions_app(self, delete_list): """ Deletes function application for the service. :param delete_list: :return: """ for resource in delete_list: if resource.resource_type == "FunctionsApplication": fn_app_id = resource.identifier try: # Delete the function application self.fn_composite_operations.delete_application_and_wait_for_state( application_id = fn_app_id, wait_for_states = [fn_models.Application.LIFECYCLE_STATE_DELETED] ) except Exception as e: print("Error while deleting the Functions application with id {}, Error message {}".format(fn_app_id, str(e))) def delete_event_rules(self, delete_list): """ Deletes the event rules for the service. :param delete_list: :return: """ for resource in delete_list: if resource.resource_type == "EventRule": event_rule_id = resource.identifier event_rule_name = resource.display_name try: print("Deleting event rule {}".format(event_rule_name)) self.events_client_composite_operations.delete_rule_and_wait_for_state( rule_id=event_rule_id, wait_for_states=[event_models.Rule.LIFECYCLE_STATE_DELETED] ) print("Deleted the event rule {} successfully!".format(event_rule_name)) except Exception as e: print("Error while deleting the event rule id {}, Error message {}".format(event_rule_id, str(e))) def delete_autoscaling_logs(self, delete_list): """ Deletes the event rule invoke log for the service. :param delete_list: :return: """ # trim the extra hyphen char if present from the service_name_prefix service_name = self.service_name_prefix[:-1] if self.service_name_prefix[ -1] == '-' else self.service_name_prefix self.delete_log(delete_list, name="{0}_event_rule_invoke_log".format(service_name)) self.delete_log(delete_list, name="{0}_autoscaling_log".format(service_name)) def predestroy_autoscaling_resources(self, delete_list): """ Deletes pre-destroy resources associated with autoscaling. This is to be invoked when user needs to delete only the resources created via API during provisioning outside of terraform preferably prior to running terraform destroy action. :param delete_list: :return: """ print("Executing pre-destroy of resources created for autoscaling feature") self.delete_functions(delete_list) self.delete_event_rules(delete_list) self.delete_autoscaling_logs(delete_list) def delete_alarms(self, delete_list): """ Delete all alarms created for autoscaling for the service. :param delete_list: :return: """ for resource in delete_list: if resource.resource_type == "Alarm": alarm_id = resource.identifier alarm_name = resource.display_name try: print("Deleting alarm {}".format(alarm_name)) self.monitoring_composite_operations.delete_alarm_and_wait_for_state( alarm_id=alarm_id, wait_for_states=[monitoring_models.Alarm.LIFECYCLE_STATE_DELETED] ) print("Deleted the alarm {} successfully!".format(alarm_name)) except Exception as e: print("Error while deleting the alarm id {}, Error message {}".format(alarm_id, str(e))) def delete_container_repos(self, delete_list): """ Deletes container repos created for autoscaling for the service. Deleting the repos also deletes the container images in those repos. :param delete_list: :return: """ for resource in delete_list: if resource.resource_type == "ContainerRepo": repo_id = resource.identifier repo_name = resource.display_name try: print("Deleting container repo {}".format(repo_name)) self.artifacts_composite_operations.delete_container_repository_and_wait_for_state( repository_id=repo_id, wait_for_states=[artifacts_models.ContainerRepository.LIFECYCLE_STATE_DELETED] ) print("Deleted the container repo {} successfully!".format(repo_name)) except Exception as e: print("Error while deleting the container repo id {}, Error message {}".format(repo_id, str(e))) def delete_notification_topics(self, delete_list): """ Deletes the notification topics created for autoscaling for the service. Deleting the notification topics also deletes the associated subscriptions for those topics. :param delete_list: :return: """ for resource in delete_list: if resource.resource_type == "OnsTopic": topic_id = resource.identifier topic_name = resource.display_name try: print("Deleting notification topic {}".format(topic_name)) self.ons_control_plane_client.delete_topic( topic_id=topic_id ) oci.wait_until(self.ons_control_plane_client, self.ons_control_plane_client.get_topic(topic_id), 'lifecycle_state', ons_models.NotificationTopic.LIFECYCLE_STATE_DELETING) print("Deleted the notification topic {} successfully!".format(topic_name)) except Exception as e: print( "Error while deleting the notification topic id {}, Error message {}".format(topic_id, str(e))) def delete_all_autoscaling_resources(self, delete_list): """ Deletes all resources created when autoscaling feature is enabled. :param delete_list: :return: """ self.predestroy_autoscaling_resources(delete_list) self.delete_container_repos(delete_list) self.delete_alarms(delete_list) self.delete_notification_topics(delete_list) self.delete_functions_app(delete_list) def main(): parser = argparse.ArgumentParser(description=textwrap.dedent(''' This script is used for: - pre-destroying autoscaling resources (prior to running destroy job on the stack from OCI Console) - delete all infra resources created for the stack (identified by its service name prefix) '''), formatter_class=argparse.RawDescriptionHelpFormatter) # Required position params parser.add_argument('command', choices=['delete', 'list', 'pre-destroy'], help='Command') parser.add_argument('service_name_prefix', help='Stack service name') # Optional params parser.add_argument('-r', '--region', default='', help='Region other than home region must be specified.') parser.add_argument('-f', '--feature', default='autoscaling', help='Feature to run pre-destroy for.') args = parser.parse_args() command = args.command service_prefix = args.service_name_prefix global REGION REGION = args.region feature = args.feature print("Service prefix name:" + service_prefix) if len(service_prefix) >= 16: service_prefix = service_prefix[0:16] service_prefix = service_prefix + "-" cleanup_util = CleanUpResources(service_prefix) if command == 'list': print("Listing all resources with service prefix name " + service_prefix) cleanup_resources = cleanup_util.list_all_resources() elif command == 'delete': print("Deleting all resources with service prefix name " + service_prefix) cleanup_resources = cleanup_util.list_all_resources() cleanup_util.cleanup_resources(cleanup_resources) cleanup_util.delete_dynamic_group() elif command == 'pre-destroy' and feature == 'autoscaling': print('Deleting pre-destroy autoscaling resources with service prefix name ' + service_prefix) cleanup_resources = cleanup_util.list_all_resources() cleanup_util.predestroy_autoscaling_resources(cleanup_resources) if __name__ == '__main__': main()