diff --git a/testutils/clustermanager/prow-cluster-operation/README.md b/testutils/clustermanager/prow-cluster-operation/README.md
new file mode 100644
index 000000000..ac587c0a3
--- /dev/null
+++ b/testutils/clustermanager/prow-cluster-operation/README.md
@@ -0,0 +1,53 @@
+## prow-cluster-operation
+
+prow-cluster-operation is a tool for creating, deleting, getting a GKE
+cluster
+
+## Prerequisite
+
+- `GOOGLE_APPLICATION_CREDENTIALS` set
+
+## Usage
+
+This tool can be invoked from command line with following parameters:
+
+- `--min-nodes`: minumum number of nodes, default 1
+- `--max-nodes`: maximum number of nodes, default 3
+- `--node-type`: GCE node type, default "n1-standard-4"
+- `--region`: GKE region, default "us-central1"
+- `--zone`: GKE zone, default empty
+- `--project`: GCP project, default empty
+- `--name`: cluster name, default empty
+- `--backup-regions`: backup regions to be used if cluster creation in primary
+  region failed, comma separated list, default "us-west1,us-east1"
+- `--addons`: GKE addons, comma separated list, default empty
+
+## Flow
+
+### Create
+
+1. Acquiring cluster if kubeconfig already points to it
+1. Get GCP project name if not provided as a parameter:
+    - [In Prow] Acquire from Boskos
+    - [Not in Prow] Read from gcloud config
+
+    Failed obtaining project name will fail the tool
+1. Get default cluster name if not provided as a parameter
+1. Delete cluster if cluster with same name and location already exists in GKE
+1. Create cluster
+1. Write cluster metadata to `${ARTIFACT}/metadata.json`
+
+### Delete
+
+1. Acquiring cluster if kubeconfig already points to it
+1. If cluster name is defined then getting cluster by its name
+1. If no cluster is found from previous step then it fails
+1. Delete:
+    - [In Prow] Release Boskos project
+    - [Not in Prow] Delete cluster
+
+### Get
+
+1. Acquiring cluster if kubeconfig already points to it
+1. If cluster name is defined then getting cluster by its name
+1. If no cluster is found from previous step then it fails
diff --git a/testutils/clustermanager/prow-cluster-operation/actions/create.go b/testutils/clustermanager/prow-cluster-operation/actions/create.go
new file mode 100644
index 000000000..207a19084
--- /dev/null
+++ b/testutils/clustermanager/prow-cluster-operation/actions/create.go
@@ -0,0 +1,108 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package actions
+
+import (
+	"log"
+	"strconv"
+
+	container "google.golang.org/api/container/v1beta1"
+	"knative.dev/pkg/test/gke"
+	clm "knative.dev/pkg/testutils/clustermanager/e2e-tests"
+	"knative.dev/pkg/testutils/clustermanager/e2e-tests/common"
+	"knative.dev/pkg/testutils/clustermanager/prow-cluster-operation/options"
+	"knative.dev/pkg/testutils/metahelper/client"
+)
+
+const (
+	// Keys to be written into metadata.json
+	e2eRegionKey      = "E2E:Region"
+	e2eZoneKey        = "E2E:Zone"
+	clusterNameKey    = "E2E:Machine"
+	clusterVersionKey = "E2E:Version"
+	minNodesKey       = "E2E:MinNodes"
+	maxNodesKey       = "E2E:MaxNodes"
+	projectKey        = "E2E:Project"
+)
+
+func writeMetaData(cluster *container.Cluster, project string) {
+	// Set up metadata client for saving metadata
+	c, err := client.NewClient("")
+	if err != nil {
+		log.Fatal(err)
+	}
+	log.Printf("Writing metadata to: %q", c.Path)
+	// Get minNodes and maxNodes counts from default-pool, this is
+	// usually the case in tests in Prow
+	var minNodes, maxNodes string
+	for _, np := range cluster.NodePools {
+		if np.Name == "default-pool" {
+			minNodes = strconv.FormatInt(np.InitialNodeCount, 10)
+			// maxNodes is equal to minNodes if autoscaling isn't on
+			maxNodes = minNodes
+			if np.Autoscaling != nil {
+				minNodes = strconv.FormatInt(np.Autoscaling.MinNodeCount, 10)
+				maxNodes = strconv.FormatInt(np.Autoscaling.MaxNodeCount, 10)
+			} else {
+				log.Printf("DEBUG: nodepool is default-pool but autoscaling is not on: '%+v'", np)
+			}
+			break
+		}
+	}
+
+	e2eRegion, e2eZone := gke.RegionZoneFromLoc(cluster.Location)
+	for key, val := range map[string]string{
+		e2eRegionKey:      e2eRegion,
+		e2eZoneKey:        e2eZone,
+		clusterNameKey:    cluster.Name,
+		clusterVersionKey: cluster.InitialClusterVersion,
+		minNodesKey:       minNodes,
+		maxNodesKey:       maxNodes,
+		projectKey:        project,
+	} {
+		if err = c.Set(key, val); err != nil {
+			log.Fatalf("Failed saving metadata %q:%q: '%v'", key, val, err)
+		}
+	}
+	log.Println("Done writing metadata")
+}
+
+func Create(o *options.RequestWrapper) {
+	o.Prep()
+
+	gkeClient := clm.GKEClient{}
+	clusterOps := gkeClient.Setup(o.Request)
+	gkeOps := clusterOps.(*clm.GKECluster)
+	if err := gkeOps.Acquire(); err != nil || gkeOps.Cluster == nil {
+		log.Fatalf("failed acquiring GKE cluster: '%v'", err)
+	}
+
+	// At this point we should have a cluster ready to run test. Need to save
+	// metadata so that following flow can understand the context of cluster, as
+	// well as for Prow usage later
+	writeMetaData(gkeOps.Cluster, gkeOps.Project)
+
+	// set up kube config points to cluster
+	// TODO(chaodaiG): this probably should also be part of clustermanager lib
+	if out, err := common.StandardExec("gcloud", "beta", "container", "clusters", "get-credentials",
+		gkeOps.Cluster.Name, "--region", gkeOps.Cluster.Location, "--project", gkeOps.Project); err != nil {
+		log.Fatalf("Failed connecting to cluster: %q, '%v'", out, err)
+	}
+	if out, err := common.StandardExec("gcloud", "config", "set", "project", gkeOps.Project); err != nil {
+		log.Fatalf("Failed setting gcloud: %q, '%v'", out, err)
+	}
+}
diff --git a/testutils/clustermanager/prow-cluster-operation/actions/delete.go b/testutils/clustermanager/prow-cluster-operation/actions/delete.go
new file mode 100644
index 000000000..6f316ec67
--- /dev/null
+++ b/testutils/clustermanager/prow-cluster-operation/actions/delete.go
@@ -0,0 +1,50 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package actions
+
+import (
+	"log"
+
+	clm "knative.dev/pkg/testutils/clustermanager/e2e-tests"
+	"knative.dev/pkg/testutils/clustermanager/prow-cluster-operation/options"
+)
+
+func Delete(o *options.RequestWrapper) {
+	o.Request.NeedsCleanup = true
+	o.Request.SkipCreation = true
+
+	gkeClient := clm.GKEClient{}
+	clusterOps := gkeClient.Setup(o.Request)
+	gkeOps := clusterOps.(*clm.GKECluster)
+	if err := gkeOps.Acquire(); err != nil || gkeOps.Cluster == nil {
+		log.Fatalf("Failed identifying cluster for cleanup: '%v'", err)
+	}
+	log.Printf("Identified project %q and cluster %q for removal", gkeOps.Project, gkeOps.Cluster.Name)
+	var err error
+	if err = gkeOps.Delete(); err != nil {
+		log.Fatalf("Failed deleting cluster: '%v'", err)
+	}
+	// TODO: uncomment the lines below when previous Delete command becomes
+	// async operation
+	// // Unset context with best effort. The first command only unsets current
+	// // context, but doesn't delete the entry from kubeconfig, and should return it's
+	// // context if succeeded, which can be used by the second command to
+	// // delete it from kubeconfig
+	// if out, err := common.StandardExec("kubectl", "config", "unset", "current-context"); err != nil {
+	// 	common.StandardExec("kubectl", "config", "unset", "contexts."+string(out))
+	// }
+}
diff --git a/testutils/clustermanager/prow-cluster-operation/actions/get.go b/testutils/clustermanager/prow-cluster-operation/actions/get.go
new file mode 100644
index 000000000..fe3c47cf6
--- /dev/null
+++ b/testutils/clustermanager/prow-cluster-operation/actions/get.go
@@ -0,0 +1,29 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package actions
+
+import (
+	"knative.dev/pkg/testutils/clustermanager/prow-cluster-operation/options"
+)
+
+func Get(o *options.RequestWrapper) {
+	o.Prep()
+	o.Request.SkipCreation = true
+	// Reuse `Create` for getting operation, so that we can reuse the same logic
+	// such as protected project/cluster etc.
+	Create(o)
+}
diff --git a/testutils/clustermanager/prow-cluster-operation/main.go b/testutils/clustermanager/prow-cluster-operation/main.go
new file mode 100644
index 000000000..161701663
--- /dev/null
+++ b/testutils/clustermanager/prow-cluster-operation/main.go
@@ -0,0 +1,53 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"flag"
+	"log"
+
+	"knative.dev/pkg/testutils/clustermanager/prow-cluster-operation/actions"
+	"knative.dev/pkg/testutils/clustermanager/prow-cluster-operation/options"
+)
+
+var (
+	create bool
+	delete bool
+	get    bool
+)
+
+func main() {
+	flag.BoolVar(&create, "create", false, "Create cluster")
+	flag.BoolVar(&delete, "delete", false, "Delete cluster")
+	flag.BoolVar(&get, "get", false, "Get existing cluster from kubeconfig or gcloud")
+	o := options.NewRequestWrapper()
+	flag.Parse()
+
+	if (create && delete) || (create && get) || (delete && get) {
+		log.Fatal("--create, --delete, --get are mutually exclusive")
+	}
+	switch {
+	case create:
+		actions.Create(o)
+	case delete:
+		actions.Delete(o)
+	case get:
+		actions.Get(o)
+	default:
+		log.Fatal("Must pass one of --create, --delete, --get")
+	}
+}
diff --git a/testutils/clustermanager/prow-cluster-operation/options/options.go b/testutils/clustermanager/prow-cluster-operation/options/options.go
new file mode 100644
index 000000000..0eb4b6605
--- /dev/null
+++ b/testutils/clustermanager/prow-cluster-operation/options/options.go
@@ -0,0 +1,61 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package options
+
+import (
+	"flag"
+	"strings"
+
+	clm "knative.dev/pkg/testutils/clustermanager/e2e-tests"
+)
+
+type RequestWrapper struct {
+	Request          clm.GKERequest
+	BackupRegionsStr string
+	AddonsStr        string
+	NoWait           bool
+}
+
+func NewRequestWrapper() *RequestWrapper {
+	rw := &RequestWrapper{
+		Request: clm.GKERequest{},
+	}
+	rw.addOptions()
+	return rw
+}
+
+func (rw *RequestWrapper) Prep() {
+	if rw.BackupRegionsStr != "" {
+		rw.Request.BackupRegions = strings.Split(rw.BackupRegionsStr, ",")
+	}
+	if rw.AddonsStr != "" {
+		rw.Request.Addons = strings.Split(rw.AddonsStr, ",")
+	}
+}
+
+func (rw *RequestWrapper) addOptions() {
+	flag.Int64Var(&rw.Request.MinNodes, "min-nodes", 0, "minimal number of nodes")
+	flag.Int64Var(&rw.Request.MaxNodes, "max-nodes", 0, "maximal number of nodes")
+	flag.StringVar(&rw.Request.NodeType, "node-type", "", "node type")
+	flag.StringVar(&rw.Request.Region, "region", "", "GCP region")
+	flag.StringVar(&rw.Request.Zone, "zone", "", "GCP zone")
+	flag.StringVar(&rw.Request.Project, "project", "", "GCP project")
+	flag.StringVar(&rw.Request.ClusterName, "name", "", "cluster name")
+	flag.StringVar(&rw.BackupRegionsStr, "backup-regions", "", "GCP regions as backup, separated by comma")
+	flag.StringVar(&rw.AddonsStr, "addons", "", "addons to be added, separated by comma")
+	flag.BoolVar(&rw.Request.SkipCreation, "skip-creation", false, "should skip creation or not")
+}