Python-kubernetes模块—k8s集群资源管理方法

 2024/11/16 

pip3 install kubernetes

和其他API一样，python访问k8s也需要API授权，也就是~/.kube/config文件，也可以自己写一个专门用来给python访问的config文件

我为了方便就直接用管理员权限的config了

使用read方法获取的资源信息的对象，格式上和kubectl get pod harbor-test -o yaml 这样返回是一样的，写的时候可以在第二终端打开来进行参考

查看k8s资源

#coding=utf-8
from kubernetes import client,config
# 加载权限文件，用于连接k8s集群
config.load_kube_config(config_file='config')
# 打开连接，创建客户端对象
client_k8s=client.CoreV1Api()

# 获取node资源
nodes=client_k8s.list_node()
for node in nodes.items:
    print(f"node name: {node.metadata.name}")

# node name: ws-k8s-master1
# node name: ws-k8s-master2
# node name: ws-k8s-master3
# node name: ws-k8s-node1
# node name: ws-k8s-node2
# node name: ws-k8s-node3

# 获取pods资源
pods=client_k8s.list_pod_for_all_namespaces()
for pod in pods.items:
    print(f"pod name：{pod.metadata.name}")

管理Pod资源

创建、更新与删除pod

from kubernetes import  client,config

config.load_kube_config(config_file='config')
# 打开连接，创建客户端对象
client_k8s=client.CoreV1Api()

# 创建一个pod的对象
pod_create=client.V1Pod(
    metadata=client.V1ObjectMeta(name='my-pod',labels={'api':'python'}),
    spec=client.V1PodSpec(
        containers=[client.V1Container(
            name='my-container',
            image='docker.io/library/nginx:1.9.1',
            image_pull_policy='IfNotPresent'
        )]
    )
)
# 调用创建方法，传入pod信息的对象
client_k8s.create_namespaced_pod(namespace='default',body=pod_create)

# 用pod接收返回的值
pod=client_k8s.read_namespaced_pod(name='my-pod',namespace='default')
print(pod.spec.containers)
# 修改pod中的image字段
pod.spec.containers[0].image='nginx:latest'

# 替换镜像
client_k8s.replace_namespaced_pod(name='my-pod',namespace='default',body=pod)

# 删除pod
client_k8s.delete_namespaced_pod(name='my-pod',namespace='default')

定时监控pod状态

from kubernetes import config,client
from time import sleep
config.load_kube_config(config_file="config")
client_k8s=client.CoreV1Api()
namespace='default'
while True:
    # 获取所有ns下的pod：pods=client_k8s.list_pod_for_all_namespaces()
    pods=client_k8s.list_namespaced_pod(namespace=namespace)
    for pod in pods.items:
        print(f"namespace:{pod.metadata.namespace},pod:{pod.metadata.name},status:{pod.status.phase}")
    sleep(5)

for创建pod

import time
from kubernetes import config,client
config.load_kube_config('config')
client_k8s=client.CoreV1Api()

for i in range(5):
    pod_body=client.V1Pod(
        # 定义pod体时，每一级都需要重新调用client.V1的接口，然后接入数据
        metadata=client.V1ObjectMeta(name=f"pod-{i}"),
        spec=client.V1PodSpec(
            containers=[client.V1Container(
                name=f'container-{i}',
                image='nginx:latest',
                image_pull_policy='IfNotPresent'
            )]
        )
    )
    # 批量创建
    client_k8s.create_namespaced_pod(namespace='default',body=pod_body)

获取pod日志

from kubernetes import config,client

pod_name='pod-1'
namespace_name='default'

config.load_kube_config('config')
client_k8s=client.CoreV1Api()

logs=client_k8s.read_namespaced_pod_log(name=pod_name,namespace=namespace_name)
print(logs)

管理service资源-创建，列出，删除

from kubernetes import client,config

config.load_kube_config(config_file='config')
client_k8s=client.CoreV1Api()

# 定义一个pod
pod_body=client.V1Pod(
    metadata=client.V1ObjectMeta(name='my-pod',labels={'api':'python'}),
    spec=client.V1PodSpec(
        containers=[client.V1Container(
            name='my-container',
            image='nginx:latest',
            image_pull_policy='IfNotPresent'
        )]
    )
)
# 创建pod
client_k8s.create_namespaced_pod(namespace='default',body=pod_body)

# 定义service对象
service_body=client.V1Service(
    metadata=client.V1ObjectMeta(name='my-service'),
    spec=client.V1ServiceSpec(
        selector={'api':'python'},
        ports=[client.V1ServicePort(port=80,target_port=80)],
        type='ClusterIP'
    )
)
# 创建service
client_k8s.create_namespaced_service(namespace='default',body=service_body)

# 列出default ns下的service资源
service_list=client_k8s.list_namespaced_service(namespace='default')
print("namespace:default")
for svc in service_list.items:
    print(f"service_name:{svc.metadata.name}")

# 列出所有ns下的service
service_list=client_k8s.list_service_for_all_namespaces()
for svc in service_list.items:
    print(f"service_name:{svc.metadata.name},namespace:{svc.metadata.namespace}")

# 删除service
client_k8s.delete_namespaced_service(name='my-service',namespace='default')
# 删除pod
client_k8s.delete_namespaced_pod(name='my-pod',namespace='default')

管理deployment-创建、更新、列出、删除

