Python で Google Kubernetes Engine (GKE) 用の Kubernetes クライアントを構築する

このブログ投稿では、Python で GKE 用の Kubernetes クライアントを作成する効果的な方法を紹介します。 google-cloud-container、google-auth、kubernetes ライブラリを活用すると、アプリケーションがローカルで実行されているか、Google Cloud 上で実行されているかに関係なく、同じコードを使用して Kubernetes API とやり取りできます。この柔軟性は、Application Default Credentials (ADC) を使用して Kubernetes API の対話に必要なリクエストを認証し、動的に構築することで実現され、追加のツールや kubeconfig などの構成ファイルが不要になります。

ローカルで実行する場合、一般的なアプローチは、gcloudcontainerclustersget-credentials コマンドを使用して kubeconfig ファイルを生成し、kubectl を使用して Kubernetes API と対話することです。このワークフローはローカル設定では自然で効果的ですが、Cloud Run や他の Google Cloud サービスのような環境では実用的ではなくなります。

ADC を使用すると、Kubernetes クライアントを動的に構成することで、GKE クラスタの Kubernetes API へのアクセスを効率化できます。このアプローチにより、外部構成ファイルの管理や追加のツールのインストールのオーバーヘッドを発生させずに、クラスターに接続するための一貫した効率的な方法が保証されます。

---

前提条件

1. Google Cloudによる認証

コードをローカルで実行している場合は、次のコマンドを使用して認証するだけです:

gcloud auth application-default login

これにより、ユーザー アカウントの認証情報がアプリケーションのデフォルト認証情報 (ADC) として使用されます。

Cloud Run などの Google Cloud サービスでコードを実行している場合は、認証を手動で処理する必要はありません。サービスに、GKE クラスタにアクセスするために必要な権限が付与された、適切に構成されたサービス アカウントがあることを確認してください。

---

2. クラスターの詳細を収集する

スクリプトを実行する前に、次の詳細を確認してください:

• Google Cloud プロジェクト ID: GKE クラスターがホストされているプロジェクトの ID。
• クラスターの場所: クラスターが配置されているリージョンまたはゾーン (例: us-central1-a)。
• クラスター名: 接続する Kubernetes クラスターの名前。

---

スクリプト

以下は、GKE クラスタ用の Kubernetes クライアントをセットアップする Python 関数です。

gcloud auth application-default login

---

仕組み

1. GKE クラスタへの接続

get_k8s_client 関数は、google-cloud-container ライブラリを使用して GKE からクラスタの詳細を取得することから始まります。このライブラリは GKE サービスと対話し、クラスタの API エンドポイントや認証局 (CA) などの情報を取得できるようにします。これらの詳細は、Kubernetes クライアントの構成に不可欠です。

from google.cloud import container_v1
import google.auth
import google.auth.transport.requests
from kubernetes import client as kubernetes_client
from tempfile import NamedTemporaryFile
import base64
import yaml

def get_k8s_client(project_id: str, location: str, cluster_id: str) -> kubernetes_client.CoreV1Api:
    """
    Fetches a Kubernetes client for the specified GCP project, location, and cluster ID.

    Args:
        project_id (str): Google Cloud Project ID
        location (str): Location of the cluster (e.g., "us-central1-a")
        cluster_id (str): Name of the Kubernetes cluster

    Returns:
        kubernetes_client.CoreV1Api: Kubernetes CoreV1 API client
    """

    # Retrieve cluster information
    gke_cluster = container_v1.ClusterManagerClient().get_cluster(request={
        "name": f"projects/{project_id}/locations/{location}/clusters/{cluster_id}"
    })

    # Obtain Google authentication credentials
    creds, _ = google.auth.default()
    auth_req = google.auth.transport.requests.Request()
    # Refresh the token
    creds.refresh(auth_req)

    # Initialize the Kubernetes client configuration object
    configuration = kubernetes_client.Configuration()
    # Set the cluster endpoint
    configuration.host = f'https://{gke_cluster.endpoint}'

    # Write the cluster CA certificate to a temporary file
    with NamedTemporaryFile(delete=False) as ca_cert:
        ca_cert.write(base64.b64decode(gke_cluster.master_auth.cluster_ca_certificate))
        configuration.ssl_ca_cert = ca_cert.name

    # Set the authentication token
    configuration.api_key_prefix['authorization'] = 'Bearer'
    configuration.api_key['authorization'] = creds.token

    # Create and return the Kubernetes CoreV1 API client
    return kubernetes_client.CoreV1Api(kubernetes_client.ApiClient(configuration))


