mock: optionally support topology

This is off by default. If enabled, the mock driver returns the
capability for topology and a fixed "io.kubernetes.storage.mock/node:
some-mock-node" topology segment for the node it runs on and all
volumes.

This makes it possible to test some more code paths in
external-provisioner.

Author: pohly

cmd/mock-driver/main.go

@@ -37,6 +37,7 @@ func main() {
 	flag.StringVar(&config.DriverName, "name", service.Name, "CSI driver name.")
 	flag.Int64Var(&config.AttachLimit, "attach-limit", 2, "number of attachable volumes on a node")
 	flag.BoolVar(&config.NodeExpansionRequired, "node-expand-required", false, "Enables NodeServiceCapability_RPC_EXPAND_VOLUME capacity.")
+	flag.BoolVar(&config.EnableTopology, "enable-topology", false, "Enables PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS capability.")
 	flag.BoolVar(&config.DisableControllerExpansion, "disable-controller-expansion", false, "Disables ControllerServiceCapability_RPC_EXPAND_VOLUME capability.")
 	flag.BoolVar(&config.DisableOnlineExpansion, "disable-online-expansion", false, "Disables online volume expansion capability.")
 	flag.BoolVar(&config.PermissiveTargetPath, "permissive-target-path", false, "Allows the CO to create PublishVolumeRequest.TargetPath, which violates the CSI spec.")

mock/service/identity.go

@@ -40,22 +40,35 @@ func (s *service) GetPluginCapabilities(
 		volExpType = csi.PluginCapability_VolumeExpansion_OFFLINE
 	}
 
-	return &csi.GetPluginCapabilitiesResponse{
-		Capabilities: []*csi.PluginCapability{
-			{
-				Type: &csi.PluginCapability_Service_{
-					Service: &csi.PluginCapability_Service{
-						Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
-					},
+	capabilities := []*csi.PluginCapability{
+		{
+			Type: &csi.PluginCapability_Service_{
+				Service: &csi.PluginCapability_Service{
+					Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
 				},
 			},
-			{
-				Type: &csi.PluginCapability_VolumeExpansion_{
-					VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
-						Type: volExpType,
-					},
+		},
+		{
+			Type: &csi.PluginCapability_VolumeExpansion_{
+				VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
+					Type: volExpType,
 				},
 			},
 		},
+	}
+
+	if s.config.EnableTopology {
+		capabilities = append(capabilities,
+			&csi.PluginCapability{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			})
+	}
+
+	return &csi.GetPluginCapabilitiesResponse{
+		Capabilities: capabilities,
 	}, nil
 }

mock/service/node.go

@@ -381,6 +381,13 @@ func (s *service) NodeGetInfo(ctx context.Context,
 	if s.config.AttachLimit > 0 {
 		csiNodeResponse.MaxVolumesPerNode = s.config.AttachLimit
 	}
+	if s.config.EnableTopology {
+		csiNodeResponse.AccessibleTopology = &csi.Topology{
+			Segments: map[string]string{
+				TopologyKey: TopologyValue,
+			},
+		}
+	}
 	return csiNodeResponse, nil
 }
 

mock/service/service.go

@@ -24,6 +24,12 @@ const (
 
 	// VendorVersion is the version returned by GetPluginInfo.
 	VendorVersion = "0.3.0"
+
+	// TopologyKey simulates a per-node topology.
+	TopologyKey = Name + "/node"
+
+	// TopologyValue is the one, fixed node on which the driver runs.
+	TopologyValue = "some-mock-node"
 )
 
 // Manifest is the SP's manifest.
@@ -79,6 +85,7 @@ type Config struct {
 	DisableControllerExpansion bool
 	DisableOnlineExpansion     bool
 	PermissiveTargetPath       bool
+	EnableTopology             bool
 	ExecHooks                  *Hooks
 }
 
@@ -154,11 +161,13 @@ const (
 )
 
 func (s *service) newVolume(name string, capcity int64) csi.Volume {
-	return csi.Volume{
+	vol := csi.Volume{
 		VolumeId:      fmt.Sprintf("%d", atomic.AddUint64(&s.volsNID, 1)),
 		VolumeContext: map[string]string{"name": name},
 		CapacityBytes: capcity,
 	}
+	s.setTopology(&vol)
+	return vol
 }
 
 func (s *service) newVolumeFromSnapshot(name string, capacity int64, snapshotID int) csi.Volume {
@@ -170,6 +179,7 @@ func (s *service) newVolumeFromSnapshot(name string, capacity int64, snapshotID
 			},
 		},
 	}
+	s.setTopology(&vol)
 	return vol
 }
 
@@ -182,9 +192,22 @@ func (s *service) newVolumeFromVolume(name string, capacity int64, volumeID int)
 			},
 		},
 	}
+	s.setTopology(&vol)
 	return vol
 }
 
+func (s *service) setTopology(vol *csi.Volume) {
+	if s.config.EnableTopology {
+		vol.AccessibleTopology = []*csi.Topology{
+			&csi.Topology{
+				Segments: map[string]string{
+					TopologyKey: TopologyValue,
+				},
+			},
+		}
+	}
+}
+
 func (s *service) findVol(k, v string) (volIdx int, volInfo csi.Volume) {
 	s.volsRWL.RLock()
 	defer s.volsRWL.RUnlock()