All the samples are self contained unless they are placed inside their own folders. The samples use [Application Default Credentails (ADC)](https://cloud.google.com/docs/authentication/production#automatically) to authenticate with GCP. So make sure ADC is setup correctly _(i.e. `GOOGLE_APPLICATION_CREDENTIALS` environment variable is set)_ before running the samples. Some sample might require additional python modules to be installed.

You can run samples as follows:

python <sample_name.py> <arg1> <arg2> ...

You can run the following command to find the usage and arguments for the samples:

python <sample_name.py> -h

python quickstart.py -h

usage: quickstart.py [-h] project_id zone

positional arguments:

project_id Google Cloud project ID

zone GKE Cluster zone

optional arguments:

-h, --help show this help message and exit

- [**quickstart.py**](quickstart.py): A simple example to list the GKE clusters in a given GCP project and zone. The sample uses the [`list_clusters()`](https://cloud.google.com/python/docs/reference/container/latest/google.cloud.container_v1.services.cluster_manager.ClusterManagerClient#google_cloud_container_v1_services_cluster_manager_ClusterManagerClient_list_clusters) API to fetch the list of cluster.

The following samples are examples of operations that take a while to complete.

For example _creating a cluster_ in GKE can take a while to set up the cluster

nodes, networking and configuring Kubernetes. Thus, calls to such long running

APIs return an object of type [`Operation`](https://cloud.google.com/python/docs/reference/container/latest/google.cloud.container_v1.types.Operation). We can

then use the id of the returned operation to **poll** the [`get_operation()`](https://cloud.google.com/python/docs/reference/container/latest/google.cloud.container_v1.services.cluster_manager.ClusterManagerClient#google_cloud_container_v1_services_cluster_manager_ClusterManagerClient_get_operation) API to check for it's status. You can see the

different statuses it can be in, in [this proto definition](https://github.com/googleapis/googleapis/blob/master/google/container/v1/cluster_service.proto#L1763-L1778).

- [**create_cluster.py**](create_cluster.py): An example of creating a GKE cluster _(with mostly the defaults)_. This example shows how to handle responses of type [`Operation`](https://cloud.google.com/python/docs/reference/container/latest/google.cloud.container_v1.types.Operation) that reperesents a long running operation. The example uses the python module [`backoff`](https://github.com/litl/backoff) to handle a graceful exponential backoff retry mechanism to check if the `Operation` has completed.

- [**delete_cluster.py**](delete_cluster.py): An example of deleting a GKE cluster. This example shows how to handle responses of type [`Operation`](https://cloud.google.com/python/docs/reference/container/latest/google.cloud.container_v1.types.Operation) that reperesents a long running operation.

import argparse

import sys

from typing import Dict

import backoff

from google.cloud import container_v1

def on_success(details: Dict[str, str]) -> None:

"""

print("Successfully created cluster after {elapsed:0.1f} seconds".format(**details))

def on_failure(details: Dict[str, str]) -> None:

"""

print("Backing off {wait:0.1f} seconds after {tries} tries".format(**details))

def poll_for_op_status(

client: container_v1.ClusterManagerClient, op_id: str

) -> container_v1.Operation.Status:

"""

op = client.get_operation({"name": op_id})

return op.status

def create_cluster(project_id: str, location: str, cluster_name: str) -> None:

"""Create a new GKE cluster in the given GCP Project and Zone"""

# Initialize the Cluster management client.

client = container_v1.ClusterManagerClient()

# Create a fully qualified location identifier of form `projects/{project_id}/location/{zone}'.

cluster_location = client.common_location_path(project_id, location)

cluster_def = {

"name": cluster_name,

"initial_node_count": 2,

"node_config": {"machine_type": "e2-standard-2"},

}

# Create the request object with the location identifier.

request = {"parent": cluster_location, "cluster": cluster_def}

create_response = client.create_cluster(request)

op_identifier = f"{cluster_location}/operations/{create_response.name}"

# poll for the operation status and schedule a retry until the cluster is created

poll_for_op_status(client, op_identifier)

if __name__ == "__main__":

parser = argparse.ArgumentParser(

description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter,

)

parser.add_argument("project_id", help="Google Cloud project ID")

parser.add_argument("zone", help="GKE Cluster zone")

parser.add_argument("cluster_name", help="Name to be given to the GKE Cluster")

args = parser.parse_args()

if len(sys.argv) != 4:

parser.print_usage()

sys.exit(1)

create_cluster(args.project_id, args.zone, args.cluster_name)

import os

import uuid

import backoff

from google.cloud import container_v1 as gke

import pytest

import create_cluster as gke_create

PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]

ZONE = "us-central1-b"

CLUSTER_NAME = f"py-container-repo-test-{uuid.uuid4().hex[:10]}"

def setup_and_tear_down() -> None:

# nohing to setup here

# run the tests here

yield

# delete the cluster

client = gke.ClusterManagerClient()

cluster_location = client.common_location_path(PROJECT_ID, ZONE)

cluster_name = f"{cluster_location}/clusters/{CLUSTER_NAME}"

op = client.delete_cluster({"name": cluster_name})

op_id = f"{cluster_location}/operations/{op.name}"

# schedule a retry to ensure the cluster is deleted

@backoff.on_predicate(

def wait_for_delete() -> gke.Operation.Status:

return client.get_operation({"name": op_id}).status

wait_for_delete()

def test_create_clusters(capsys: object) -> None:

gke_create.create_cluster(PROJECT_ID, ZONE, CLUSTER_NAME)

out, _ = capsys.readouterr()

assert "Backing off " in out

assert "Successfully created cluster after" in out

client = gke.ClusterManagerClient()

cluster_location = client.common_location_path(PROJECT_ID, ZONE)

list_response = client.list_clusters({"parent": cluster_location})

list_of_clusters = []

for cluster in list_response.clusters:

list_of_clusters.append(cluster.name)

assert CLUSTER_NAME in list_of_clusters

import argparse

import sys

from typing import Dict

import backoff

from google.cloud import container_v1

def on_success(details: Dict[str, str]) -> None:

"""

print("Successfully deleted cluster after {elapsed:0.1f} seconds".format(**details))

def on_failure(details: Dict[str, str]) -> None:

"""

print("Backing off {wait:0.1f} seconds after {tries} tries".format(**details))

def poll_for_op_status(

client: container_v1.ClusterManagerClient, op_id: str

) -> container_v1.Operation.Status:

"""

op = client.get_operation({"name": op_id})

return op.status

def delete_cluster(project_id: str, location: str, cluster_name: str) -> None:

"""Delete an existing GKE cluster in the given GCP Project and Zone"""

# Initialize the Cluster management client.

client = container_v1.ClusterManagerClient()

# Create a fully qualified location identifier of form `projects/{project_id}/location/{zone}'.

cluster_location = client.common_location_path(project_id, location)

cluster_name = f"{cluster_location}/clusters/{cluster_name}"

# Create the request object with the location identifier.

request = {"name": cluster_name}

delete_response = client.delete_cluster(request)

op_identifier = f"{cluster_location}/operations/{delete_response.name}"

# poll for the operation status until the cluster is deleted

poll_for_op_status(client, op_identifier)

if __name__ == "__main__":

parser = argparse.ArgumentParser(

description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter,

)

parser.add_argument("project_id", help="Google Cloud project ID")

parser.add_argument("zone", help="GKE Cluster zone")

parser.add_argument("cluster_name", help="Name to be given to the GKE Cluster")

args = parser.parse_args()

if len(sys.argv) != 4:

parser.print_usage()

sys.exit(1)

delete_cluster(args.project_id, args.zone, args.cluster_name)