Add cluster volume support

- Write test for cluster volumes
- Add inspect test, add update command
- Add cluster volume opts to create
- Add requisite and preferred topology flags
- volume: move cluster bool in opts

Signed-off-by: Drew Erny <[email protected]>
Signed-off-by: Sebastiaan van Stijn <[email protected]>

cli/command/formatter/volume.go

@@ -1,6 +1,7 @@
 package formatter
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
@@ -12,10 +13,13 @@ const (
 	defaultVolumeQuietFormat = "{{.Name}}"
 	defaultVolumeTableFormat = "table {{.Driver}}\t{{.Name}}"
 
-	volumeNameHeader = "VOLUME NAME"
-	mountpointHeader = "MOUNTPOINT"
-	linksHeader      = "LINKS"
-	// Status header ?
+	idHeader           = "ID"
+	volumeNameHeader   = "VOLUME NAME"
+	mountpointHeader   = "MOUNTPOINT"
+	linksHeader        = "LINKS"
+	groupHeader        = "GROUP"
+	availabilityHeader = "AVAILABILITY"
+	statusHeader       = "STATUS"
 )
 
 // NewVolumeFormat returns a format for use with a volume Context
@@ -56,13 +60,17 @@ type volumeContext struct {
 func newVolumeContext() *volumeContext {
 	volumeCtx := volumeContext{}
 	volumeCtx.Header = SubHeaderContext{
-		"Name":       volumeNameHeader,
-		"Driver":     DriverHeader,
-		"Scope":      ScopeHeader,
-		"Mountpoint": mountpointHeader,
-		"Labels":     LabelsHeader,
-		"Links":      linksHeader,
-		"Size":       SizeHeader,
+		"ID":           idHeader,
+		"Name":         volumeNameHeader,
+		"Group":        groupHeader,
+		"Driver":       DriverHeader,
+		"Scope":        ScopeHeader,
+		"Availability": availabilityHeader,
+		"Mountpoint":   mountpointHeader,
+		"Labels":       LabelsHeader,
+		"Links":        linksHeader,
+		"Size":         SizeHeader,
+		"Status":       statusHeader,
 	}
 	return &volumeCtx
 }
@@ -119,3 +127,39 @@ func (c *volumeContext) Size() string {
 	}
 	return units.HumanSize(float64(c.v.UsageData.Size))
 }
+
+func (c *volumeContext) Group() string {
+	if c.v.ClusterVolume == nil {
+		return "N/A"
+	}
+
+	return c.v.ClusterVolume.Spec.Group
+}
+
+func (c *volumeContext) Availability() string {
+	if c.v.ClusterVolume == nil {
+		return "N/A"
+	}
+
+	return string(c.v.ClusterVolume.Spec.Availability)
+}
+
+func (c *volumeContext) Status() string {
+	if c.v.ClusterVolume == nil {
+		return "N/A"
+	}
+
+	if c.v.ClusterVolume.Info == nil || c.v.ClusterVolume.Info.VolumeID == "" {
+		return "pending creation"
+	}
+
+	l := len(c.v.ClusterVolume.PublishStatus)
+	switch l {
+	case 0:
+		return "created"
+	case 1:
+		return "in use (1 node)"
+	default:
+		return fmt.Sprintf("in use (%d nodes)", l)
+	}
+}

cli/command/formatter/volume_test.go

@@ -148,8 +148,8 @@ func TestVolumeContextWriteJSON(t *testing.T) {
 		{Driver: "bar", Name: "foobar_bar"},
 	}
 	expectedJSONs := []map[string]interface{}{
-		{"Driver": "foo", "Labels": "", "Links": "N/A", "Mountpoint": "", "Name": "foobar_baz", "Scope": "", "Size": "N/A"},
-		{"Driver": "bar", "Labels": "", "Links": "N/A", "Mountpoint": "", "Name": "foobar_bar", "Scope": "", "Size": "N/A"},
+		{"Availability": "N/A", "Driver": "foo", "Group": "N/A", "Labels": "", "Links": "N/A", "Mountpoint": "", "Name": "foobar_baz", "Scope": "", "Size": "N/A", "Status": "N/A"},
+		{"Availability": "N/A", "Driver": "bar", "Group": "N/A", "Labels": "", "Links": "N/A", "Mountpoint": "", "Name": "foobar_bar", "Scope": "", "Size": "N/A", "Status": "N/A"},
 	}
 	out := bytes.NewBufferString("")
 	err := VolumeWrite(Context{Format: "{{json .}}", Output: out}, volumes)

