一、python通过config文件执行api操作
https://www.cnblogs.com/zhangb8042/p/11444756.html
https://github.com/kubernetes-client/python/tree/master/examples
1）准备环境
cp .kube/config kubeconfig.yaml
pip install kubernetes
2）获取k8s资源
from kubernetes import client, config from os import path import yaml class k8sapi: def k8s_conn(self): config.kube_config.load_kube_config(config_file="kubeconfig.yaml") conn = client.CoreV1Api() return conn def namespaces(self): # 列出 namespaces conn = self.k8s_conn() ret = conn.list_pod_for_all_namespaces(watch=False) for i in ret.items: print("%s %s %s" % (i.status.pod_ip, i.metadata.namespace, i.metadata.name)) def services(self): # 列出所有的services conn = self.k8s_conn() ret = conn.list_service_for_all_namespaces(watch=False) for i in ret.items: print("%s %s %s %s %s " % ( i.kind, i.metadata.namespace, i.metadata.name, i.spec.cluster_ip, i.spec.ports)) def pods(self): # 列出所有的pod conn = self.k8s_conn() ret = conn.list_pod_for_all_namespaces(watch=False) for i in ret.items: print("%s %s %s" % (i.status.pod_ip, i.metadata.namespace, i.metadata.name)) def create(self): config.load_kube_config() with open(path.join(path.dirname(__file__), "/root/deploy.yaml")) as f: dep = yaml.safe_load(f) k8s_apps_v1 = client.AppsV1Api() resp = k8s_apps_v1.create_namespaced_deployment( body=dep, namespace="default") print("Deployment created. status='%s'" % resp.metadata.name) def delete(self): config.load_kube_config() k8s_core_v1 = client.CoreV1Api() resp = k8s_core_v1.delete_namespaced_pod(namespace="default", name='nginx-pod') print("delete Pod ") if __name__ == '__main__': myk8s = k8sapi() myk8s.pods()
3）创建pod使用的yaml文件
apiVersion: apps/v1
kind: Deployment
metadata:
name: myapp
namespace: default
labels:
app: myapp
spec:
replicas: 1
selector:
matchLabels:
app: myapp
template:
metadata:
name: myapp-pod
labels:
app: myapp
spec:
containers:
- name: myapp
image: wangyanglinux/myapp:v2
ports:
- name: http
containerPort: 80
二、获取app应用信息
获取app应用的基本信息, 有重复的
# coding: utf-8 import json,sys import time from kubernetes import client, config region = sys.argv[1] class myk8sinfo: def __init__(self): self.applists = [] def k8s_conn(self): Token = "eyJhbGciOiJSUzI1NiIsImtpZCI6IiJ9.eyJpc3MiOiJrdWJlcm5ldGVzL3NlcnZpY2VhY2NvdW50Iiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9uYW1lc3BhY2UiOiJrdWJlLXN5c3RlbSIsImt1YmVybmV0ZXMuaW8vc2VydmljZWFjY291bnQvc2VjcmV0Lm5hbWUiOiJlbmlvdC1yZWFkZXItdG9rZW4tN2wydHgiLCJrdWJlcm5ldGVzLmlvL3NlcnZpY2VhY2NvdW50L3NlcnZpY2UtYWNjb3VudC5uYW1lIjoiZW5pb3QtcmVhZGVyIiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9zZXJ2aWNlLWFjY291bnQudWlkIjoiNWFlNjVmMmUtOGQ5Ni0xMWU5LWIwNzUtMDAwZDNhYTA4OTA2Iiwic3ViIjoic3lzdGVtOnNlcnZpY2VhY2NvdW50Omt1YmUtc3lzdGVtOmVuaW90LXJlYWRlciJ9.kUjZhtjQPHvxtcjonB-EAQeAVTXSdyLoiXAirRRJ6hrSPeP2WhgAzF1h2v6BwV1uZulnWZ9g0rZCY5RKlAIvnYeDrgsy1EAZ40u7N4I5DBPKAxbljZAjUcF_jO-OSBNDfSOCrUFgY_rmC5uROzXNe6jlVQ300vI3BUAkLf0Q3_LB0uFmMXutz2ZzUTOssJgMTUKMh0qutbTuwGgiDvpyBjqTgXpZOBcY1cNzuXoI3pBPnb2WiMcVoaB4rp3ov9m04KD6k1OqwB4MeR0UCz7hCA-8J7N2GYzGdAtDYoqRniJ5tkrfH2hQYjKLak0JfKg9vBrePCKCodUxTGp24QLdjQ" APISERVER = 'https://apaas0001:8443' configuration = client.Configuration() configuration.host = APISERVER configuration.verify_ssl = False configuration.api_key = {"authorization": "Bearer " + Token} client.Configuration.set_default(configuration) conn = client.CoreV1Api() return conn def podinfo(self,namespaced): conn = self.k8s_conn() ret = conn.list_namespaced_pod(namespaced) for i in ret.items: appinfo = {} image = i.spec.containers[0].image.split(":")[-1] cpu = i.spec.containers[0].resources.limits['cpu'] memory = i.spec.containers[0].resources.limits['memory'] namespace = i.metadata.namespace app = i.metadata.labels['app'] appinfo['app'] = app appinfo['image'] = image appinfo['cpu'] = cpu appinfo['memory'] = memory appinfo['namespace'] = namespace self.applists.append(appinfo) def main(self): self.podinfo('arch') self.podinfo('enos') files = '%s.json' % region res = json.dumps(self.applists) with open(files,mode='w') as f: f.write(res)
if __name__ == '__main__':
k8s = myk8sinfo()
k8s.main()
python apps_info.py region
结果：[{"app": "arch-rule-engine", "memory": "2Gi", "namespace": "arch", "image": "tag_rule-engine_20190613_002", "cpu": "1"},]
2) 2种方式，config或token
# coding: utf-8 import json,sys import time from kubernetes import client, config region = sys.argv[1] class myk8sinfo: def __init__(self): self.applists = [] def k8s_conn(self): Token = "eyJhbGciOiJSUzI1NiIsImtpZCI6IiJ9.eyJpc3MiOiJrdWJlcm5ldGVzL3NlcnZpY2VhY2NvdW50Iiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9uYW1lc3BhY2UiOiJrdWJlLXN5c3RlbSIsImt1YmVybmV0ZXMuaW8vc2VydmljZWFjY291bnQvc2VjcmV0Lm5hbWUiOiJlbmlvdC1yZWFkZXItdG9rZW4tN2wydHgiLCJrdWJlcm5ldGVzLmlvL3NlcnZpY2VhY2NvdW50L3NlcnZpY2UtYWNjb3VudC5uYW1lIjoiZW5pb3QtcmVhZGVyIiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9zZXJ2aWNlLWFjY291bnQudWlkIjoiNWFlNjVmMmUtOGQ5Ni0xMWU5LWIwNzUtMDAwZDNhYTA4OTA2Iiwic3ViIjoic3lzdGVtOnNlcnZpY2VhY2NvdW50Omt1YmUtc3lzdGVtOmVuaW90LXJlYWRlciJ9.kUjZhtjQPHvxtcjonB-EAQeAVTXSdyLoiXAirRRJ6hrSPeP2WhgAzF1h2v6BwV1uZulnWZ9g0rZCY5RKlAIvnYeDrgsy1EAZ40u7N4I5DBPKAxbljZAjUcF_jO-OSBNDfSOCrUFgY_rmC5uROzXNe6jlVQ300vI3BUAkLf0Q3_LB0uFmMXutz2ZzUTOssJgMTUKMh0qutbTuwGgiDvpyBjqTgXpZOBcY1cNzuXoI3pBPnb2WiMcVoaB4rp3ov9m04KD6k1OqwB4MeR0UCz7hCA-8J7N2GYzGdAtDYoqRniJ5tkrfH2hQYjKLak0JfKg9vBrePCKCodUxTGp24QLdjQ" APISERVER = 'https://apaas0001:8443' configuration = client.Configuration() configuration.host = APISERVER configuration.verify_ssl = False configuration.api_key = {"authorization": "Bearer " + Token} client.Configuration.set_default(configuration) conn = client.CoreV1Api() return conn def k8s_conn2(self): config.kube_config.load_kube_config(config_file="kubeconfig.yaml") conn = client.CoreV1Api() return conn def podinfo(self,namespaced): conn = self.k8s_conn2() ret = conn.list_namespaced_pod(namespaced) for i in ret.items: appinfo = {} image = i.spec.containers[0].image.split(":")[-1] cpu = i.spec.containers[0].resources.limits['cpu'] memory = i.spec.containers[0].resources.limits['memory'] namespace = i.metadata.namespace app = i.metadata.labels['app'] appinfo['app'] = app appinfo['image'] = image appinfo['cpu'] = cpu appinfo['memory'] = memory appinfo['namespace'] = namespace self.applists.append(appinfo) def main(self): self.podinfo('arch') self.podinfo('enos') files = '%s.json' % region res = json.dumps(self.applists) with open(files,mode='w') as f: f.write(res) if __name__ == '__main__': k8s = myk8sinfo() k8s.main()
三、纯粹记录版本镜像
1）执行需给定环境
# coding: utf-8 import json,sys from kubernetes import client, config region = sys.argv[1] class myk8sinfo: def __init__(self): self.images = [] def k8s_conn(self): Token = "eyJhbGciOiJSUzI1NiIsImtpZCI6IiJ9.eyJpc3MiOiJrdWJlcm5ldGVzL3NlcnZpY2VhY2NvdW50Iiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9uYW1lc3BhY2UiOiJrdWJlLXN5c3RlbSIsImt1YmVybmV0ZXMuaW8vc2VydmljZWFjY291bnQvc2VjcmV0Lm5hbWUiOiJlbmlvdC1yZWFkZXItdG9rZW4tN2wydHgiLCJrdWJlcm5ldGVzLmlvL3NlcnZpY2VhY2NvdW50L3NlcnZpY2UtYWNjb3VudC5uYW1lIjoiZW5pb3QtcmVhZGVyIiwia3ViZXJuZXRlcy5pby9zZXJ2aWNlYWNjb3VudC9zZXJ2aWNlLWFjY291bnQudWlkIjoiNWFlNjVmMmUtOGQ5Ni0xMWU5LWIwNzUtMDAwZDNhYTA4OTA2Iiwic3ViIjoic3lzdGVtOnNlcnZpY2VhY2NvdW50Omt1YmUtc3lzdGVtOmVuaW90LXJlYWRlciJ9.kUjZhtjQPHvxtcjonB-EAQeAVTXSdyLoiXAirRRJ6hrSPeP2WhgAzF1h2v6BwV1uZulnWZ9g0rZCY5RKlAIvnYeDrgsy1EAZ40u7N4I5DBPKAxbljZAjUcF_jO-OSBNDfSOCrUFgY_rmC5uROzXNe6jlVQ300vI3BUAkLf0Q3_LB0uFmMXutz2ZzUTOssJgMTUKMh0qutbTuwGgiDvpyBjqTgXpZOBcY1cNzuXoI3pBPnb2WiMcVoaB4rp3ov9m04KD6k1OqwB4MeR0UCz7hCA-8J7N2GYzGdAtDYoqRniJ5tkrfH2hQYjKLak0JfKg9vBrePCKCodUxTGp24QLdjQ" APISERVER = 'https://apaas0001:8443' configuration = client.Configuration() configuration.host = APISERVER configuration.verify_ssl = False configuration.api_key = {"authorization": "Bearer " + Token} client.Configuration.set_default(configuration) conn = client.CoreV1Api() return conn def k8s_conn2(self): config.kube_config.load_kube_config(config_file="kubeconfig.yaml") conn = client.CoreV1Api() return conn def podinfo(self,namespaced): conn = self.k8s_conn2() ret = conn.list_namespaced_pod(namespaced) for i in ret.items: image = i.spec.containers[0].image.split(":")[-1] if image not in self.images: self.images.append(image) def main(self): self.podinfo('arch') self.podinfo('enos') self.podinfo('uscada') files = '%s.json' % region res = json.dumps(self.images) with open(files,mode='w') as f: f.write(res) if __name__ == '__main__': k8s = myk8sinfo() k8s.main()
2）对比镜像
import json,sys file1 = sys.argv[1] file2 = sys.argv[2] class Diff_images: def region_info(self,file): with open(file, 'r') as f: res = f.read() images = json.loads(res) return images def main(self): images1 = self.region_info(file1) images2 = self.region_info(file2) images1 = set(images1) images2 = set(images2) print(images2 - images1) if __name__ == '__main__': diff = Diff_images() diff.main()
3）综合接口。https://github.com/kubernetes-client/python/blob/master/kubernetes/README.md
from kubernetes import client, config class k8sCoreV1Api: def __init__(self,config_file): self.config_file=config_file self.config = config.kube_config.load_kube_config(config_file=self.config_file) self.v1 = client.CoreV1Api() def list_namespace(self): namespaces = [] for ns in self.v1.list_namespace().items: namespaces.append(ns.metadata.name) return namespaces def list_nodes(self): nodes = self.v1.list_node().items list_nodes = [] for node in nodes: nodeinfo = {} for addres in node.status.addresses: if addres.type == 'Hostname': nodeinfo['Hostname'] = addres.address if addres.type == 'InternalIP': nodeinfo['InternalIP'] = addres.address nodeinfo['cpu'] = node.status.allocatable['cpu'] nodeinfo['memory'] = node.status.allocatable['memory'] list_nodes.append(nodeinfo) return list_nodes class k8sAppsV1Api: def __init__(self,config_file): self.config_file=config_file self.config = config.kube_config.load_kube_config(config_file=self.config_file) self.v1 = client.AppsV1Api() def list_deployment(self): deployments = self.v1.list_deployment_for_all_namespaces().items list_deployments = [] for deploy in deployments: deploymentsinfo = {} namespace = deploy.metadata.namespace name = deploy.metadata.name deploymentsinfo['namespace'] = namespace deploymentsinfo['name'] = name list_deployments.append(deploymentsinfo) return list_deployments if __name__ == '__main__': config_file='config.yaml' test = k8sAppsV1Api(config_file) list_nodes = test.list_deployment() print(list_nodes)