consul: Support DeleteTree

consul/state_store.go

@@ -766,6 +766,16 @@ func (s *StateStore) KVSList(prefix string) (uint64, structs.DirEntries, error)
 
 // KVSDelete is used to delete a KVS entry
 func (s *StateStore) KVSDelete(index uint64, key string) error {
+	return s.kvsDeleteWithIndex(index, "id", key)
+}
+
+// KVSDeleteTree is used to delete all keys with a given prefix
+func (s *StateStore) KVSDeleteTree(index uint64, prefix string) error {
+	return s.kvsDeleteWithIndex(index, "id_prefix", prefix)
+}
+
+// kvsDeleteWithIndex does a delete with either the id or id_prefix
+func (s *StateStore) kvsDeleteWithIndex(index uint64, tableIndex, val string) error {
 	// Start a new txn
 	tx, err := s.kvsTable.StartTxn(false, nil)
 	if err != nil {
@@ -773,7 +783,7 @@ func (s *StateStore) KVSDelete(index uint64, key string) error {
 	}
 	defer tx.Abort()
 
-	num, err := s.kvsTable.DeleteTxn(tx, "id", key)
+	num, err := s.kvsTable.DeleteTxn(tx, tableIndex, val)
 	if err != nil {
 		return err
 	}
@@ -787,12 +797,6 @@ func (s *StateStore) KVSDelete(index uint64, key string) error {
 	return tx.Commit()
 }
 
-// KVSDeleteTree is used to delete all keys with a given prefix
-func (s *StateStore) KVSDeleteTree() error {
-	// TODO:
-	return nil
-}
-
 // KVSCheckAndSet is used to perform an atomic check-and-set
 func (s *StateStore) KVSCheckAndSet(index uint64, d *structs.DirEntry) (bool, error) {
 	// Start a new txn

consul/state_store_test.go

@@ -1163,3 +1163,49 @@ func TestKVS_List(t *testing.T) {
 		t.Fatalf("bad: %v", ents[2])
 	}
 }
+
+func TestKVSDeleteTree(t *testing.T) {
+	store, err := testStateStore()
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	defer store.Close()
+
+	// Should not exist
+	err = store.KVSDeleteTree(1000, "/web")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Create the entries
+	d := &structs.DirEntry{Key: "/web/a", Flags: 42, Value: []byte("test")}
+	if err := store.KVSSet(1000, d); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	d = &structs.DirEntry{Key: "/web/b", Flags: 42, Value: []byte("test")}
+	if err := store.KVSSet(1001, d); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	d = &structs.DirEntry{Key: "/web/sub/c", Flags: 42, Value: []byte("test")}
+	if err := store.KVSSet(1002, d); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Nuke the web tree
+	err = store.KVSDeleteTree(1010, "/web")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	// Nothing should list
+	idx, ents, err := store.KVSList("/web")
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if idx != 1010 {
+		t.Fatalf("bad: %v", idx)
+	}
+	if len(ents) != 0 {
+		t.Fatalf("bad: %v", ents)
+	}
+}