Merge pull request #44884 from verult/master

pkg/cloudprovider/providers/gce/BUILD

@@ -68,9 +68,18 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["gce_test.go"],
+    srcs = [
+        "gce_disks_test.go",
+        "gce_test.go",
+    ],
     library = ":go_default_library",
     tags = ["automanaged"],
+    deps = [
+        "//pkg/cloudprovider:go_default_library",
+        "//vendor/google.golang.org/api/compute/v1:go_default_library",
+        "//vendor/google.golang.org/api/googleapi:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+    ],
 )
 
 filegroup(

pkg/cloudprovider/providers/gce/gce.go

@@ -91,6 +91,25 @@ type GCECloud struct {
 	nodeInstancePrefix       string   // If non-"", an advisory prefix for all nodes in the cluster
 	useMetadataServer        bool
 	operationPollRateLimiter flowcontrol.RateLimiter
+	manager                  ServiceManager
+}
+
+type ServiceManager interface {
+	// Creates a new persistent disk on GCE with the given disk spec.
+	CreateDisk(project string, zone string, disk *compute.Disk) (*compute.Operation, error)
+
+	// Gets the persistent disk from GCE with the given diskName.
+	GetDisk(project string, zone string, diskName string) (*compute.Disk, error)
+
+	// Deletes the persistent disk from GCE with the given diskName.
+	DeleteDisk(project string, zone string, disk string) (*compute.Operation, error)
+
+	// Waits until GCE reports the given operation in the given zone as done.
+	WaitForZoneOp(op *compute.Operation, zone string, mc *metricContext) error
+}
+
+type GCEServiceManager struct {
+	gce *GCECloud
 }
 
 type Config struct {
@@ -230,7 +249,7 @@ func CreateGCECloud(projectID, region, zone string, managedZones []string, netwo
 
 	operationPollRateLimiter := flowcontrol.NewTokenBucketRateLimiter(10, 100) // 10 qps, 100 bucket size.
 
-	return &GCECloud{
+	gce := &GCECloud{
 		service:                  service,
 		serviceBeta:              serviceBeta,
 		containerService:         containerService,
@@ -244,7 +263,10 @@ func CreateGCECloud(projectID, region, zone string, managedZones []string, netwo
 		nodeInstancePrefix:       nodeInstancePrefix,
 		useMetadataServer:        useMetadataServer,
 		operationPollRateLimiter: operationPollRateLimiter,
-	}, nil
+	}
+
+	gce.manager = &GCEServiceManager{gce}
+	return gce, nil
 }
 
 // Initialize takes in a clientBuilder and spawns a goroutine for watching the clusterid configmap.
@@ -383,3 +405,31 @@ func newOauthClient(tokenSource oauth2.TokenSource) (*http.Client, error) {
 
 	return oauth2.NewClient(oauth2.NoContext, tokenSource), nil
 }
+
+func (manager *GCEServiceManager) CreateDisk(
+	project string,
+	zone string,
+	disk *compute.Disk) (*compute.Operation, error) {
+
+	return manager.gce.service.Disks.Insert(project, zone, disk).Do()
+}
+
+func (manager *GCEServiceManager) GetDisk(
+	project string,
+	zone string,
+	diskName string) (*compute.Disk, error) {
+
+	return manager.gce.service.Disks.Get(project, zone, diskName).Do()
+}
+
+func (manager *GCEServiceManager) DeleteDisk(
+	project string,
+	zone string,
+	diskName string) (*compute.Operation, error) {
+
+	return manager.gce.service.Disks.Delete(project, zone, diskName).Do()
+}
+
+func (manager *GCEServiceManager) WaitForZoneOp(op *compute.Operation, zone string, mc *metricContext) error {
+	return manager.gce.waitForZoneOp(op, zone, mc)
+}

pkg/cloudprovider/providers/gce/gce_disks.go

@@ -244,7 +244,7 @@ func (gce *GCECloud) CreateDisk(
 	}
 
 	mc := newDiskMetricContext("create", zone)
-	createOp, err := gce.service.Disks.Insert(gce.projectID, zone, diskToCreate).Do()
+	createOp, err := gce.manager.CreateDisk(gce.projectID, zone, diskToCreate)
 	if isGCEError(err, "alreadyExists") {
 		glog.Warningf("GCE PD %q already exists, reusing", name)
 		return nil
@@ -252,7 +252,7 @@ func (gce *GCECloud) CreateDisk(
 		return mc.Observe(err)
 	}
 
-	err = gce.waitForZoneOp(createOp, zone, mc)
+	err = gce.manager.WaitForZoneOp(createOp, zone, mc)
 	if isGCEError(err, "alreadyExists") {
 		glog.Warningf("GCE PD %q already exists, reusing", name)
 		return nil
@@ -323,7 +323,7 @@ func (gce *GCECloud) GetAutoLabelsForPD(name string, zone string) (map[string]st
 // If not found, returns (nil, nil)
 func (gce *GCECloud) findDiskByName(diskName string, zone string) (*GCEDisk, error) {
 	mc := newDiskMetricContext("get", zone)
-	disk, err := gce.service.Disks.Get(gce.projectID, zone, diskName).Do()
+	disk, err := gce.manager.GetDisk(gce.projectID, zone, diskName)
 	if err == nil {
 		d := &GCEDisk{
 			Zone: lastComponent(disk.Zone),
@@ -410,12 +410,12 @@ func (gce *GCECloud) doDeleteDisk(diskToDelete string) error {
 
 	mc := newDiskMetricContext("delete", disk.Zone)
 
-	deleteOp, err := gce.service.Disks.Delete(gce.projectID, disk.Zone, disk.Name).Do()
+	deleteOp, err := gce.manager.DeleteDisk(gce.projectID, disk.Zone, disk.Name)
 	if err != nil {
 		return mc.Observe(err)
 	}
 
-	return gce.waitForZoneOp(deleteOp, disk.Zone, mc)
+	return gce.manager.WaitForZoneOp(deleteOp, disk.Zone, mc)
 }
 
 // Converts a Disk resource to an AttachedDisk resource.