Kubernetes / 运维笔记

Kubernetes Python API

Einic Yeo · 1月4日 · 2019年 · ·

一、用户管理

1、创建用户

vi CreateServiceAccount.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: admin-user
  namespace: kube-system
kubectl create -f CreateServiceAccount.yaml

2、用户授权

vi RoleBinding.yaml
apiVersion: rbac.authorization.k8s.io/v1beta1
kind: ClusterRoleBinding
metadata:
  name: admin-user
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: cluster-admin
subjects:
- kind: ServiceAccount
  name: admin-user
  namespace: kube-system
kubectl create -f RoleBinding.yaml

3、获取Token

kubectl describe secret $(kubectl get secret -n kube-system | grep ^admin-user | awk '{print $1}') -n kube-system | grep -E '^token'| awk '{print $2}'

二、安装kubernetes python sdk

1、模块安装

pip install kubernetes

2、测试demo

from kubernetes import client, config

ApiToken = "xxxxx"  #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj  = client.CoreV1Api(client.ApiClient(configuration))
ret = k8s_api_obj.list_namespaced_pod("dev")
print(ret)

三、接口操作案例

1、创建namespace

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/d版权声明：本文遵循 CC 4.0 BY-SA 版权协议，若要转载请务必附上原文出处链接及本声明，谢谢合作！ocs/CoreV1Api.md#create_namespace

from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj  = client.CoreV1Api(client.ApiClient(configuration))
body = {
    "apiVersion": "v1",
    "kind": "Namespace",
    "metadata": {
        "name": "test123",
    }
}

ret = k8s_api_obj.create_namespace(body=body)
print (ret)

2、删除namespace

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md

from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj  = client.CoreV1Api(client.ApiClient(configuration))
body = client.V1DeleteOptions()
body.api_version = "v1"
body.grace_period_seconds = 0
ret = k8s_api_obj.delete_namespace("test123", body=body)
print(ret)

3、查看namespace列表

官方文档：https://github.com/kubernetes-client/python/blob/master/版权声明：本文遵循 CC 4.0 BY-SA 版权协议，若要转载请务必附上原文出处链接及本声明，谢谢合作！kubernetes/docs/CoreV1Api.md#list_namespace

from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.CoreV1Api()
limit = 56                                  #返回最大值,可选参数可以不写
timeout_seconds = 56                                #超时时间可选参数
watch = False                                   #监听资源，可选参数可以不填
try:
    api_response = k8s_api_obj.list_namespace(limit=limit,timeout_seconds=timeout_seconds, watch=watch)
    for  namespace in api_response.items:
        print(namespace.metadata.name)
except ApiException as e:
    print("Exception when calling CoreV1Api->list_namespace: %s\n" % e)

4、创建pod

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1beta2Api.md

from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.AppsV1beta2Api(client.ApiClient(configuration))
body=eval("{'kind': 'Deployment', 'spec': {'replicas': 1, 'template': {'spec': {'containers': [{'image': 'nginx:1.7.9', 'name': 'nginx', 'ports': [{'contain
erPort': 80}]}]}, 'metadata': {'labels': {'app': 'nginx-deployment'}}}, 'selector': {'matchLabels': {'app': 'nginx-deployment'}}}, 'apiVersion': 'apps/v1beta2', 'metadata': {'labels': {'app': 'nginx-deployment'}, 'namespace': 'default', 'name': 'nginx-deployment'}}")

resp = k8s_api_obj.create_namespaced_deployment(body=body, namespace="default")
print(resp)

5、删除pod

官方文档：https://githu版权声明：本文遵循 CC 4.0 BY-SA 版权协议，若要转载请务必附上原文出处链接及本声明，谢谢合作！b.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1beta2Api.md#delete_namespaced_deployment

from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)


k8s_api_obj = client.AppsV1beta2Api(client.ApiClient(configuration))
resp = k8s_api_obj.delete_namespaced_deployment(name="nginx-deployment",
                                               namespace="default",
                                               body=client.V1DeleteOptions(
                                                       propagation_policy='Foreground',
                                                       grace_period_seconds=0)
                                               )
print(resp)

6、更新pod

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1beta2Api.md#patch_namespaced_deployment

版权声明：本文遵循 CC 4.0 BY-SA 版权协议，若要转载请务必附上原文出处链接及本声明，谢谢合作！ 
from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.AppsV1beta2Api(client.ApiClient(configuration))

