如何通过Python Kubernetes Client获取所有CR以实现批量删除？

核心思路 #

要实现kubectl get crd -A的等效功能，需先通过ApiextensionsV1Api获取所有自定义资源定义（CRD），再根据每个CRD的元数据，用CustomObjectsApi遍历获取对应的*自定义资源（CR）*实例，最后执行删除操作。

代码实现 #

from kubernetes import client, config

# 加载K8s配置（本地开发用kubeconfig，集群内运行用服务账号）
config.load_kube_config()  # 集群内环境可替换为 config.load_incluster_config()

# 初始化API客户端
apiextensions_api = client.ApiextensionsV1Api()
custom_objects_api = client.CustomObjectsApi()

def delete_all_custom_resources():
    # 1. 获取所有CRD（完全等效于kubectl get crd -A）
    crds = apiextensions_api.list_custom_resource_definition().items
    
    for crd in crds:
        # 提取CRD核心标识：组、版本、复数名称
        group = crd.spec.group
        version = crd.spec.versions[0].name  # 默认取第一个版本，可按需筛选特定版本
        plural = crd.spec.names.plural
        
        print(f"开始处理CRD: {plural}.{group}/{version}")
        
        try:
            # 2. 获取该CRD下的所有CR实例（集群范围）
            cr_list = custom_objects_api.list_cluster_custom_object(
                group=group,
                version=version,
                plural=plural,
                watch=False
            )
            
            # 3. 遍历删除每个CR实例
            for cr in cr_list.get('items', []):
                cr_metadata = cr.get('metadata', {})
                namespace = cr_metadata.get('namespace')
                cr_name = cr_metadata['name']
                
                if namespace:
                    # 处理命名空间级CR
                    custom_objects_api.delete_namespaced_custom_object(
                        group=group,
                        version=version,
                        namespace=namespace,
                        plural=plural,
                        name=cr_name,
                        # 可选：设置删除传播策略，确保依赖资源先被清理
                        body=client.V1DeleteOptions(propagation_policy='Foreground')
                    )
                    print(f"已删除命名空间[{namespace}]中的CR: {cr_name}")
                else:
                    # 处理集群级CR
                    custom_objects_api.delete_cluster_custom_object(
                        group=group,
                        version=version,
                        plural=plural,
                        name=cr_name,
                        body=client.V1DeleteOptions(propagation_policy='Foreground')
                    )
                    print(f"已删除集群级CR: {cr_name}")
                    
        except client.exceptions.ApiException as e:
            print(f"处理CRD {plural}.{group}时出错: {e.reason} (状态码: {e.status})")

if __name__ == "__main__":
    delete_all_custom_resources()

关键细节说明 #

• CRD获取逻辑：list_custom_resource_definition()方法直接返回集群内所有CRD，无需提前知晓任何CRD的具体信息，完全替代kubectl get crd -A的功能。
• 资源类型区分：通过CR实例的metadata.namespace字段判断是命名空间级还是集群级资源，调用对应的删除方法。
• 异常处理：捕获ApiException可精准处理权限不足、资源不存在等API调用异常，比子进程调用kubectl的错误处理更灵活可控。
• 版本适配：若CRD存在多个版本，可遍历crd.spec.versions列表，筛选出需要操作的版本名称替换代码中的version变量。

内容的提问来源于stack exchange，提问作者zappa