[TIL] Kubernetes Python API로 파이썬 어플리케이션에서 클러스터 정보 가져오기

cjkangme·2025년 10월 12일

TIL 쿠버네티스

TIL

목록 보기

42/43

Kubernetes Python API

종종 쿠버네티스 클러스터에 배포된 pod안에서 모니터링 등을 위해 클러스터에 배포된 다른 파드나 노드 정보에 접근해야 할 때가 있습니다.

쿠퍼네티스는 이런 상황에서 사용할 수 있도록 프로그래밍 언어별로 쿠버네티스 클라이언트 API를 제공하는데요. 이번에 직접 사용해보면서 배운점과 겪었던 문제점들을 바탕으로 정리해보려고 합니다.

Python Kubernetes Client 설치

Kubernetes는 다양한 프로그래밍 언어를 위한 공식 클라이언트 라이브러리를 제공합니다. Python의 경우 PyPI에서 kubernetes 패키지를 설치할 수 있습니다.

pip install kubernetes

Config 로드 방법

쿠버네티스 파이썬 클라이언트를 통해 쿠버네티스 API를 사용하려면 먼저 config를 로드해야 합니다.

# 공식 예제 (https://github.com/kubernetes-client/python/)
from kubernetes import client, config

# Configs can be set in Configuration class directly or using helper utility
config.load_kube_config()

v1 = client.CoreV1Api()
print("Listing pods with their IPs:")
ret = v1.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))

이 때 config를 불러오는 방법에는 크게 세 가지가 있습니다.

load_kube_config(): 로컬 환경에서 사용. ~/.kube/config 파일을 자동으로 읽어옵니다.
load_incluster_config(): Pod 내부에서 사용. Pod에 마운트되어있는Service Account 정보를 사용합니다.
Configuration() 수동 생성: 위 두 함수들은 Configuration을 자동으로 읽어오는 함수이고, 직접 클래스를 생성한 뒤 설정하는 방법이 있습니다. 발급된 ClusterRole에 대한 token key와 클러스터 주소 등을 직접 설정하면 되는데, 값을 직접 넣어주는 것이다 보니 추후 에러가 발생할 여지가 많아 여기서 고려하지는 않겠습니다.

Pod 내부에서 실행하므로 저는 load_incluster_config()를 사용하겠습니다.

from kubernetes import client, config

# load_incluster_config() 사용
config.load_incluster_config()

v1 = client.CoreV1Api()
custom_api = client.CustomObjectsApi()

클라이언트 별로 사용가능한 API는 공식 레포의 README 파일을 참조하시면 됩니다.

권한 문제

하지만 아무런 사전 설정 없이 Pod 내부에서 쿠버네티스 API를 호출하면 다음과 같은 403 Forbidden 에러가 발생할 수 있습니다.

kubernetes.client.exceptions.ApiException: (403)
Reason: Forbidden
message: "nodes.metrics.k8s.io is forbidden: User \"system:serviceaccount:dev:default\" 
cannot list resource \"nodes\" in API group \"metrics.k8s.io\" in the namespace \"dev\""

이는 파드가 사용하는 ServiceAccount에 해당 API에 접근 할 수 있는 권한이 없기 때문입니다.

Service Account와 권한 구조

쿠버네티스에서 파드는 ServiceAccount를 통해 API 서버에 인증합니다. 각 네임스페이스에는 기본값으로 default라는 ServiceAccount가 존재합니다. 네임스페이스는 최소 1개의 ServiceAccount가 항상 존재해야 합니다.

하지만 ServiceAccount 자체는 권한을 갖고 있지 않습니다. 권한 관리에는 다음 3가지 리소스가 필요합니다.

ServiceAccount: 파드의 인증 주체
ClusterRole (또는 Role): 실제 권한을 정의
ClusterRoleBinding (또는 RoleBinding): ServiceAccount와 ClusterRole을 연결

즉 권한을 실제로 설정하는 것은 ClusterRole이고, ClusterRoleBinding을 통해 Role을 ServiceAccount에 연결하는 것입니다.
마지막으로 파드가 ServiceAccount를 사용하도록 spec에 명시하면, 해당 파드가 ClusterRole에 명시된 권한을 얻게 됩니다.

권한 설정 예시

먼저 ServiceAccount를 생성합니다.

apiVersion: v1
kind: ServiceAccount
metadata:
  name: node-reader
  namespace: dev

다음으로 필요한 권한을 정의하는 ClusterRole을 생성합니다.
노드에 접근할 수 있는 권한을 설정하겠습니다.

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: node-reader-role
rules:
- apiGroups: ["metrics.k8s.io"]
  resources: ["nodes"]
  verbs: ["get", "list"]

ClusterRole은 클러스터 자체에 대한 권한이므로 namespace와는 상관없는 전역적인 자원입니다.

마지막으로 ClusterRoleBinding을 생성해 ServiceAccount와 ClusterRole을 연결합니다.

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: node-reader-binding
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: node-reader-role
subjects:
- kind: ServiceAccount
  name: node-reader
  namespace: dev

마지막으로 Pod 생성 시 serviceAccountName을 명시합니다.

apiVersion: v1
kind: Pod
metadata:
  name: my-pod
spec:
  serviceAccountName: node-reader
  containers:
  - name: workspace
    image: ubuntu:22.04

이제 파드 내부에 배포되는 파이썬 프로그램에서 load_incluster_config()를 사용하면 쿠버네티스 API를 통해 노드 정보를 획득할 수 있습니다.

from kubernetes import client, config

config.load_incluster_config()
custom_api = client.CustomObjectsApi()

def get_node_metrics():
    # API 사용 (권한이 없다면 여기서 403 Forbidden Error)
    metrics = custom_api.list_custom_object_for_all_namespaces(
        group="metrics.k8s.io",
        version="v1beta1",
        resource_plural="nodes",
    )

    for item in metrics["items"]:
        name = item["metadata"]["name"]
        cpu_usage = item["usage"]["cpu"]
        memory_usage = item["usage"]["memory"]
        
        print(f"Name: {name}, CPU: {cpu_usage}, Memory: {memory_usage}\n")
        
		cpu_postfix = cpu_usage[-1]
        if cpu_postfix == "m":
            cpu_cores = float(cpu_usage[:-1]) / 1000
        elif cpu_postfix == "n":
            cpu_cores = float(cpu_usage[:-1]) / 1_000_000_000
        else:
            cpu_cores = float(cpu_usage)

        print(f"Name: {name}, CPU: {cpu_cores}, Memory: {memory_usage}\n")