Posted inPython

Tự động hóa Kubernetes bằng Python: Đừng chỉ dừng lại ở kubectl

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Tại sao phải dùng Python thay vì kubectl?

Kubectl là công cụ tuyệt vời cho các tác vụ đơn lẻ như check log hay restart một Pod. Tuy nhiên, hãy thử tưởng tượng bạn phải quét 50 cluster để tìm các Pod ngốn trên 80% RAM, hoặc dọn dẹp hàng trăm Namespace rác sau mỗi đợt testing. Lúc này, gõ lệnh tay là một cơn ác mộng vận hành.

Thực tế, mình từng bắt đầu với một script Python chỉ vỏn vẹn 200 dòng để dọn dẹp tài nguyên. Sau một năm, nó phát triển thành một hệ thống 2.000 dòng, tích hợp cả Slack bot để báo cáo sự cố theo thời gian thực. Bài học rút ra rất đơn giản: Đừng chỉ dùng script để chạy lệnh shell. Hãy tận dụng Python Kubernetes Client để tương tác trực tiếp với API Server.

Cài đặt và cấu hình môi trường

Chỉ cần một dòng lệnh để cài đặt thư viện chính thức từ Kubernetes Team:

pip install kubernetes

Thư viện này cần file kubeconfig (thường ở ~/.kube/config) để xác thực. Nếu lệnh kubectl get nodes của bạn đang chạy bình thường, script Python cũng sẽ hoạt động trơn tru.

Kết nối với Cluster: Local vs In-Cluster

Đây là bước đầu tiên và cũng là nơi nhiều bạn mới hay nhầm lẫn. Script của bạn có hai cách để “nói chuyện” với Kubernetes tùy vào nơi nó thực thi.

  • External: Chạy từ máy cá nhân hoặc server ngoài cluster, sử dụng file config cục bộ.
  • Internal: Chạy như một Pod nằm ngay trong Cluster, sử dụng Service Account được cấp quyền.
from kubernetes import client, config

def connect_k8s():
    try:
        # Ưu tiên load cấu hình nếu script đang chạy bên trong một Pod
        config.load_incluster_config()
        print("Đang chạy mode: In-Cluster")
    except config.ConfigException:
        # Tự động tìm file ~/.kube/config nếu chạy ở máy local
        config.load_kube_config()
        contexts, active_context = config.list_kube_config_contexts()
        print(f"Đang chạy mode: Local | Context: {active_context['name']}")

connect_k8s()

Quản lý Pods với CoreV1Api

Trong thế giới Kubernetes API, CoreV1Api quản lý các tài nguyên cơ bản nhất như Pods, Nodes và Namespaces. Giả sử bạn cần liệt kê trạng thái của tất cả Pod trong namespace “default” để kiểm tra sức khỏe hệ thống:

v1 = client.CoreV1Api()
print("Danh sách Pod và IP tương ứng:")
# Thêm label_selector để lọc chính xác ứng dụng cần tìm
ret = v1.list_namespaced_pod(namespace="default", label_selector="app=nginx")
for i in ret.items:
    print(f"{i.status.pod_ip:15} | {i.metadata.name:30} | {i.status.phase}")

Kinh nghiệm của mình là luôn kiểm tra trường status.phase. Một Pod tồn tại không có nghĩa là nó đang chạy. Nó có thể kẹt ở Pending hoặc CrashLoopBackOff cả ngày nếu bạn không bắt lỗi logic ở bước này.

Tự động hóa Deployments với AppsV1Api

Nếu CoreV1Api xử lý các đối tượng đơn lẻ, thì AppsV1Api là “tổng tư lệnh” điều phối Deployments và StatefulSets. Đây là công cụ đắc lực để thực hiện Rolling Update tự động mà không cần can thiệp thủ công.

Ví dụ, đoạn code sau sẽ cập nhật image mới cho một Deployment trong môi trường production:

def update_deployment_image(name, namespace, new_image):
    apps_v1 = client.AppsV1Api()
    
    # Đọc cấu hình hiện tại của deployment
    deployment = apps_v1.read_namespaced_deployment(name=name, namespace=namespace)
    
    # Cập nhật image cho container đầu tiên trong danh sách
    deployment.spec.template.spec.containers[0].image = new_image
    
    # Gửi lệnh patch để API Server thực hiện cập nhật
    apps_v1.patch_namespaced_deployment(name=name, namespace=namespace, body=deployment)
    print(f"[OK] {name} đã lên đời image: {new_image}")

# Sử dụng thực tế
# update_deployment_image("api-gateway", "prod", "my-repo/api:v2.0.1")

Lắng nghe sự kiện (Watch Events) theo thời gian thực

Thay vì tạo vòng lặp vô tận để hỏi API Server “Pod xong chưa?”, hãy sử dụng cơ chế watch. Cách này giúp script nhận thông báo ngay lập tức khi có thay đổi, giảm tải cho hệ thống và tăng tốc độ phản ứng.

from kubernetes import watch

v1 = client.CoreV1Api()
w = watch.Watch()

print("Đang streaming sự kiện từ Cluster...")
for event in w.stream(v1.list_namespaced_pod, namespace="default", timeout_seconds=60):
    pod_name = event['object'].metadata.name
    event_type = event['type']
    
    if event_type == 'DELETED':
        print(f"🔥 CẢNH BÁO: Pod {pod_name} vừa bị xóa!")
    elif event_type == 'MODIFIED':
        print(f"⚙️ Pod {pod_name} vừa thay đổi trạng thái.")

Nâng cấp cấu trúc code: Từ Script đến System

Đừng lặp lại sai lầm của mình bằng việc nhồi nhét mọi thứ vào một file main.py. Khi hệ thống automation lớn dần, việc duy trì sẽ trở nên cực kỳ khó khăn. Hãy lưu ý ba nguyên tắc vàng sau:

  1. Đóng gói (Encapsulation): Gom các logic xử lý K8s vào Class như K8sManager để tái sử dụng.
  2. Xử lý ngoại lệ: API Server thường trả về lỗi 404 (không tìm thấy) hoặc 403 (sai quyền). Luôn bao bọc code trong try-except với ApiException.
  3. Ghi log chuyên nghiệp: Thay print() bằng thư viện logging. Bạn sẽ cần các file log này khi script chạy ngầm (background job) bị lỗi.

Chuyển từ tư duy “gõ lệnh” sang “lập trình hệ thống” giúp bạn làm chủ hạ tầng hoàn toàn. Kubernetes không chỉ là công cụ điều phối, nó là một nền tảng lập trình đầy quyền năng.

Hãy thử viết một script nhỏ tự động xóa các Pod bị Evicted để dọn dẹp cluster của bạn ngay hôm nay. Chúc bạn thành công!

Share:
Tags: