CSI: implement support for topology

.changelog/12129.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+csi: Added support for storage topology
+```

api/csi.go

@@ -208,11 +208,21 @@ type CSISecrets map[string]string
 
 // CSIVolume is used for serialization, see also nomad/structs/csi.go
 type CSIVolume struct {
-	ID             string
-	Name           string
-	ExternalID     string `mapstructure:"external_id" hcl:"external_id"`
-	Namespace      string
-	Topologies     []*CSITopology
+	ID         string
+	Name       string
+	ExternalID string `mapstructure:"external_id" hcl:"external_id"`
+	Namespace  string
+
+	// RequestedTopologies are the topologies submitted as options to
+	// the storage provider at the time the volume was created. After
+	// volumes are created, this field is ignored.
+	RequestedTopologies *CSITopologyRequest `hcl:"topology_request"`
+
+	// Topologies are the topologies returned by the storage provider,
+	// based on the RequestedTopologies and what the storage provider
+	// could support. This value cannot be set by the user.
+	Topologies []*CSITopology
+
 	AccessMode     CSIVolumeAccessMode     `hcl:"access_mode"`
 	AttachmentMode CSIVolumeAttachmentMode `hcl:"attachment_mode"`
 	MountOptions   *CSIMountOptions        `hcl:"mount_options"`

api/nodes.go

@@ -614,8 +614,13 @@ type NodeReservedNetworkResources struct {
 	ReservedHostPorts string
 }
 
+type CSITopologyRequest struct {
+	Required  []*CSITopology `hcl:"required"`
+	Preferred []*CSITopology `hcl:"preferred"`
+}
+
 type CSITopology struct {
-	Segments map[string]string
+	Segments map[string]string `hcl:"segments"`
 }
 
 // CSINodeInfo is the fingerprinted data from a CSI Plugin that is specific to

client/csi_endpoint.go

@@ -216,6 +216,12 @@ func (c *CSI) ControllerCreateVolume(req *structs.ClientCSIControllerCreateVolum
 	resp.CapacityBytes = cresp.Volume.CapacityBytes
 	resp.VolumeContext = cresp.Volume.VolumeContext
 
+	resp.Topologies = make([]*nstructs.CSITopology, len(cresp.Volume.AccessibleTopology))
+	for _, topo := range cresp.Volume.AccessibleTopology {
+		resp.Topologies = append(resp.Topologies,
+			&nstructs.CSITopology{Segments: topo.Segments})
+	}
+
 	return nil
 }
 

client/structs/csi.go

@@ -211,17 +211,16 @@ type ClientCSIControllerDetachVolumeResponse struct{}
 // Nomad client to tell a CSI controller plugin on that client to perform
 // CreateVolume
 type ClientCSIControllerCreateVolumeRequest struct {
-	Name               string
-	VolumeCapabilities []*structs.CSIVolumeCapability
-	MountOptions       *structs.CSIMountOptions
-	Parameters         map[string]string
-	Secrets            structs.CSISecrets
-	CapacityMin        int64
-	CapacityMax        int64
-	SnapshotID         string
-	CloneID            string
-	// TODO: topology is not yet supported
-	// TopologyRequirement
+	Name                string
+	VolumeCapabilities  []*structs.CSIVolumeCapability
+	MountOptions        *structs.CSIMountOptions
+	Parameters          map[string]string
+	Secrets             structs.CSISecrets
+	CapacityMin         int64
+	CapacityMax         int64
+	SnapshotID          string
+	CloneID             string
+	RequestedTopologies *structs.CSITopologyRequest
 
 	CSIControllerQuery
 }
@@ -237,8 +236,10 @@ func (req *ClientCSIControllerCreateVolumeRequest) ToCSIRequest() (*csi.Controll
 			CloneID:    req.CloneID,
 			SnapshotID: req.SnapshotID,
 		},
-		// TODO: topology is not yet supported
-		AccessibilityRequirements: &csi.TopologyRequirement{},
+		AccessibilityRequirements: &csi.TopologyRequirement{
+			Requisite: []*csi.Topology{},
+			Preferred: []*csi.Topology{},
+		},
 	}
 
 	// The CSI spec requires that at least one of the fields in CapacityRange
@@ -258,16 +259,29 @@ func (req *ClientCSIControllerCreateVolumeRequest) ToCSIRequest() (*csi.Controll
 		}
 		creq.VolumeCapabilities = append(creq.VolumeCapabilities, ccap)
 	}
+
+	if req.RequestedTopologies != nil {
+		for _, topo := range req.RequestedTopologies.Required {
+			creq.AccessibilityRequirements.Requisite = append(
+				creq.AccessibilityRequirements.Requisite, &csi.Topology{
+					Segments: topo.Segments,
+				})
+		}
+		for _, topo := range req.RequestedTopologies.Preferred {
+			creq.AccessibilityRequirements.Preferred = append(
+				creq.AccessibilityRequirements.Preferred, &csi.Topology{
+					Segments: topo.Segments,
+				})
+		}
+	}
 	return creq, nil
 }
 
 type ClientCSIControllerCreateVolumeResponse struct {
 	ExternalVolumeID string
 	CapacityBytes    int64
 	VolumeContext    map[string]string
-
-	// TODO: topology is not yet supported
-	// AccessibleTopology []*Topology
+	Topologies       []*structs.CSITopology
 }
 
 // ClientCSIControllerDeleteVolumeRequest the RPC made from the server to a

command/agent/csi_endpoint.go

@@ -500,6 +500,13 @@ func structsCSIVolumeToApi(vol *structs.CSIVolume) *api.CSIVolume {
 		ModifyIndex:         vol.ModifyIndex,
 	}
 
+	if vol.RequestedTopologies != nil {
+		out.RequestedTopologies = &api.CSITopologyRequest{
+			Preferred: structsCSITopolgiesToApi(vol.RequestedTopologies.Preferred),
+			Required:  structsCSITopolgiesToApi(vol.RequestedTopologies.Required),
+		}
+	}
+
 	// WriteAllocs and ReadAllocs will only ever contain the Allocation ID,
 	// with a null value for the Allocation; these IDs are mapped to
 	// allocation stubs in the Allocations field. This indirection is so the
@@ -717,9 +724,11 @@ func structsTaskEventToApi(te *structs.TaskEvent) *api.TaskEvent {
 func structsCSITopolgiesToApi(tops []*structs.CSITopology) []*api.CSITopology {
 	out := make([]*api.CSITopology, 0, len(tops))
 	for _, t := range tops {
-		out = append(out, &api.CSITopology{
-			Segments: t.Segments,
-		})
+		if t != nil {
+			out = append(out, &api.CSITopology{
+				Segments: t.Segments,
+			})
+		}
 	}
 
 	return out

command/plugin_status_csi.go

@@ -118,6 +118,12 @@ func (c *PluginStatusCommand) csiFormatPlugin(plug *api.CSIPlugin) (string, erro
 			full = append(full, c.Colorize().Color("\n[bold]Node Capabilities[reset]"))
 			full = append(full, nodeCaps)
 		}
+		topos := c.formatTopology(plug.Nodes)
+		if topos != "" {
+			full = append(full, c.Colorize().Color("\n[bold]Accessible Topologies[reset]"))
+			full = append(full, topos)
+		}
+
 	}
 
 	// Format the allocs
@@ -183,6 +189,9 @@ func (c *PluginStatusCommand) formatControllerCaps(controllers map[string]*api.C
 func (c *PluginStatusCommand) formatNodeCaps(nodes map[string]*api.CSIInfo) string {
 	caps := []string{}
 	for _, node := range nodes {
+		if node.RequiresTopologies {
+			caps = append(caps, "VOLUME_ACCESSIBILITY_CONSTRAINTS")
+		}
 		switch info := node.NodeInfo; {
 		case info.RequiresNodeStageVolume:
 			caps = append(caps, "STAGE_UNSTAGE_VOLUME")
@@ -207,3 +216,21 @@ func (c *PluginStatusCommand) formatNodeCaps(nodes map[string]*api.CSIInfo) stri
 
 	return "  " + strings.Join(sort.StringSlice(caps), "\n  ")
 }
+
+func (c *PluginStatusCommand) formatTopology(nodes map[string]*api.CSIInfo) string {
+	rows := []string{"Node ID|Accessible Topology"}
+	for nodeID, node := range nodes {
+		if node.NodeInfo.AccessibleTopology != nil {
+			segments := node.NodeInfo.AccessibleTopology.Segments
+			segmentPairs := make([]string, 0, len(segments))
+			for k, v := range segments {
+				segmentPairs = append(segmentPairs, fmt.Sprintf("%s=%s", k, v))
+			}
+			rows = append(rows, fmt.Sprintf("%s|%s", nodeID[:8], strings.Join(segmentPairs, ",")))
+		}
+	}
+	if len(rows) == 1 {
+		return ""
+	}
+	return formatList(rows)
+}

command/volume_init.go

@@ -133,7 +133,7 @@ capacity_max = "20G"
 # capabilities to validate. Registering an existing volume will record but
 # ignore these fields.
 capability {
-  access_mode = "single-node-writer"
+  access_mode     = "single-node-writer"
   attachment_mode = "file-system"
 }
 
@@ -150,6 +150,18 @@ mount_options {
   mount_flags = ["ro"]
 }
 
+# Optional: specify one or more locations where the volume must be accessible
+# from. Refer to the plugin documentation for what segment values are supported.
+topology_request {
+  preferred {
+    topology { segments { rack = "R1" } }
+  }
+  required {
+    topology { segments { rack = "R1" } }
+    topology { segments { rack = "R2", zone = "us-east-1a" } }
+  }
+}
+
 # Optional: provide any secrets specified by the plugin.
 secrets {
   example_secret = "xyzzy"
@@ -201,6 +213,34 @@ var defaultJsonVolumeSpec = strings.TrimSpace(`
       ]
     }
   ],
+  "topology_request": {
+    "preferred": [
+      {
+        "topology": {
+          "segments": {
+            "rack": "R1"
+          }
+        }
+      }
+    ],
+    "required": [
+      {
+        "topology": {
+          "segments": {
+            "rack": "R1"
+          }
+        }
+      },
+      {
+        "topology": {
+          "segments": {
+            "rack": "R2",
+            "zone": "us-east-1a"
+          }
+        }
+      }
+    ]
+  },
   "parameters": [
     {
       "skuname": "Premium_LRS"

command/volume_register_csi.go

@@ -48,6 +48,7 @@ func csiDecodeVolume(input *ast.File) (*api.CSIVolume, error) {
 	delete(m, "mount_options")
 	delete(m, "capacity_max")
 	delete(m, "capacity_min")
+	delete(m, "topology_request")
 	delete(m, "type")
 
 	// Decode the rest
@@ -116,6 +117,48 @@ func csiDecodeVolume(input *ast.File) (*api.CSIVolume, error) {
 		}
 	}
 
+	requestedTopos := list.Filter("topology_request")
+	if len(requestedTopos.Items) > 0 {
+
+		vol.RequestedTopologies = &api.CSITopologyRequest{}
+
+		for _, o := range requestedTopos.Elem().Items {
+			if err := helper.CheckHCLKeys(o.Val, []string{"preferred", "required"}); err != nil {
+				return nil, err
+			}
+			ot, ok := o.Val.(*ast.ObjectType)
+			if !ok {
+				break
+			}
+
+			// topology_request -> required|preferred -> []topology -> []segments (kv)
+			decoded := map[string][]map[string][]map[string][]map[string]string{}
+			if err := hcl.DecodeObject(&decoded, ot.List); err != nil {
+				return nil, err
+			}
+
+			getTopologies := func(topKey string) []*api.CSITopology {
+				for _, topo := range decoded[topKey] {
+					var topos []*api.CSITopology
+					for _, segments := range topo["topology"] {
+						for _, segment := range segments["segments"] {
+							if len(segment) > 0 {
+								topos = append(topos, &api.CSITopology{Segments: segment})
+							}
+						}
+					}
+					if len(topos) > 0 {
+						return topos
+					}
+				}
+				return nil
+			}
+
+			vol.RequestedTopologies.Required = getTopologies("required")
+			vol.RequestedTopologies.Preferred = getTopologies("preferred")
+		}
+	}
+
 	return vol, nil
 }
 

command/volume_register_test.go

@@ -84,6 +84,17 @@ capability {
   access_mode     = "single-node-reader-only"
   attachment_mode = "block-device"
 }
+
+topology_request {
+  preferred {
+    topology { segments {rack = "R1"} }
+  }
+
+  required {
+    topology { segments {rack = "R1"} }
+    topology { segments {rack = "R2", zone = "us-east-1a"} }
+  }
+}
 `,
 		expected: &api.CSIVolume{
 			ID:                   "testvolume",
@@ -108,6 +119,16 @@ capability {
 			},
 			Parameters: map[string]string{"skuname": "Premium_LRS"},
 			Secrets:    map[string]string{"password": "xyzzy"},
+			RequestedTopologies: &api.CSITopologyRequest{
+				Required: []*api.CSITopology{
+					{Segments: map[string]string{"rack": "R1"}},
+					{Segments: map[string]string{"rack": "R2", "zone": "us-east-1a"}},
+				},
+				Preferred: []*api.CSITopology{
+					{Segments: map[string]string{"rack": "R1"}},
+				},
+			},
+			Topologies: nil, // this is left empty
 		},
 		err: "",
 	}, {
@@ -124,6 +145,19 @@ capability {
   access_mode     = "single-node-writer"
   attachment_mode = "file-system"
 }
+
+topology_request {
+  # make sure we safely handle empty blocks even
+  # if they're invalid
+  preferred {
+    topology {}
+    topology { segments {} }
+  }
+
+  required {
+    topology { segments { rack = "R2", zone = "us-east-1a"} }
+  }
+}
 `,
 		expected: &api.CSIVolume{
 			ID:         "testvolume",
@@ -136,6 +170,13 @@ capability {
 					AttachmentMode: api.CSIVolumeAttachmentModeFilesystem,
 				},
 			},
+			RequestedTopologies: &api.CSITopologyRequest{
+				Required: []*api.CSITopology{
+					{Segments: map[string]string{"rack": "R2", "zone": "us-east-1a"}},
+				},
+				Preferred: nil,
+			},
+			Topologies: nil,
 		},
 		err: "",
 	},