add ability for gce to bulk verify attached disks

pkg/volume/gcepd/attacher.go

@@ -141,6 +141,41 @@ func (attacher *gcePersistentDiskAttacher) VolumesAreAttached(specs []*volume.Sp
 	return volumesAttachedCheck, nil
 }
 
+func (attacher *gcePersistentDiskAttacher) BulkVerifyVolumes(volumesByNode map[types.NodeName][]*volume.Spec) (map[types.NodeName]map[*volume.Spec]bool, error) {
+	volumesAttachedCheck := make(map[types.NodeName]map[*volume.Spec]bool)
+	diskNamesByNode := make(map[types.NodeName][]string)
+	volumeSpecToDiskName := make(map[*volume.Spec]string)
+
+	for nodeName, volumeSpecs := range volumesByNode {
+		diskNames := []string{}
+		for _, spec := range volumeSpecs {
+			volumeSource, _, err := getVolumeSource(spec)
+			if err != nil {
+				klog.Errorf("Error getting volume (%q) source : %v", spec.Name(), err)
+				continue
+			}
+			diskNames = append(diskNames, volumeSource.PDName)
+			volumeSpecToDiskName[spec] = volumeSource.PDName
+		}
+		diskNamesByNode[nodeName] = diskNames
+	}
+
+	attachedDisksByNode, err := attacher.gceDisks.BulkDisksAreAttached(diskNamesByNode)
+	if err != nil {
+		return nil, err
+	}
+
+	for nodeName, volumeSpecs := range volumesByNode {
+		volumesAreAttachedToNode := make(map[*volume.Spec]bool)
+		for _, spec := range volumeSpecs {
+			diskName := volumeSpecToDiskName[spec]
+			volumesAreAttachedToNode[spec] = attachedDisksByNode[nodeName][diskName]
+		}
+		volumesAttachedCheck[nodeName] = volumesAreAttachedToNode
+	}
+	return volumesAttachedCheck, nil
+}
+
 // search Windows disk number by LUN
 func getDiskID(pdName string, exec mount.Exec) (string, error) {
 	// TODO: replace Get-GcePdName with native windows support of Get-Disk, see issue #74674

pkg/volume/gcepd/attacher_test.go

@@ -405,6 +405,10 @@ func (testcase *testcase) DisksAreAttached(diskNames []string, nodeName types.No
 	return nil, errors.New("Not implemented")
 }
 
+func (testcase *testcase) BulkDisksAreAttached(diskByNodes map[types.NodeName][]string) (map[types.NodeName]map[string]bool, error) {
+	return nil, errors.New("Not implemented")
+}
+
 func (testcase *testcase) CreateDisk(name string, diskType string, zone string, sizeGb int64, tags map[string]string) error {
 	return errors.New("Not implemented")
 }

pkg/volume/gcepd/gce_pd.go

@@ -109,7 +109,7 @@ func (plugin *gcePersistentDiskPlugin) SupportsMountOption() bool {
 }
 
 func (plugin *gcePersistentDiskPlugin) SupportsBulkVolumeVerification() bool {
-	return false
+	return true
 }
 
 func (plugin *gcePersistentDiskPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {

staging/src/k8s.io/legacy-cloud-providers/gce/gce_disks.go

@@ -424,6 +424,10 @@ type Disks interface {
 	// to the node with the specified NodeName.
 	DisksAreAttached(diskNames []string, nodeName types.NodeName) (map[string]bool, error)
 
+	// BulkDisksAreAttached is a batch function to check if all corresponding disks are attached to the
+	// nodes specified with nodeName.
+	BulkDisksAreAttached(diskByNodes map[types.NodeName][]string) (map[types.NodeName]map[string]bool, error)
+
 	// CreateDisk creates a new PD with given properties. Tags are serialized
 	// as JSON into Description field.
 	CreateDisk(name string, diskType string, zone string, sizeGb int64, tags map[string]string) error
@@ -620,6 +624,37 @@ func (g *Cloud) DisksAreAttached(diskNames []string, nodeName types.NodeName) (m
 	return attached, nil
 }
 
+// BulkDisksAreAttached is a batch function to check if all corresponding disks are attached to the
+// nodes specified with nodeName.
+func (g *Cloud) BulkDisksAreAttached(diskByNodes map[types.NodeName][]string) (map[types.NodeName]map[string]bool, error) {
+	instanceNames := []string{}
+	for nodeName := range diskByNodes {
+		instanceNames = append(instanceNames, mapNodeNameToInstanceName(nodeName))
+	}
+
+	// List all instances with the given instance names
+	// Then for each instance listed, add the disks attached to that instance to a map
+	listedInstances, err := g.getFoundInstanceByNames(instanceNames)
+	if err != nil {
+		return nil, fmt.Errorf("error listing instances: %v", err)
+	}
+	listedInstanceNamesToDisks := make(map[string][]*compute.AttachedDisk)
+	for _, instance := range listedInstances {
+		listedInstanceNamesToDisks[instance.Name] = instance.Disks
+	}
+
+	verifyDisksAttached := make(map[types.NodeName]map[string]bool)
+
+	// For each node and its desired attached disks that needs to be verified
+	for nodeName, disksToVerify := range diskByNodes {
+		instanceName := canonicalizeInstanceName(mapNodeNameToInstanceName(nodeName))
+		disksActuallyAttached := listedInstanceNamesToDisks[instanceName]
+		verifyDisksAttached[nodeName] = verifyDisksAttachedToNode(disksToVerify, disksActuallyAttached)
+	}
+
+	return verifyDisksAttached, nil
+}
+
 // CreateDisk creates a new Persistent Disk, with the specified name &
 // size, in the specified zone. It stores specified tags encoded in
 // JSON in Description field.
@@ -990,3 +1025,20 @@ func isGCEError(err error, reason string) bool {
 	}
 	return false
 }
+
+// verifyDisksAttachedToNode takes in an slice of disks that should be attached to an instance, and the
+// slice of disks actually attached to it. It returns a map verifying if the disks are actually attached.
+func verifyDisksAttachedToNode(disksToVerify []string, disksActuallyAttached []*compute.AttachedDisk) map[string]bool {
+	verifiedDisks := make(map[string]bool)
+	diskNamesActuallyAttached := sets.NewString()
+	for _, disk := range disksActuallyAttached {
+		diskNamesActuallyAttached.Insert(disk.DeviceName)
+	}
+
+	// For every disk that's supposed to be attached, verify that it is
+	for _, diskName := range disksToVerify {
+		verifiedDisks[diskName] = diskNamesActuallyAttached.Has(diskName)
+	}
+
+	return verifiedDisks
+}