Building a Kubernetes Client for Google Kubernetes Engine (GKE) in Python

This blog post introduces an effective method for creating a Kubernetes client for GKE in Python. By leveraging the google-cloud-container, google-auth, and kubernetes libraries, you can use the same code to interact with the Kubernetes API regardless of whether your application is running locally or on Google Cloud. This flexibility comes from using Application Default Credentials (ADC) to authenticate and dynamically construct the requests needed for Kubernetes API interactions, eliminating the need for additional tools or configuration files like kubeconfig.

When running locally, a common approach is to use the gcloud container clusters get-credentials command to generate a kubeconfig file and interact with the Kubernetes API using kubectl. While this workflow is natural and effective for local setups, it becomes less practical in environments like Cloud Run or other Google Cloud services.

With ADC, you can streamline access to the Kubernetes API for GKE clusters by dynamically configuring the Kubernetes client. This approach ensures a consistent, efficient way to connect to your cluster without the overhead of managing external configuration files or installing extra tools.

---

Prerequisites

1. Authentication with Google Cloud

If you're running the code locally, simply authenticate using the following command:

gcloud auth application-default login

This will use your user account credentials as the Application Default Credentials (ADC).

If you're running the code on Google Cloud services like Cloud Run, you don’t need to handle authentication manually. Just ensure that the service has a properly configured service account attached with the necessary permissions to access the GKE cluster.

---

2. Gather Your Cluster Details

Before running the script, make sure you have the following details:

• Google Cloud Project ID: The ID of the project where your GKE cluster is hosted.
• Cluster Location: The region or zone where your cluster is located (e.g., us-central1-a).
• Cluster Name: The name of the Kubernetes cluster you want to connect to.

---

The Script

Below is the Python function that sets up a Kubernetes client for a GKE cluster.

gcloud auth application-default login

---

How It Works

1. Connecting to the GKE Cluster

The get_k8s_client function begins by fetching cluster details from GKE using the google-cloud-container library. This library interacts with the GKE service, allowing you to retrieve information such as the cluster's API endpoint and certificate authority (CA). These details are essential for configuring the Kubernetes client.

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()

It’s important to note that the google-cloud-container library is designed for interacting with GKE as a service, not directly with Kubernetes APIs. For example, while you can use this library to retrieve cluster information, upgrade clusters, or configure maintenance policies—similar to what you can do with the gcloud container clusters command—you cannot use it to directly obtain a Kubernetes API client. This distinction is why the function constructs a Kubernetes client separately after fetching the necessary cluster details from GKE.

---

2. Authenticating with Google Cloud

To interact with GKE and Kubernetes APIs, the function uses Google Cloud’s Application Default Credentials (ADC) to authenticate. Here's how each step of the authentication process works:

google.auth.default()

This function retrieves the ADC for the environment in which the code is running. Depending on the context, it may return:

• User account credentials (e.g., from gcloud auth application-default login in a local development setup).
• Service account credentials (e.g., when running in a Google Cloud environment like Cloud Run).

It also returns the associated project ID if available, although in this case, only the credentials are used.

google.auth.transport.requests.Request()

This creates an HTTP request object for handling authentication-related network requests. It uses Python's requests library internally and provides a standardized way to refresh credentials or request access tokens.

creds.refresh(auth_req)

When ADC is retrieved using google.auth.default(), the credentials object does not initially include an access token (at least in a local environment). The refresh() method explicitly obtains an access token and attaches it to the credentials object, enabling it to authenticate API requests.

The following code demonstrates how you can verify this behavior:

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

example output:

# 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}")

Before calling refresh(), the token attribute is None. After refresh() is invoked, the credentials are populated with a valid access token and its expiry time.

---

3. Configuring the Kubernetes Client

The Kubernetes client is configured using the cluster’s API endpoint, a temporary file for the CA certificate, and the refreshed Bearer token. This ensures that the client can securely authenticate and communicate with the cluster.

gcloud auth application-default login

The CA certificate is stored temporarily and referenced by the client for secure SSL communication. With these settings, the Kubernetes client is fully configured and ready to interact with the cluster.

---

Example Output

Here’s an example of the YAML output for the kube-system Namespace:

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()

---

Conclusion

This approach highlights the portability of using the same code to interact with the Kubernetes API, whether running locally or on a Google Cloud service like Cloud Run. By leveraging Application Default Credentials (ADC), we’ve demonstrated a flexible method to dynamically generate a Kubernetes API client without relying on pre-generated configuration files or external tools. This makes it easy to build applications that can seamlessly adapt to different environments, simplifying both development and deployment workflows.

The above is the detailed content of Building a Kubernetes Client for Google Kubernetes Engine (GKE) in Python. For more information, please follow other related articles on the PHP Chinese website!