import copy import datetime import hashlib import json import os import re import pandas as pd from dateutil.tz import tzutc from google.cloud import bigquery from pandas._libs.missing import NAType from analyst.libs import common_cloud_resource_analyzer, metadata, util from analyst.libs.common_cloud_resource_analyzer import parse_env_region_by_3layer_k8s_name from analyst.libs.consts import UsedBy from analyst.resource_leak.settings import AWS_ASSET_SNAPSHOT_ANALYSIS_RESULT_FILE, ASSET_FILE_DIR AWS_EC2_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_ec2_instances.pkl') AWS_EBS_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_ec2_ebs_volumes.pkl') AWS_ELB_V1_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_elbv1_load_balancers.pkl') AWS_ELB_V2_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_elbv2_load_balancers.pkl') AWS_NAT_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_ec2_nat_gateways.pkl') AWS_ASG_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_autoscaling_groups.pkl') AWS_NODE_GROUP_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_eks_cluster_node_groups.pkl') AWS_EKS_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_eks_clusters.pkl') AWS_VPC_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_ec2_vpcs.pkl') AWS_SG_PKL_FILE = os.path.join(ASSET_FILE_DIR, 'aws_ec2_security_groups.pkl') # StakeHolder_type: ['tenant','cluster'] # vpc_type: ['base','seed','shoot'] # ebs usage: ['etcd','monitoring'] result_header = { # time 'datetime': 0, 'created_at': 0, 'hold_time': 0, # aws/gcp cloud resource control handle 'billing_key': 'null', 'region': 'null', 'account_id': 'null', 'id': 'null', 'name': 'null', # resource dashboard 'service_type': 'null', 'usedby': 'null', 'network_status': 'null', 'vpc_name': 'null', 'k8s': 'null', # because it's always be recorded in tag 'network_namespace': 'null', 'network_name': 'null', 'cluster_status': 'null', 'product': 'null', 'component': 'null', # tidbcloud label standard 'env': 'null', 'tenant_id': 'null', 'project_id': 'null', 'cluster_id': 'null', 'member': '', # 'component': 'null', # 'usedby': 'null', # other resource attribute 'vpc_type': 'null', 'shoot_id': 'null', # for eks/gke network, shoot_id isn't important 'vpc_id': 'null', 'ebs_size': 0, 'ec2_instance_type': 'null', 'ec2_instance_state': 'null', 'ec2_wg_id': 'null', 'ebs_pvc_name': "null", 'ebs_state': 'null', 'ebs_real_k8s': 'null', 'ebs_ec2_mount': 'null', 'elb_service_name': 'null', 'nat_public_ip': 'null', 'eip_allocation_id': '', # tags for cross resource analysis 'tags': 'null', 'extra_info': {}, } result_header['datetime'] = int( datetime.datetime.now().replace(tzinfo=tzutc()).strftime('%s')) # Analysis class DatetimeEncoder(json.JSONEncoder): def default(self, obj): try: return super().default(obj) except TypeError: return str(obj) def convert_cloudquery_tags_to_aws_tags(tags): tags = json.loads(tags) aws_tags = [] for item in tags.items(): aws_tags.append({'Key': item[0], 'Value': item[-1]}) return aws_tags def convert_cloudquery_tags_to_tag_dict(tags): tags = json.loads(tags) tag_dict = {} for item in tags.items(): tag_dict[item[0]] = item[-1] return tag_dict def extract_information_in_ec2(vpc_list, md: metadata.Metadata): ec2_results = [] for _, instance in pd.read_pickle(AWS_EC2_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'ec2', 'region': instance['region'], 'billing_key': instance['arn'], 'ec2_instance_type': instance['instance_type'], 'id': instance['instance_id'], # for example, ip-172-30-29-161.ec2.internal 'hostname': instance['private_dns_name'], 'account_id': instance['account_id'], 'tags': instance['tags'], }) result['ec2_instance_state'] = json.loads(instance['state'])['Name'] if result['ec2_instance_state'] not in ['terminated', 'shutting-down']: result['vpc_id'] = instance['vpc_id'] for item in list(filter(lambda item: item['id'] == result['vpc_id'], vpc_list)): result['vpc_name'] = item['name'] # launch time result['created_at'] = int( instance['launch_time'].to_pydatetime().timestamp()) delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - instance['launch_time'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 # tags for tag in convert_cloudquery_tags_to_aws_tags(instance['tags']): if tag['Key'] == 'Name': result['name'] = tag['Value'] if 'kubernetes.io/cluster/' in tag['Key']: result['k8s'] = tag['Key'].replace( 'kubernetes.io/cluster/', '') if tag['Key'] == 'gardener.wg.prefix.name': wg_prefix_name = tag['Value'].split('-') if 'tidb-lightning' not in tag['Value']: try: result['component'] = wg_prefix_name[1] result['cluster_id'] = wg_prefix_name[0].replace( 'tidb', '') except Exception as e: print(e) print(wg_prefix_name) else: result['component'] = 'tidb-lightning' if tag['Key'] == 'tenant': result['tenant_id'] = tag['Value'] if tag['Key'] == 'project' and tag['Value'].isnumeric(): result['project_id'] = tag['Value'] if tag['Key'] == 'cluster' and tag['Value'] != '': result['cluster_id'] = tag['Value'] if tag['Key'] == 'env': result['environment'] = tag['Value'] if tag['Key'] == 'worker.gardener.cloud/pool': result['ec2_wg_id'] = tag['Value'] if tag['Key'] == 'servicetype': result['service_type'] = tag['Value'] if tag['Key'] == 'usedby': result['usedby'] = tag['Value'] if tag['Key'] == 'usedby' and tag['Value'] == 'metadb': result['component'] = tag['Value'] common_cloud_resource_analyzer.analyze_by_pod_node_relation(result, result['hostname'], md) ec2_results.append(result) return ec2_results def extract_information_in_ebs(pvc_lists, ec2_list, md: metadata.Metadata): instance_id2ec2 = {} for each in ec2_list: instance_id2ec2[each['id']] = each ebs_results = [] for _, volume in pd.read_pickle(AWS_EBS_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'ebs', 'region': volume['region'], 'account_id': volume['account_id'], 'tags': volume['tags'], 'id': volume['volume_id'], 'billing_key': volume['arn'], 'ebs_type': volume['volume_type'], 'ebs_state': volume['state'], 'ebs_size': volume['size'], 'ebs_iops': volume['iops'], 'ebs_throughput': volume['throughput'], 'created_at': int(volume['create_time'].to_pydatetime().timestamp()), }) if isinstance(result['ebs_throughput'], NAType): result['ebs_throughput'] = '' if isinstance(result['ebs_iops'], NAType): result['ebs_iops'] = '' # k8s analysis for attch_status in json.loads(volume['attachments']): result['ebs_ec2_mount'] = attch_status['InstanceId'] for item in list(filter(lambda item: item['id'] == attch_status['InstanceId'], ec2_list)): result['vpc_name'] = item['vpc_name'] delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - volume['create_time'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 # tags volume_tags = convert_cloudquery_tags_to_aws_tags(volume['tags']) if volume['tags'] == '{}' and result['ebs_ec2_mount'] != 'null': for item in list(filter(lambda item: item['id'] == result['ebs_ec2_mount'], ec2_list)): volume_tags = convert_cloudquery_tags_to_aws_tags(item['tags']) for tag in volume_tags: if tag['Key'] == 'Name': result['name'] = tag['Value'] if 'kubernetes.io/cluster/' in tag['Key'] and result['k8s'] == 'null': result['k8s'] = tag['Key'].replace( 'kubernetes.io/cluster/', '') if tag['Key'] == 'kubernetes.io/created-for/pvc/name': result['ebs_pvc_name'] = tag['Value'] if tag['Key'] == 'kubernetes.io/created-for/pvc/namespace': result['ebs_pvc_namespace'] = tag['Value'] if 'shoot' in tag['Value']: result['ebs_real_k8s'] = result['k8s'] result['k8s'] = tag['Value'] if 'tidb' in tag['Value'] and tag['Value'] != 'tidb-admin': result['cluster_id'] = tag['Value'].replace('tidb', '') if 'tiflow-' in tag['Value'] and result['k8s'] == '': result['k8s'] = tag['Value'].replace('tiflow-', '') if tag['Key'] == 'worker.gardener.cloud/pool': result['ec2_wg_id'] = tag['Value'] # TBC if tag['Key'] == 'gardener.wg.prefix.name': wg_prefix_name = tag['Value'].split('-') try: result['component'] = wg_prefix_name[1] result['cluster_id'] = wg_prefix_name[0].replace( 'tidb', '') except Exception as e: print(e) if tag['Key'] == 'tenant': result['tenant_id'] = tag['Value'] if tag['Key'] == 'project': result['project_id'] = tag['Value'] if tag['Key'] == 'cluster' and tag['Value'].isnumeric(): result['cluster_id'] = tag['Value'] if tag['Key'] == 'worker.gardener.cloud/pool': result['ec2_wg_id'] = tag['Value'] if tag['Key'] == 'CSIVolumeName': volumename = tag['Value'] if volumename in pvc_lists.keys(): result['ebs_pvc_name'] = pvc_lists[volumename]['pvc_name'] pvc_namespace = pvc_lists[volumename]['namespace'] if 'tidb' in pvc_namespace and pvc_namespace != 'tidb-admin': result['cluster_id'] = pvc_namespace.replace( 'tidb', '') # else: # print('pvc not found, volumeid: %s' % # volumename) tag_values = tag['Value'].split('--') if len(tag_values) > 1: result['env'] = tag_values[1] result['component'] = 'csi' if tag_values[0] == 'pv-shoot': result['vpc_type'] = 'shoot' result['shoot_id'] = tag_values[2].split('-')[0] result['service_type'] = 'dedicated' result['k8s'] = "shoot--%s--%s" % ( result['env'], result['shoot_id']) if tag['Key'] == 'servicetype': result['service_type'] = tag['Value'] if tag['Key'] == 'component' and tag['Value'] == 'dbaas-control-plane-base': result['vpc_type'] = 'base' if tag['Key'] == 'component' and tag['Value'] == 'dbaas-control-plane-shot': result['usedby'] = 'dbaas-control-plane-shot' if tag['Key'] == 'usedby' and '-f0' in tag['Value']: result['k8s'] = tag['Value'] result['vpc_type'] = 'serverless' result['service_type'] = 'serverless' elif tag['Key'] == 'usedby' and tag['Value'] in ['dbaas-control-plane-shot', 'customer-tidb']: result['usedby'] = tag['Value'] if result['k8s'] == 'null': for ec2 in ec2_list: if ec2['id'] == result['ebs_ec2_mount']: result['k8s'] = ec2['k8s'] if ec2['component'] == 'bastion': result.update({ 'component': 'bastion', 'usedby': 'dbaas-control-plane-base', 'service_type': 'dedicated', 'tenant_id': 'admin', 'project_id': 'admin', 'cluster_id': 'admin' }) # sync info from PVC util.set_member_name_from_pvc_name(result, result.get('ebs_pvc_name', '')) util.set_component_from_pvc_name(result, result.get('ebs_pvc_name', '')) util.set_cluster_id_from_tc_namespace(result, result.get('ebs_pvc_namespace', '')) # sync info from node node = instance_id2ec2.get(result['ebs_ec2_mount']) if node: common_cloud_resource_analyzer.sync_from_node_to_ebs(result, node) ebs_results.append(result) return ebs_results def extract_information_in_elb(vpc_result): elb_results = [] # v1 for _, lb in pd.read_pickle(AWS_ELB_V1_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'clb', 'region': lb['region'], 'id': lb['load_balancer_name'], 'billing_key': lb['arn'], 'account_id': lb['account_id'], 'created_at': int(lb['created_time'].to_pydatetime().timestamp()), 'tags': lb['tags'], 'vpc_id': lb['vpc_id'], }) delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - lb['created_time'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 for item in list(filter(lambda item: item['id'] == lb['vpc_id'], vpc_result)): result['vpc_name'] = item['name'] result['k8s'] = item['name'] for tag in convert_cloudquery_tags_to_aws_tags(lb['tags']): if 'kubernetes.io/cluster/' in tag['Key']: result['k8s'] = tag['Key'].replace( 'kubernetes.io/cluster/', '') if tag['Key'] == 'servicetype': result['service_type'] = tag['Value'] if tag['Key'] == "kubernetes.io/service-name": result['elb_service_name'] = tag['Value'] if 'ingress.k8s.aws/stack' == tag['Key']: result['elb_service_name'] = tag['Value'] if tag['Key'] == 'app': result['name'] = tag['Value'] if tag['Key'] == 'elbv2.k8s.aws/cluster': result['k8s'] = tag['Value'] tag_dict = convert_cloudquery_tags_to_tag_dict(lb['tags']) result['usedby'] = tag_dict.get('usedby', 'dbaas-control-plane-shoot') result['tenant_id'] = tag_dict.get('tenant', 'admin') result['project_id'] = tag_dict.get('project', 'admin') result['cluster_id'] = tag_dict.get('cluster', 'admin') elb_results.append(result) for _, lb in pd.read_pickle(AWS_ELB_V2_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'nlb', 'region': lb['region'], 'id': lb['load_balancer_name'], 'billing_key': lb['arn'], 'account_id': lb['account_id'], 'created_at': int(lb['created_time'].to_pydatetime().timestamp()), 'tags': lb['tags'], 'vpc_id': lb['vpc_id'], }) delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - lb['created_time'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 for item in list(filter(lambda item: item['id'] == lb['vpc_id'], vpc_result)): result['vpc_name'] = item['name'] result['k8s'] = item['name'] for tag in convert_cloudquery_tags_to_aws_tags(lb['tags']): if 'kubernetes.io/cluster/' in tag['Key']: result['k8s'] = tag['Key'].replace( 'kubernetes.io/cluster/', '') if tag['Key'] == "kubernetes.io/service-name": result['elb_service_name'] = tag['Value'] if 'elbv2.k8s.aws/cluster' == tag['Key']: result['k8s'] = tag['Value'] if 'ingress.k8s.aws/stack' == tag['Key']: result['elb_service_name'] = tag['Value'] if tag['Key'] == 'servicetype': result['service_type'] = tag['Value'] tag_dict = convert_cloudquery_tags_to_tag_dict(lb['tags']) result['usedby'] = tag_dict.get('usedby', 'dbaas-control-plane-shoot') result['tenant_id'] = tag_dict.get('tenant', 'admin') result['project_id'] = tag_dict.get('project', 'admin') result['cluster_id'] = tag_dict.get('cluster', 'admin') elb_results.append(result) return elb_results def extract_information_in_nat(vpc_result): nat_results = [] for _, nat in pd.read_pickle(AWS_NAT_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'nat', 'id': nat['nat_gateway_id'], 'region': nat['region'], 'billing_key': nat['arn'], 'account_id': nat['account_id'], 'tags': nat['tags'], 'vpc_id': nat['vpc_id'], }) try: for item in json.loads(nat['nat_gateway_addresses']): result['nat_public_ip'] = item.get('PublicIp', '') result['eip_allocation_id'] = item.get('AllocationId', '') except: pass # creation time result['created_at'] = int( nat['create_time'].to_pydatetime().timestamp()) delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - nat['create_time'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 for item in list(filter(lambda item: item['id'] == nat['vpc_id'], vpc_result)): result['vpc_name'] = item['name'] result['k8s'] = item['k8s'] nat_tags = convert_cloudquery_tags_to_aws_tags(nat['tags']) if nat['tags'] == '{}': for item in list(filter(lambda item: item['vpc_id'] == nat['vpc_id'], vpc_result)): nat_tags = convert_cloudquery_tags_to_aws_tags(item['tags']) for tag in nat_tags: if 'kubernetes.io/cluster/' in tag['Key']: result['k8s'] = tag['Key'].replace( 'kubernetes.io/cluster/', '') if tag['Key'] == "component": result['component'] = tag['Value'] if tag['Key'] == 'environment': result['env'] = tag['Value'] if tag['Key'] == 'servicetype': result['service_type'] = tag['Value'] # temp resources if tag['Key'] == 'Name': result['name'] = tag['Value'] if tag['Key'] == 'app' and tag['Value'] == 'jira': result['component'] = 'jira' result['tenant_id'] = 'admin' result['project_id'] = 'admin' result['cluster_id'] = 'admin' result['usedby'] = 'dbaas-control-plane-base' result['service_type'] = 'dedicated' if tag['Key'] == 'project' and tag['Value'].isnumeric(): result['project_id'] = tag['Value'] nat_results.append(result) return nat_results # AutoScalingGroup def extract_information_in_asg(eks_result): name2eks = {each['name']: each for each in eks_result} asg_results = [] for _, asg in pd.read_pickle(AWS_ASG_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'asg', 'billing_key': asg['arn'], 'id': asg['auto_scaling_group_name'], 'name': asg['auto_scaling_group_name'], 'region': asg['region'], 'account_id': asg['account_id'], 'tags': asg['tags'], 'created_at': int(asg['created_time'].to_pydatetime().timestamp()), }) result['extra_info'].update({ 'min_size': asg['min_size'], 'max_size': asg['max_size'], 'desired_capacity': asg['desired_capacity'], }) delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - asg['created_time'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 tags = convert_cloudquery_tags_to_tag_dict(asg['tags']) if tags.get("eks:cluster-name"): result['k8s'] = tags["eks:cluster-name"] if tags.get("component"): result['component'] = tags['component'] if tags.get("environment"): result['env'] = tags['environment'] if tags.get("servicetype"): result['service_type'] = tags['servicetype'] # get info from eks k8s_name = result['k8s'] if k8s_name and name2eks.get(k8s_name): eks = name2eks[k8s_name] util.set_if_empty(result, 'service_type', eks['service_type']) util.set_if_empty(result, 'tenant_id', eks['tenant_id']) util.set_if_empty(result, 'project_id', eks['project_id']) util.set_if_empty(result, 'usedby', eks['usedby']) util.set_if_empty(result, 'env', eks['env']) util.set_if_empty(result, 'vpc_name', eks['vpc_name']) asg_results.append(result) return asg_results # AutoScalingGroup def extract_information_in_nodegroup(eks_result): name2eks = {each['name']: each for each in eks_result} results = [] for _, ng in pd.read_pickle(AWS_NODE_GROUP_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'node_group', 'billing_key': ng['arn'], 'id': ng['nodegroup_name'], 'name': ng['nodegroup_name'], 'region': ng['region'], 'account_id': ng['account_id'], 'k8s': ng['cluster_name'], 'tags': ng['tags'], 'created_at': int(ng['created_at'].to_pydatetime().timestamp()), }) # fill extra info result['extra_info'].update({ 'ami_type': ng['ami_type'], 'capacity_type': ng['capacity_type'], 'instance_types': [each for each in ng['instance_types']], 'version': ng['version'], }) # scaling_config: {"DesiredSize":3,"MaxSize":10,"MinSize":3} if ng['scaling_config']: try: sc = json.loads(ng['scaling_config']) result['extra_info'].update({ 'min_size': sc['MinSize'], 'max_size': sc['MaxSize'], 'desired_capacity': sc['DesiredSize'], }) except Exception as e: pass delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - ng['created_at'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 # get info from tags # {"Name":"tidbcloud-prod-1372813089454550833-eks-ap-northeast-1-8a4a4d36","Vendor":"PingCAP", # "project":"1372813089454550833","servicetype":"dedicated","tenant":"1372813089194211283", # "tidbcloud.com/nodegroup-name":"tidbcloud-prod-1372813089454550833-eks-ap-northeast-1-8a4a4d36", # "usedby":"dbaas-control-plane-shoot"} tags = convert_cloudquery_tags_to_tag_dict(ng['tags']) result['usedby'] = tags.get('usedby', '') result['component'] = tags.get('component', '') result['vpc_name'] = tags.get('vpc_name', '') result['tenant_id'] = tags.get('tenant', '') result['project_id'] = tags.get('project', '') result['cluster_id'] = tags.get('cluster', '') if result['usedby'] == 'dbaas-control-plane-shoot' and result['cluster_id'] == '': result['cluster_id'] = 'admin' result['service_type'] = tags.get('servicetype', '') result['env'] = tags.get('environment', '') # get info from eks k8s_name = result['k8s'] if k8s_name and name2eks.get(k8s_name): eks = name2eks[k8s_name] util.set_if_empty(result, 'service_type', eks['service_type']) util.set_if_empty(result, 'tenant_id', eks['tenant_id']) util.set_if_empty(result, 'project_id', eks['project_id']) util.set_if_empty(result, 'usedby', eks['usedby']) util.set_if_empty(result, 'env', eks['env']) util.set_if_empty(result, 'vpc_name', eks['vpc_name']) results.append(result) return results # def extract_information_in_s3(s3, region, account_id, network_list): # s3_results = [] # for bucket in s3: # if isinstance(bucket, str): # continue # result = copy.deepcopy(result_header) # result.update({ # 'product': 's3', # 'billing_key': bucket['billing_key'], # 'account_id': account_id, # }) # result['created_at'] = int(dateutil.parser.parse( # bucket['CreationDate']).strftime('%s')) # delta_time = datetime.datetime.now().replace( # tzinfo=tzutc()) - dateutil.parser.parse(bucket['CreationDate']) # result['hold_time'] = delta_time.days*24 + delta_time.seconds//3600 # name_keys = re.findall( # "dbaas-(.*)-(.*)-(.*)-(.*)-(\d+)", bucket['Name']) # if len(name_keys) > 0: # result['env'] = name_keys[0][0] # result['project_id'] = name_keys[0][-1] # result['region'] = '-'.join(list(name_keys[0][1:-1])) # result['env'] = result['env'].replace('nightly', 'dev') # if result['env'] in network_list: # env_network_list = network_list[result['env']] # for key, shoot in env_network_list.items(): # if shoot['project'] == result['project_id'] and shoot['region'] == result['region']: # result['tenant_id'] = shoot['tenant'] # result['shoot_id'] = key # result['network_existed'] = shoot['network_exist'] # result['attribute_network_status'] = shoot['hibernated'] # result['vpc_type'] = 'shoot' # break # s3_results.append(result) # return s3_results def extract_information_in_eks(vpc_result): eks_results = [] for _, cluster in pd.read_pickle(AWS_EKS_PKL_FILE, compression="gzip").iterrows(): if isinstance(cluster, str): continue tag_dict = convert_cloudquery_tags_to_tag_dict(cluster['tags']) result = copy.deepcopy(result_header) result.update({ 'product': 'eks', 'name': cluster['name'], 'region': cluster['region'], 'billing_key': cluster['arn'], 'account_id': cluster['account_id'], 'vpc_id': json.loads(cluster['resources_vpc_config'])['VpcId'], 'k8s': cluster['name'], 'env': tag_dict.get('environment', ''), 'tags': cluster['tags'], }) if not result['env']: result['env'] = cluster['name'].split('-')[0].replace('nightly', 'dev') for item in list( filter(lambda item: item['id'] == json.loads(cluster['resources_vpc_config'])['VpcId'], vpc_result)): result['vpc_name'] = item['name'] result['created_at'] = int( cluster['created_at'].to_pydatetime().timestamp()) delta_time = datetime.datetime.now().replace( tzinfo=tzutc()) - cluster['created_at'].to_pydatetime() result['hold_time'] = delta_time.days * 24 + delta_time.seconds // 3600 tags = convert_cloudquery_tags_to_tag_dict(cluster['tags']) result['usedby'] = tags.get('usedby', '') result['tenant_id'] = tags.get('tenant', '') result['project_id'] = tags.get('project', '') result['service_type'] = tags.get('servicetype', '') result['env'] = tags.get('environment', '') eks_results.append(result) return eks_results def extract_information_in_vpc(): vpc_results = [] for _, vpc in pd.read_pickle(AWS_VPC_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'vpc', 'region': vpc['region'], 'billing_key': vpc['vpc_id'], 'id': vpc['vpc_id'], 'vpc_id': vpc['vpc_id'], 'account_id': vpc['account_id'], 'tags': vpc['tags'], 'created_at': 0, # no VPC creation timestamp 'cidr_block': vpc['cidr_block'], }) for tag in convert_cloudquery_tags_to_aws_tags(vpc['tags']): if 'kubernetes.io/cluster/' in tag['Key']: result['k8s'] = tag['Key'].replace( 'kubernetes.io/cluster/', '') if tag['Key'] == "component": result['component'] = tag['Value'] if tag['Key'] == 'environment': result['env'] = tag['Value'] if tag['Key'] == 'servicetype': result['service_type'] = tag['Value'] if tag['Key'] == 'usedby': result['usedby'] = tag['Value'] # temp resources if tag['Key'] == 'Name': result['name'] = tag['Value'] result['vpc_name'] = tag['Value'] result['k8s'] = tag['Value'] if tag['Key'] == 'project' and tag['Value'].isnumeric(): result['project_id'] = tag['Value'] # amend info if result['usedby'] == 'gardener-shoot': result['usedby'] = UsedBy.DBaaSControlPlaneShoot vpc_results.append(result) return vpc_results # Security Group def extract_information_in_sg(): sg_results = [] for _, sg in pd.read_pickle(AWS_SG_PKL_FILE, compression="gzip").iterrows(): result = copy.deepcopy(result_header) result.update({ 'product': 'security_group', 'region': sg['region'], 'billing_key': sg['arn'], 'id': sg['group_id'], 'account_id': sg['account_id'], 'tags': sg['tags'], 'vpc_id': sg['vpc_id'], 'created_at': 0, # no VPC creation timestamp }) tag_dict = convert_cloudquery_tags_to_tag_dict(sg['tags']) result['service_type'] = tag_dict.get('servicetype', '') result['component'] = tag_dict.get('component', '') result['usedby'] = tag_dict.get('usedby', 'dbaas-control-plane-shoot') result['env'] = tag_dict.get('environment', '') result['tenant_id'] = tag_dict.get('tenant', 'admin') result['project_id'] = tag_dict.get('project', 'admin') result['cluster_id'] = tag_dict.get('cluster', 'admin') result['k8s'] = tag_dict.get('elbv2.k8s.aws/cluster', '') # add: get ks8 from eks cluster name or sg name, common analyse can then fill project/tenant info if result['k8s'] == '': result['k8s'] = tag_dict.get('aws:eks:cluster-name', '') if not re.match(r'(prod|staging|nightly-ms)', result['k8s']): result['k8s'] = '' if result['k8s'] == '': sg_name = tag_dict.get('Name', '') if re.match(r'(prod|staging|nightly-ms).*-[a-zA-Z0-9]{8}$', sg_name): result['k8s'] = sg_name sg_results.append(result) return sg_results def extract_information_for_common(resource, network_list, md: metadata.Metadata): # kube-apiserver if 'kube-apiserver' in resource['elb_service_name']: resource.update( { 'env': resource['elb_service_name'].split('--')[1], 'shoot_id': resource['elb_service_name'].split('--')[2].replace('/kube-apiserver', ''), 'vpc_type': 'shoot', 'usedby': 'dbaas-control-plane-shoot', 'cluster_id': 'admin', 'component': 'kube-apiserver', 'service_type': "dedicated", 'k8s': resource['elb_service_name'].split('/')[0] } ) if 'shoot' in resource['k8s']: resource.update( { 'env': resource['k8s'].split('--')[1].replace('nightly', 'dev'), 'shoot_id': resource['k8s'].split('--')[2], 'vpc_type': 'shoot', 'usedby': 'dbaas-control-plane-shoot', 'service_type': "dedicated", 'vpc_name': resource['k8s'] } ) # todo(shujia) Gardener's network CR can no longer be retrieved, therefore resource['shoot_id'] will inevitably fail to match. for network in network_list: # "vpc": "vpc-03cf886ca121aae4c" # "dataplane": "prod-1372813089188301282-eks-ap-northeast-1-14af0acb" if resource['shoot_id'] == network['shoot_name'] \ or resource['vpc_name'] == network.get('vpc') \ or resource['shoot_id'] == network.get('dataplane'): resource.update( { 'tenant_id': network['tenant'], 'project_id': network['project'], 'network_status': network['status'], 'network_name': network['network_name'], 'network_namespace': network['namespace'], } ) break # 3-layer structure elif parse_env_region_by_3layer_k8s_name(resource['k8s']) != None: resource['env'], resource['region'] = parse_env_region_by_3layer_k8s_name(resource['k8s']) resource.update({ 'usedby': "dbaas-control-plane-shoot", 'vpc_type': "shoot", 'service_type': "dedicated", }) if resource['project_id'] not in ['null', ''] and resource['project_id'].isnumeric(): resource['network_name'] = 'eks-' + resource['region'] resource['namespace'] = resource['env'] + '-' + resource['project_id'] for network in network_list: if (network['network_name'] == resource['network_name'] and network['namespace'] == resource['namespace']): resource.update( { 'tenant_id': network['tenant'], 'network_status': network['status'], } ) break # cloud-native resources elif re.search(r'(prod|dev|staging).*-[a-zA-Z0-9]{8}$', resource['k8s']): resource.update({ 'usedby': "dbaas-control-plane-shoot", 'vpc_type': "shoot", 'service_type': "dedicated", }) for network in network_list: if network['network_name'] in resource['k8s'] and network['namespace'] == \ resource['k8s'].split(network['network_name'])[0][:-1]: resource.update( { 'tenant_id': network['tenant'], 'project_id': network['project'], 'network_status': network['status'], 'network_name': network['network_name'], 'network_namespace': network['namespace'], } ) break # Serverless resources elif re.search(r'^(prod|staging|dev)-.*-f0\d+$', resource.get('k8s', '')): #print(resource['k8s']) resource.update({ 'service_type': 'serverless', 'usedby': 'serverless', 'vpc_type': 'shoot', }) elif 'nightly-ms-' in resource['k8s']: resource.update({ 'usedby': "dbaas-control-plane-shoot", 'env': 'dev', 'service_type': "dedicated", }) # todo(shujia): We need a better way to identify these invalid k8s names，currently keep to avoid excessive tag updates elif re.search("(.*)-(.*)-(?=.*[0-9])(?=.*[a-zA-Z])(.{8,})", resource['k8s']) != None: resource.update({ 'usedby': "dbaas-control-plane-shoot", 'vpc_type': "shoot", 'service_type': "dedicated", }) # customer-tidb if resource['cluster_id'] != 'null' and resource['cluster_id'] != 'admin': for network in network_list: if resource['cluster_id'] in network['cluster_list'].keys(): resource.update( { 'tenant_id': network['tenant'], 'project_id': network['project'], 'network_status': network['status'], 'network_name': network['network_name'], 'network_namespace': network['namespace'], 'cluster_status': network['cluster_list'][resource['cluster_id']], 'usedby': "customer-tidb", 'service_type': "dedicated", } ) # the following k8s_name build logic is not applicable to 3-layer architecture. if resource['vpc_name'] == 'null' and 'dedicated-' not in resource['k8s']: vpc_name_hash_id = hashlib.md5( ("%s/%s" % (network['namespace'], network['network_name'])).encode('utf-8')).hexdigest()[:8] vpc_name = "%s-%s-%s" % (network['namespace'], network['network_name'], vpc_name_hash_id) resource.update( { 'vpc_name': vpc_name, 'k8s': vpc_name, } ) break common_cloud_resource_analyzer.set_env_by_account(resource) if resource['component'] == 'Gardener-ShootMaster': resource['component'] = 'dbaas-control-plane-shoot' # tidbmonitor and slowlog should be after tidb, so that these labels can overwrite the wrong label for key in ['pd', 'tikv', 'tidb', 'tiflash', 'tidbmonitor', 'slowlog']: if key in resource['ebs_pvc_name']: resource['component'] = key if resource['cluster_id'] == 'null': resource['cluster_id'] = 'admin' # if 'null' in result['k8s']: # result['component'] = 'unmanaged' if 'tidb' in resource['elb_service_name'] and 'tidb-admin' not in resource['elb_service_name']: resource['cluster_id'] = resource['elb_service_name'].split( '/')[0].replace('tidb', '') resource['usedby'] = 'customer-tidb' if resource['name'] == 'null' and resource['id'] != 'null': resource['name'] = resource['id'] if resource['service_type'] == 'serverless': resource['usedby'] = "serverless" resource['env'] = resource['env'].replace('nightly', 'dev') common_cloud_resource_analyzer.update_by_keyword(resource) util.set_component_from_member_name(resource, resource['member']) common_cloud_resource_analyzer.sync_from_network_by_k8s_name(resource, resource['k8s'], md) # in aws using vpc_id (vpc-xxxxx) common_cloud_resource_analyzer.sync_from_network_by_vpc_name(resource, resource['vpc_id'], md) common_cloud_resource_analyzer.set_by_vpc_name(resource, md) if resource['vpc_type'] == 'shoot' and resource['component'] in ['pd', 'tidb', 'tikv', 'tiflash', 'ticdc', 'tidb-lightning']: resource['usedby'] = 'customer-tidb' if resource['vpc_type'] == 'base' and resource['component'] == 'null': resource['component'] = 'dbaas-control-plane-base' if resource['cluster_id'] == 'admin': resource['cluster_status'] = resource['network_status'] util.set_if_empty(resource, 'component', 'admin') if 'tiflash' not in resource['ebs_pvc_name'] and '-db' in resource['ebs_pvc_name']: resource['component'] = resource['ebs_pvc_name'].split('-')[0] # if 'null' in result['k8s']: # result['component'] = 'unmanaged' if resource['cluster_status'] == 'Normal': resource['cluster_status'] = 'Running' def analysis_resources_list(network_list, pvc_lists, md: metadata.Metadata): results = [] vpc_result = extract_information_in_vpc() results += vpc_result ec2_result = extract_information_in_ec2(vpc_result, md) results += ec2_result results += extract_information_in_ebs(pvc_lists, ec2_result, md) results += extract_information_in_elb(vpc_result) results += extract_information_in_nat(vpc_result) eks_result = extract_information_in_eks(vpc_result) results += eks_result results += extract_information_in_sg() results += extract_information_in_asg(eks_result) results += extract_information_in_nodegroup(eks_result) # results += extract_information_in_s3( # resource['s3'], resource['region'], resource['account'], network_list) for resource in results: extract_information_for_common(resource, network_list, md) return results def write_result_to_bigquery(result): # write to bigquery client = bigquery.Client() table_id = "pingcap-gardener.aws_cost_data_us_west1.aws_dev_resources_tree" errors = client.insert_rows_json(table_id, result) # Make an API request. if errors == []: print("New rows have been added.") else: print("Encountered errors while inserting rows: {}".format(errors)) def main(): md = metadata.build_metadata() if os.path.exists('new_network_result.json'): network_list = json.load( open('new_network_result.json', 'r')) else: network_list = {} if os.path.exists('kube_persistentvolumeclaim_info_asset_snapshot_analysis_result.json'): pvc_list = json.load( open('kube_persistentvolumeclaim_info_asset_snapshot_analysis_result.json', 'r')) else: pvc_list = {} results = analysis_resources_list(network_list, pvc_list, md) json.dump(results, fp=open( AWS_ASSET_SNAPSHOT_ANALYSIS_RESULT_FILE, 'w'), indent=4) if __name__ == "__main__": main()