moby/daemon/cluster/volumes.go
Drew Erny 240a9fcb83
Add Swarm cluster volume supports
Adds code to support Cluster Volumes in Swarm using CSI drivers.

Signed-off-by: Drew Erny <derny@mirantis.com>
2022-05-13 00:55:44 +02:00

140 lines
4.1 KiB
Go

package cluster // import "github.com/docker/docker/daemon/cluster"
import (
"context"
"fmt"
volumetypes "github.com/docker/docker/api/types/volume"
"github.com/docker/docker/daemon/cluster/convert"
"github.com/docker/docker/errdefs"
swarmapi "github.com/moby/swarmkit/v2/api"
"google.golang.org/grpc"
)
// GetVolume returns a volume from the swarm cluster.
func (c *Cluster) GetVolume(nameOrID string) (volumetypes.Volume, error) {
var volume *swarmapi.Volume
if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
v, err := getVolume(ctx, state.controlClient, nameOrID)
if err != nil {
return err
}
volume = v
return nil
}); err != nil {
return volumetypes.Volume{}, err
}
return convert.VolumeFromGRPC(volume), nil
}
// GetVolumes returns all of the volumes matching the given options from a swarm cluster.
func (c *Cluster) GetVolumes(options volumetypes.ListOptions) ([]*volumetypes.Volume, error) {
var volumes []*volumetypes.Volume
if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
r, err := state.controlClient.ListVolumes(
ctx, &swarmapi.ListVolumesRequest{},
grpc.MaxCallRecvMsgSize(defaultRecvSizeForListResponse),
)
if err != nil {
return err
}
volumes = make([]*volumetypes.Volume, 0, len(r.Volumes))
for _, volume := range r.Volumes {
v := convert.VolumeFromGRPC(volume)
volumes = append(volumes, &v)
}
return nil
}); err != nil {
return nil, err
}
return volumes, nil
}
// CreateVolume creates a new cluster volume in the swarm cluster.
//
// Returns the volume ID if creation is successful, or an error if not.
func (c *Cluster) CreateVolume(v volumetypes.CreateOptions) (*volumetypes.Volume, error) {
var resp *swarmapi.CreateVolumeResponse
if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
volumeSpec := convert.VolumeCreateToGRPC(&v)
r, err := state.controlClient.CreateVolume(
ctx, &swarmapi.CreateVolumeRequest{Spec: volumeSpec},
)
if err != nil {
return err
}
resp = r
return nil
}); err != nil {
return nil, err
}
createdVol, err := c.GetVolume(resp.Volume.ID)
if err != nil {
// If there's a failure of some sort in this operation the user would
// get a very unhelpful "not found" error on a create, which is not
// very helpful at all. Instead, before returning the error, add some
// context, and change this to a system-type error, because it's
// nothing the user did wrong.
return nil, errdefs.System(fmt.Errorf("unable to retrieve created volume: %w", err))
}
return &createdVol, nil
}
// RemoveVolume removes a volume from the swarm cluster.
func (c *Cluster) RemoveVolume(nameOrID string, force bool) error {
return c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
volume, err := getVolume(ctx, state.controlClient, nameOrID)
if err != nil {
return err
}
req := &swarmapi.RemoveVolumeRequest{
VolumeID: volume.ID,
Force: force,
}
_, err = state.controlClient.RemoveVolume(ctx, req)
return err
})
}
// UpdateVolume updates a volume in the swarm cluster.
func (c *Cluster) UpdateVolume(nameOrID string, version uint64, volume volumetypes.UpdateOptions) error {
return c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
v, err := getVolume(ctx, state.controlClient, nameOrID)
if err != nil {
return err
}
// For now, the only thing we can update is availability. Instead of
// converting the whole spec, just pluck out the availability if it has
// been set.
if volume.Spec != nil {
switch volume.Spec.Availability {
case volumetypes.AvailabilityActive:
v.Spec.Availability = swarmapi.VolumeAvailabilityActive
case volumetypes.AvailabilityPause:
v.Spec.Availability = swarmapi.VolumeAvailabilityPause
case volumetypes.AvailabilityDrain:
v.Spec.Availability = swarmapi.VolumeAvailabilityDrain
}
// if default empty value, change nothing.
}
_, err = state.controlClient.UpdateVolume(
ctx, &swarmapi.UpdateVolumeRequest{
VolumeID: nameOrID,
VolumeVersion: &swarmapi.Version{
Index: version,
},
Spec: &v.Spec,
},
)
return err
})
}