PV NodeAffinity

.gitignore

@@ -41,6 +41,7 @@ flycheck_*.el
 
 .idea/
 /.project
+*.iml
 
 # ignore build directory
 _output/

pkg/controller/controller.go

@@ -296,7 +296,7 @@ func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration, needSn
 		// Check whether plugin supports create snapshot
 		// If not, create volume from snapshot cannot proceed
 		if !capabilities.Has(ControllerCapability_CREATE_DELETE_SNAPSHOT) {
-			return nil, fmt.Errorf("no create/delete snapshot support detected. Cannot create volume from shapshot")
+			return nil, fmt.Errorf("no create/delete snapshot support detected. Cannot create volume from snapshot")
 		}
 	}
 
@@ -496,6 +496,7 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	if len(fsType) == 0 {
 		fsType = defaultFSType
 	}
+
 	pv := &v1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: pvName,
@@ -521,6 +522,10 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		},
 	}
 
+	if driverState.capabilities.Has(PluginCapability_ACCESSIBILITY_CONSTRAINTS) {
+		pv.Spec.NodeAffinity = GenerateVolumeNodeAffinity(rep.Volume.AccessibleTopology)
+	}
+
 	glog.Infof("successfully created PV %+v", pv.Spec.PersistentVolumeSource)
 
 	return pv, nil

pkg/controller/controller_test.go

@@ -25,7 +25,7 @@ import (
 	"testing"
 	"time"
 
-	csi "github.com/container-storage-interface/spec/lib/go/csi/v0"
+	"github.com/container-storage-interface/spec/lib/go/csi/v0"
 	"github.com/golang/mock/gomock"
 	"github.com/kubernetes-csi/csi-test/driver"
 	crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1"
@@ -494,7 +494,7 @@ func provisionMockServerSetupExpectations(identityServer *driver.MockIdentitySer
 func provisionFromSnapshotMockServerSetupExpectations(identityServer *driver.MockIdentityServer, controllerServer *driver.MockControllerServer) {
 	identityServer.EXPECT().GetPluginCapabilities(gomock.Any(), gomock.Any()).Return(&csi.GetPluginCapabilitiesResponse{
 		Capabilities: []*csi.PluginCapability{
-			&csi.PluginCapability{
+			{
 				Type: &csi.PluginCapability_Service_{
 					Service: &csi.PluginCapability_Service{
 						Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
@@ -505,14 +505,14 @@ func provisionFromSnapshotMockServerSetupExpectations(identityServer *driver.Moc
 	}, nil).Times(1)
 	controllerServer.EXPECT().ControllerGetCapabilities(gomock.Any(), gomock.Any()).Return(&csi.ControllerGetCapabilitiesResponse{
 		Capabilities: []*csi.ControllerServiceCapability{
-			&csi.ControllerServiceCapability{
+			{
 				Type: &csi.ControllerServiceCapability_Rpc{
 					Rpc: &csi.ControllerServiceCapability_RPC{
 						Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 					},
 				},
 			},
-			&csi.ControllerServiceCapability{
+			{
 				Type: &csi.ControllerServiceCapability_Rpc{
 					Rpc: &csi.ControllerServiceCapability_RPC{
 						Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
@@ -1496,3 +1496,68 @@ func TestProvisionFromSnapshot(t *testing.T) {
 		}
 	}
 }
+
+// TestProvisionWithTopology is a basic test of provisioner integration with topology functions.
+func TestProvisionWithTopology(t *testing.T) {
+	accessibleTopology := []*csi.Topology{
+		{
+			Segments: map[string]string{
+				"com.example.csi/zone": "zone1",
+				"com.example.csi/rack": "rack2",
+			},
+		},
+	}
+	expectedNodeAffinity := &v1.VolumeNodeAffinity{
+		Required: &v1.NodeSelector{
+			NodeSelectorTerms: []v1.NodeSelectorTerm{
+				{
+					MatchExpressions: []v1.NodeSelectorRequirement{
+						{
+							Key:      "com.example.csi/zone",
+							Operator: v1.NodeSelectorOpIn,
+							Values:   []string{"zone1"},
+						},
+						{
+							Key:      "com.example.csi/rack",
+							Operator: v1.NodeSelectorOpIn,
+							Values:   []string{"rack2"},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	const requestBytes = 100
+	mockController, driver, identityServer, controllerServer, csiConn, err := createMockServer(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer mockController.Finish()
+	defer driver.Stop()
+
+	clientSet := fakeclientset.NewSimpleClientset()
+	csiProvisioner := NewCSIProvisioner(clientSet, driver.Address(), 5*time.Second, "test-provisioner", "test", 5, csiConn.conn, nil)
+
+	out := &csi.CreateVolumeResponse{
+		Volume: &csi.Volume{
+			CapacityBytes:      requestBytes,
+			Id:                 "test-volume-id",
+			AccessibleTopology: accessibleTopology,
+		},
+	}
+
+	provisionWithTopologyMockServerSetupExpectations(identityServer, controllerServer)
+	controllerServer.EXPECT().CreateVolume(gomock.Any(), gomock.Any()).Return(out, nil).Times(1)
+
+	pv, err := csiProvisioner.Provision(controller.VolumeOptions{
+		PVC: createFakePVC(requestBytes), // dummy PVC
+	})
+	if err != nil {
+		t.Errorf("got error from Provision call: %v", err)
+	}
+
+	if !volumeNodeAffinitiesEqual(pv.Spec.NodeAffinity, expectedNodeAffinity) {
+		t.Errorf("expected node affinity %v; got: %v", expectedNodeAffinity, pv.Spec.NodeAffinity)
+	}
+}

pkg/controller/topology.go

@@ -0,0 +1,53 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"github.com/container-storage-interface/spec/lib/go/csi/v0"
+	"k8s.io/api/core/v1"
+)
+
+func GenerateVolumeNodeAffinity(accessibleTopology []*csi.Topology) *v1.VolumeNodeAffinity {
+	if len(accessibleTopology) == 0 {
+		return nil
+	}
+
+	var terms []v1.NodeSelectorTerm
+	for _, topology := range accessibleTopology {
+		if len(topology.Segments) == 0 {
+			continue
+		}
+
+		var expressions []v1.NodeSelectorRequirement
+		for k, v := range topology.Segments {
+			expressions = append(expressions, v1.NodeSelectorRequirement{
+				Key:      k,
+				Operator: v1.NodeSelectorOpIn,
+				Values:   []string{v},
+			})
+		}
+		terms = append(terms, v1.NodeSelectorTerm{
+			MatchExpressions: expressions,
+		})
+	}
+
+	return &v1.VolumeNodeAffinity{
+		Required: &v1.NodeSelector{
+			NodeSelectorTerms: terms,
+		},
+	}
+}