body=eval("{'kind': 'Deployment', 'spec': {'replicas': 1, 'template': {'spec': {'containers': [{'image': 'nginx', 'name': 'nginx', 'ports': [{'containerPort
': 80}]}]}, 'metadata': {'labels': {'app': 'nginx-deployment'}}}, 'selector': {'matchLabels': {'app': 'nginx-deployment'}}}, 'apiVersion': 'apps/v1beta2', 'metadata': {'labels': {'app': 'nginx-deployment'}, 'namespace': 'default', 'name': 'nginx-deployment'}}")

resp = k8s_api_obj.patch_namespaced_deployment(
                name="nginx-deployment",
                namespace="default",
                body=body
            )
print(resp)

7、查询pod

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_pod

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj  = client.CoreV1Api()
resp = k8s_api_obj.list_namespaced_pod("default", label_selector="app=" + "nginx-deployment")
print(resp)

8、创建svc

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#create_namespaced_service

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.CoreV1Api()
namespace = "default"
body = {'apiVersion': 'v1', 'kind': 'Service', 'metadata': {'name': 'nginx-service', 'labels': {'app': 'nginx'}}, 'spec': {'ports': [{'port': 80, 'targetPor
t': 80}], 'selector': {'app': 'nginx'}}}
try:
    api_response = k8s_api_obj.create_namespaced_service(namespace , body)
    print(api_response)
except ApiException as e:
    print("Exception when calling CoreV1Api->create_namespaced_service: %s\n" % e)

9、删除svc

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#delete_namespaced_service

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.CoreV1Api()
name = 'nginx-service'                              #要删除svc名称
namespace = 'default'                               #命名空间
grace_period_seconds = 0                            #延迟时间,0立即删除
body = client.V1DeleteOptions()                         #删除选项
try:
    api_response = k8s_api_obj.delete_namespaced_service(name, namespace,body,  grace_period_seconds=grace_period_seconds)
    print(api_response)
except ApiException as e:
    print("Exception when calling CoreV1Api->delete_namespaced_service: %s\n" % e)

10、pod列表

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/AppsV1beta2Api.md#list_namespaced_deployment

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.AppsV1beta2Api(client.ApiClient(configuration))
namespace = 'dev'                               #命名空间
try:
    api_response = k8s_api_obj.list_namespaced_deployment(namespace)
    for deployment in api_response.items:
        print(deployment.metadata.name)
except ApiException as e:
    print("Exception when calling AppsV1beta2Api->list_namespaced_deployment: %s\n" % e)

11、创建ConfigMap

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#create_namespaced_config_map

from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj  = client.CoreV1Api()
body = {
        'apiVersion': 'v1',
        'kind': 'ConfigMap',
        'metadata': {
                'name': 'filebeat-configmap',
                'namespace': 'default'
        },
        'data': {
                'filebeat.yml': 'filebeat.prospectors: \n - input_type: log\ n paths: \n - "/mnt/*/logs/app/app.log"\n tags: ["json"]\ n json.keys_under_roo
t: true\ n json.overwrite_keys: true\ noutput.elasticsearch: \n hosts: ["xx.xx.xx.xx:9200"]\ n username: "elastic"\n password: "elastic"\n template.enabled: false\ n index: "dev_namespace_name_java_log-%{+yyyy.MM.dd}"\n '}     
   }


resp = k8s_api_obj.create_namespaced_config_map(
                body=body, namespace="default")
print(resp)

12、删除ConfigMap

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#delete_namespaced_config_map

from kubernetes import client, config

ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj  = client.CoreV1Api()
resp = k8s_api_obj.delete_namespaced_config_map(
                name="filebeat-configmap",
                namespace="default",
                body=client.V1DeleteOptions()
            )
print(resp)

13、查看ConfigMap

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#list_namespaced_config_map

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.CoreV1Api()
namespace = 'default'
try:
    api_response = k8s_api_obj.list_namespaced_config_map(namespace)
    for config_map in  api_response.items:
        print(config_map.metadata.name)
except ApiException as e:
    print("Exception when calling CoreV1Api->list_namespaced_config_map: %s\n" % e)

14、获取Node节点信息

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernete版权声明：本文遵循 CC 4.0 BY-SA 版权协议，若要转载请务必附上原文出处链接及本声明，谢谢合作！s/docs/CoreV1Api.md#read_node

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

k8s_api_obj = client.CoreV1Api()
exact = True
export = True
name = "192.168.1.50"               #此处填写node名称
try:
    api_response = k8s_api_obj.read_node(name, exact=exact, export=export)
    print(api_response)
except ApiException as e:
    print("Exception when calling CoreV1Api->read_node: %s\n" % e)

15、获取Node状态信息

官方文档：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md#read_node_status

from kubernetes import client, config
from kubernetes.client.rest import ApiException
ApiToken = "xxxxx"                              #ApiToken
configuration = client.Configuration()
setattr(configuration, 'verify_ssl', False)
client.Configuration.set_default(configuration)
configuration.host = "https://xxxx:6443"                    #ApiHost
configuration.verify_ssl = False
configuration.debug = True
configuration.api_key = {"authorization": "Bearer " + ApiToken}
client.Configuration.set_default(configuration)

name = "192.168.1.50"                               #此处填写node名称
k8s_api_obj = client.CoreV1Api()
try:
    api_response = k8s_api_obj.read_node_status(name, pretty=True)
    print(api_response)
except ApiException as e:
    print("Exception when calling CoreV1Api->read_node_status: %s\n" % e)

参考文献

https://github.com/kubernetes-client/python/tree/master/kubernetes/docs

https://www.servicemesher.com/blog/kubernetes-operator-in-python/

相关文章

虚机网格(istio)管理实战篇

11月17日 · 2023年

Kubeflow Volcano 实现典型 AI 训练任务

10月18日 · 2023年

Cilium 网络模型之关键配置

9月24日 · 2023年

Cilium Pod-to-Service 实地探索转发路径及 BPF 处理逻辑

7月11日 · 2023年
0 条回应
© 2017~ Infvie Envoy. All Rights Reserved.
IPV6 | GITHUB