如何通过Python Kubernetes客户端获取PersistentVolumeClaim的Used By信息

kubectl describe pvc显示的"Used By"字段并非PVC资源本身的内置属性，而是K8s客户端通过遍历集群中引用该PVC的关联资源生成的结果。因此需要通过Python Kubernetes客户端主动查询相关资源来实现同样的效果。

实现思路 #

核心逻辑是：遍历集群中可能引用PVC的资源（Pod、Deployment、StatefulSet等），检查这些资源的配置中是否存在对目标PVC的引用，收集所有匹配的资源作为"Used By"结果。

代码实现 #

from kubernetes import client, config

# 加载K8s配置（本地开发用kubeconfig，集群内部署用in-cluster配置）
config.load_kube_config()
core_api = client.CoreV1Api()
apps_api = client.AppsV1Api()

def get_pvc_used_by(namespace: str, pvc_name: str) -> list:
    used_by_resources = []

    # 1. 查找直接引用PVC的Pod
    pods = core_api.list_namespaced_pod(namespace=namespace)
    for pod in pods.items:
        for volume in pod.spec.volumes:
            if volume.persistent_volume_claim and volume.persistent_volume_claim.claim_name == pvc_name:
                used_by_resources.append(f"Pod: {pod.metadata.name}")
                break  # 一个Pod多次引用同一PVC，只记录一次

    # 2. 查找模板中引用PVC的StatefulSet
    statefulsets = apps_api.list_namespaced_stateful_set(namespace=namespace)
    for sts in statefulsets.items:
        # 检查VolumeClaimTemplate（动态生成PVC的情况）
        if sts.spec.volume_claim_templates:
            for template in sts.spec.volume_claim_templates:
                if template.metadata.name == pvc_name:
                    used_by_resources.append(f"StatefulSet: {sts.metadata.name} (via claim template)")
                    break
        # 检查Pod模板直接引用的PVC
        for volume in sts.spec.template.spec.volumes:
            if volume.persistent_volume_claim and volume.persistent_volume_claim.claim_name == pvc_name:
                used_by_resources.append(f"StatefulSet: {sts.metadata.name}")
                break

    # 3. 查找Pod模板中引用PVC的Deployment
    deployments = apps_api.list_namespaced_deployment(namespace=namespace)
    for deploy in deployments.items:
        for volume in deploy.spec.template.spec.volumes:
            if volume.persistent_volume_claim and volume.persistent_volume_claim.claim_name == pvc_name:
                used_by_resources.append(f"Deployment: {deploy.metadata.name}")
                break

    return used_by_resources

# 示例调用
if __name__ == "__main__":
    target_namespace = "default"
    target_pvc = "my-test-pvc"
    result = get_pvc_used_by(target_namespace, target_pvc)
    
    print(f"Used By for PVC [{target_pvc}] in namespace [{target_namespace}]:")
    for item in result:
        print(f"- {item}")

补充说明 #

• 代码覆盖了最常见的三类资源：Pod、Deployment、StatefulSet，你可以根据集群实际使用的资源类型（如DaemonSet、ReplicaSet）扩展查询逻辑。
• 如果是在K8s集群内部运行的应用，将config.load_kube_config()替换为config.load_incluster_config()即可。
• 结果逻辑与kubectl describe pvc完全一致，包含直接引用的Pod，以及通过模板间接关联的控制器资源。

内容的提问来源于stack exchange，提问作者Muhammad Ali Khan