#!/usr/bin/env python3 from absl import app from absl import flags import os import io import csv import argparse import logging import time import yaml from kubernetes import client, config, dynamic import warnings FLAGS = flags.FLAGS flags.DEFINE_string( 'kubeconfig', '', 'path to the kubeconfig file of the infra apiserver') flags.DEFINE_string('candidates_file', '', 'candidate networks to delete') flags.DEFINE_string('result_file', '', 'deletion result file') flags.DEFINE_bool('use_kubeconfig_in_candidate_file', True, 'If true, use kubeconfig specified in the candidates file') flags.DEFINE_bool( 'dryrun', True, 'If true, only print networks to delete, do noto take action') flags.DEFINE_integer('deletion_timeout_secs', 1800, 'Timeout waiting for deletions, in seconds') # https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#list-and-watch-filtering GCP_SELECTOR = 'cloud.pingcap.com/provider-type in (gcp,gke)' RECONCILER_LABEL = "cloud.pingcap.com/reconciler" def get_ns(obj): # every object should contain the metadata return obj["metadata"].get("namespace", "default") def get_name(obj): # every object should contain the metadata return obj["metadata"]["name"] def gen_net_cluster_map(clusters): """ gen_net_cluster_map generates the network to cluster map with the key is the namespaced name of the network and the value is the cluster object. """ return {( get_ns(cluster), cluster["spec"]["networkName"], ): cluster for cluster in clusters} def remove_backups(backup_cli, net): """remove_backups removes all backups under the same namespace of the net.""" # get all backups associated with the network net_ns = get_ns(net) bps = backup_cli.get(namespace=net_ns).items for bp in bps: bp_name = get_name(bp) try: backup_cli.delete(name=bp_name, namespace=net_ns, async_request=True) logging.info(f"bp({net_ns}/{bp_name}) is being deleted") except Exception as e: logging.error( f"failed to delete the vpcendpoint({net_ns}/{bp_name}):{e}") def remove_vpcendpoints(vpcendpoint_cli, net, vps): """remove_vpcendpoints removes all vpcendpoints associated with the net.""" net_vpc = net["spec"]["forProvider"]["vpcId"] for vp in vps: vp_name = get_name(vp) vp_vpc = vp["spec"]["forProvider"]["vpcId"] if vp_vpc == net_vpc: logging.info(f"vpcendpoint({vp_name}) is being deleted") try: if vp_vpc == net_vpc: vpcendpoint_cli.delete(name=vp_name, async_request=True) logging.info(f"vpcendpoint({vp_name}) is being deleted") except Exception as e: logging.error( f"failed to delete the vpcendpoint({vp_name}):{e}") def remove_networkpeers(networkpeer_cli, net): """remove_networkpeers removes all network peers associated with the net.""" net_ns, net_name = get_ns(net), get_name(net) nps = networkpeer_cli.get(namespace=net_ns).items for np in nps: np_name = get_name(np) # The vpc peering request will be initialized from the TiDB cloud side source = np["spec"]["source"] if source == net_name: try: networkpeer_cli.delete( name=np_name, namespace=net_ns, async_request=True, ) logging.info( f"networkpeer({net_ns}/{np_name}) is being deleted") except Exception as e: logging.error( f"failed to delete the networkpeer({net_ns}/{np_name}):{e}" ) def remove_reconciler_label(net_cli, net): """remove_reconciler_label removes the reconciler label if exist.""" if "labels" not in net["metadata"].keys(): # return immediately if the network does not contain any label. return old_rcsl = net["metadata"]["labels"].pop(RECONCILER_LABEL, None) net_ns, net_name = get_ns(net), get_name(net) if not old_rcsl: # return immediately if the reconciler label does not exist. return logging.info( f"change reconciler({old_rcsl} -> nightly-eks-provider) for network({net_ns}/{net_name})" ) body = {"metadata": {"labels": {RECONCILER_LABEL: None}}} net_cli.patch( body, namespace=net_ns, name=net_name, content_type="application/merge-patch+json", ) def remove_network(net_cli, net): """ remove_network removes the network using dynamic client. see https://github.com/kubernetes-client/python-base/blob/3815cdaca8857931a275460f68593b9f4cb78813/dynamic/client.py#L212 for the full list of args. """ net_ns, net_name = get_ns(net), get_name(net) # to ensure the nightly-eks-provider will take over the network object, # remove the reconciler label. try: remove_reconciler_label(net_cli, net) except Exception as e: logging.error( f"failed to change the reconciler of the network({net_ns}/{net_name}): {e}" ) return # delete the network object try: net_cli.delete(name=net_name, namespace=net_ns, async_request=True) logging.info(f"network({net_ns}/{net_name}) is being deleted") except Exception as e: # capture the exception, as we want to remove other network logging.error( f"failed to delete the network({net_ns}/{net_name}): {e}") def cleanup(kubeconfig, rows, dryrun, deletion_timeout_secs): """main is the main entry of the script.""" config.load_kube_config(kubeconfig) cfg = client.Configuration.get_default_copy() cfg.verify_ssl = False dyn_cli = dynamic.DynamicClient(client.ApiClient(configuration=cfg)) net_cli = dyn_cli.resources.get(api_version="cloud.pingcap.com/v1", kind="Network") backup_cli = dyn_cli.resources.get(api_version="cloud.pingcap.com/v1", kind="Backup") cluster_cli = dyn_cli.resources.get(api_version="cloud.pingcap.com/v1", kind="Cluster") networkpeer_cli = dyn_cli.resources.get(api_version="cloud.pingcap.com/v1", kind="NetworkPeer") vpcendpoint_cli = dyn_cli.resources.get( api_version="ec2.aws.crossplane.io/v1alpha1", kind="VPCEndpoint") # get the vpc endpoint list vpe_lst = vpcendpoint_cli.get().items clusters = cluster_cli.get(label_selector=GCP_SELECTOR).items nets = net_cli.get(label_selector=GCP_SELECTOR).items nets_dict = {} for net in nets: nets_dict[(get_ns(net), get_name(net))] = net net_cluster_dict = gen_net_cluster_map(clusters) for row in rows: if row['provider_type'] not in ['gcp', 'gke']: row['result'] = 'IGNORED_BY_PROVIDER_TYPE' continue key = (row['namespace'], row['name']) if key in net_cluster_dict: row['result'] = 'SKIP_NON_EMPTY' continue if key not in nets_dict: row['result'] = 'NOT_FOUND' continue rn = nets_dict[key] if dryrun: row['result'] = 'DRYRUN_TO_DELETE' else: row['result'] = 'DELETING' remove_backups(backup_cli, rn) remove_networkpeers(networkpeer_cli, rn) remove_vpcendpoints(vpcendpoint_cli, rn, vpe_lst) remove_network(net_cli, rn) now = time.time() while time.time() - now < deletion_timeout_secs: deleting_networks = {} for row in rows: if row['result'] == 'DELETING': key = (row['namespace'], row['name']) deleting_networks[key] = row if not deleting_networks: break existing_networks = {} nets = net_cli.get(label_selector=GCP_SELECTOR).items for n in nets: existing_networks[(get_ns(n), get_name(n))] = n for key, row in deleting_networks.items(): if key not in existing_networks: row['result'] = 'DELETED' logging.info(f'{key} deleted.') else: row['phase'] = n.status.phase logging.info(f'{key} still exists, phase={n.status.phase}') time.sleep(10) for row in rows: if row['result'] == 'DELETING': row['result'] = 'DELETION_FAILED_TIMEOUT' def main(argv): if not FLAGS.use_kubeconfig_in_candidate_file and not FLAGS.kubeconfig: print( 'if use_kubeconfig_in_candidate_file is false, kubeconfig has to be given.' ) os.exit(1) csv_reader = csv.DictReader(open(FLAGS.candidates_file)) records_by_kubeconfig = {} fieldnames = None for row in csv_reader: if not fieldnames: fieldnames = list(row.keys()) if FLAGS.use_kubeconfig_in_candidate_file: kc = row['kubeconfig_file'] else: kc = FLAGS.kubeconfig records_by_kubeconfig.setdefault(kc, []).append(row) final_result = [] for kubeconfig, rows in records_by_kubeconfig.items(): cleanup(kubeconfig, rows, FLAGS.dryrun, FLAGS.deletion_timeout_secs) final_result.extend(rows) if not final_result: return # Writes result file output = io.StringIO() if 'result' not in fieldnames: fieldnames.append('result') writer = csv.DictWriter(output, fieldnames=fieldnames) writer.writeheader() for row in final_result: writer.writerow(row) if FLAGS.result_file: with open(FLAGS.result_file, 'w') as fp: fp.write(output.getvalue()) print(output.getvalue()) if __name__ == "__main__": warnings.simplefilter("ignore") logging.getLogger().setLevel(logging.INFO) app.run(main)