def main():
    project_id = "your-project-id"  # Google Cloud Project ID
    location = "your-cluster-location"  # Cluster region (e.g., "us-central1-a")
    cluster_id = "your-cluster-id"  # Cluster name

    # Retrieve the Kubernetes client
    core_v1_api = get_k8s_client(project_id, location, cluster_id)

    # Fetch the kube-system Namespace
    namespace = core_v1_api.read_namespace(name="kube-system")

    # Output the Namespace resource in YAML format
    yaml_output = yaml.dump(namespace.to_dict(), default_flow_style=False)
    print(yaml_output)

if __name__ == "__main__":
    main()

google-cloud-container ライブラリは、Kubernetes API と直接ではなく、サービスとして GKE と対話するように設計されていることに注意することが重要です。たとえば、このライブラリを使用してクラスタ情報の取得、クラスタのアップグレード、またはメンテナンス ポリシーの構成を行うことはできますが、これは gcloudcontainerclusters コマンドで実行できることと同様ですが、Kubernetes API クライアントを直接取得するために使用することはできません。この違いが、この関数が必要なクラスタの詳細を GKE から取得した後に Kubernetes クライアントを個別に構築する理由です。

---

2. Google Cloud での認証

GKE および Kubernetes API と通信するために、この関数は Google Cloud のアプリケーションデフォルト認証情報 (ADC) を使用して認証します。認証プロセスの各ステップがどのように機能するかは次のとおりです:

google.auth.default()

この関数は、コードが実行されている環境の ADC を取得します。コンテキストに応じて、次のものが返される場合があります:

• ユーザー アカウントの認証情報 (例: ローカル開発設定の gcloud auth application-default ログインから)。
• サービス アカウントの認証情報 (例: Cloud Run などの Google Cloud 環境で実行する場合)。

利用可能な場合は関連するプロジェクト ID も返しますが、この場合は資格情報のみが使用されます。

google.auth.transport.requests.Request()

これにより、認証関連のネットワーク要求を処理するための HTTP 要求オブジェクトが作成されます。 Python のリクエスト ライブラリを内部で使用し、資格情報を更新したり、アクセス トークンをリクエストしたりするための標準化された方法を提供します。

creds.refresh(auth_req)

google.auth.default() を使用して ADC が取得される場合、資格情報オブジェクトには最初はアクセス トークンが含まれません (少なくともローカル環境では)。 fresh() メソッドは、アクセス トークンを明示的に取得し、資格情報オブジェクトに添付して、API リクエストを認証できるようにします。

次のコードは、この動作を確認する方法を示しています。

gke_cluster = container_v1.ClusterManagerClient().get_cluster(request={
    "name": f"projects/{project_id}/locations/{location}/clusters/{cluster_id}"
})

出力例:

# Obtain Google authentication credentials
creds, _ = google.auth.default()
auth_req = google.auth.transport.requests.Request()

# Inspect credentials before refreshing
print(f"Access Token (before refresh()): {creds.token}")
print(f"Token Expiry (before refresh()): {creds.expiry}")

# Refresh the token
creds.refresh(auth_req)

# Inspect credentials after refreshing
print(f"Access Token (after): {creds.token}")
print(f"Token Expiry (after): {creds.expiry}")

