376 lines
13 KiB
Go
376 lines
13 KiB
Go
package csimanager
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
|
|
grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
|
|
"github.com/hashicorp/go-hclog"
|
|
"github.com/hashicorp/go-multierror"
|
|
"github.com/hashicorp/nomad/helper/mount"
|
|
"github.com/hashicorp/nomad/nomad/structs"
|
|
"github.com/hashicorp/nomad/plugins/csi"
|
|
)
|
|
|
|
var _ VolumeMounter = &volumeManager{}
|
|
|
|
const (
|
|
DefaultMountActionTimeout = 2 * time.Minute
|
|
StagingDirName = "staging"
|
|
AllocSpecificDirName = "per-alloc"
|
|
)
|
|
|
|
// volumeManager handles the state of attached volumes for a given CSI Plugin.
|
|
//
|
|
// volumeManagers outlive the lifetime of a given allocation as volumes may be
|
|
// shared by multiple allocations on the same node.
|
|
//
|
|
// volumes are stored by an enriched volume usage struct as the CSI Spec requires
|
|
// slightly different usage based on the given usage model.
|
|
type volumeManager struct {
|
|
logger hclog.Logger
|
|
eventer TriggerNodeEvent
|
|
plugin csi.CSIPlugin
|
|
|
|
usageTracker *volumeUsageTracker
|
|
|
|
// mountRoot is the root of where plugin directories and mounts may be created
|
|
// e.g /opt/nomad.d/statedir/csi/my-csi-plugin/
|
|
mountRoot string
|
|
|
|
// containerMountPoint is the location _inside_ the plugin container that the
|
|
// `mountRoot` is bound in to.
|
|
containerMountPoint string
|
|
|
|
// requiresStaging shows whether the plugin requires that the volume manager
|
|
// calls NodeStageVolume and NodeUnstageVolume RPCs during setup and teardown
|
|
requiresStaging bool
|
|
}
|
|
|
|
func newVolumeManager(logger hclog.Logger, eventer TriggerNodeEvent, plugin csi.CSIPlugin, rootDir, containerRootDir string, requiresStaging bool) *volumeManager {
|
|
return &volumeManager{
|
|
logger: logger.Named("volume_manager"),
|
|
eventer: eventer,
|
|
plugin: plugin,
|
|
mountRoot: rootDir,
|
|
containerMountPoint: containerRootDir,
|
|
requiresStaging: requiresStaging,
|
|
usageTracker: newVolumeUsageTracker(),
|
|
}
|
|
}
|
|
|
|
func (v *volumeManager) stagingDirForVolume(root string, volID string, usage *UsageOptions) string {
|
|
return filepath.Join(root, StagingDirName, volID, usage.ToFS())
|
|
}
|
|
|
|
func (v *volumeManager) allocDirForVolume(root string, volID, allocID string) string {
|
|
return filepath.Join(root, AllocSpecificDirName, allocID, volID)
|
|
}
|
|
|
|
func (v *volumeManager) targetForVolume(root string, volID, allocID string, usage *UsageOptions) string {
|
|
return filepath.Join(root, AllocSpecificDirName, allocID, volID, usage.ToFS())
|
|
}
|
|
|
|
// ensureStagingDir attempts to create a directory for use when staging a volume
|
|
// and then validates that the path is not already a mount point for e.g an
|
|
// existing volume stage.
|
|
//
|
|
// Returns whether the directory is a pre-existing mountpoint, the staging path,
|
|
// and any errors that occurred.
|
|
func (v *volumeManager) ensureStagingDir(vol *structs.CSIVolume, usage *UsageOptions) (string, bool, error) {
|
|
stagingPath := v.stagingDirForVolume(v.mountRoot, vol.ID, usage)
|
|
|
|
// Make the staging path, owned by the Nomad User
|
|
if err := os.MkdirAll(stagingPath, 0700); err != nil && !os.IsExist(err) {
|
|
return "", false, fmt.Errorf("failed to create staging directory for volume (%s): %v", vol.ID, err)
|
|
|
|
}
|
|
|
|
// Validate that it is not already a mount point
|
|
m := mount.New()
|
|
isNotMount, err := m.IsNotAMountPoint(stagingPath)
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("mount point detection failed for volume (%s): %v", vol.ID, err)
|
|
}
|
|
|
|
return stagingPath, !isNotMount, nil
|
|
}
|
|
|
|
// ensureAllocDir attempts to create a directory for use when publishing a volume
|
|
// and then validates that the path is not already a mount point (e.g when reattaching
|
|
// to existing allocs).
|
|
//
|
|
// Returns whether the directory is a pre-existing mountpoint, the publish path,
|
|
// and any errors that occurred.
|
|
func (v *volumeManager) ensureAllocDir(vol *structs.CSIVolume, alloc *structs.Allocation, usage *UsageOptions) (string, bool, error) {
|
|
allocPath := v.allocDirForVolume(v.mountRoot, vol.ID, alloc.ID)
|
|
|
|
// Make the alloc path, owned by the Nomad User
|
|
if err := os.MkdirAll(allocPath, 0700); err != nil && !os.IsExist(err) {
|
|
return "", false, fmt.Errorf("failed to create allocation directory for volume (%s): %v", vol.ID, err)
|
|
}
|
|
|
|
// Validate that the target is not already a mount point
|
|
targetPath := v.targetForVolume(v.mountRoot, vol.ID, alloc.ID, usage)
|
|
|
|
m := mount.New()
|
|
isNotMount, err := m.IsNotAMountPoint(targetPath)
|
|
|
|
switch {
|
|
case errors.Is(err, os.ErrNotExist):
|
|
// ignore; path does not exist and as such is not a mount
|
|
case err != nil:
|
|
return "", false, fmt.Errorf("mount point detection failed for volume (%s): %v", vol.ID, err)
|
|
}
|
|
|
|
return targetPath, !isNotMount, nil
|
|
}
|
|
|
|
func volumeCapability(vol *structs.CSIVolume, usage *UsageOptions) (*csi.VolumeCapability, error) {
|
|
var opts *structs.CSIMountOptions
|
|
if vol.MountOptions == nil {
|
|
opts = usage.MountOptions
|
|
} else {
|
|
opts = vol.MountOptions.Copy()
|
|
opts.Merge(usage.MountOptions)
|
|
}
|
|
|
|
capability, err := csi.VolumeCapabilityFromStructs(usage.AttachmentMode, usage.AccessMode, opts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return capability, nil
|
|
}
|
|
|
|
// stageVolume prepares a volume for use by allocations. When a plugin exposes
|
|
// the STAGE_UNSTAGE_VOLUME capability it MUST be called once-per-volume for a
|
|
// given usage mode before the volume can be NodePublish-ed.
|
|
func (v *volumeManager) stageVolume(ctx context.Context, vol *structs.CSIVolume, usage *UsageOptions, publishContext map[string]string) error {
|
|
logger := hclog.FromContext(ctx)
|
|
logger.Trace("Preparing volume staging environment")
|
|
hostStagingPath, isMount, err := v.ensureStagingDir(vol, usage)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
pluginStagingPath := v.stagingDirForVolume(v.containerMountPoint, vol.ID, usage)
|
|
|
|
logger.Trace("Volume staging environment", "pre-existing_mount", isMount, "host_staging_path", hostStagingPath, "plugin_staging_path", pluginStagingPath)
|
|
|
|
if isMount {
|
|
logger.Debug("re-using existing staging mount for volume", "staging_path", hostStagingPath)
|
|
return nil
|
|
}
|
|
|
|
capability, err := volumeCapability(vol, usage)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
req := &csi.NodeStageVolumeRequest{
|
|
ExternalID: vol.RemoteID(),
|
|
PublishContext: publishContext,
|
|
StagingTargetPath: pluginStagingPath,
|
|
VolumeCapability: capability,
|
|
Secrets: vol.Secrets,
|
|
VolumeContext: vol.Context,
|
|
}
|
|
|
|
// CSI NodeStageVolume errors for timeout, codes.Unavailable and
|
|
// codes.ResourceExhausted are retried; all other errors are fatal.
|
|
return v.plugin.NodeStageVolume(ctx, req,
|
|
grpc_retry.WithPerRetryTimeout(DefaultMountActionTimeout),
|
|
grpc_retry.WithMax(3),
|
|
grpc_retry.WithBackoff(grpc_retry.BackoffExponential(100*time.Millisecond)),
|
|
)
|
|
}
|
|
|
|
func (v *volumeManager) publishVolume(ctx context.Context, vol *structs.CSIVolume, alloc *structs.Allocation, usage *UsageOptions, publishContext map[string]string) (*MountInfo, error) {
|
|
logger := hclog.FromContext(ctx)
|
|
var pluginStagingPath string
|
|
if v.requiresStaging {
|
|
pluginStagingPath = v.stagingDirForVolume(v.containerMountPoint, vol.ID, usage)
|
|
}
|
|
|
|
hostTargetPath, isMount, err := v.ensureAllocDir(vol, alloc, usage)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
pluginTargetPath := v.targetForVolume(v.containerMountPoint, vol.ID, alloc.ID, usage)
|
|
|
|
if isMount {
|
|
logger.Debug("Re-using existing published volume for allocation")
|
|
return &MountInfo{Source: hostTargetPath}, nil
|
|
}
|
|
|
|
capabilities, err := volumeCapability(vol, usage)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// CSI NodePublishVolume errors for timeout, codes.Unavailable and
|
|
// codes.ResourceExhausted are retried; all other errors are fatal.
|
|
err = v.plugin.NodePublishVolume(ctx, &csi.NodePublishVolumeRequest{
|
|
ExternalID: vol.RemoteID(),
|
|
PublishContext: publishContext,
|
|
StagingTargetPath: pluginStagingPath,
|
|
TargetPath: pluginTargetPath,
|
|
VolumeCapability: capabilities,
|
|
Readonly: usage.ReadOnly,
|
|
Secrets: vol.Secrets,
|
|
VolumeContext: vol.Context,
|
|
},
|
|
grpc_retry.WithPerRetryTimeout(DefaultMountActionTimeout),
|
|
grpc_retry.WithMax(3),
|
|
grpc_retry.WithBackoff(grpc_retry.BackoffExponential(100*time.Millisecond)),
|
|
)
|
|
|
|
return &MountInfo{Source: hostTargetPath}, err
|
|
}
|
|
|
|
// MountVolume performs the steps required for using a given volume
|
|
// configuration for the provided allocation.
|
|
// It is passed the publishContext from remote attachment, and specific usage
|
|
// modes from the CSI Hook.
|
|
// It then uses this state to stage and publish the volume as required for use
|
|
// by the given allocation.
|
|
func (v *volumeManager) MountVolume(ctx context.Context, vol *structs.CSIVolume, alloc *structs.Allocation, usage *UsageOptions, publishContext map[string]string) (mountInfo *MountInfo, err error) {
|
|
logger := v.logger.With("volume_id", vol.ID, "alloc_id", alloc.ID)
|
|
ctx = hclog.WithContext(ctx, logger)
|
|
|
|
if v.requiresStaging {
|
|
err = v.stageVolume(ctx, vol, usage, publishContext)
|
|
}
|
|
|
|
if err == nil {
|
|
mountInfo, err = v.publishVolume(ctx, vol, alloc, usage, publishContext)
|
|
}
|
|
|
|
if err == nil {
|
|
v.usageTracker.Claim(alloc.ID, vol.ID, usage)
|
|
}
|
|
|
|
event := structs.NewNodeEvent().
|
|
SetSubsystem(structs.NodeEventSubsystemStorage).
|
|
SetMessage("Mount volume").
|
|
AddDetail("volume_id", vol.ID)
|
|
if err == nil {
|
|
event.AddDetail("success", "true")
|
|
} else {
|
|
event.AddDetail("success", "false")
|
|
event.AddDetail("error", err.Error())
|
|
}
|
|
|
|
v.eventer(event)
|
|
|
|
return mountInfo, err
|
|
}
|
|
|
|
// unstageVolume is the inverse operation of `stageVolume` and must be called
|
|
// once for each staging path that a volume has been staged under.
|
|
// It is safe to call multiple times and a plugin is required to return OK if
|
|
// the volume has been unstaged or was never staged on the node.
|
|
func (v *volumeManager) unstageVolume(ctx context.Context, volID, remoteID string, usage *UsageOptions) error {
|
|
logger := hclog.FromContext(ctx)
|
|
logger.Trace("Unstaging volume")
|
|
stagingPath := v.stagingDirForVolume(v.containerMountPoint, volID, usage)
|
|
|
|
// CSI NodeUnstageVolume errors for timeout, codes.Unavailable and
|
|
// codes.ResourceExhausted are retried; all other errors are fatal.
|
|
return v.plugin.NodeUnstageVolume(ctx,
|
|
remoteID,
|
|
stagingPath,
|
|
grpc_retry.WithPerRetryTimeout(DefaultMountActionTimeout),
|
|
grpc_retry.WithMax(3),
|
|
grpc_retry.WithBackoff(grpc_retry.BackoffExponential(100*time.Millisecond)),
|
|
)
|
|
}
|
|
|
|
func combineErrors(maybeErrs ...error) error {
|
|
var result *multierror.Error
|
|
for _, err := range maybeErrs {
|
|
if err == nil {
|
|
continue
|
|
}
|
|
|
|
result = multierror.Append(result, err)
|
|
}
|
|
|
|
return result.ErrorOrNil()
|
|
}
|
|
|
|
func (v *volumeManager) unpublishVolume(ctx context.Context, volID, remoteID, allocID string, usage *UsageOptions) error {
|
|
pluginTargetPath := v.targetForVolume(v.containerMountPoint, volID, allocID, usage)
|
|
|
|
// CSI NodeUnpublishVolume errors for timeout, codes.Unavailable and
|
|
// codes.ResourceExhausted are retried; all other errors are fatal.
|
|
rpcErr := v.plugin.NodeUnpublishVolume(ctx, remoteID, pluginTargetPath,
|
|
grpc_retry.WithPerRetryTimeout(DefaultMountActionTimeout),
|
|
grpc_retry.WithMax(3),
|
|
grpc_retry.WithBackoff(grpc_retry.BackoffExponential(100*time.Millisecond)),
|
|
)
|
|
|
|
hostTargetPath := v.targetForVolume(v.mountRoot, volID, allocID, usage)
|
|
if _, err := os.Stat(hostTargetPath); os.IsNotExist(err) {
|
|
if rpcErr != nil && strings.Contains(rpcErr.Error(), "no mount point") {
|
|
// host target path was already destroyed, nothing to do here.
|
|
// this helps us in the case that a previous GC attempt cleaned
|
|
// up the volume on the node but the controller RPCs failed
|
|
rpcErr = fmt.Errorf("%w: %v", structs.ErrCSIClientRPCIgnorable, rpcErr)
|
|
}
|
|
return rpcErr
|
|
}
|
|
|
|
// Host Target Path was not cleaned up, attempt to do so here. If it's still
|
|
// a mount then removing the dir will fail and we'll return any rpcErr and the
|
|
// file error.
|
|
rmErr := os.Remove(hostTargetPath)
|
|
if rmErr != nil {
|
|
return combineErrors(rpcErr, rmErr)
|
|
}
|
|
|
|
// We successfully removed the directory, return any rpcErrors that were
|
|
// encountered, but because we got here, they were probably flaky or was
|
|
// cleaned up externally.
|
|
return fmt.Errorf("%w: %v", structs.ErrCSIClientRPCIgnorable, rpcErr)
|
|
}
|
|
|
|
func (v *volumeManager) UnmountVolume(ctx context.Context, volID, remoteID, allocID string, usage *UsageOptions) (err error) {
|
|
logger := v.logger.With("volume_id", volID, "alloc_id", allocID)
|
|
ctx = hclog.WithContext(ctx, logger)
|
|
|
|
err = v.unpublishVolume(ctx, volID, remoteID, allocID, usage)
|
|
|
|
if err == nil || errors.Is(err, structs.ErrCSIClientRPCIgnorable) {
|
|
canRelease := v.usageTracker.Free(allocID, volID, usage)
|
|
if v.requiresStaging && canRelease {
|
|
err = v.unstageVolume(ctx, volID, remoteID, usage)
|
|
}
|
|
}
|
|
|
|
if errors.Is(err, structs.ErrCSIClientRPCIgnorable) {
|
|
logger.Trace("unmounting volume failed with ignorable error", "error", err)
|
|
err = nil
|
|
}
|
|
|
|
event := structs.NewNodeEvent().
|
|
SetSubsystem(structs.NodeEventSubsystemStorage).
|
|
SetMessage("Unmount volume").
|
|
AddDetail("volume_id", volID)
|
|
if err == nil {
|
|
event.AddDetail("success", "true")
|
|
} else {
|
|
event.AddDetail("success", "false")
|
|
event.AddDetail("error", err.Error())
|
|
}
|
|
|
|
v.eventer(event)
|
|
|
|
return err
|
|
}
|