from kubernetes import client,config
from time import sleep
config.load_kube_config(config_file='config')
# 因为deployment使用的是apps/v1，所以需要使用AppsV1Api实例化
client_k8s=client.AppsV1Api()

# 创建deploy对象
deploy_body=client.V1Deployment(
    metadata=client.V1ObjectMeta(name='my-deployment',namespace='default'),
    spec=client.V1DeploymentSpec(
        replicas=3,
        selector=client.V1LabelSelector(
            match_labels={'app':'nginx'}
        ),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={'app':'nginx'}),
            spec=client.V1PodSpec(
                containers=[client.V1Container(
                    name='my-containers',
                    image='nginx:1.18',
                    image_pull_policy='IfNotPresent',
                    ports=[client.V1ContainerPort(container_port=80)]
                )]
            )
        )
    )
)

# 创建
print("开始创建Deployment")
client_k8s.create_namespaced_deployment(namespace='default',body=deploy_body)
sleep(10)
print('deployment创建完成')

print("开始获取Deployment信息,并更新nginx镜像")
deploy_update=client_k8s.read_namespaced_deployment(name='my-deployment',namespace='default')
deploy_update.spec.template.spec.containers[0].image='nginx:latest'
#更新deployment
client_k8s.patch_namespaced_deployment(name='my-deployment',namespace='default',body=deploy_update)
print('更新deployment镜像完成')

# 列出deployment
deploy_list=client_k8s.list_namespaced_deployment(namespace='default')
for deploy in deploy_list.items:
    print(f"namespace:{deploy.metadata.namespace},deployment_name:{deploy.metadata.name}")

# 删除deployment
client_k8s.delete_namespaced_deployment(name='my-deployment',namespace='default')
print('deployment已删除')

管理statefulset资源

from kubernetes import config,client
from time import sleep
config.load_kube_config(config_file='config')
# 实例化不同api版本的对象
client_v1=client.CoreV1Api()
client_apps=client.AppsV1Api()

service_body=client.V1Service(
    metadata=client.V1ObjectMeta(name='my-service'),
    spec=client.V1ServiceSpec(
        selector={'app':'statefulset'},
        ports=[client.V1ServicePort(port=80,target_port=80)],
        type='ClusterIP'
    )
)
client_v1.create_namespaced_service(namespace='default',body=service_body)
print("service创建完成")

statefulset_body=client.V1StatefulSet(
    metadata=client.V1ObjectMeta(name='my-statefulset',namespace='default'),
    spec=client.V1StatefulSetSpec(
        replicas=5,
        service_name='my-service',
        selector=client.V1LabelSelector(
            match_labels={'app':"statefulset"}),
    template=client.V1PodTemplateSpec(
        metadata=client.V1ObjectMeta(labels={'app':"statefulset"}),
        spec=client.V1PodSpec(
            containers=[client.V1Container(
                name='my-container',
                image='nginx:1.18',
                image_pull_policy='IfNotPresent',
                ports=[client.V1ContainerPort(container_port=80)]
            )]
        )
    )
    )
)

client_apps.create_namespaced_stateful_set(namespace='default',body=statefulset_body)
print('statefulest已创建完成')
sleep(10)
# 更新statefulset
print('开始获取statefulest信息并修改')
statefulset_info=client_apps.read_namespaced_stateful_set(name='my-statefulset',namespace='default')
statefulset_info.spec.template.spec.containers[0].image='nginx:latest'

client_apps.replace_namespaced_stateful_set(name='my-statefulset',namespace='default',body=statefulset_info)
print(f"statefulset镜像已修改")

client_v1.delete_namespaced_service(name='my-service',namespace='default')
print("service已删除")

client_apps.delete_namespaced_stateful_set(name='my-statefulset',namespace='default')
print('statefulest已删除')

管理Configmap资源

from kubernetes import client,config
config.load_kube_config(config_file='config')

client_v1=client.CoreV1Api()

config_body=client.V1ConfigMap(
    metadata=client.V1ObjectMeta(
        name='my-configmap'),
        data={'a':'b'}
    )

client_v1.create_namespaced_config_map(namespace='default',body=config_body)
print('my-configmap 创建完成')

cm_text=client_v1.read_namespaced_config_map(namespace='default',name='my-configmap')
print(f"configmap的内容:{cm_text.data}")

cm_list=client_v1.list_namespaced_config_map(namespace='default')
print('default ns下的cm:')
for cm in cm_list.items:
    print(cm.metadata.name)

print('定义新的data内容')
configmap_update=client.V1ConfigMap(
    data={'aa':'bb'})
client_v1.patch_namespaced_config_map(namespace='default',name='my-configmap',body=configmap_update)
print('configmap内容新增完成')

cm_text=client_v1.read_namespaced_config_map(namespace='default',name='my-configmap')
print(f"configmap的内容:{cm_text.data}")

client_v1.delete_namespaced_config_map(namespace='default',name='my-configmap')
print('configmap已删除')

原文作者：王盛

原文链接：https://akemi.zj.cn/2024/11/16/Python-Module-kubernetes-1/

发表日期：November 16th 2024, 6:42:18 pm

更新日期：February 20th 2025, 6:37:27 pm

Next Post

Python-kubernetes模块案例—根据不同ns更新pod副本数
Previous Post

Python socket模块案例—简单网络编程

CATALOG

1. 查看k8s资源
2. 管理Pod资源
3. 管理service资源-创建，列出，删除
4. 管理deployment-创建、更新、列出、删除
5. 管理statefulset资源
6. 管理Configmap资源