refresh() を呼び出す前、トークン属性は None です。 fresh() が呼び出された後、認証情報には有効なアクセス トークンとその有効期限が設定されます。

---

3. Kubernetes クライアントの構成

Kubernetes クライアントは、クラスターの API エンドポイント、CA 証明書の一時ファイル、および更新されたベアラー トークンを使用して構成されます。これにより、クライアントが安全に認証され、クラスターと通信できるようになります。

gcloud auth application-default login

CA 証明書は一時的に保存され、安全な SSL 通信のためにクライアントによって参照されます。これらの設定により、Kubernetes クライアントは完全に構成され、クラスターと対話する準備が整いました。

---

出力例

kube-system 名前空間の YAML 出力の例を次に示します。

from google.cloud import container_v1
import google.auth
import google.auth.transport.requests
from kubernetes import client as kubernetes_client
from tempfile import NamedTemporaryFile
import base64
import yaml

def get_k8s_client(project_id: str, location: str, cluster_id: str) -> kubernetes_client.CoreV1Api:
    """
    Fetches a Kubernetes client for the specified GCP project, location, and cluster ID.

    Args:
        project_id (str): Google Cloud Project ID
        location (str): Location of the cluster (e.g., "us-central1-a")
        cluster_id (str): Name of the Kubernetes cluster

    Returns:
        kubernetes_client.CoreV1Api: Kubernetes CoreV1 API client
    """

    # Retrieve cluster information
    gke_cluster = container_v1.ClusterManagerClient().get_cluster(request={
        "name": f"projects/{project_id}/locations/{location}/clusters/{cluster_id}"
    })

    # Obtain Google authentication credentials
    creds, _ = google.auth.default()
    auth_req = google.auth.transport.requests.Request()
    # Refresh the token
    creds.refresh(auth_req)

    # Initialize the Kubernetes client configuration object
    configuration = kubernetes_client.Configuration()
    # Set the cluster endpoint
    configuration.host = f'https://{gke_cluster.endpoint}'

    # Write the cluster CA certificate to a temporary file
    with NamedTemporaryFile(delete=False) as ca_cert:
        ca_cert.write(base64.b64decode(gke_cluster.master_auth.cluster_ca_certificate))
        configuration.ssl_ca_cert = ca_cert.name

    # Set the authentication token
    configuration.api_key_prefix['authorization'] = 'Bearer'
    configuration.api_key['authorization'] = creds.token

    # Create and return the Kubernetes CoreV1 API client
    return kubernetes_client.CoreV1Api(kubernetes_client.ApiClient(configuration))


def main():
    project_id = "your-project-id"  # Google Cloud Project ID
    location = "your-cluster-location"  # Cluster region (e.g., "us-central1-a")
    cluster_id = "your-cluster-id"  # Cluster name

    # Retrieve the Kubernetes client
    core_v1_api = get_k8s_client(project_id, location, cluster_id)

    # Fetch the kube-system Namespace
    namespace = core_v1_api.read_namespace(name="kube-system")

    # Output the Namespace resource in YAML format
    yaml_output = yaml.dump(namespace.to_dict(), default_flow_style=False)
    print(yaml_output)

if __name__ == "__main__":
    main()

---

結論

このアプローチは、ローカルで実行されているのか、Cloud Run などの Google Cloud サービス上で実行されているのかに関係なく、同じコードを使用して Kubernetes API と対話する移植性を強調しています。アプリケーションのデフォルト認証情報 (ADC) を活用することで、事前に生成された構成ファイルや外部ツールに依存せずに、Kubernetes API クライアントを動的に生成する柔軟な方法を実証しました。これにより、さまざまな環境にシームレスに適応できるアプリケーションの構築が容易になり、開発と展開の両方のワークフローが簡素化されます。

以上がPython で Google Kubernetes Engine (GKE) 用の Kubernetes クライアントを構築するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。