Repository: kubernetes-sigs/gcp-filestore-csi-driver
Pull request #56 (2020-08-04 23:00:49)

Topology feature support for Filestore CSI driver
1. Support Topology constraints feature in FS CSI driver
2. Use Storage Class allowedTopology field to provide input to
   FS driver for instance creation.

<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
> /kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
/kind feature
> /kind flake

**What this PR does / why we need it**:

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

**Special notes for your reviewer**:
This patch removes the need to 'location' parameter in storage class, and uses CreateVolumeRequest.accessibility_requirements  field to determine a zone for instance creation. Further the CreateVolume response does not put any restriction on the volume, since it can be accessed across all zones on the same VPC.

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note
Topology feature support for Filestore CSI driver
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/pull/56#" title="Author self-approved">saikat-royc</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-filestore-csi-driver).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/blob/master/OWNERS)~~ [saikat-royc]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
saikat-royc: /assign @msau42 @mattcary 

Patch:
diff --git a/README.md b/README.md
index d39d5731..9146af76 100644
--- a/README.md
+++ b/README.md
@@ -41,13 +41,27 @@ volume. Customizable parameters for volume creation include:
 | ---------------   | ----------------------- |-----------                             | ----------- |
 | tier              | "standard"<br>"premium" | "standard"                             | storage performance tier |
 | network           | string                  | "default"                              | VPC name |