cli/command/volume/cmd.go

@@ -21,6 +21,7 @@ func NewVolumeCommand(dockerCli command.Cli) *cobra.Command {
 		newListCommand(dockerCli),
 		newRemoveCommand(dockerCli),
 		NewPruneCommand(dockerCli),
+		newUpdateCommand(dockerCli),
 	)
 	return cmd
 }

cli/command/volume/create.go

@@ -3,6 +3,7 @@ package volume
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/docker/cli/cli"
 	"github.com/docker/cli/cli/command"
@@ -11,19 +12,36 @@ import (
 	"github.com/docker/docker/api/types/volume"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 )
 
 type createOptions struct {
 	name       string
 	driver     string
 	driverOpts opts.MapOpts
 	labels     opts.ListOpts
+
+	// options for cluster volumes only
+	cluster           bool
+	group             string
+	scope             string
+	sharing           string
+	availability      string
+	secrets           opts.MapOpts
+	requiredBytes     opts.MemBytes
+	limitBytes        opts.MemBytes
+	accessType        string
+	requisiteTopology opts.ListOpts
+	preferredTopology opts.ListOpts
 }
 
 func newCreateCommand(dockerCli command.Cli) *cobra.Command {
 	options := createOptions{
-		driverOpts: *opts.NewMapOpts(nil, nil),
-		labels:     opts.NewListOpts(opts.ValidateLabel),
+		driverOpts:        *opts.NewMapOpts(nil, nil),
+		labels:            opts.NewListOpts(opts.ValidateLabel),
+		secrets:           *opts.NewMapOpts(nil, nil),
+		requisiteTopology: opts.NewListOpts(nil),
+		preferredTopology: opts.NewListOpts(nil),
 	}
 
 	cmd := &cobra.Command{
@@ -37,6 +55,7 @@ func newCreateCommand(dockerCli command.Cli) *cobra.Command {
 				}
 				options.name = args[0]
 			}
+			options.cluster = hasClusterVolumeOptionSet(cmd.Flags())
 			return runCreate(dockerCli, options)
 		},
 		ValidArgsFunction: completion.NoComplete,
@@ -48,16 +67,110 @@ func newCreateCommand(dockerCli command.Cli) *cobra.Command {
 	flags.VarP(&options.driverOpts, "opt", "o", "Set driver specific options")
 	flags.Var(&options.labels, "label", "Set metadata for a volume")
 
+	// flags for cluster volumes only
+	flags.StringVarP(&options.group, "group", "g", "", "Cluster Volume group (cluster volumes)")
+	flags.StringVar(&options.scope, "scope", "single", `Cluster Volume access scope ("single"|"multi")`)
+	flags.StringVar(&options.sharing, "sharing", "none", `Cluster Volume access sharing ("none"|"readonly"|"onewriter"|"all")`)
+	flags.StringVar(&options.availability, "availability", "active", `Cluster Volume availability ("active"|"pause"|"drain")`)
+	flags.StringVar(&options.accessType, "type", "block", `Cluster Volume access type ("mount"|"block")`)
+	flags.Var(&options.secrets, "secret", "Cluster Volume secrets")
+	flags.Var(&options.limitBytes, "limit-bytes", "Minimum size of the Cluster Volume in bytes (default 0 for undefined)")
+	flags.Var(&options.requiredBytes, "required-bytes", "Maximum size of the Cluster Volume in bytes (default 0 for undefined)")
+	flags.Var(&options.requisiteTopology, "topology-required", "A topology that the Cluster Volume must be accessible from")
+	flags.Var(&options.preferredTopology, "topology-preferred", "A topology that the Cluster Volume would be preferred in")
+
 	return cmd
 }
 
