Use zone from node for topology aware volume creation

pkg/volume/awsebs/BUILD

@@ -52,9 +52,11 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//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/legacy-cloud-providers/aws:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )

pkg/volume/awsebs/aws_ebs_test.go

@@ -23,9 +23,11 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes/fake"
 	utiltesting "k8s.io/client-go/util/testing"
 	"k8s.io/kubernetes/pkg/util/mount"
@@ -376,3 +378,37 @@ func TestMountOptions(t *testing.T) {
 		t.Errorf("Expected mount options to be %v got %v", expectedMountOptions, mountOptions)
 	}
 }
+
+func TestGetCandidateZone(t *testing.T) {
+	const testZone = "my-zone-1a"
+
+	// TODO: add test case for immediate bind volume when we have a good way to mock Cloud outside cloudprovider
+	tests := []struct {
+		cloud         *aws.Cloud
+		node          *v1.Node
+		expectedZones sets.String
+	}{
+		{
+			cloud: nil,
+			node: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						v1.LabelZoneFailureDomain: testZone,
+					},
+				},
+			},
+			expectedZones: sets.NewString(),
+		},
+		{
+			cloud:         nil,
+			node:          &v1.Node{},
+			expectedZones: sets.NewString(),
+		},
+	}
+
+	for _, test := range tests {
+		zones, err := getCandidateZones(test.cloud, test.node)
+		assert.Nil(t, err)
+		assert.Equal(t, test.expectedZones, zones)
+	}
+}

pkg/volume/awsebs/aws_util.go

@@ -29,7 +29,7 @@ import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/sets"
-	cloudprovider "k8s.io/cloud-provider"
+	"k8s.io/cloud-provider"
 	volumehelpers "k8s.io/cloud-provider/volume/helpers"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
@@ -86,9 +86,9 @@ func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner, node *
 
 	capacity := c.options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
 
-	zonesWithNodes, err := cloud.GetCandidateZonesForDynamicVolume()
+	zonesWithNodes, err := getCandidateZones(cloud, node)
 	if err != nil {
-		return "", 0, nil, "", fmt.Errorf("error querying for all zones: %v", err)
+		return "", 0, nil, "", fmt.Errorf("error finding candidate zone for pvc: %v", err)
 	}
 
 	volumeOptions, err := populateVolumeOptions(c.plugin.GetPluginName(), c.options.PVC.Name, capacity, tags, c.options.Parameters, node, allowedTopologies, zonesWithNodes)
@@ -125,6 +125,20 @@ func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner, node *
 	return name, volumeOptions.CapacityGB, labels, fstype, nil
 }
 
+// getCandidateZones finds possible zones that a volume can be created in
+func getCandidateZones(cloud *aws.Cloud, selectedNode *v1.Node) (sets.String, error) {
+	if selectedNode != nil {
+		// For topology aware volume provisioning, node is already selected so we use the zone from
+		// selected node directly instead of candidate zones.
+		// We can assume the information is always available as node controller shall maintain it.
+		return sets.NewString(), nil
+	}
+
+	// For non-topology-aware volumes (those that binds immediately), we fall back to original logic to query
+	// cloud provider for possible zones
+	return cloud.GetCandidateZonesForDynamicVolume()
+}
+
 // returns volumeOptions for EBS based on storageclass parameters and node configuration
 func populateVolumeOptions(pluginName, pvcName string, capacityGB resource.Quantity, tags map[string]string, storageParams map[string]string, node *v1.Node, allowedTopologies []v1.TopologySelectorTerm, zonesWithNodes sets.String) (*aws.VolumeOptions, error) {
 	requestGiB, err := volumehelpers.RoundUpToGiBInt(capacityGB)