-| location          | string                  | zone where the plugin<br>is running in | zone |
 | reserved-ipv4-cidr| string		              | ""                                     | CIDR range to allocate Filestore IP Ranges from.<br>The CIDR must be large enough to accommodate multiple Filestore IP Ranges of /29 each |
 
 For Kubernetes clusters, these parameters are specified in the StorageClass.
 
 Note that non-default networks require extra [firewall setup](https://cloud.google.com/filestore/docs/configuring-firewall)
 
+## Current supported Features
+* Volume resizing: CSI Filestore driver supports volume expansion for all supported Filestore tiers.
+* Topology preferences: Filestore performance and network usage is affected by topology. For example, it is recommended to run
+  workloads in the same zone where the Cloud Filestore instance is provisioned in. The following table describes how provisioning can be tuned by topology. The volumeBindingMode is specified in the StorageClass used for provisioning. 'strict-topology' is a flag passed to the CSI provisioner sidecar. 'allowedTopology' is also specified in the StorageClass. The Filestore driver will use the first topology in the preferred list, or if empty the first in the requisite list. If topology feature is not enabled in CSI provisioner (--feature-gates=Topology=false), CreateVolume.accessibility_requirements will be nil, and the driver simply creates the instance in the zone where the driver deployment running.
+
+
+  | SC Bind Mode         | 'strict-topology' | SC allowedTopology  | CSI provisioner Behavior    |
+  | -------------------- | ----------------- | ------------------- | --------------------------- |
+  | WaitForFirstCustomer |       true        |        Present      | If the topology of the node selected by the schedule is not in allowedTopology, provisioning fails and the scheduler will continue with a different node. Otherwise, CreateVolume is called with requisite and preferred topologies set to that of the selected node |
+  | WaitForFirstCustomer |       false       |        Present      | If the topology of the node selected by the schedule is not in allowedTopology, provisioning fails and the scheduler will continue with a different node. Otherwise, CreateVolume is called with requisite set to allowedTopology and preferred set to allowedTopology rearranged with the selected node topology as the first parameter |
+  | WaitForFirstCustomer |       true        |        Not Present  | Call CreateVolume with requisite set to selected node topology, and preferred set to the same |
+  | WaitForFirstCustomer |       false       |        Not Present  | Call CreateVolume with requisite set to aggregated topology across all nodes, which matches the topology of the selected node, and preferred is set to the sorted and shifted version of requisite, with selected node topology as the first parameter |
+  | Immediate            |       N/A         |        Present      | Call CreateVolume with requisite set to allowedTopology and preferred set to the sorted and shifted version of requisite at a randomized index |
+  | Immediate            |       N/A         |        Not Present  | Call CreateVolume with requisite = aggregated topology across nodes which contain the topology keys of CSINode objects, preferred = sort and shift requisite at a randomized index |
+
 ## Future Features
 * Non-root access: By default, GCFS instances are only writable by the root user
   and readable by all users. Provide a CreateVolume parameter to set non-root
@@ -58,11 +72,6 @@ Note that non-default networks require extra [firewall setup](https://cloud.goog
   [nfs-client](https://github.com/kubernetes-incubator/external-storage/tree/master/nfs-client)
   external provisioner can be used to provide similar functionality for
   Kubernetes clusters.
-* Volume resizing: CSI Filestore driver supports volume expansion for all supported Filestore tiers.
-* Topology preferences: For better performance, it is recommended to run
-  workloads in the same zone where the Cloud Filestore instance is provisioned in. In the
-  future, the location where to create a Cloud Filestore instance could be automatically
-  influenced by where the workload is scheduled.
 
 ## Kubernetes User Guide
 1. One-time per project: Create GCP service account for the CSI driver and set the Cloud
diff --git a/cmd/main.go b/cmd/main.go
index cfbfd10b..7c859e69 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -47,10 +47,15 @@ func main() {
 	var err error
 	if *runController {
 		provider, err = cloud.NewCloud(version)
-		if err != nil {
-			glog.Fatalf("Failed to initialize cloud provider: %v", err)
-		}
+	} else {
+		// For driver running in worker nodes, initialize cloud provider with only medata service.
+		provider, err = cloud.NewCloudWithMetadataService()
 	}
+
+	if err != nil {
+		glog.Fatalf("Failed to initialize cloud provider: %v", err)
+	}
+
 	mounter := mount.New("")
 	config := &driver.GCFSDriverConfig{
 		Name:          driverName,
diff --git a/deploy/kubernetes/manifests/controller.yaml b/deploy/kubernetes/manifests/controller.yaml
index b63756b9..7bdbcc72 100644
--- a/deploy/kubernetes/manifests/controller.yaml
+++ b/deploy/kubernetes/manifests/controller.yaml
@@ -24,6 +24,7 @@ spec:
             - "--v=5"
             - "--csi-address=/csi/csi.sock"
             - "--timeout=250s"
+            - "--feature-gates=Topology=true"
           volumeMounts:
             - name: socket-dir
               mountPath: /csi
diff --git a/examples/kubernetes/demo-sc.yaml b/examples/kubernetes/demo-sc.yaml
index 91c3bdbd..9445bd8a 100644
--- a/examples/kubernetes/demo-sc.yaml
+++ b/examples/kubernetes/demo-sc.yaml
@@ -4,8 +4,6 @@ metadata:
   name: csi-filestore
 provisioner: filestore.csi.storage.gke.io
 parameters:
-  # Available locations can be found at: TODO
-  location: us-central1-c
   # "CIDR range to allocate Filestore IP Ranges from"
   # reserved-ipv4-cidr: 192.168.92.22/26
   # # standard (default) or premier
diff --git a/examples/kubernetes/sc-latebind.yaml b/examples/kubernetes/sc-latebind.yaml
index 229a0ed2..a236169e 100644
--- a/examples/kubernetes/sc-latebind.yaml
+++ b/examples/kubernetes/sc-latebind.yaml
@@ -3,7 +3,5 @@ kind: StorageClass
 metadata:
   name: csi-filestore
 provisioner: filestore.csi.storage.gke.io
-parameters:
-  location: us-central1-c
 volumeBindingMode: WaitForFirstConsumer
 allowVolumeExpansion: true
diff --git a/examples/kubernetes/topology/delayed-binding/demo-deployment-delayed-allowedtopo.yaml b/examples/kubernetes/topology/delayed-binding/demo-deployment-delayed-allowedtopo.yaml
new file mode 100644
index 00000000..8a96cfee
--- /dev/null
+++ b/examples/kubernetes/topology/delayed-binding/demo-deployment-delayed-allowedtopo.yaml
@@ -0,0 +1,38 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: web-server-delayed-binding-allowedtopo
+  labels:
+    app: nginx
+spec:
+  replicas: 5
+  selector:
+    matchLabels:
+      app: nginx
+  template:
+    metadata:
+      labels:
+        app: nginx
+    spec:
+      containers:
+      - name: nginx
+        image: nginx
+        volumeMounts:
+        - mountPath: /usr/share/nginx/html
+          name: mypvc
+      volumes:
+      - name: mypvc
+        persistentVolumeClaim:
+          claimName: test-pvc-fs-delayed-binding-allowedtopo
+---
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: test-pvc-fs-delayed-binding-allowedtopo
+spec:
+  accessModes:
+    - ReadWriteMany
+  storageClassName: csi-filestore-delayed-binding-allowedtopo
+  resources:
+    requests:
+      storage: 1Ti
diff --git a/examples/kubernetes/topology/delayed-binding/sc-delayed-allowedtopo.yaml b/examples/kubernetes/topology/delayed-binding/sc-delayed-allowedtopo.yaml
new file mode 100644
index 00000000..ab00db4f
--- /dev/null
+++ b/examples/kubernetes/topology/delayed-binding/sc-delayed-allowedtopo.yaml
@@ -0,0 +1,14 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: csi-filestore-delayed-binding-allowedtopo
+provisioner: filestore.csi.storage.gke.io
+volumeBindingMode: WaitForFirstConsumer
+allowVolumeExpansion: true
+allowedTopologies:
+- matchLabelExpressions:
+  - key: topology.gke.io/zone
+    # Replace the values with a subset of zones of the regional cluster
+    values:
+    - us-central1-a
+    - us-central1-b
diff --git a/examples/kubernetes/topology/immediate-binding/demo-deployment-immediate-allowedtopo.yaml b/examples/kubernetes/topology/immediate-binding/demo-deployment-immediate-allowedtopo.yaml
new file mode 100644
index 00000000..20029741
--- /dev/null
+++ b/examples/kubernetes/topology/immediate-binding/demo-deployment-immediate-allowedtopo.yaml
@@ -0,0 +1,41 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: web-server-immediate-allowedtopo
+  labels:
+    app: nginx
+spec:
+  replicas: 5
+  selector:
+    matchLabels:
+      app: nginx
+  template:
+    metadata:
+      labels:
+        app: nginx
+    spec:
+      containers:
+      - name: nginx
+        image: nginx
+        volumeMounts:
+        - mountPath: /usr/share/nginx/html
+          name: mypvc
+      nodeSelector:
+        topology.gke.io/zone: us-central1-a
+      volumes:
+      - name: mypvc
+        persistentVolumeClaim:
+          claimName: test-pvc-fs-immediate-binding-allowedtopo
+
+---
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: test-pvc-fs-immediate-binding-allowedtopo
+spec:
+  accessModes:
+    - ReadWriteMany
+  storageClassName: csi-filestore-immediate-binding-allowedtopo
+  resources:
+    requests:
+      storage: 1Ti
diff --git a/examples/kubernetes/topology/immediate-binding/sc-immediate-allowedtopo.yaml b/examples/kubernetes/topology/immediate-binding/sc-immediate-allowedtopo.yaml
new file mode 100644
index 00000000..e1b38def
--- /dev/null
+++ b/examples/kubernetes/topology/immediate-binding/sc-immediate-allowedtopo.yaml
@@ -0,0 +1,12 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: csi-filestore-immediate-binding-allowedtopo
+provisioner: filestore.csi.storage.gke.io
+volumeBindingMode: Immediate
+allowVolumeExpansion: true
+allowedTopologies:
+- matchLabelExpressions:
+  - key: topology.gke.io/zone
+    values:
+    - us-central1-a
diff --git a/pkg/cloud_provider/cloud.go b/pkg/cloud_provider/cloud.go
index 99a50ce2..64909533 100644
--- a/pkg/cloud_provider/cloud.go
+++ b/pkg/cloud_provider/cloud.go
@@ -43,3 +43,14 @@ func NewCloud(version string) (*Cloud, error) {
 		Meta: meta,
 	}, nil
 }
+
+func NewCloudWithMetadataService() (*Cloud, error) {
+	meta, err := metadata.NewMetadataService()
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize Metadata service: %v", err)
+	}
+
+	return &Cloud{
+		Meta: meta,
+	}, nil
+}
diff --git a/pkg/csi_driver/controller.go b/pkg/csi_driver/controller.go
index 324eddda..925e696a 100644
--- a/pkg/csi_driver/controller.go
+++ b/pkg/csi_driver/controller.go
@@ -38,6 +38,9 @@ const (
 
 	defaultTier    = "standard"
 	defaultNetwork = "default"
+
+	// Keys for Topology.
+	TopologyKeyZone = "topology.gke.io/zone"
 )
 
 // Volume attributes
@@ -91,7 +94,7 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 	}
 
 	glog.V(5).Infof("Using capacity bytes %q for volume %q", capBytes, name)
-	newFiler, err := s.generateNewFileInstance(name, capBytes, req.GetParameters())
+	newFiler, err := s.generateNewFileInstance(name, capBytes, req.GetParameters(), req.GetAccessibilityRequirements())
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -101,6 +104,7 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 	if err != nil && !file.IsNotFoundErr(err) {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
+
 	if filer != nil {
 		// Instance already exists, check if it meets the request
 		if err = file.CompareInstances(newFiler, filer); err != nil {
@@ -268,22 +272,24 @@ func getRequestCapacity(capRange *csi.CapacityRange) (int64, error) {
 
 // generateNewFileInstance populates the GCFS Instance object using
 // CreateVolume parameters
-func (s *controllerServer) generateNewFileInstance(name string, capBytes int64, params map[string]string) (*file.ServiceInstance, error) {
+func (s *controllerServer) generateNewFileInstance(name string, capBytes int64, params map[string]string, topo *csi.TopologyRequirement) (*file.ServiceInstance, error) {
+	location, err := s.pickZone(topo)
+	if err != nil {
+		return nil, fmt.Errorf("invalid topology error %v", err.Error())
+	}
+
 	// Set default parameters
 	tier := defaultTier
 	network := defaultNetwork
-	location := s.config.metaService.GetZone()
+
 	// Validate parameters (case-insensitive).
 	for k, v := range params {
 		switch strings.ToLower(k) {
 		// Cloud API will validate these
 		case paramTier:
 			tier = v
-		case paramLocation:
-			location = v
 		case paramNetwork:
 			network = v
-
 		// Ignore the cidr flag as it is not passed to the cloud provider
 		// It will be used to get unreserved IP in the reserveIPV4Range function
 		case paramReservedIPV4CIDR:
@@ -351,3 +357,66 @@ func (s *controllerServer) ControllerExpandVolume(ctx context.Context, req *csi.
 		NodeExpansionRequired: false,
 	}, nil
 }
+
+func (s *controllerServer) pickZone(top *csi.TopologyRequirement) (string, error) {
+	if top == nil {
+		return s.config.metaService.GetZone(), nil
+	}
+
+	return pickZoneFromTopology(top)
+}
+
+// Pick the first available topology from preferred list or requisite list in that order.
+func pickZoneFromTopology(top *csi.TopologyRequirement) (string, error) {
+	reqZones, err := getZonesFromTopology(top.GetRequisite())
+	if err != nil {
+		return "", fmt.Errorf("could not get zones from requisite topology: %v", err)
+	}
+
+	prefZones, err := getZonesFromTopology(top.GetPreferred())
+	if err != nil {
+		return "", fmt.Errorf("could not get zones from preferred topology: %v", err)
+	}
+
+	if len(prefZones) == 0 && len(reqZones) == 0 {
+		return "", fmt.Errorf("both requisite and preferred topology list empty")
+	}
+
+	if len(prefZones) != 0 {
+		return prefZones[0], nil
+	}
+	return reqZones[0], nil
+}
+
+func getZonesFromTopology(topList []*csi.Topology) ([]string, error) {
+	zones := []string{}
+	for _, top := range topList {
+		if top.GetSegments() == nil {
+			return nil, fmt.Errorf("topologies specified but no segments")
+		}
+
+		zone, err := getZoneFromSegment(top.GetSegments())
+		if err != nil {
+			return nil, fmt.Errorf("could not get zone from topology: %v", err)
+		}
+		zones = append(zones, zone)
+	}
+	return zones, nil
+}
+
+func getZoneFromSegment(seg map[string]string) (string, error) {
+	var zone string
+	for k, v := range seg {
+		switch k {
+		case TopologyKeyZone:
+			zone = v
+		default:
+			return "", fmt.Errorf("topology segment has unknown key %v", k)
+		}
+	}
+
+	if len(zone) == 0 {
+		return "", fmt.Errorf("topology specified but could not find zone in segment: %v", seg)
+	}
+	return zone, nil
+}
diff --git a/pkg/csi_driver/gcfs_driver.go b/pkg/csi_driver/gcfs_driver.go
index 42d6e09b..e34253cf 100644
--- a/pkg/csi_driver/gcfs_driver.go
+++ b/pkg/csi_driver/gcfs_driver.go
@@ -82,7 +82,7 @@ func NewGCFSDriver(config *GCFSDriverConfig) (*GCFSDriver, error) {
 	// Setup RPC servers
 	driver.ids = newIdentityServer(driver)
 	if config.RunNode {
-		driver.ns = newNodeServer(driver, config.Mounter)
+		driver.ns = newNodeServer(driver, config.Mounter, config.Cloud.Meta)
 	}
 	if config.RunController {
 		csc := []csi.ControllerServiceCapability_RPC_Type{
diff --git a/pkg/csi_driver/identity.go b/pkg/csi_driver/identity.go
index b02e757d..8ab2afab 100644
--- a/pkg/csi_driver/identity.go
+++ b/pkg/csi_driver/identity.go
@@ -60,6 +60,13 @@ func (s *identityServer) GetPluginCapabilities(ctx context.Context, req *csi.Get
 					},
 				},
 			},
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			},
 		},
 	}, nil
 }
diff --git a/pkg/csi_driver/node.go b/pkg/csi_driver/node.go
index f0f186e5..a2f7970d 100644
--- a/pkg/csi_driver/node.go
+++ b/pkg/csi_driver/node.go
@@ -27,6 +27,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/utils/mount"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/metadata"
 )
 
 const (
@@ -41,14 +42,16 @@ var (
 
 // nodeServer handles mounting and unmounting of GCFS volumes on a node
 type nodeServer struct {
-	driver  *GCFSDriver
-	mounter mount.Interface
+	driver      *GCFSDriver
+	mounter     mount.Interface
+	metaService metadata.Service
 }
 
-func newNodeServer(driver *GCFSDriver, mounter mount.Interface) csi.NodeServer {
+func newNodeServer(driver *GCFSDriver, mounter mount.Interface, metaService metadata.Service) csi.NodeServer {
 	return &nodeServer{
-		driver:  driver,
-		mounter: mounter,
+		driver:      driver,
+		mounter:     mounter,
+		metaService: metaService,
 	}
 }
 
@@ -157,6 +160,9 @@ func (s *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpub
 func (s *nodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
 	return &csi.NodeGetInfoResponse{
 		NodeId: s.driver.config.NodeID,
+		AccessibleTopology: &csi.Topology{
+			Segments: map[string]string{TopologyKeyZone: s.metaService.GetZone()},
+		},
 	}, nil
 }
 


Test patch:
diff --git a/pkg/csi_driver/controller_test.go b/pkg/csi_driver/controller_test.go
index f75f7e87..3a9b16b8 100644
--- a/pkg/csi_driver/controller_test.go
+++ b/pkg/csi_driver/controller_test.go
@@ -22,6 +22,7 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/file"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/metadata"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
@@ -309,11 +310,12 @@ func TestGenerateNewFileInstance(t *testing.T) {
 	cases := []struct {
 		name      string
 		params    map[string]string
+		toporeq   *csi.TopologyRequirement
 		instance  *file.ServiceInstance
 		expectErr bool
 	}{
 		{
-			name: "default params",
+			name: "default params, nil topology requirement",
 			instance: &file.ServiceInstance{
 				Project:  testProject,
 				Name:     testCSIVolume,
@@ -329,10 +331,99 @@ func TestGenerateNewFileInstance(t *testing.T) {
 			},
 		},
 		{
-			name: "custom params",
+			name: "custom params, non-nil topology requirement",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "foo-location",
+						},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "foo-location",
+						},
+					},
+				},
+			},
+			params: map[string]string{
+				paramTier:                       "foo-tier",
+				paramNetwork:                    "foo-network",
+				"csiProvisionerSecretName":      "foo-secret",
+				"csiProvisionerSecretNamespace": "foo-namespace",
+			},
+			instance: &file.ServiceInstance{
+				Project:  testProject,
+				Name:     testCSIVolume,
+				Location: "foo-location",
+				Tier:     "foo-tier",
+				Network: file.Network{
+					Name: "foo-network",
+				},
+				Volume: file.Volume{
+					Name:      newInstanceVolume,
+					SizeBytes: testBytes,
+				},
+			},
+		},
+		{
+			name: "custom params, preferred topology requirement",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "foo-location",
+						},
+					},
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "bar-location",
+						},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "bar-location",
+						},
+					},
+				},
+			},
+			params: map[string]string{
+				paramTier:                       "foo-tier",
+				paramNetwork:                    "foo-network",
+				"csiProvisionerSecretName":      "foo-secret",
+				"csiProvisionerSecretNamespace": "foo-namespace",
+			},
+			instance: &file.ServiceInstance{
+				Project:  testProject,
+				Name:     testCSIVolume,
+				Location: "bar-location",
+				Tier:     "foo-tier",
+				Network: file.Network{
+					Name: "foo-network",
+				},
+				Volume: file.Volume{
+					Name:      newInstanceVolume,
+					SizeBytes: testBytes,
+				},
+			},
+		},
+		{
+			name: "custom params, requisite topology requirement only",
+			toporeq: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{
+							TopologyKeyZone: "foo-location",
+						},
+					},
+				},
+			},
 			params: map[string]string{
 				paramTier:                       "foo-tier",
-				paramLocation:                   "foo-location",
 				paramNetwork:                    "foo-network",
 				"csiProvisionerSecretName":      "foo-secret",
 				"csiProvisionerSecretNamespace": "foo-namespace",
@@ -367,7 +458,7 @@ func TestGenerateNewFileInstance(t *testing.T) {
 			t.Fatalf("couldn't get internal controller")
 		}
 
-		filer, err := internalServer.generateNewFileInstance(testCSIVolume, testBytes, test.params)
+		filer, err := internalServer.generateNewFileInstance(testCSIVolume, testBytes, test.params, test.toporeq)
 		if !test.expectErr && err != nil {
 			t.Errorf("test %q failed: %v", test.name, err)
 		}
@@ -379,3 +470,176 @@ func TestGenerateNewFileInstance(t *testing.T) {
 		}
 	}
 }
