# -*- coding: utf-8 -*- from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_tea_util import models as util_models from alibabacloud_tea_util.client import Client as UtilClient from alibabacloud_sts20150401.client import Client as Sts20150401Client from alibabacloud_sts20150401 import models as sts_20150401_models from alibabacloud_vpc20160428.client import Client as Vpc20160428Client from alibabacloud_vpc20160428 import models as vpc_20160428_models from alibabacloud_alb20200616.client import Client as Alb20200616Client from alibabacloud_alb20200616 import models as alb_20200616_models from alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Client from alibabacloud_bssopenapi20171214 import models as bss_open_api_20171214_models from alibabacloud_ecs20140526.client import Client as Ecs20140526Client from alibabacloud_ecs20140526 import models as ecs_20140526_models from datetime import timedelta, datetime rd_management_account_access_key_id = 'yourAccessKeyId' rd_management_account_access_key_secret = 'yourAccessKeySecret' rd_management_account_readonly_role_name = 'readOnly' rd_management_account_id = 'yourRdManagementAccountId' rd_member_account_id_list = ['yourRdMemberAccountId', 'yourRdMemberAccountId'] region_id_list = ['cn-shanghai', 'cn-hangzhou'] class OpenAPI: def __init__(self): pass @staticmethod def create_sts_client( access_key_id: str, access_key_secret: str, ) -> Sts20150401Client: """ 使用AK&SK初始化账号Client @param access_key_id: @param access_key_secret: @return: Client @throws Exception """ config = open_api_models.Config( # 您的 AccessKey ID, access_key_id=access_key_id, # 您的 AccessKey Secret, access_key_secret=access_key_secret ) # 访问的域名 config.endpoint = f'sts.cn-shanghai.aliyuncs.com' return Sts20150401Client(config) @staticmethod def create_credentials_by_assume_role(account_id, access_key_id, access_key_secret, role_name): client = OpenAPI.create_sts_client(access_key_id, access_key_secret) assume_role_request = sts_20150401_models.AssumeRoleRequest( role_arn='acs:ram::' + account_id + ':role/' + role_name, role_session_name='management-account-programmaticUser' ) runtime = util_models.RuntimeOptions() try: resp = client.assume_role_with_options(assume_role_request, runtime) body_dict = resp.body.to_map() return body_dict['Credentials'] except Exception as error: print(UtilClient.assert_as_string(error)) @staticmethod def create_api_models_config_by_assume_role(account_id, access_key_id, access_key_secret, role_name): rd_role_credentials = OpenAPI.create_credentials_by_assume_role(account_id, access_key_id, access_key_secret, role_name) config = open_api_models.Config() config.access_key_id = rd_role_credentials['AccessKeyId'] config.access_key_secret = rd_role_credentials['AccessKeySecret'] config.security_token = rd_role_credentials['SecurityToken'] return config class IdleResourceSample: def __init__(self): pass @staticmethod def create_ecs_client( account_id: str, access_key_id: str, access_key_secret: str, role_name: str, region_id: str, ) -> Ecs20140526Client: config = OpenAPI.create_api_models_config_by_assume_role(account_id, access_key_id, access_key_secret, role_name) config.endpoint = f'ecs.{region_id}.aliyuncs.com' return Ecs20140526Client(config) @staticmethod def query_ecs_disk( region_id: str, status: str, client: Ecs20140526Client ): describe_disks_request = ecs_20140526_models.DescribeDisksRequest( region_id=region_id, status=status ) runtime = util_models.RuntimeOptions() try: resp = client.describe_disks_with_options(describe_disks_request, runtime) body_dict = resp.body.to_map() # print(body_dict) return body_dict['Disks']['Disk'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_idle_ecs_disk(account_id, access_key_id, access_key_secret, role_name): idle_resource_list = [] print('[Idle ECS Disk] AccountId: ' + account_id) for region_id in region_id_list: output_str = 'Region:' + region_id + ' ' client = IdleResourceSample.create_ecs_client(account_id, access_key_id, access_key_secret, role_name, region_id) result_list = IdleResourceSample.query_ecs_disk(region_id, 'Available', client) for disk in result_list: # 状态为待挂载，闲置 output_str += f' ' idle_resource_list.append(disk) print(output_str) print() return idle_resource_list @staticmethod def create_bss_client( account_id: str, access_key_id: str, access_key_secret: str, role_name: str, ) -> BssOpenApi20171214Client: config = OpenAPI.create_api_models_config_by_assume_role(account_id, access_key_id, access_key_secret, role_name) # 访问的域名 config.endpoint = f'business.aliyuncs.com' return BssOpenApi20171214Client(config) @staticmethod def query_bss_resource_usage_detail( start_period: str, end_period: str, period_type: str, resource_type: str, client: BssOpenApi20171214Client, ): describe_resource_usage_detail_request = bss_open_api_20171214_models.DescribeResourceUsageDetailRequest( start_period=start_period, end_period=end_period, period_type=period_type, resource_type=resource_type, max_results=300 ) runtime = util_models.RuntimeOptions() try: resp = client.describe_resource_usage_detail_with_options(describe_resource_usage_detail_request, runtime) body_dict = resp.body.to_map() return body_dict['Data']['Items'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_idle_ri(rd_account_id, access_key_id, access_key_secret, role_name): client = IdleResourceSample.create_bss_client(rd_account_id, access_key_id, access_key_secret, role_name) # 2022-12-27 22:00:00 today_fixed_time = datetime.today().replace(hour=22, minute=0, second=0, microsecond=0) # 2022-12-24 20:00:00 - 2022-12-24 22:00:00 end_period = today_fixed_time + timedelta(days=-3) start_period = end_period + timedelta(hours=-2) ri_list = IdleResourceSample.query_bss_resource_usage_detail(start_period.strftime("%Y-%m-%d %H:%M:%S"), end_period.strftime("%Y-%m-%d %H:%M:%S"), 'HOUR', 'RI', client) # print(ri_list) id_set = set() idle_ri_list = [] usage_percentage_threshold = 0.1 output_str = '[Idle Reserved Instance Voucher] Region:All\n' for ri in ri_list: if ri['Status'] != 'Valid': continue ri_id = ri['ResourceInstanceId'] if ri_id not in id_set and ri['UsagePercentage'] < usage_percentage_threshold: # 查询时段内预留实例券利用率小于阈值，闲置 output_str += f' ' idle_ri_list.append(ri) id_set.add(ri_id) print(output_str + '\n') return idle_ri_list @staticmethod def create_alb_client( account_id: str, access_key_id: str, access_key_secret: str, role_name: str, region_id: str, ) -> Alb20200616Client: config = OpenAPI.create_api_models_config_by_assume_role(account_id, access_key_id, access_key_secret, role_name) config.endpoint = f'alb.{region_id}.aliyuncs.com' return Alb20200616Client(config) @staticmethod def query_alb_instance(client: Alb20200616Client): list_load_balancers_request = alb_20200616_models.ListLoadBalancersRequest( max_results=100 ) runtime = util_models.RuntimeOptions() try: resp = client.list_load_balancers_with_options(list_load_balancers_request, runtime) body_dict = resp.body.to_map() return body_dict['LoadBalancers'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_alb_listener( client: Alb20200616Client, load_balancer_ids: list, ): list_listeners_request = alb_20200616_models.ListListenersRequest( load_balancer_ids=load_balancer_ids, max_results=100 ) runtime = util_models.RuntimeOptions() try: resp = client.list_listeners_with_options(list_listeners_request, runtime) body_dict = resp.body.to_map() return body_dict['Listeners'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_alb_forward_rule( client: Alb20200616Client, load_balancer_ids: list, ): list_rules_request = alb_20200616_models.ListRulesRequest( load_balancer_ids=load_balancer_ids, max_results=100, ) runtime = util_models.RuntimeOptions() try: resp = client.list_rules_with_options(list_rules_request, runtime) body_dict = resp.body.to_map() return body_dict['Rules'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_alb_server_group( client: Alb20200616Client, server_group_ids: list, ): list_server_groups_request = alb_20200616_models.ListServerGroupsRequest( server_group_ids=server_group_ids, max_results=100, ) runtime = util_models.RuntimeOptions() try: resp = client.list_server_groups_with_options(list_server_groups_request, runtime) body_dict = resp.body.to_map() return body_dict['ServerGroups'] except Exception as error: # 如有需要，请打印 error UtilClient.assert_as_string(error) @staticmethod def query_idle_alb(account_id, access_key_id, access_key_secret, role_name): idle_resource_list = [] print('[Idle ALB] AccountId: ' + account_id) for region_id in region_id_list: output_str = 'Region:' + region_id + ' ' client = IdleResourceSample.create_alb_client(account_id, access_key_id, access_key_secret, role_name, region_id) result_list = IdleResourceSample.query_alb_instance(client) for alb in result_list: alb_id = alb['LoadBalancerId'] # print(alb_id) # 查询监听器 listeners = IdleResourceSample.query_alb_listener(client, [alb_id]) # print(listeners) if len(listeners) == 0: # 无监听器，闲置 output_str += f' ' idle_resource_list.append(alb) continue all_listener_stopped = True # 转发服务器组ID列表 forward_server_group_ids = [] for listener in listeners: # 默认转发规则的服务器组ID forward_server_group_ids.append( listener['DefaultActions'][0]['ForwardGroupConfig']['ServerGroupTuples'][0]['ServerGroupId']) if listener['ListenerStatus'] != 'Stopped': all_listener_stopped = False break if all_listener_stopped: # 所有监听器停止，闲置 output_str += f' ' idle_resource_list.append(alb) continue # 查询转发规则 rules = IdleResourceSample.query_alb_forward_rule(client, [alb_id]) for rule in rules: # 添加转发规则中的服务器组ID forward_server_group_ids.append( rule['RuleActions'][0]['ForwardGroupConfig']['ServerGroupTuples'][0]['ServerGroupId']) # 查询服务器组信息 server_groups = IdleResourceSample.query_alb_server_group(client, list(set(forward_server_group_ids))) all_server_group_is_empty = True for server_group in server_groups: if server_group['ServerCount'] > 0: all_server_group_is_empty = False break if all_server_group_is_empty: # 所有服务器组都无后端服务器，闲置 output_str += f' ' idle_resource_list.append(alb) print(output_str) print() return idle_resource_list @staticmethod def create_vpc_client( account_id, access_key_id, access_key_secret, role_name ) -> Vpc20160428Client: config = OpenAPI.create_api_models_config_by_assume_role(account_id, access_key_id, access_key_secret, role_name) config.endpoint = f'vpc.aliyuncs.com' return Vpc20160428Client(config) @staticmethod def query_common_bandwidth_package( client: Vpc20160428Client, region_id: str ): describe_common_bandwidth_packages_request = vpc_20160428_models.DescribeCommonBandwidthPackagesRequest( region_id=region_id, page_size=50, page_number=1 ) runtime = util_models.RuntimeOptions() try: resp = client.describe_common_bandwidth_packages_with_options(describe_common_bandwidth_packages_request, runtime) body_dict = resp.body.to_map() return body_dict['CommonBandwidthPackages']['CommonBandwidthPackage'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_idle_common_bandwidth_package(account_id, access_key_id, access_key_secret, role_name): client = IdleResourceSample.create_vpc_client(account_id, access_key_id, access_key_secret, role_name) idle_resource_list = [] print('[Idle Common Bandwidth Package] AccountId: ' + account_id) for region_id in region_id_list: output_str = 'Region:' + region_id + ' ' result_list = IdleResourceSample.query_common_bandwidth_package(client, region_id) # print(result_list) for cbp in result_list: if cbp['BusinessStatus'] != 'Normal': continue if cbp['PublicIpAddresses'] is None or len(cbp['PublicIpAddresses']['PublicIpAddresse']) == 0: output_str += f' ' idle_resource_list.append(cbp) print(output_str) print() return idle_resource_list @staticmethod def query_nat_gateway(client: Vpc20160428Client, region_id: str): describe_nat_gateways_request = vpc_20160428_models.DescribeNatGatewaysRequest( region_id=region_id, page_size=50, ) runtime = util_models.RuntimeOptions() try: resp = client.describe_nat_gateways_with_options(describe_nat_gateways_request, runtime) body_dict = resp.body.to_map() # print(body_dict) return body_dict['NatGateways']['NatGateway'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_nat_gateway_snat_entry(client: Vpc20160428Client, region_id: str, nat_gateway_id: str): describe_snat_table_entries_request = vpc_20160428_models.DescribeSnatTableEntriesRequest( nat_gateway_id=nat_gateway_id, region_id=region_id ) runtime = util_models.RuntimeOptions() try: resp = client.describe_snat_table_entries_with_options(describe_snat_table_entries_request, runtime) body_dict = resp.body.to_map() # print(body_dict) return body_dict['SnatTableEntries']['SnatTableEntry'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_nat_gateway_dnat_entry(client: Vpc20160428Client, region_id: str, nat_gateway_id: str): describe_forward_table_entries_request = vpc_20160428_models.DescribeForwardTableEntriesRequest( region_id=region_id, nat_gateway_id=nat_gateway_id ) runtime = util_models.RuntimeOptions() try: resp = client.describe_forward_table_entries_with_options(describe_forward_table_entries_request, runtime) body_dict = resp.body.to_map() # print(body_dict) return body_dict['ForwardTableEntries']['ForwardTableEntry'] except Exception as error: UtilClient.assert_as_string(error) @staticmethod def query_idle_nat_gateway(account_id, access_key_id, access_key_secret, role_name): client = IdleResourceSample.create_vpc_client(account_id, access_key_id, access_key_secret, role_name) idle_resource_list = [] print('[Idle NAT Gateway] AccountId: ' + account_id) for region_id in region_id_list: output_str = 'Region:' + region_id + ' ' result_list = IdleResourceSample.query_nat_gateway(client, region_id) # print(eip_list) for nat_gateway in result_list: if nat_gateway['NetworkType'] == 'internet' and len(nat_gateway['IpLists']['IpList']) == 0: # 公网NAT网关未绑定EIP，闲置 output_str += f' ' idle_resource_list.append(nat_gateway) continue snat_entries = IdleResourceSample.query_nat_gateway_snat_entry(client, region_id, nat_gateway['NatGatewayId']) if len(snat_entries) != 0: continue else: dnat_entries = IdleResourceSample.query_nat_gateway_dnat_entry(client, region_id, nat_gateway['NatGatewayId']) if len(dnat_entries) == 0: # 无SNAT及DNAT条目，闲置 output_str += f' ' idle_resource_list.append(nat_gateway) print(output_str) print() return idle_resource_list @staticmethod def query_eip(client: Vpc20160428Client, region_id: str): describe_eip_addresses_request = vpc_20160428_models.DescribeEipAddressesRequest( region_id=region_id, page_size=100, page_number=1 ) runtime = util_models.RuntimeOptions() try: resp = client.describe_eip_addresses_with_options(describe_eip_addresses_request, runtime) body_dict = resp.body.to_map() # print(body_dict) return body_dict['EipAddresses']['EipAddress'] except Exception as error: # 如有需要，请打印 error UtilClient.assert_as_string(error) @staticmethod def query_idle_eip(account_id, access_key_id, access_key_secret, role_name): client = IdleResourceSample.create_vpc_client(account_id, access_key_id, access_key_secret, role_name) idle_eip_list = [] print('[Idle EIP] AccountId: ' + account_id) for region_id in region_id_list: output_str = 'Region:' + region_id + ' ' eip_list = IdleResourceSample.query_eip(client, region_id) # print(eip_list) for eip in eip_list: bind_instance_id = str(eip['InstanceId']) if len(bind_instance_id) == 0 or len(bind_instance_id.strip()) == 0: output_str += f' ' idle_eip_list.append(eip) print(output_str) print() return idle_eip_list @staticmethod def query_idle_resource(): rd_member_account_rd_account_access_role = 'ResourceDirectoryAccountAccessRole' # EIP try: IdleResourceSample.query_idle_eip(rd_management_account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_management_account_readonly_role_name) except Exception as error: print(f'AccountId {rd_management_account_id} query exception:{repr(error)}') for account_id in rd_member_account_id_list: try: IdleResourceSample.query_idle_eip(account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_member_account_rd_account_access_role) except Exception as error: # traceback.print_exc() print(f'AccountId {account_id} query exception:{repr(error)}') # 共享带宽包 try: IdleResourceSample.query_idle_common_bandwidth_package(rd_management_account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_management_account_readonly_role_name) except Exception as error: print(f'AccountId {rd_management_account_id} query exception:{repr(error)}') for account_id in rd_member_account_id_list: try: IdleResourceSample.query_idle_common_bandwidth_package(account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_member_account_rd_account_access_role) except Exception as error: # traceback.print_exc() print(f'AccountId {account_id} query exception:{repr(error)}') # ALB try: IdleResourceSample.query_idle_alb(rd_management_account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_management_account_readonly_role_name) except Exception as error: print(f'AccountId {rd_management_account_id} query exception:{repr(error)}') for account_id in rd_member_account_id_list: try: IdleResourceSample.query_idle_alb(account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_member_account_rd_account_access_role) except Exception as error: print(f'AccountId {account_id} query exception:{repr(error)}') # NAT网关 try: IdleResourceSample.query_idle_nat_gateway(rd_management_account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_management_account_readonly_role_name) except Exception as error: print(f'AccountId {rd_management_account_id} query exception:{repr(error)}') for account_id in rd_member_account_id_list: try: IdleResourceSample.query_idle_nat_gateway(account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_member_account_rd_account_access_role) except Exception as error: print(f'AccountId {account_id} query exception:{repr(error)}') # ECS云盘 try: IdleResourceSample.query_idle_ecs_disk(rd_management_account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_management_account_readonly_role_name) except Exception as error: print(f'AccountId {rd_management_account_id} query exception:{repr(error)}') for account_id in rd_member_account_id_list: try: IdleResourceSample.query_idle_ecs_disk(account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_member_account_rd_account_access_role) except Exception as error: print(f'AccountId {account_id} query exception:{repr(error)}') # 预留实例券RI try: IdleResourceSample.query_idle_ri(rd_management_account_id, rd_management_account_access_key_id, rd_management_account_access_key_secret, rd_management_account_readonly_role_name) except Exception as error: print(f'AccountId {rd_management_account_id} query exception:{repr(error)}') if __name__ == '__main__': IdleResourceSample.query_idle_resource()