通過將身份認證令牌直接傳給 API 服務(wù)器，可以避免使用 kubectl 代理，像這樣：使用 grep/cut 方式：

# 查看所有的集群，因為你的 .kubeconfig 文件中可能包含多個上下文kubectl config view -o jsonpath=’{'Cluster nametServern'}{range .clusters[*]}{.name}{'t'}{.cluster.server}{'n'}{end}’# 從上述命令輸出中選擇你要與之交互的集群的名稱export CLUSTER_NAME='some_server_name'# 指向引用該集群名稱的 API 服務(wù)器APISERVER=$(kubectl config view -o jsonpath='{.clusters[?(@.name=='$CLUSTER_NAME')].cluster.server}')# 獲得令牌TOKEN=$(kubectl get secrets -o jsonpath='{.items[?(@.metadata.annotations[’kubernetes.io/service-account.name’]==’default’)].data.token}'|base64 -d)# 使用令牌玩轉(zhuǎn) APIcurl -X GET $APISERVER/api --header 'Authorization: Bearer $TOKEN' --insecure

客戶端庫：https://kubernetes.io/zh/docs/reference/using-api/client-libraries/

python舉例：

目錄結(jié)構(gòu)

配置文件兩種方式

1、將集群中的~/.kube/config,重命名為kubeconfig.yaml代碼：

from kubernetes import client,configfrom kubernetes.stream import streamimport yamlconfig_file = r'D:UsersJackHePycharmProjectsJJk8sauthkubeconfig.yaml'config.kube_config.load_kube_config(config_file=config_file)Api_Instance = client.CoreV1Api()Api_Batch = client.BatchV1Api()#列出所有的namesapcefor ns in Api_Instance.list_namespace().items: print(ns.metadata.name)#列出所有的nodesdef 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 datanodes = list_node()print(nodes)

2、使用token形式，獲取命令上文所示。代碼：

# -*- coding: utf-8 -*-from kubernetes.client import api_clientfrom kubernetes.client.apis import core_v1_apifrom kubernetes import client,configclass KubernetesTools(object): def __init__(self):self.k8s_url = ’https://192.168.1.56:6443’ def get_token(self):'''獲取token:return:'''with open(r’D:UsersJackHePycharmProjectsJJk8sauthtoken’, ’r’) as file: Token = file.read().strip(’n’) return Token def get_api(self):'''獲取API的CoreV1Api版本對象:return:'''configuration = client.Configuration()configuration.host = self.k8s_urlconfiguration.verify_ssl = Falseconfiguration.api_key = {'authorization': 'Bearer ' + self.get_token()}client1 = api_client.ApiClient(configuration=configuration)api = core_v1_api.CoreV1Api(client1)return api def get_namespace_list(self):'''獲取命名空間列表:return:'''api = self.get_api()namespace_list = []for ns in api.list_namespace().items: # print(ns.metadata.name) namespace_list.append(ns.metadata.name)return namespace_list def get_pod_list(self): api = self.get_api() print('Listing pods with their IPs:') ret = api.list_pod_for_all_namespaces(watch=False) for i in ret.items: print('%st%st%s' % (i.status.pod_ip, i.metadata.namespace, i.metadata.name)) def get_service_list(self):api = self.get_api()ret = api.list_service_for_all_namespaces(watch=False)for i in ret.items: print('%s t%s t%s t%s t%s n' %(i.kind,i.metadata.namespace,i.metadata.name,i.spec.cluster_ip,i.spec.ports))if __name__ == ’__main__’: namespace_list = KubernetesTools().get_namespace_list() pod_list = KubernetesTools().get_pod_list() service = KubernetesTools().get_service_list() print(namespace_list) print(pod_list) print(service)

到此這篇關(guān)于Python使用Kubernetes API訪問集群的文章就介紹到這了,更多相關(guān)Python Kubernetes API訪問集群內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！