Python 은 Kubernetes API 를 사용 하여 클 러 스 터 에 접근 합 니 다.

인증 토 큰 을 API 서버 에 직접 전달 함으로써 kubectl 프 록 시 를 사용 하 는 것 을 피 할 수 있 습 니 다.이렇게:
grep/cut 방식 사용:

#        ,     .kubeconfig             
kubectl config view -o jsonpath='{"Cluster name\tServer
"}{range .clusters[*]}{.name}{"\t"}{.cluster.server}{"
"}{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) # API curl -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,config
from kubernetes.stream import stream
import yaml
config_file = r"D:\Users\JackHe\PycharmProjects\JJ\k8s\auth\kubeconfig.yaml"
config.kube_config.load_kube_config(config_file=config_file)
Api_Instance = client.CoreV1Api()
Api_Batch = client.BatchV1Api()

#     namesapce
for ns in Api_Instance.list_namespace().items:
    print(ns.metadata.name)

#     nodes
def 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 data
nodes = list_node()
print(nodes)

2.token 형식 을 사용 하여 명령 에 표 시 된 것 을 가 져 옵 니 다.
코드:

# -*- coding: utf-8 -*-
from kubernetes.client import api_client
from kubernetes.client.apis import core_v1_api
from kubernetes import client,config


class KubernetesTools(object):
    def __init__(self):
        self.k8s_url = 'https://192.168.1.56:6443'

    def get_token(self):
        """
          token
        :return:
        """
        with open(r'D:\Users\JackHe\PycharmProjects\JJ\k8s\auth\token', 'r') as file:
            Token = file.read().strip('
') return Token def get_api(self): """ API CoreV1Api :return: """ configuration = client.Configuration() configuration.host = self.k8s_url configuration.verify_ssl = False configuration.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("%s\t%s\t%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
" %(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)
파 이 썬 이 Kubernetes API 를 사용 하여 클 러 스 터 를 방문 하 는 것 에 관 한 이 글 은 여기까지 소개 되 었 습 니 다.더 많은 파 이 썬 Kubernetes API 방문 클 러 스 터 내용 은 우리 의 이전 글 을 검색 하거나 아래 의 관련 글 을 계속 조회 하 십시오.앞으로 많은 지원 을 바 랍 니 다!

좋은 웹페이지 즐겨찾기