k8s client (python 模块）

k8s python库

安装

sudo pip3 install kubernetes

认证

首先引入SDK支持库。然后将 ~/.kube 的config文件的内容复制到本地目录，保存为文件kubeconfig.yaml，然后运行下面的python代码。

#使用
from kubernetes import client, config
from kubernetes.stream import stream
import yaml
# kubeconfig.yaml保存的路径
config_file="/home/ics/kubeconfig.yaml" 
config.kube_config.load_kube_config(config_file=config_file)
#获取API的CoreV1Api和BatchV1Api版本对象
Api_Instance = client.CoreV1Api()
Api_Batch = client.BatchV1Api()

api使用

1、列出资源信息

#列出 namespaces
for ns in Api_Instance.list_namespace().items:
    print(ns.metadata.name)

#列出所有的nodes
def list_node():
    api_response = Api_Instance.list_node()
    data = {}
    for i in api_response.items:
        data[i.metadata.name] = {"name": i.metadata.name,
                                "status": i.status.conditions[-1].type if i.status.conditions[-1].status == "True" else "NotReady",
                                "ip": i.status.addresses[0].address,
                                "kubelet_version": i.status.node_info.kubelet_version,
                                "os_image": i.status.node_info.os_image,
                                 }
    return data
    
#列出所有的services
def list_service():
    api_response = Api_Instance.list_service_for_all_namespaces()
    return api_response
    
#列出所有的pod
def list_pod():
    api_response = Api_Instance.list_pod_for_all_namespaces()
    data = {}
    for i in api_response.items:
        data[i.metadata.name] = {"ip": i.status.pod_ip, "namespace": i.metadata.namespace}
    return data

#列出所有job
def list_job_for_all_namespaces():
    api_response = Api_Batch.list_job_for_all_namespaces()
    return api_response

def list_job(namespace="default"):
    api_response = Api_Batch.list_namespaced_job(namespace)
    return api_response
    
##列出其他资源和以上类似，不懂可以查看(kubectl  api-resources)

2、创建k8s资源对象

# 创建pod
def read_pod_log(name, namespace="default"):
    api_response = Api_Instance.read_namespaced_pod_log(name, namespace)
    return api_response

# 创建service
def create_service(file, namespace="default"):
    with open(file) as f:
        dep = yaml.safe_load(f)
        api_response = Api_Instance.create_namespaced_service(namespace, body=dep)
        return api_response

# 创建job
def create_job(file, namespace="default"):
    with open(file) as f:
        dep = yaml.safe_load(f)
        api_response = Api_Batch.create_namespaced_job(namespace, body=dep)
        return api_response

3、删除资源

# 删除pod
def delete_pod(name, namespace="default"):
    api_response = Api_Instance.delete_namespaced_pod(name, namespace)
    return api_response

# 删除service
def delete_service(name, namespace="default"):
    api_response = Api_Instance.delete_namespaced_service(name, namespace)
    return api_response

# 删除job
def delete_job(name, namespace="default"):
    api_response = Api_Batch.delete_namespaced_job(name, namespace)
    return api_response

4、查看资源

# 查看pod
def read_pod(name, namespace="default"):  # 可获得：status.pod_ip
    api_response = Api_Instance.read_namespaced_pod(name, namespace)
    return api_response
#指定信息
print(api_response.spec.containers[0].image) #pod的镜像名
print(api_response.status.pod_ip)  #pod所在的node ip

# 查看service
def read_service(name, namespace="default"):  # 获取服务端口号
    api_response = Api_Instance.read_namespaced_service(name, namespace)
    return api_response

# 查看job
def read_job(name, namespace="default"):
    api_response = Api_Batch.read_namespaced_job(name, namespace)
    return api_response

5、修改

# 修改pod
def replace_pod(file, name, namespace="default"):
    # replace: 替换容器
    # patch: 可以在容器运行时直接修改属性，不删除容器
    with open(file) as f:
        dep = yaml.safe_load(f)
        api_response = Api_Instance.replace_namespaced_pod(name, namespace, body=dep)
    return api_response
或
old_resp = Api_Instance.read_namespaced_pod(namespace="default",name='nginx-pod')
old_resp.spec.containers[0].image = "nginx:alpine" #只修改镜像
new_resp = Api_Instance.patch_namespaced_pod(namespace="default",name='nginx-pod',body=old_resp)

# 修改job
def replace_job(file, name, namespace="default"):
    with open(file) as f:
        dep = yaml.safe_load(f)
        api_response = Api_Batch.replace_namespaced_job(name, namespace, body=dep)
    return api_response

# 修改service
def replace_service(file, name, namespace="default"):
    with open(file) as f:
        dep = yaml.safe_load(f)
        api_response = Api_Instance.replace_namespaced_service(name, namespace, body=dep)
    return api_response

6、查看日志

# 查看pod日志
def read_pod_log(name, namespace="default"):
    api_response = Api_Instance.read_namespaced_pod_log(name, namespace)
    return api_response

7、查看pod运行的服务信息

# 查看ssh、jupyter等服务的端口
使用read_service函数

# 获取jupyter的token
def exec_pod(name, command, namespace="default"):
    # command = ["/bin/sh","-c","jupyter notebook list"]
    container_stream = stream(self.Api_Instance.connect_get_namespaced_pod_exec, name=name, namespace=namespace, command=command, stderr=True, stdin=True, stdout=True, tty=True, _preload_content=True)
    return container_stream