Repository: puppetlabs/wash
Pull request #567 (2019-11-05 17:56:33)

Implement Delete for volumes
Supports #479.

Signed-off-by: Enis Inan <enis.inan@puppet.com>


Patch:
diff --git a/plugin/docker/volume.go b/plugin/docker/volume.go
index 663a64b..ddef60e 100644
--- a/plugin/docker/volume.go
+++ b/plugin/docker/volume.go
@@ -7,6 +7,7 @@ import (
 	"errors"
 	"io"
 	"io/ioutil"
+	"strings"
 	"time"
 
 	"github.com/docker/docker/api/types"
@@ -90,12 +91,11 @@ func (v *volume) createContainer(ctx context.Context, cmd []string) (string, err
 	return created.ID, nil
 }
 
-func (v *volume) VolumeList(ctx context.Context, path string) (volpkg.DirMap, error) {
-	// Use a larger maxdepth because volumes have relatively few files and VolumeList is slow.
-	maxdepth := 10
-
-	// Create a container that mounts a volume and inspects it. Run it and capture the output.
-	cid, err := v.createContainer(ctx, volpkg.StatCmd(mountpoint+path, maxdepth))
+// Runs cmd in a temporary container. If the exit code is 0, then it returns the cmd's output.
+// Otherwise, it wraps the cmd's output in an error object.
+func (v *volume) runInTemporaryContainer(ctx context.Context, cmd []string) ([]byte, error) {
+	// Create a container that mounts a volume and deletes its file. Run rm -rf on it.
+	cid, err := v.createContainer(ctx, cmd)
 	if err != nil {
 		return nil, err
 	}
@@ -134,15 +134,24 @@ func (v *volume) VolumeList(ctx context.Context, path string) (volpkg.DirMap, er
 		activity.Record(ctx, "Closed log for %v: %v", cid, output.Close())
 	}()
 
+	bytes, err := ioutil.ReadAll(output)
+	if err != nil {
+		return nil, err
+	}
 	if statusCode != 0 {
-		bytes, err := ioutil.ReadAll(output)
-		if err != nil {
-			return nil, err
-		}
-		return nil, errors.New(string(bytes))
+		return nil, errors.New(strings.Trim(string(bytes), "\n"))
 	}
+	return bytes, nil
+}
 
-	return volpkg.StatParseAll(output, mountpoint, path, maxdepth)
+func (v *volume) VolumeList(ctx context.Context, path string) (volpkg.DirMap, error) {
+	// Use a larger maxdepth because volumes have relatively few files and VolumeList is slow.
+	maxdepth := 10
+	output, err := v.runInTemporaryContainer(ctx, volpkg.StatCmd(mountpoint+path, maxdepth))
+	if err != nil {
+		return nil, err
+	}
+	return volpkg.StatParseAll(bytes.NewReader(output), mountpoint, path, maxdepth)
 }
 
 func (v *volume) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, error) {
@@ -226,6 +235,14 @@ func (v *volume) VolumeStream(ctx context.Context, path string) (io.ReadCloser,
 	return plugin.CleanupReader{ReadCloser: output, Cleanup: killAndDelete}, nil
 }
 
+func (v *volume) VolumeDelete(ctx context.Context, path string) (bool, error) {
+	_, err := v.runInTemporaryContainer(ctx, []string{"rm", "-rf", mountpoint + path})
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 const volumeDescription = `
 This is a Docker volume. We create a temporary Docker container whenever
 Wash invokes a currently uncached List/Read/Stream action on it or one of
diff --git a/plugin/kubernetes/pvc.go b/plugin/kubernetes/pvc.go
index 98a4c75..e93ff3d 100644
--- a/plugin/kubernetes/pvc.go
+++ b/plugin/kubernetes/pvc.go
@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"strings"
 	"time"
 
 	"github.com/puppetlabs/wash/activity"
@@ -143,12 +144,11 @@ func (v *pvc) waitForPod(ctx context.Context, pid string) error {
 	}
 }
 
-func (v *pvc) VolumeList(ctx context.Context, path string) (volume.DirMap, error) {
-	// Use a larger maxdepth because volumes have relatively few files and VolumeList is slow.
-	maxdepth := 10
-
-	// Create a container that mounts a pvc and inspects it. Run it and capture the output.
-	pid, err := v.createPod(volume.StatCmd(mountpoint+path, maxdepth))
+// Runs cmd in a temporary pod. If the exit code is 0, then it returns the cmd's output.
+// Otherwise, it wraps the cmd's output in an error object.
+func (v *pvc) runInTemporaryPod(ctx context.Context, cmd []string) ([]byte, error) {
+	// Create a pod that mounts a pvc and inspects it. Run it and capture the output.
+	pid, err := v.createPod(cmd)
 	if err != nil {
 		return nil, err
 	}
@@ -171,54 +171,32 @@ func (v *pvc) VolumeList(ctx context.Context, path string) (volume.DirMap, error
 		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
 	}()
 
+	bytes, readErr := ioutil.ReadAll(output)
+	if readErr != nil {
+		return nil, readErr
+	}
 	if err == errPodTerminated {
-		bytes, err := ioutil.ReadAll(output)
-		if err != nil {
-			return nil, err
-		}
-		return nil, errors.New(string(bytes))
+		return nil, errors.New(strings.Trim(string(bytes), "\n"))
 	}
-
-	return volume.StatParseAll(output, mountpoint, path, maxdepth)
+	return bytes, nil
 }
 
-func (v *pvc) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, error) {
-	// Create a container that mounts a pvc and output the file.
-	pid, err := v.createPod([]string{"cat", mountpoint + path})
-	activity.Record(ctx, "Reading from: %v", mountpoint+path)
-	if err != nil {
-		return nil, err
-	}
-	defer func() {
-		activity.Record(ctx, "Deleted temporary pod %v: %v", pid, v.podi.Delete(pid, &metav1.DeleteOptions{}))
-	}()
-
-	activity.Record(ctx, "Waiting for pod %v", pid)
-	// Start watching for new events related to the pod we created.
-	if err = v.waitForPod(ctx, pid); err != nil && err != errPodTerminated {
-		return nil, err
-	}
-	podErr := err
-
-	activity.Record(ctx, "Gathering log for %v", pid)
-	output, err := v.podi.GetLogs(pid, &corev1.PodLogOptions{}).Stream()
+func (v *pvc) VolumeList(ctx context.Context, path string) (volume.DirMap, error) {
+	// Use a larger maxdepth because volumes have relatively few files and VolumeList is slow.
+	maxdepth := 10
+	output, err := v.runInTemporaryPod(ctx, volume.StatCmd(mountpoint+path, maxdepth))
 	if err != nil {
 		return nil, err
 	}
-	defer func() {
-		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
-	}()
+	return volume.StatParseAll(bytes.NewReader(output), mountpoint, path, maxdepth)
+}
 
-	bits, err := ioutil.ReadAll(output)
+func (v *pvc) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, error) {
+	output, err := v.runInTemporaryPod(ctx, []string{"cat", mountpoint + path})
 	if err != nil {
 		return nil, err
 	}
-	activity.Record(ctx, "Read: %v", bits)
-
-	if podErr == errPodTerminated {
-		return nil, errors.New(string(bits))
-	}
-	return bytes.NewReader(bits), nil
+	return bytes.NewReader(output), nil
 }
 
 func (v *pvc) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
@@ -265,6 +243,14 @@ func (v *pvc) VolumeStream(ctx context.Context, path string) (io.ReadCloser, err
 	return plugin.CleanupReader{ReadCloser: output, Cleanup: delete}, nil
 }
 
+func (v *pvc) VolumeDelete(ctx context.Context, path string) (bool, error) {
+	_, err := v.runInTemporaryPod(ctx, []string{"rm", "-rf", mountpoint + path})
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 const pvcDescription = `
 This is a Kubernetes persistent volume claim. We create a temporary Kubernetes
 pod whenever Wash invokes a currently uncached List/Read/Stream action on it or
diff --git a/volume/core.go b/volume/core.go
index 27bd9d9..b28a19e 100644
--- a/volume/core.go
+++ b/volume/core.go
@@ -9,6 +9,8 @@ package volume
 import (
 	"context"
 	"io"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/puppetlabs/wash/plugin"
@@ -29,9 +31,11 @@ type Interface interface {
 	VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, error)
 	// Accepts a path and streams updates to the content associated with that path.
 	VolumeStream(ctx context.Context, path string) (io.ReadCloser, error)
+	// Deletes the volume node at the specified path. Mirrors plugin.Deletable#Delete
+	VolumeDelete(ctx context.Context, path string) (bool, error)
 }
 
-// Children represents a directory's children. It is a map of <child_path> => <child_attributes>.
+// Children represents a directory's children. It is a map of <child_basename> => <child_attributes>.
 type Children = map[string]plugin.EntryAttributes
 
 // A DirMap is a map of <dir_path> => <children>. If <children> is nil, then that means
@@ -39,6 +43,13 @@ type Children = map[string]plugin.EntryAttributes
 // <dir_path>.
 type DirMap = map[string]Children
 
+// dirMap is a thread-safe wrapper to a DirMap object. It is needed to properly implement
+// Delete.
+type dirMap struct {
+	mp  DirMap
+	mux sync.RWMutex
+}
+
 // ChildSchemas returns a volume's child schema
 func ChildSchemas() []*plugin.EntrySchema {
 	return []*plugin.EntrySchema{
@@ -61,3 +72,28 @@ func List(ctx context.Context, impl Interface) ([]plugin.Entry, error) {
 // ListTTL represents the List op's TTL. The entry implementing volume.Interface should
 // set the List op's TTL to this value.
 const ListTTL = 30 * time.Second
+
+// delete is a keyword, so we use deleteNode instead. Note that this implementation is
+// symmetric with plugin.Delete except that we are managing a dirmap instead of a cache.
+func deleteNode(ctx context.Context, impl Interface, path string, dirmap *dirMap) (deleted bool, err error) {
+	deleted, err = impl.VolumeDelete(ctx, path)
+	if err != nil {
+		return
+	}
+	if !deleted {
+		return
+	}
+
+	// The node was deleted so remove it from the dirmap and from its parent's children
+	dirmap.mux.Lock()
+	defer dirmap.mux.Unlock()
+
+	delete(dirmap.mp, path)
+	segments := strings.Split(path, "/")
+	parentPath := strings.Join(segments[:len(segments)-1], "/")
+	if parentChildren, ok := dirmap.mp[parentPath]; ok {
+		basename := segments[len(segments)-1]
+		delete(parentChildren, basename)
+	}
+	return
+}
diff --git a/volume/dir.go b/volume/dir.go
index f6065ee..7a9466d 100644
--- a/volume/dir.go
+++ b/volume/dir.go
@@ -11,7 +11,7 @@ type dir struct {
 	plugin.EntryBase
 	impl   Interface
 	path   string
-	dirmap DirMap
+	dirmap *dirMap
 }
 
 // newDir creates a dir populated from dirs.
@@ -35,22 +35,27 @@ func (v *dir) Schema() *plugin.EntrySchema {
 
 // Generate children using the provided DirMap. The dir may not have a dirmap
 // stored if it's a source because it should dynamically generate it.
-func (v *dir) generateChildren(dirmap DirMap) []plugin.Entry {
-	parent := dirmap[v.path]
+func (v *dir) generateChildren(dirmap *dirMap) []plugin.Entry {
+	dirmap.mux.RLock()
+	defer dirmap.mux.RUnlock()
+
+	parent := dirmap.mp[v.path]
 	entries := make([]plugin.Entry, 0, len(parent))
 	for name, attr := range parent {
 		subpath := v.path + "/" + name
 		if attr.Mode().IsDir() {
 			newEntry := newDir(name, attr, v.impl, subpath)
 			newEntry.SetTTLOf(plugin.ListOp, ListTTL)
-			if d, ok := dirmap[subpath]; ok && d != nil {
+			if d, ok := dirmap.mp[subpath]; ok && d != nil {
 				newEntry.dirmap = dirmap
 				newEntry.Prefetched()
 				newEntry.DisableCachingFor(plugin.ListOp)
 			}
 			entries = append(entries, newEntry)
 		} else {
-			entries = append(entries, newFile(name, attr, v.impl, subpath))
+			newEntry := newFile(name, attr, v.impl, subpath)
+			newEntry.dirmap = dirmap
+			entries = append(entries, newEntry)
 		}
 	}
 	return entries
@@ -69,7 +74,11 @@ func (v *dir) List(ctx context.Context) ([]plugin.Entry, error) {
 		return nil, err
 	}
 
-	return v.generateChildren(dirmap), nil
+	return v.generateChildren(&dirMap{mp: dirmap}), nil
+}
+
+func (v *dir) Delete(ctx context.Context) (bool, error) {
+	return deleteNode(ctx, v.impl, v.path, v.dirmap)
 }
 
 const dirDescription = `
diff --git a/volume/file.go b/volume/file.go
index e2679e2..55b9827 100644
--- a/volume/file.go
+++ b/volume/file.go
@@ -11,8 +11,9 @@ import (
 // file represents a file in a volume that has content we can access.
 type file struct {
 	plugin.EntryBase
-	impl Interface
-	path string
+	impl   Interface
+	path   string
+	dirmap *dirMap
 }
 
 // newFile creates a VolumeFile.
@@ -41,6 +42,10 @@ func (v *file) Stream(ctx context.Context) (io.ReadCloser, error) {
 	return v.impl.VolumeStream(ctx, v.path)
 }
 
+func (v *file) Delete(ctx context.Context) (bool, error) {
+	return deleteNode(ctx, v.impl, v.path, v.dirmap)
+}
+
 const fileDescription = `
 This is a file on a remote volume or a container/VM.
 `
diff --git a/volume/fs.go b/volume/fs.go
index aad56ef..7f67c15 100644
--- a/volume/fs.go
+++ b/volume/fs.go
@@ -165,6 +165,17 @@ func (d *FS) VolumeStream(ctx context.Context, path string) (io.ReadCloser, erro
 	return r, nil
 }
 
+// VolumeDelete satisfies the Interface required by Delete to delete volume nodes.
+func (d *FS) VolumeDelete(ctx context.Context, path string) (bool, error) {
+	activity.Record(ctx, "Deleting %v on %v", path, plugin.ID(d.executor))
+	_, err := exec(ctx, d.executor, []string{"rm", "-rf", path})
+	if err != nil {
+		activity.Record(ctx, "Exec error running 'rm -rf %v' in VolumeDelete: %v", path, err)
+		return false, err
+	}
+	return true, nil
+}
+
 const fsDescription = `
 This represents the root directory of a container/VM. It lets you navigate
 and interact with that container/VM's filesystem as if you were logged into


Test patch:
diff --git a/volume/core_test.go b/volume/core_test.go
new file mode 100644
index 0000000..57d4e08
--- /dev/null
+++ b/volume/core_test.go
@@ -0,0 +1,74 @@
+package volume
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/puppetlabs/wash/plugin"
+	"github.com/stretchr/testify/suite"
+)
+
+type coreTestSuite struct {
+	suite.Suite
+}
+
+func (s *coreTestSuite) TestDeleteNode_ReturnsVolumeDeleteError() {
+	ctx := context.Background()
+	mockImpl := &mockDirEntry{EntryBase: plugin.NewEntry("foo")}
+	path := "bar"
+
+	expectedErr := fmt.Errorf("failed to delete")
+	mockImpl.On("VolumeDelete", ctx, path).Return(false, expectedErr)
+
+	_, err := deleteNode(ctx, mockImpl, path, &dirMap{})
+	s.EqualError(expectedErr, err.Error())
+}
+
+func (s *coreTestSuite) TestDeleteNode_NodeDeletionInProgress_LeavesDirMapAlone() {
+	ctx := context.Background()
+	mockImpl := &mockDirEntry{EntryBase: plugin.NewEntry("foo")}
+	path := "bar/baz"
+	dirMap := &dirMap{
+		mp: map[string]Children{
+			"bar/baz": map[string]plugin.EntryAttributes{
+				"baz": plugin.EntryAttributes{},
+			},
+		},
+	}
+
+	mockImpl.On("VolumeDelete", ctx, path).Return(false, nil)
+
+	deleted, err := deleteNode(ctx, mockImpl, path, dirMap)
+	if s.NoError(err) {
+		s.False(deleted)
+		s.Contains(dirMap.mp, "bar/baz")
+		s.Contains(dirMap.mp["bar/baz"], "baz")
+	}
+}
+
+func (s *coreTestSuite) TestDeleteNode_DeletedNode_UpdatesDirMap() {
+	ctx := context.Background()
+	mockImpl := &mockDirEntry{EntryBase: plugin.NewEntry("foo")}
+	path := "bar/baz"
+	dirMap := &dirMap{
+		mp: map[string]Children{
+			"bar/baz": map[string]plugin.EntryAttributes{
+				"baz": plugin.EntryAttributes{},
+			},
+		},
+	}
+
+	mockImpl.On("VolumeDelete", ctx, path).Return(true, nil)
+
+	deleted, err := deleteNode(ctx, mockImpl, path, dirMap)
+	if s.NoError(err) {
+		s.True(deleted)
+		s.NotContains(dirMap.mp, "bar/baz")
+		s.NotContains(dirMap.mp["bar/baz"], "baz")
+	}
+}
+
+func TestCore(t *testing.T) {
+	suite.Run(t, new(coreTestSuite))
+}
diff --git a/volume/dir_test.go b/volume/dir_test.go
index 83a4267..150867a 100644
--- a/volume/dir_test.go
+++ b/volume/dir_test.go
@@ -31,6 +31,12 @@ func (m *mockDirEntry) VolumeStream(context.Context, string) (io.ReadCloser, err
 	return nil, nil
 }
 
+func (m *mockDirEntry) VolumeDelete(ctx context.Context, path string) (bool, error) {
+	// deleteNode's tests use this entry, so we need to implement VolumeDelete for them
+	args := m.Called(ctx, path)
+	return args.Get(0).(bool), args.Error(1)
+}
+
 func (m *mockDirEntry) Schema() *plugin.EntrySchema {
 	return nil
 }
@@ -62,7 +68,7 @@ func TestVolumeDir(t *testing.T) {
 
 	assert.NotNil(t, dmap[RootPath]["path2"])
 	vd = newDir("path", dmap[RootPath]["path2"], &entry, "/path2")
-	vd.dirmap = dmap
+	vd.dirmap = &dirMap{mp: dmap}
 	entries, err = vd.List(ctx)
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(entries))
diff --git a/volume/file_test.go b/volume/file_test.go
index bfa03b6..dea28ef 100644
--- a/volume/file_test.go
+++ b/volume/file_test.go
@@ -37,6 +37,10 @@ func (m *mockFileEntry) VolumeStream(context.Context, string) (io.ReadCloser, er
 	return ioutil.NopCloser(strings.NewReader(m.content)), nil
 }
 
+func (m *mockFileEntry) VolumeDelete(context.Context, string) (bool, error) {
+	return true, nil
+}
+
 func (m *mockFileEntry) Schema() *plugin.EntrySchema {
 	return nil
 }
diff --git a/volume/fs_test.go b/volume/fs_test.go
index 0a4c0ea..36a8267 100644
--- a/volume/fs_test.go
+++ b/volume/fs_test.go
@@ -227,6 +227,21 @@ func (suite *fsTestSuite) TestFSRead() {
 	suite.Equal("hello", string(buf))
 }
 
+func (suite *fsTestSuite) TestVolumeDelete() {
+	exec := suite.createExec(varLogFixture, varLogDepth)
+	fs := NewFS("fs", exec, varLogDepth)
+	// ID would normally be set when listing FS within the parent instance.
+	fs.SetTestID("/instance/fs")
+
+	execResult := suite.createResult("deleted")
+	exec.On("Exec", mock.Anything, "rm", []string{"-rf", "/var/log/path1/a file"}, plugin.ExecOptions{Elevate: true}).Return(execResult, nil)
+	deleted, err := fs.VolumeDelete(context.Background(), "/var/log/path1/a file")
+	if suite.NoError(err) {
+		suite.True(deleted)
+		exec.AssertCalled(suite.T(), "Exec", mock.Anything, "rm", []string{"-rf", "/var/log/path1/a file"}, plugin.ExecOptions{Elevate: true})
+	}
+}
+
 func TestFS(t *testing.T) {
 	suite.Run(t, new(fsTestSuite))
 }