csi: add -force flag to volume deregister

api/csi.go

@@ -1,6 +1,8 @@
 package api
 
 import (
+	"fmt"
+	"net/url"
 	"sort"
 	"time"
 )
@@ -53,8 +55,8 @@ func (v *CSIVolumes) Register(vol *CSIVolume, w *WriteOptions) (*WriteMeta, erro
 	return meta, err
 }
 
-func (v *CSIVolumes) Deregister(id string, w *WriteOptions) error {
-	_, err := v.client.delete("/v1/volume/csi/"+id, nil, w)
+func (v *CSIVolumes) Deregister(id string, force bool, w *WriteOptions) error {
+	_, err := v.client.delete(fmt.Sprintf("/v1/volume/csi/%v?purge=%t", url.PathEscape(id), force), nil, w)
 	return err
 }
 

api/csi_test.go

@@ -86,7 +86,7 @@ func TestCSIVolumes_CRUD(t *testing.T) {
 	require.Equal(t, "bar", vol.Topologies[0].Segments["foo"])
 
 	// Deregister the volume
-	err = v.Deregister(id, wpts)
+	err = v.Deregister(id, true, wpts)
 	require.NoError(t, err)
 
 	// Successful empty result

command/agent/csi_endpoint.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -122,8 +123,19 @@ func (s *HTTPServer) csiVolumeDelete(id string, resp http.ResponseWriter, req *h
 		return nil, CodedError(405, ErrInvalidMethod)
 	}
 
+	raw := req.URL.Query().Get("force")
+	var force bool
+	if raw != "" {
+		var err error
+		force, err = strconv.ParseBool(raw)
+		if err != nil {
+			return nil, CodedError(400, "invalid force value")
+		}
+	}
+
 	args := structs.CSIVolumeDeregisterRequest{
 		VolumeIDs: []string{id},
+		Force:     force,
 	}
 	s.parseWriteRequest(req, &args.WriteRequest)
 

command/volume_deregister.go

@@ -20,13 +20,23 @@ Usage: nomad volume deregister [options] <id>
 
 General Options:
 
-  ` + generalOptionsUsage()
+  ` + generalOptionsUsage() + `
 
+Volume Deregister Options:
+
+  -force
+    Force deregistration of the volume and immediately drop claims for
+    terminal allocations. Returns an error if the volume has running
+    allocations. This does not detach the volume from client nodes.
+`
 	return strings.TrimSpace(helpText)
 }
 
 func (c *VolumeDeregisterCommand) AutocompleteFlags() complete.Flags {
-	return c.Meta.AutocompleteFlags(FlagSetClient)
+	return mergeAutocompleteFlags(c.Meta.AutocompleteFlags(FlagSetClient),
+		complete.Flags{
+			"-force": complete.PredictNothing,
+		})
 }
 
 func (c *VolumeDeregisterCommand) AutocompleteArgs() complete.Predictor {
@@ -52,8 +62,10 @@ func (c *VolumeDeregisterCommand) Synopsis() string {
 func (c *VolumeDeregisterCommand) Name() string { return "volume deregister" }
 
 func (c *VolumeDeregisterCommand) Run(args []string) int {
+	var force bool
 	flags := c.Meta.FlagSet(c.Name(), FlagSetClient)
 	flags.Usage = func() { c.Ui.Output(c.Help()) }
+	flags.BoolVar(&force, "force", false, "Force deregister and drop claims")
 
 	if err := flags.Parse(args); err != nil {
 		c.Ui.Error(fmt.Sprintf("Error parsing arguments %s", err))
@@ -76,9 +88,32 @@ func (c *VolumeDeregisterCommand) Run(args []string) int {
 		return 1
 	}
 
+	// Confirm the -force flag
+	if force {
+		question := fmt.Sprintf("Are you sure you want to force deregister volume %q? [y/N]", volID)
+		answer, err := c.Ui.Ask(question)
+		if err != nil {
+			c.Ui.Error(fmt.Sprintf("Failed to parse answer: %v", err))
+			return 1
+		}
+
+		if answer == "" || strings.ToLower(answer)[0] == 'n' {
+			// No case
+			c.Ui.Output("Cancelling volume deregister")
+			return 0
+		} else if strings.ToLower(answer)[0] == 'y' && len(answer) > 1 {
+			// Non exact match yes
+			c.Ui.Output("For confirmation, an exact ‘y’ is required.")
+			return 0
+		} else if answer != "y" {
+			c.Ui.Output("No confirmation detected. For confirmation, an exact 'y' is required.")
+			return 1
+		}
+	}
+
 	// Deregister only works on CSI volumes, but could be extended to support other
 	// network interfaces or host volumes
-	err = client.CSIVolumes().Deregister(volID, nil)
+	err = client.CSIVolumes().Deregister(volID, force, nil)
 	if err != nil {
 		c.Ui.Error(fmt.Sprintf("Error deregistering volume: %s", err))
 		return 1

e2e/csi/csi.go

@@ -221,7 +221,7 @@ func (tc *CSIVolumesTest) AfterEach(f *framework.F) {
 	}
 	// Deregister all volumes in test
 	for _, id := range tc.volumeIDs {
-		nomadClient.CSIVolumes().Deregister(id, nil)
+		nomadClient.CSIVolumes().Deregister(id, true, nil)
 	}
 	// Deregister all plugin jobs in test
 	for _, id := range tc.pluginJobIDs {

nomad/fsm.go

@@ -1154,7 +1154,7 @@ func (n *nomadFSM) applyCSIVolumeDeregister(buf []byte, index uint64) interface{
 	}
 	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_csi_volume_deregister"}, time.Now())
 
-	if err := n.state.CSIVolumeDeregister(index, req.RequestNamespace(), req.VolumeIDs); err != nil {
+	if err := n.state.CSIVolumeDeregister(index, req.RequestNamespace(), req.VolumeIDs, req.Force); err != nil {
 		n.logger.Error("CSIVolumeDeregister failed", "error", err)
 		return err
 	}

nomad/state/state_store.go

@@ -2114,7 +2114,7 @@ func (s *StateStore) CSIVolumeClaim(index uint64, namespace, id string, claim *s
 }
 
 // CSIVolumeDeregister removes the volume from the server
-func (s *StateStore) CSIVolumeDeregister(index uint64, namespace string, ids []string) error {
+func (s *StateStore) CSIVolumeDeregister(index uint64, namespace string, ids []string, force bool) error {
 	txn := s.db.Txn(true)
 	defer txn.Abort()
 
@@ -2133,8 +2133,14 @@ func (s *StateStore) CSIVolumeDeregister(index uint64, namespace string, ids []s
 			return fmt.Errorf("volume row conversion error: %s", id)
 		}
 
+		// The common case for a volume deregister is when the volume is
+		// unused, but we can also let an operator intervene in the case where
+		// allocations have been stopped but claims can't be freed because
+		// ex. the plugins have all been removed.
 		if vol.InUse() {
-			return fmt.Errorf("volume in use: %s", id)
+			if !force || !s.volSafeToForce(vol) {
+				return fmt.Errorf("volume in use: %s", id)
+			}
 		}
 
 		if err = txn.Delete("csi_volumes", existing); err != nil {
@@ -2150,6 +2156,28 @@ func (s *StateStore) CSIVolumeDeregister(index uint64, namespace string, ids []s
 	return nil
 }
 
+// volSafeToForce checks if the any of the remaining allocations
+// are in a non-terminal state.
+func (s *StateStore) volSafeToForce(v *structs.CSIVolume) bool {
+	ws := memdb.NewWatchSet()
+	vol, err := s.CSIVolumeDenormalize(ws, v)
+	if err != nil {
+		return false
+	}
+
+	for _, alloc := range vol.ReadAllocs {
+		if alloc != nil && !alloc.TerminalStatus() {
+			return false
+		}
+	}
+	for _, alloc := range vol.WriteAllocs {
+		if alloc != nil && !alloc.TerminalStatus() {
+			return false
+		}
+	}
+	return true
+}
+
 // CSIVolumeDenormalizePlugins returns a CSIVolume with current health and plugins, but
 // without allocations
 // Use this for current volume metadata, handling lists of volumes

nomad/state/state_store_test.go

@@ -2991,7 +2991,12 @@ func TestStateStore_CSIVolume(t *testing.T) {
 	require.Error(t, err, fmt.Sprintf("volume exists: %s", vol0))
 	// as is deregistration
 	index++
-	err = state.CSIVolumeDeregister(index, ns, []string{vol0})
+	err = state.CSIVolumeDeregister(index, ns, []string{vol0}, false)
+	require.Error(t, err, fmt.Sprintf("volume in use: %s", vol0))
+
+	// even if forced, because we have a non-terminal claim
+	index++
+	err = state.CSIVolumeDeregister(index, ns, []string{vol0}, true)
 	require.Error(t, err, fmt.Sprintf("volume in use: %s", vol0))
 
 	// release claims to unblock deregister
@@ -3006,7 +3011,7 @@ func TestStateStore_CSIVolume(t *testing.T) {
 	require.NoError(t, err)
 
 	index++
-	err = state.CSIVolumeDeregister(index, ns, []string{vol0})
+	err = state.CSIVolumeDeregister(index, ns, []string{vol0}, false)
 	require.NoError(t, err)
 
 	// List, now omitting the deregistered volume

nomad/structs/csi.go

@@ -599,6 +599,7 @@ type CSIVolumeRegisterResponse struct {
 
 type CSIVolumeDeregisterRequest struct {
 	VolumeIDs []string
+	Force     bool
 	WriteRequest
 }
 

nomad/volumewatcher/volumes_watcher_test.go

@@ -284,7 +284,7 @@ func TestVolumeWatch_RegisterDeregister(t *testing.T) {
 
 	// deregistering the volume doesn't cause an update that triggers
 	// a watcher; we'll clean up this watcher in a GC later
-	err = srv.State().CSIVolumeDeregister(index, vol.Namespace, []string{vol.ID})
+	err = srv.State().CSIVolumeDeregister(index, vol.Namespace, []string{vol.ID}, false)
 	require.NoError(err)
 	require.Equal(1, len(watcher.watchers))
 	require.False(watcher.watchers[vol.ID+vol.Namespace].isRunning())

website/pages/api-docs/volumes.mdx

@@ -318,11 +318,16 @@ The table below shows this endpoint's support for
   volume. This must be the full ID. This is specified as part of the
   path.
 
+- `force` `(bool: false)`- Force deregistration of the volume and immediately
+  drop claims for terminal allocations. Returns an error if the volume has
+  running allocations. This does not detach the volume from client nodes.
+  This is specified as a query string parameter.
+
 ### Sample Request
 
 ```shell-session
 $ curl \
     --request DELETE \
     --data @payload.json \
-    https://localhost:4646/v1/volume/csi/volume-id1
+    https://localhost:4646/v1/volume/csi/volume-id1?force=false
 ```