Fix zookeeper backend so that properly deletes/lists secrets.

This patch fixes two bugs in Zookeeper backends:
 * backend was determining if the node is a leaf or not basing on the number
   of the childer given node has. This is incorrect if you consider the fact
   that deleteing nested node can leave empty prefixes/dirs behind which have
   neither children nor data inside. The fix changes this situation by testing
   if the node has any data set - if not then it is not a leaf.
 * zookeeper does not delete nodes that do not have childern just like consul
   does and this leads to leaving empty nodes behind. In order to fix it, we
   scan the logical path of a secret being deleted for empty dirs/prefixes and
   remove them up until first non-empty one.

physical/zookeeper.go

@@ -10,7 +10,7 @@ import (
 
 	log "github.com/mgutz/logxi/v1"
 
-	"github.com/armon/go-metrics"
+	metrics "github.com/armon/go-metrics"
 	"github.com/samuel/go-zookeeper/zk"
 )
 
@@ -159,7 +159,39 @@ func (c *ZookeeperBackend) ensurePath(path string, value []byte) error {
 	return nil
 }
 
-// nodePath returns an etcd filepath based on the given key.
+// cleanupLogicalPath is used to remove all empty nodes, begining with deepest one,
+// aborting on first non-empty one, up to top-level node.
+func (c *ZookeeperBackend) cleanupLogicalPath(path string) error {
+	nodes := strings.Split(path, "/")
+	for i := len(nodes) - 1; i > 0; i-- {
+		fullPath := c.path + strings.Join(nodes[:i], "/")
+
+		_, stat, err := c.client.Exists(fullPath)
+		if err != nil {
+			return fmt.Errorf("Failed to acquire node data: %s", err)
+		}
+
+		if stat.DataLength > 0 && stat.NumChildren > 0 {
+			msgFmt := "Node %s is both of data and leaf type ??"
+			panic(fmt.Sprintf(msgFmt, fullPath))
+		} else if stat.DataLength > 0 {
+			msgFmt := "Node %s is a data node, this is either a bug or " +
+				"backend data is corrupted"
+			panic(fmt.Sprintf(msgFmt, fullPath))
+		} else if stat.NumChildren > 0 {
+			return nil
+		} else {
+			// Empty node, lets clean it up!
+			if err := c.client.Delete(fullPath, -1); err != nil {
+				msgFmt := "Removal of node `%s` failed: `%v`"
+				return fmt.Errorf(msgFmt, fullPath, err)
+			}
+		}
+	}
+	return nil
+}
+
+// nodePath returns an zk path based on the given key.
 func (c *ZookeeperBackend) nodePath(key string) string {
 	return filepath.Join(c.path, filepath.Dir(key), ZKNodeFilePrefix+filepath.Base(key))
 }
@@ -215,9 +247,12 @@ func (c *ZookeeperBackend) Delete(key string) error {
 	err := c.client.Delete(fullPath, -1)
 
 	// Mask if the node does not exist
-	if err == zk.ErrNoNode {
-		err = nil
+	if err != nil && err != zk.ErrNoNode {
+		return fmt.Errorf("Failed to remove `%s`: %v", fullPath, err)
 	}
+
+	err = c.cleanupLogicalPath(key)
+
 	return err
 }
 
@@ -233,15 +268,28 @@ func (c *ZookeeperBackend) List(prefix string) ([]string, error) {
 	// If the path nodes are missing, no children!
 	if err == zk.ErrNoNode {
 		return []string{}, nil
+	} else if err != nil {
+		return []string{}, err
 	}
 
 	children := []string{}
 	for _, key := range result {
-		// Check if this entry has any child entries,
+		childPath := fullPath + "/" + key
+		_, stat, err := c.client.Exists(childPath)
+		if err != nil {
+			// Node is ought to exists, so it must be something different
+			return []string{}, err
+		}
+
+		// Check if this entry is a leaf of a node,
 		// and append the slash which is what Vault depends on
 		// for iteration
-		nodeChildren, _, _ := c.client.Children(fullPath + "/" + key)
-		if nodeChildren != nil && len(nodeChildren) > 0 {
+		if stat.DataLength > 0 && stat.NumChildren > 0 {
+			msgFmt := "Node %s is both of data and leaf type ??"
+			panic(fmt.Sprintf(msgFmt, childPath))
+		} else if stat.DataLength == 0 {
+			// No, we cannot differentiate here on number of children as node
+			// can have all it leafs remoed, and it still is a node.
 			children = append(children, key+"/")
 		} else {
 			children = append(children, key[1:])

physical/zookeeper_test.go

@@ -33,6 +33,9 @@ func TestZookeeperBackend(t *testing.T) {
 	}
 
 	defer func() {
+		client.Delete(randPath+"/foo/nested1/nested2/nested3", -1)
+		client.Delete(randPath+"/foo/nested1/nested2", -1)
+		client.Delete(randPath+"/foo/nested1", -1)
 		client.Delete(randPath+"/foo/bar/baz", -1)
 		client.Delete(randPath+"/foo/bar", -1)
 		client.Delete(randPath+"/foo", -1)