+
+func TestGetZoneFromSegment(t *testing.T) {
+	cases := []struct {
+		name         string
+		seg          map[string]string
+		expectErr    bool
+		expectedZone string
+	}{
+		// Error cases
+		{
+			name:      "Empty segment map",
+			seg:       make(map[string]string),
+			expectErr: true,
+		},
+		{
+			name: "Missing zone key in segment map",
+			seg: map[string]string{
+				"zonekey": "z1",
+			},
+			expectErr: true,
+		},
+		{
+			name: "Unknown zone key in segment map",
+			seg: map[string]string{
+				TopologyKeyZone: "z1",
+				"unknown_key":   "z2",
+			},
+			expectErr: true,
+		},
+		// Successful cases
+		{
+			name: "Found expected zone",
+			seg: map[string]string{
+				TopologyKeyZone: "z1",
+			},
+			expectedZone: "z1",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			z, err := getZoneFromSegment(tc.seg)
+			if tc.expectErr && err == nil {
+				t.Errorf("Expected error, got none")
+			}
+
+			if !tc.expectErr && err != nil {
+				t.Errorf("Unexpected error %v", err)
+			}
+
+			if z != tc.expectedZone {
+				t.Errorf("Unexpected zone %v, expected zone %v", z, tc.expectedZone)
+			}
+		})
+	}
+}
+
+func TestGetZonesFromTopology(t *testing.T) {
+	cases := []struct {
+		name          string
+		topo          []*csi.Topology
+		expectErr     bool
+		expectedZones []string
+	}{
+		// Error cases
+		{
+			name:          "nil topology list",
+			topo:          nil,
+			expectedZones: make([]string, 0),
+		},
+		{
+			name:          "Empty topology list",
+			topo:          make([]*csi.Topology, 0),
+			expectedZones: make([]string, 0),
+		},
+		{
+			name:      "Non-Empty topology list with missing segment",
+			topo:      make([]*csi.Topology, 1),
+			expectErr: true,
+		},
+		{
+			name: "Non-Empty topology list with segment missing zone key",
+			topo: []*csi.Topology{
+				{
+					Segments: map[string]string{},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "Non-Empty topology list with segment unknown zone key",
+			topo: []*csi.Topology{
+				{
+					Segments: map[string]string{
+						"unknown_zone_key": "z1",
+					},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "Non-Empty topology list with empty segment map",
+			topo: []*csi.Topology{
+				{
+					Segments: make(map[string]string),
+				},
+			},
+			expectErr: true,
+		},
+		// two elements, one with error.
+		{
+			name: "Non-Empty topology list with error in one of the elements",
+			topo: []*csi.Topology{
+				{
+					Segments: map[string]string{
+						"unknown_key": "z1",
+					},
+				},
+				{
+					Segments: map[string]string{
+						TopologyKeyZone: "z2",
+					},
+				},
+			},
+			expectErr: true,
+		},
+		// Success cases
+		{
+			name: "Non-Empty topology list with valid segment",
+			topo: []*csi.Topology{
+				{
+					Segments: map[string]string{
+						TopologyKeyZone: "z1",
+					},
+				},
+			},
+			expectedZones: []string{"z1"},
+		},
+		{
+			name: "Non-Empty topology list with multiple zones",
+			topo: []*csi.Topology{
+				{
+					Segments: map[string]string{
+						TopologyKeyZone: "z1",
+					},
+				},
+				{
+					Segments: map[string]string{
+						TopologyKeyZone: "z2",
+					},
+				},
+			},
+			expectedZones: []string{"z1", "z2"},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			z, err := getZonesFromTopology(tc.topo)
+			if tc.expectErr && err == nil {
+				t.Errorf("Expected error, got none")
+			}
+
+			if !tc.expectErr && err != nil {
+				t.Errorf("Unexpected error %v", err)
+			}
+
+			if !sets.NewString(z...).Equal(sets.NewString(tc.expectedZones...)) {
+				t.Errorf("Unexpected zone list %v, expected zone list %v", z, tc.expectedZones)
+			}
+		})
+	}
+}
diff --git a/pkg/csi_driver/gcfs_driver_test.go b/pkg/csi_driver/gcfs_driver_test.go
index aa84315d..5689865c 100644
--- a/pkg/csi_driver/gcfs_driver_test.go
+++ b/pkg/csi_driver/gcfs_driver_test.go
@@ -20,14 +20,21 @@ import (
 	"testing"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
 )
 
 func initTestDriver(t *testing.T) *GCFSDriver {
+	c, err := cloud.NewFakeCloud()
+	if err != nil {
+		t.Fatalf("Failed to init cloud")
+	}
+
 	config := &GCFSDriverConfig{
 		Name:    "test-driver",
 		NodeID:  "test-node",
 		Version: "test-version",
 		RunNode: true,
+		Cloud:   c,
 	}
 	driver, err := NewGCFSDriver(config)
 	if err != nil {
diff --git a/pkg/csi_driver/identity_test.go b/pkg/csi_driver/identity_test.go
index d740e7f0..efa5c64a 100644
--- a/pkg/csi_driver/identity_test.go
+++ b/pkg/csi_driver/identity_test.go
@@ -65,7 +65,7 @@ func TestGetPluginCapabilities(t *testing.T) {
 		t.Fatalf("GetPluginCapabilities resp is nil")
 	}
 
-	if len(resp.Capabilities) != 3 {
+	if len(resp.Capabilities) != 4 {
 		t.Fatalf("returned %v capabilities", len(resp.Capabilities))
 	}
 
@@ -99,6 +99,15 @@ func TestGetPluginCapabilities(t *testing.T) {
 	if expansiontype := volumeexpansion.GetType(); expansiontype != csi.PluginCapability_VolumeExpansion_OFFLINE {
 		t.Fatalf("returned %v capability", expansiontype)
 	}
+
+	service = resp.Capabilities[3].GetService()
+	if service == nil {
+		t.Fatalf("returned nil capability service")
+	}
+
+	if serviceType := service.GetType(); serviceType != csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS {
+		t.Fatalf("returned %v capability service", serviceType)
+	}
 }
 
 func TestProbe(t *testing.T) {
diff --git a/pkg/csi_driver/node_test.go b/pkg/csi_driver/node_test.go
index e783e846..1a953c36 100644
--- a/pkg/csi_driver/node_test.go
+++ b/pkg/csi_driver/node_test.go
@@ -25,6 +25,7 @@ import (
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
 	"k8s.io/utils/mount"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/metadata"
 )
 
 var (
@@ -58,8 +59,12 @@ type nodeServerTestEnv struct {
 func initTestNodeServer(t *testing.T) *nodeServerTestEnv {
 	// TODO: make a constructor in FakeMmounter library
 	mounter := &mount.FakeMounter{MountPoints: []mount.MountPoint{}}
+	metaserice, err := metadata.NewFakeService()
+	if err != nil {
+		t.Fatalf("Failed to init metadata service")
+	}
 	return &nodeServerTestEnv{
-		ns: newNodeServer(initTestDriver(t), mounter),
+		ns: newNodeServer(initTestDriver(t), mounter, metaserice),
 		fm: mounter,
 	}
 }