如何用Kubernetes Python API获取Pod的secretName并生成Prometheus指标？

一、解析Projected卷的secretName字段 #

Kubernetes Python客户端返回的Pod对象中，Projected卷的Secret信息并非直接暴露，而是嵌套在pod.spec.volumes下的projected.sources数组里。每个source如果是secret类型，其secret_name字段（蛇形命名，而非kubectl展示的驼峰）就是你要找的内容。

代码示例：提取Pod镜像与SecretName #

from kubernetes import client, config

def fetch_cluster_pod_data(kube_config_path):
    config.load_kube_config(config_file=kube_config_path)
    v1 = client.CoreV1Api()
    
    pod_records = []
    for pod in v1.list_pod_for_all_namespaces().items:
        record = {
            "namespace": pod.metadata.namespace,
            "pod_name": pod.metadata.name,
            "images": [],
            "secret_names": []
        }
        
        # 提取容器镜像
        for container in pod.spec.containers:
            record["images"].append(container.image)
        
        # 提取Projected卷关联的SecretName
        for volume in pod.spec.volumes:
            if volume.projected:
                for source in volume.projected.sources:
                    if source.secret:
                        secret_name = source.secret.secret_name
                        if secret_name not in record["secret_names"]:
                            record["secret_names"].append(secret_name)
        
        pod_records.append(record)
    return pod_records

这段代码会加载指定集群的kubeconfig，遍历所有Pod，提取镜像列表和去重后的Projected卷SecretName。

二、生成Prometheus格式指标 #

VictoriaMetrics兼容标准Prometheus指标格式，我们可以生成两类核心指标：

• pod_image_info：记录Pod的镜像关联关系
• pod_secret_info：记录Pod与Projected卷Secret的关联关系

代码示例：生成指标 #

def build_prometheus_metrics(pod_records, cluster_label):
    metrics_lines = []
    
    # 镜像指标
    metrics_lines.append("# HELP pod_image_info Pod使用的镜像信息")
    metrics_lines.append("# TYPE pod_image_info gauge")
    for pod in pod_records:
        for image in pod["images"]:
            line = f'pod_image_info{{cluster="{cluster_label}",namespace="{pod["namespace"]}",pod="{pod["pod_name"]}",image="{image}"}} 1'
            metrics_lines.append(line)
    
    # Secret指标
    metrics_lines.append("\n# HELP pod_secret_info Pod关联的Projected卷Secret名称")
    metrics_lines.append("# TYPE pod_secret_info gauge")
    for pod in pod_records:
        for secret in pod["secret_names"]:
            line = f'pod_secret_info{{cluster="{cluster_label}",namespace="{pod["namespace"]}",pod="{pod["pod_name"]}",secret_name="{secret}"}} 1'
            metrics_lines.append(line)
    
    return "\n".join(metrics_lines)

生成的指标可以写入文件，或通过HTTP接口暴露，VictoriaMetrics直接抓取即可。

三、集群差异对比与告警 #

对比两个集群的Pod数据，找出镜像或Secret不一致的项，触发告警：

def compare_cluster_pods(cluster1_data, cluster2_data, cluster1_name, cluster2_name):
    # 构建集群1的Pod索引：(命名空间, Pod名) -> (镜像集合, Secret集合)
    cluster1_index = {
        (pod["namespace"], pod["pod_name"]): (set(pod["images"]), set(pod["secret_names"]))
        for pod in cluster1_data
    }
    
    alerts = []
    # 检查集群2的Pod在集群1中的差异
    for pod in cluster2_data:
        key = (pod["namespace"], pod["pod_name"])
        if key not in cluster1_index:
            alerts.append(f"Pod {key[1]}({key[0]})仅存在于{cluster2_name}，{cluster1_name}中无此Pod")
            continue
        
        c1_images, c1_secrets = cluster1_index[key]
        c2_images = set(pod["images"])
        c2_secrets = set(pod["secret_names"])
        
        # 镜像差异检查
        if c1_images != c2_images:
            diff = c1_images.symmetric_difference(c2_images)
            alerts.append(f"Pod {key[1]}({key[0]})镜像不一致：{cluster1_name}与{cluster2_name}差异项：{', '.join(diff)}")
        
        # Secret差异检查
        if c1_secrets != c2_secrets:
            diff = c1_secrets.symmetric_difference(c2_secrets)
            alerts.append(f"Pod {key[1]}({key[0]})关联Secret不一致：{cluster1_name}与{cluster2_name}差异项：{', '.join(diff)}")
    
    # 检查集群1有但集群2没有的Pod
    for key in cluster1_index:
        if not any((pod["namespace"], pod["pod_name"]) == key for pod in cluster2_data):
            alerts.append(f"Pod {key[1]}({key[0]})仅存在于{cluster1_name}，{cluster2_name}中无此Pod")
    
    # 输出告警（实际场景可替换为邮件、企业微信等推送方式）
    if alerts:
        print("=== 集群差异告警 ===")
        for alert in alerts:
            print(f"- {alert}")

完整工作流程示例 #

if __name__ == "__main__":
    # 两个集群的kubeconfig路径
    cluster1_cfg = "./cluster1-kubeconfig"
    cluster2_cfg = "./cluster2-kubeconfig"
    
    # 获取Pod数据
    cluster1_data = fetch_cluster_pod_data(cluster1_cfg)
    cluster2_data = fetch_cluster_pod_data(cluster2_cfg)
    
    # 生成并保存指标
    cluster1_metrics = build_prometheus_metrics(cluster1_data, "prod-cluster")
    cluster2_metrics = build_prometheus_metrics(cluster2_data, "staging-cluster")
    
    with open("./prod_metrics.prom", "w") as f:
        f.write(cluster1_metrics)
    with open("./staging_metrics.prom", "w") as f:
        f.write(cluster2_metrics)
    
    # 对比并触发告警
    compare_cluster_pods(cluster1_data, cluster2_data, "prod-cluster", "staging-cluster")

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