+// hasClusterVolumeOptionSet returns true if any of the cluster-specific
+// options are set.
+func hasClusterVolumeOptionSet(flags *pflag.FlagSet) bool {
+	return flags.Changed("group") || flags.Changed("scope") ||
+		flags.Changed("sharing") || flags.Changed("availability") ||
+		flags.Changed("type") || flags.Changed("secrets") ||
+		flags.Changed("limit-bytes") || flags.Changed("required-bytes")
+}
+
 func runCreate(dockerCli command.Cli, options createOptions) error {
-	vol, err := dockerCli.Client().VolumeCreate(context.Background(), volume.CreateOptions{
+	volOpts := volume.CreateOptions{
 		Driver:     options.driver,
 		DriverOpts: options.driverOpts.GetAll(),
 		Name:       options.name,
 		Labels:     opts.ConvertKVStringsToMap(options.labels.GetAll()),
-	})
+	}
+	if options.cluster {
+		volOpts.ClusterVolumeSpec = &volume.ClusterVolumeSpec{
+			Group: options.group,
+			AccessMode: &volume.AccessMode{
+				Scope:   volume.Scope(options.scope),
+				Sharing: volume.SharingMode(options.sharing),
+			},
+			Availability: volume.Availability(options.availability),
+		}
+
+		if options.accessType == "mount" {
+			volOpts.ClusterVolumeSpec.AccessMode.MountVolume = &volume.TypeMount{}
+		} else if options.accessType == "block" {
+			volOpts.ClusterVolumeSpec.AccessMode.BlockVolume = &volume.TypeBlock{}
+		}
+
+		vcr := &volume.CapacityRange{}
+		if r := options.requiredBytes.Value(); r >= 0 {
+			vcr.RequiredBytes = r
+		}
+
+		if l := options.limitBytes.Value(); l >= 0 {
+			vcr.LimitBytes = l
+		}
+		volOpts.ClusterVolumeSpec.CapacityRange = vcr
+
+		for key, secret := range options.secrets.GetAll() {
+			volOpts.ClusterVolumeSpec.Secrets = append(
+				volOpts.ClusterVolumeSpec.Secrets,
+				volume.Secret{
+					Key:    key,
+					Secret: secret,
+				},
+			)
+		}
+
+		// TODO(dperny): ignore if no topology specified
+		topology := &volume.TopologyRequirement{}
+		for _, top := range options.requisiteTopology.GetAll() {
+			// each topology takes the form segment=value,segment=value
+			// comma-separated list of equal separated maps
+			segments := map[string]string{}
+			for _, segment := range strings.Split(top, ",") {
+				parts := strings.SplitN(segment, "=", 2)
+				// TODO(dperny): validate topology syntax
+				segments[parts[0]] = parts[1]
+			}
+			topology.Requisite = append(
+				topology.Requisite,
+				volume.Topology{Segments: segments},
+			)
+		}
+
+		for _, top := range options.preferredTopology.GetAll() {
+			// each topology takes the form segment=value,segment=value
+			// comma-separated list of equal separated maps
+			segments := map[string]string{}
+			for _, segment := range strings.Split(top, ",") {
+				parts := strings.SplitN(segment, "=", 2)
+				// TODO(dperny): validate topology syntax
+				segments[parts[0]] = parts[1]
+			}
+
+			topology.Preferred = append(
+				topology.Preferred,
+				volume.Topology{Segments: segments},
+			)
+		}
+
+		volOpts.ClusterVolumeSpec.AccessibilityRequirements = topology
+	}
+
+	vol, err := dockerCli.Client().VolumeCreate(context.Background(), volOpts)
 	if err != nil {
 		return err
 	}

cli/command/volume/create_test.go

@@ -123,3 +123,100 @@ func TestVolumeCreateWithFlags(t *testing.T) {
 	assert.NilError(t, cmd.Execute())
 	assert.Check(t, is.Equal(name, strings.TrimSpace(cli.OutBuffer().String())))
 }
+
+func TestVolumeCreateCluster(t *testing.T) {
+	cli := test.NewFakeCli(&fakeClient{
+		volumeCreateFunc: func(body volume.CreateOptions) (volume.Volume, error) {
+			if body.Driver == "csi" && body.ClusterVolumeSpec == nil {
+				return volume.Volume{}, errors.New("expected ClusterVolumeSpec, but none present")
+			}
+			if body.Driver == "notcsi" && body.ClusterVolumeSpec != nil {
+				return volume.Volume{}, errors.New("expected no ClusterVolumeSpec, but present")
+			}
+			return volume.Volume{}, nil
+		},
+	})
+
+	cmd := newCreateCommand(cli)
+	cmd.Flags().Set("type", "block")
+	cmd.Flags().Set("group", "gronp")
+	cmd.Flags().Set("driver", "csi")
+	cmd.SetArgs([]string{"name"})
+
+	assert.NilError(t, cmd.Execute())
+
+	cmd = newCreateCommand(cli)
+	cmd.Flags().Set("driver", "notcsi")
+	cmd.SetArgs([]string{"name"})
+
+	assert.NilError(t, cmd.Execute())
+}
+
+func TestVolumeCreateClusterOpts(t *testing.T) {
+	expectedBody := volume.CreateOptions{
+		Name:       "name",
+		Driver:     "csi",
+		DriverOpts: map[string]string{},
+		Labels:     map[string]string{},
+		ClusterVolumeSpec: &volume.ClusterVolumeSpec{
+			Group: "gronp",
+			AccessMode: &volume.AccessMode{
+				Scope:   volume.ScopeMultiNode,
+				Sharing: volume.SharingOneWriter,
+				// TODO(dperny): support mount options
+				MountVolume: &volume.TypeMount{},
+			},
+			// TODO(dperny): topology requirements
+			CapacityRange: &volume.CapacityRange{
+				RequiredBytes: 1234,
+				LimitBytes:    567890,
+			},
+			Secrets: []volume.Secret{
+				{Key: "key1", Secret: "secret1"},
+				{Key: "key2", Secret: "secret2"},
+			},
+			Availability: volume.AvailabilityActive,
+			AccessibilityRequirements: &volume.TopologyRequirement{
+				Requisite: []volume.Topology{
+					{Segments: map[string]string{"region": "R1", "zone": "Z1"}},
+					{Segments: map[string]string{"region": "R1", "zone": "Z2"}},
+					{Segments: map[string]string{"region": "R1", "zone": "Z3"}},
+				},
+				Preferred: []volume.Topology{
+					{Segments: map[string]string{"region": "R1", "zone": "Z2"}},
+					{Segments: map[string]string{"region": "R1", "zone": "Z3"}},
+				},
+			},
+		},
+	}
+
+	cli := test.NewFakeCli(&fakeClient{
+		volumeCreateFunc: func(body volume.CreateOptions) (volume.Volume, error) {
+			assert.DeepEqual(t, body, expectedBody)
+			return volume.Volume{}, nil
+		},
+	})
+
+	cmd := newCreateCommand(cli)
+	cmd.SetArgs([]string{"name"})
+	cmd.Flags().Set("driver", "csi")
+	cmd.Flags().Set("group", "gronp")
+	cmd.Flags().Set("scope", "multi")
+	cmd.Flags().Set("sharing", "onewriter")
+	cmd.Flags().Set("type", "mount")
+	cmd.Flags().Set("sharing", "onewriter")
+	cmd.Flags().Set("required-bytes", "1234")
+	cmd.Flags().Set("limit-bytes", "567890")
+
+	cmd.Flags().Set("secret", "key1=secret1")
+	cmd.Flags().Set("secret", "key2=secret2")
+
+	cmd.Flags().Set("topology-required", "region=R1,zone=Z1")
+	cmd.Flags().Set("topology-required", "region=R1,zone=Z2")
+	cmd.Flags().Set("topology-required", "region=R1,zone=Z3")
+
+	cmd.Flags().Set("topology-preferred", "region=R1,zone=Z2")
+	cmd.Flags().Set("topology-preferred", "region=R1,zone=Z3")
+
+	cmd.Execute()
+}