通過將身份認證令牌直接傳給 API 服務器，可以避免使用 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 服務器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)# 使用令牌玩轉 APIcurl -X GET $APISERVER/api --header 'Authorization: Bearer $TOKEN' --insecure

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

python舉例：

目錄結構

配置文件兩種方式

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)

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