• 动机
• 脚本
• 使用实例
  • 获取指定命名空间
  • 获取所有命名空间
  • 生成csv文件

动机

最近，由于 Docker Hub 镜像的失效，在重新启动 Pod 时，拉取镜像失败，导致 Pod 无法正常启动。因此，我需要批量检查集群中有哪些 Pod 使用了官方的 Docker Hub 镜像，并将这些镜像保存到本地的 Harbor 仓库中。为此，通过python编写了一个脚本，可以导出所有命名空间中的镜像信息，或者根据需要导出特定命名空间中的镜像信息。

脚本

import sys
import logging
import argparse
from kubernetes import client, config
from kubernetes.config.config_exception import ConfigException


logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger()

config_file_path = "config路径"
logger.info(f"使用的 kubeconfig 文件路径: {config_file_path}")

def load_kube_config(config_file_path):
    """加载kubeconfig文件"""
    try:
        config.load_kube_config(config_file=config_file_path)
        logger.info("kubeconfig 加载成功")
    except ConfigException as e:
        logger.error(f"加载kubeconfig失败: {e}")
        sys.exit(1)

def get_namespaces(namespace_name):
    """根据命名空间名称获取命名空间列表"""
    if namespace_name == 'all':
        try:
            namespaces = client.CoreV1Api().list_namespace().items
        except client.rest.ApiException as e:
            logger.error(f"获取命名空间失败: {e}")
            sys.exit(1)
    else:
        try:
            namespaces = [client.CoreV1Api().read_namespace(name=namespace_name)]
        except client.rest.ApiException as e:
            logger.error(f"读取命名空间失败: {e}")
            sys.exit(1)
    return namespaces

def get_pods(namespace):
    """获取指定命名空间的Pod列表"""
    try:
        return client.CoreV1Api().list_namespaced_pod(namespace).items
    except client.rest.ApiException as e:
        logger.error(f"获取{namespace}命名空间的Pod失败: {e}")
        sys.exit(1)

#打印命名空间、pod名称、镜像名称
def print_pod_images(namespaces):
    """打印Pod及其使用的镜像"""
    print("namespace,pod,container,image")
    for ns in  namespaces:
        ns_name = ns.metadata.name
        pods = get_pods(ns_name)
        for pod in pods:
            for container in pod.spec.containers:
                print(f"命名空间：{ns_name},pod名称：{pod.metadata.name},容器名称：{container.name},镜像名称：{container.image}")

def validate_namespace(namespace):
    """验证命名空间是否有效。"""
    if namespace == "all":
        return True
    return True

def  main():
    # 命令行参数解析
    parser = argparse.ArgumentParser(description='列出Kubernetes集群中指定命名空间的Pod镜像。')
    #提示用户需要输入的传输必须是namespace的名称和all
    parser.add_argument('namespace', help='<namespace>|all', type=str)
    #解析参数
    args = parser.parse_args()
    #接收namespace的参数，也就是用户传入的参数
    namespace = args.namespace

    if not validate_namespace(namespace):
        logger.error(f"无效的命名空间'{namespace}'")
        sys.exit(1)

    logger.info(f"开始查询命名空间'{namespace}'的Pod镜像")

    try:
        load_kube_config()
        namespaces = get_namespaces(namespace)
        print_pod_images(namespaces)
    except Exception as e:
        logger.error(f"查询过程中遇到错误: {e}")
        sys.exit(1)
    logger.info("查询完成")

if __name__ == "__main__":
    main()

使用实例

python3 get_images_list.py  namespace|all

获取指定命名空间

python3 get_images_list.py  kube-system

获取所有命名空间

python3 get_images_list.py  all

生成csv文件

python3 get_images_list.py  all > pod_images.csv