Merge pull request #81036 from…

pkg/volume/gcepd/BUILD

@@ -59,6 +59,7 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/util/testing:go_default_library",
+        "//staging/src/k8s.io/cloud-provider:go_default_library",
         "//staging/src/k8s.io/cloud-provider/volume:go_default_library",
         "//staging/src/k8s.io/cloud-provider/volume/helpers:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",

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/gce_pd.go

@@ -112,7 +112,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
+}

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

@@ -426,6 +426,19 @@ func (g *Cloud) AddAliasToInstance(nodeName types.NodeName, alias *net.IPNet) er
 // Gets the named instances, returning cloudprovider.InstanceNotFound if any
 // instance is not found
 func (g *Cloud) getInstancesByNames(names []string) ([]*gceInstance, error) {
+	foundInstances, err := g.getFoundInstanceByNames(names)
+	if err != nil {
+		return nil, err
+	}
+	if len(foundInstances) != len(names) {
+		return nil, cloudprovider.InstanceNotFound
+	}
+	return foundInstances, nil
+}
+
+// Gets the named instances, returning a list of gceInstances it was able to find from the provided
+// list of names.
+func (g *Cloud) getFoundInstanceByNames(names []string) ([]*gceInstance, error) {
 	ctx, cancel := cloud.ContextWithCallTimeout()
 	defer cancel()
 
@@ -472,20 +485,17 @@ func (g *Cloud) getInstancesByNames(names []string) ([]*gceInstance, error) {
 		}
 	}
 
-	if remaining > 0 {
-		var failed []string
-		for k := range found {
-			if found[k] == nil {
-				failed = append(failed, k)
-			}
+	var ret []*gceInstance
+	var failed []string
+	for name, instance := range found {
+		if instance != nil {
+			ret = append(ret, instance)
+		} else {
+			failed = append(failed, name)
 		}
-		klog.Errorf("Failed to retrieve instances: %v", failed)
-		return nil, cloudprovider.InstanceNotFound
 	}
-
-	var ret []*gceInstance
-	for _, instance := range found {
-		ret = append(ret, instance)
+	if len(failed) > 0 {
+		klog.Errorf("Failed to retrieve instances: %v", failed)
 	}
 
 	return ret, nil