Posted inPython

PythonでKubernetesを自動化：kubectlの先へ進もう

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

なぜkubectlではなくPythonを使うのか？

kubectlは、ログの確認やPodの再起動といった単発のタスクには非常に優れたツールです。しかし、50個のクラスターをスキャンしてRAM使用率が80%を超えているPodを探したり、テスト後に不要になった数百のNamespaceをクリーンアップしたりすることを想像してみてください。このような場合、手動でコマンドを打ち続けるのは運用上の悪夢となります。

実際、私も最初はリソースを整理するためのわずか200行のPythonスクリプトから始めました。1年後、それは2,000行のシステムへと成長し、Slackボットと連携してリアルタイムで障害を報告するまでになりました。そこから得た教訓はシンプルです。単にシェルコマンドを実行するためだけにスクリプトを使うのではなく、Python Kubernetes Clientを活用してAPIサーバーと直接やり取りしましょう。

環境の構築と設定

Kubernetesチームが提供している公式ライブラリをインストールするには、以下のコマンドを実行するだけです：

pip install kubernetes

このライブラリは、認証のためにkubeconfigファイル（通常は~/.kube/config）を必要とします。もしkubectl get nodesコマンドが正常に動作している環境であれば、Pythonスクリプトも問題なく動作します。

クラスターへの接続：Local vs In-Cluster

これは最初のステップであり、初心者が間違いやすいポイントでもあります。スクリプトが実行される場所によって、Kubernetesと「対話」する方法は2つあります。

  • External: ローカルPCやクラスター外のサーバーから実行し、ローカルの設定ファイルを使用します。
  • Internal: クラスター内のPodとして実行し、付与されたService Accountの権限を使用します。
from kubernetes import client, config

def connect_k8s():
    try:
        # スクリプトがPod内で実行されている場合に設定をロード
        config.load_incluster_config()
        print("実行モード: In-Cluster")
    except config.ConfigException:
        # ローカル環境で実行されている場合は~/.kube/configを自動的に検索
        config.load_kube_config()
        contexts, active_context = config.list_kube_config_contexts()
        print(f"実行モード: Local | コンテキスト: {active_context['name']}")

connect_k8s()

CoreV1ApiによるPod管理

Kubernetes APIの世界では、CoreV1ApiがPod、Node、Namespaceといった最も基本的なリソースを管理します。例えば、システムの健全性を確認するために、”default”ネームスペース内のすべてのPodの状態をリスト化する必要があるとしましょう。

v1 = client.CoreV1Api()
print("Podリストと対応するIP:")
# 特定のアプリケーションをフィルタリングするためにlabel_selectorを追加
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}")

私の経験上、常にstatus.phaseフィールドを確認することをお勧めします。Podが存在しているからといって、必ずしも正常に動作しているとは限りません。この段階でロジックのミスをキャッチしないと、Podが一日中PendingやCrashLoopBackOffで止まっているのを見逃す可能性があります。

AppsV1ApiによるDeploymentの自動化

CoreV1Apiが個別のオブジェクトを扱うのに対し、AppsV1ApiはDeploymentやStatefulSetを制御する「総司令官」のような役割を担います。これは、手動操作なしで自動ローリングアップデートを実行するための強力なツールになります。

例えば、以下のコードは本番環境のDeploymentに新しいイメージを適用します：

def update_deployment_image(name, namespace, new_image):
    apps_v1 = client.AppsV1Api()
    
    # Deploymentの現在の設定を読み込む
    deployment = apps_v1.read_namespaced_deployment(name=name, namespace=namespace)
    
    # リスト内の最初のコンテナのイメージを更新
    deployment.spec.template.spec.containers[0].image = new_image
    
    # APIサーバーにパッチを送信して更新を実行
    apps_v1.patch_namespaced_deployment(name=name, namespace=namespace, body=deployment)
    print(f"[OK] {name} のイメージを更新しました: {new_image}")

# 実際の使用例
# update_deployment_image("api-gateway", "prod", "my-repo/api:v2.0.1")

リアルタイムでのイベント監視（Watch Events）

APIサーバーに対して「Podは完了したか？」と何度も問い合わせる無限ループを作る代わりに、watchメカニズムを使用しましょう。これにより、スクリプトは変更が発生した瞬間に通知を受け取ることができ、システム負荷を抑えつつレスポンス速度を向上させることができます。

from kubernetes import watch

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

print("クラスターからのイベントをストリーミング中...")
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"🔥 警告: Pod {pod_name} が削除されました！")
    elif event_type == 'MODIFIED':
        print(f"⚙️ Pod {pod_name} の状態が変更されました。")

コード構造のアップグレード：スクリプトからシステムへ

すべてを一つのmain.pyファイルに詰め込むという、私と同じ過ちを繰り返さないでください。自動化システムが成長するにつれ、メンテナンスは非常に困難になります。以下の3つの黄金律を意識しましょう：

  1. カプセル化 (Encapsulation): 再利用性を高めるため、Kubernetes의 処理ロジックをK8sManagerのようなクラスにまとめます。
  2. 例外処理: APIサーバーは、404 (Not Found) や 403 (Forbidden) などのエラーを返すことがあります。常にコードをtry-exceptで囲み、ApiExceptionを適切に処理しましょう。
  3. プロフェッショナルなロギング: print()の代わりにloggingライブラリを使用します。スクリプトをバックグラウンドジョブとして実行する場合、エラー発生時の調査にログファイルが不可欠です。

「コマンドを打つ」という考え方から「システムをプログラミングする」という考え方にシフトすることで、インフラを完全に制御できるようになります。Kubernetesは単なるオーケストレーションツールではなく、強力なプログラミングプラットフォームなのです。

まずは、クラスターを整理するためにEvicted状態のPodを自動削除する小さなスクリプトから書いてみてください。成功を祈っています！

Share:
Tags: