From c7a11a86c685c884c66742c04198a8e734ff099b Mon Sep 17 00:00:00 2001
From: Tim Gross <tgross@hashicorp.com>
Date: Thu, 21 Jul 2022 16:13:52 -0400
Subject: [PATCH] block deleting namespaces if the namespace contains a volume
 (#13880)

When we delete a namespace, we check to ensure that there are no non-terminal
jobs, which effectively covers evals, allocs, etc. CSI volumes are also
namespaced, so extend this check to cover CSI volumes.
---
 .changelog/13880.txt            |  3 +++
 nomad/state/state_store.go      | 16 +++++++++++++++
 nomad/state/state_store_test.go | 35 +++++++++++++++++++++++++++++++++
 3 files changed, 54 insertions(+)
 create mode 100644 .changelog/13880.txt

diff --git a/.changelog/13880.txt b/.changelog/13880.txt
new file mode 100644
index 000000000..31cdaca0f
--- /dev/null
+++ b/.changelog/13880.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+namespaces: Fixed a bug that allowed deleting a namespace that contained a CSI volume
+```
diff --git a/nomad/state/state_store.go b/nomad/state/state_store.go
index 7069f66f4..ed6242496 100644
--- a/nomad/state/state_store.go
+++ b/nomad/state/state_store.go
@@ -2440,6 +2440,11 @@ func (s *StateStore) CSIVolumesByNodeID(ws memdb.WatchSet, prefix, nodeID string
 func (s *StateStore) CSIVolumesByNamespace(ws memdb.WatchSet, namespace, prefix string) (memdb.ResultIterator, error) {
 	txn := s.db.ReadTxn()
 
+	return s.csiVolumesByNamespaceImpl(txn, ws, namespace, prefix)
+}
+
+func (s *StateStore) csiVolumesByNamespaceImpl(txn *txn, ws memdb.WatchSet, namespace, prefix string) (memdb.ResultIterator, error) {
+
 	iter, err := txn.Get("csi_volumes", "id_prefix", namespace, prefix)
 	if err != nil {
 		return nil, fmt.Errorf("volume lookup failed: %v", err)
@@ -6336,6 +6341,17 @@ func (s *StateStore) DeleteNamespaces(index uint64, names []string) error {
 			}
 		}
 
+		vIter, err := s.csiVolumesByNamespaceImpl(txn, nil, name, "")
+		if err != nil {
+			return err
+		}
+		rawVol := vIter.Next()
+		if rawVol != nil {
+			vol := rawVol.(*structs.CSIVolume)
+			return fmt.Errorf("namespace %q contains at least one CSI volume %q. "+
+				"All CSI volumes in namespace must be deleted before it can be deleted", name, vol.ID)
+		}
+
 		// Delete the namespace
 		if err := txn.Delete(TableNamespaces, existing); err != nil {
 			return fmt.Errorf("namespace deletion failed: %v", err)
diff --git a/nomad/state/state_store_test.go b/nomad/state/state_store_test.go
index 6e37c4b55..04bc87b5a 100644
--- a/nomad/state/state_store_test.go
+++ b/nomad/state/state_store_test.go
@@ -1009,6 +1009,41 @@ func TestStateStore_DeleteNamespaces_NonTerminalJobs(t *testing.T) {
 	require.False(t, watchFired(ws))
 }
 
+func TestStateStore_DeleteNamespaces_CSIVolumes(t *testing.T) {
+	ci.Parallel(t)
+
+	state := testStateStore(t)
+
+	ns := mock.Namespace()
+	require.NoError(t, state.UpsertNamespaces(1000, []*structs.Namespace{ns}))
+
+	plugin := mock.CSIPlugin()
+	vol := mock.CSIVolume(plugin)
+	vol.Namespace = ns.Name
+
+	require.NoError(t, state.UpsertCSIVolume(1001, []*structs.CSIVolume{vol}))
+
+	// Create a watchset so we can test that delete fires the watch
+	ws := memdb.NewWatchSet()
+	_, err := state.NamespaceByName(ws, ns.Name)
+	require.NoError(t, err)
+
+	err = state.DeleteNamespaces(1002, []string{ns.Name})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "one CSI volume")
+	require.False(t, watchFired(ws))
+
+	ws = memdb.NewWatchSet()
+	out, err := state.NamespaceByName(ws, ns.Name)
+	require.NoError(t, err)
+	require.NotNil(t, out)
+
+	index, err := state.Index(TableNamespaces)
+	require.NoError(t, err)
+	require.EqualValues(t, 1000, index)
+	require.False(t, watchFired(ws))
+}
+
 func TestStateStore_Namespaces(t *testing.T) {
 	ci.Parallel(t)