Kubernetes Python API中文使用说明

k8s集群操作:
创建用户:
vi CreateServiceAccount.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: admin-user
  namespace: kube-system
kubectl create -f CreateServiceAccount.yaml
用户授权:
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
获取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}’
安装python模块kubernetes
模块安装:
pip install kubernetes
验证:
from kubernetes import client, config
import urllib3  #在urllib3时代,官方强制验证https的安全证书,如果没有通过是不能通过请求的,虽然添加忽略验证的参数,但是依然会 给出醒目的 Warning,这一点没毛病。
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)
urllib3.disable_warnings() #禁用 urllib3
k8s_api_obj  = client.CoreV1Api(client.ApiClient(configuration))
ret = k8s_api_obj.list_namespaced_pod(“dev”) #NameSpace
print(ret)
注意:
/usr/local/lib/python3.6/site-packages/urllib3/connectionpool.py:851: InsecureRequestWarning: Unverified HTTPS request is being made. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings
  InsecureRequestWarning)
出现这个错误的原因是:
requests 库其实是基于 urllib 编写的,对 urllib 进行了封装,使得使用时候的体验好了很多,现在 urllib 已经出到了3版本,功能和性能自然是提升了不少。 所以,requests最新版本也是基于最新的 urllib3 进行封装。 
在urllib2时代对https的处理非常简单,只需要在请求的时候加上 verify=False 即可,这个参数的意思是忽略https安全证书的验证,也就是不验证证书的可靠性,直接请求, 这其实是不安全的,因为证书可以伪造,不验证的话就不能保证数据的真实性。 
在urllib3时代,官方强制验证https的安全证书,如果没有通过是不能通过请求的,虽然添加忽略验证的参数,但是依然会 给出醒目的 Warning,这一点没毛病。
解决办法:添加两行代码  禁用 urllib3 
import urllib3
urllib3.disable_warnings()
代码案例
node操作
获取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: %sn” % e)
获取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)
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: %sn” % e)
namespace操作
查看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: %sn” % e)
创建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)
删除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 = client.V1DeleteOptions()
body.api_version = “v1”
body.grace_period_seconds = 0
ret = k8s_api_obj.delete_namespace(“test123”, body=body)
print(ret)
pod操作
查询所有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.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: %sn” % e)
查询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)
创建pod:
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)
更新pod:
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)
删除pod:
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)
svc操作
创建svc:
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: %sn” % e)
删除svc:
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: %sn” % e)
configmap操作
查看configmap:
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: %sn” % e)
创建configmap:
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)
删除configmap:
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)

发表评论

您的电子邮箱地址不会被公开。