Repository: puppetlabs/wash
Pull request #218 (2019-04-18 19:57:17)

Speed up volume listing, enable refresh of volume data, and fix /cache


Patch:
diff --git a/api/cache.go b/api/cache.go
index 2c98786..ba143b3 100644
--- a/api/cache.go
+++ b/api/cache.go
@@ -24,7 +24,7 @@ import (
 //       200:
 //       500: errorResp
 var cacheHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
-	path, errResp := getPathFromRequest(r)
+	path, errResp := getWashPathFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/error.go b/api/error.go
index 33c920d..67dda24 100644
--- a/api/error.go
+++ b/api/error.go
@@ -197,3 +197,11 @@ func relativePathResponse(path string) *errorResponse {
 
 	return &errorResponse{http.StatusBadRequest, body}
 }
+
+func nonWashPathResponse(path string) *errorResponse {
+	return &errorResponse{http.StatusBadRequest, newErrorObj(
+		apitypes.NonWashPath,
+		"Request path is not managed by wash",
+		apitypes.ErrorFields{"path": path},
+	)}
+}
diff --git a/api/exec.go b/api/exec.go
index 985631d..3ef3755 100644
--- a/api/exec.go
+++ b/api/exec.go
@@ -8,8 +8,8 @@ import (
 	"strings"
 	"time"
 
-	apitypes "github.com/puppetlabs/wash/api/types"
 	"github.com/puppetlabs/wash/activity"
+	apitypes "github.com/puppetlabs/wash/api/types"
 	"github.com/puppetlabs/wash/plugin"
 
 	log "github.com/sirupsen/logrus"
@@ -88,7 +88,7 @@ func streamExitCode(ctx context.Context, w *json.Encoder, exitCodeCB func() (int
 //       500: errorResp
 var execHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
 	ctx := r.Context()
-	entry, path, errResp := getEntryFromRequest(ctx, r)
+	entry, path, errResp := getEntryFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/helpers.go b/api/helpers.go
index 56e745a..fd52276 100644
--- a/api/helpers.go
+++ b/api/helpers.go
@@ -62,30 +62,59 @@ func findEntry(ctx context.Context, root plugin.Entry, segments []string) (plugi
 	return curEntry, nil
 }
 
+// Common helper to get path query param from request and validate it.
 func getPathFromRequest(r *http.Request) (string, *errorResponse) {
 	paths := r.URL.Query()["path"]
 	if len(paths) != 1 {
 		return "", invalidPathsResponse(paths)
 	}
-	return paths[0], nil
-}
-
-func getEntryFromRequest(ctx context.Context, r *http.Request) (plugin.Entry, string, *errorResponse) {
-	path, errResp := getPathFromRequest(r)
-	if errResp != nil {
-		return nil, "", errResp
-	}
+	path := paths[0]
 
 	if path == "" {
 		panic("path should never be empty")
 	}
 	if !filepath.IsAbs(path) {
-		return nil, "", relativePathResponse(path)
+		return "", relativePathResponse(path)
 	}
 
+	return path, nil
+}
+
+// Common subset used by getEntryFromRequest and getWashPathFromRequest.
+// getEntryFromRequest needs both the path and the wash path.
+func toWashPath(ctx context.Context, path string) (string, *errorResponse) {
 	mountpoint := ctx.Value(mountpointKey).(string)
 	trimmedPath := strings.TrimPrefix(path, mountpoint)
 	if trimmedPath == path {
+		return path, nonWashPathResponse(path)
+	}
+
+	return trimmedPath, nil
+}
+
+// Simpler function to transform the path when an actual entry isn't needed.
+func getWashPathFromRequest(r *http.Request) (string, *errorResponse) {
+	path, err := getPathFromRequest(r)
+	if err != nil {
+		return "", err
+	}
+
+	return toWashPath(r.Context(), path)
+}
+
+func getEntryFromRequest(r *http.Request) (plugin.Entry, string, *errorResponse) {
+	path, errResp := getPathFromRequest(r)
+	if errResp != nil {
+		return nil, "", errResp
+	}
+
+	ctx := r.Context()
+	trimmedPath, errResp := toWashPath(ctx, path)
+	if errResp != nil {
+		if errResp.body.Kind != apitypes.NonWashPath {
+			panic("Unexpected error from getWashPathFromFullPath")
+		}
+
 		// Local file/directory, so convert it to a Wash entry
 		//
 		// TODO: The code here means that the Wash server cannot be
diff --git a/api/info.go b/api/info.go
index f5270c5..6931de2 100644
--- a/api/info.go
+++ b/api/info.go
@@ -26,7 +26,7 @@ import (
 //       500: errorResp
 var infoHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
 	ctx := r.Context()
-	entry, path, errResp := getEntryFromRequest(ctx, r)
+	entry, path, errResp := getEntryFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/list.go b/api/list.go
index 4a36256..641368f 100644
--- a/api/list.go
+++ b/api/list.go
@@ -5,8 +5,8 @@ import (
 	"fmt"
 	"net/http"
 
-	apitypes "github.com/puppetlabs/wash/api/types"
 	"github.com/puppetlabs/wash/activity"
+	apitypes "github.com/puppetlabs/wash/api/types"
 	"github.com/puppetlabs/wash/plugin"
 )
 
@@ -35,7 +35,7 @@ type entryList struct {
 //       500: errorResp
 var listHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
 	ctx := r.Context()
-	entry, path, errResp := getEntryFromRequest(ctx, r)
+	entry, path, errResp := getEntryFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/metadata.go b/api/metadata.go
index ac2ed3b..78198fb 100644
--- a/api/metadata.go
+++ b/api/metadata.go
@@ -32,7 +32,7 @@ type entryMetadata struct {
 //       500: errorResp
 var metadataHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
 	ctx := r.Context()
-	entry, path, errResp := getEntryFromRequest(ctx, r)
+	entry, path, errResp := getEntryFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/read.go b/api/read.go
index 9dadebb..57b60c5 100644
--- a/api/read.go
+++ b/api/read.go
@@ -27,7 +27,7 @@ import (
 //       500: errorResp
 var readHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
 	ctx := r.Context()
-	entry, path, errResp := getEntryFromRequest(ctx, r)
+	entry, path, errResp := getEntryFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/stream.go b/api/stream.go
index 89b0ad1..0befab3 100644
--- a/api/stream.go
+++ b/api/stream.go
@@ -29,7 +29,7 @@ import (
 //       500: errorResp
 var streamHandler handler = func(w http.ResponseWriter, r *http.Request) *errorResponse {
 	ctx := r.Context()
-	entry, path, errResp := getEntryFromRequest(ctx, r)
+	entry, path, errResp := getEntryFromRequest(r)
 	if errResp != nil {
 		return errResp
 	}
diff --git a/api/types/error.go b/api/types/error.go
index ea697e5..86057eb 100644
--- a/api/types/error.go
+++ b/api/types/error.go
@@ -36,4 +36,5 @@ const (
 	RelativePath       = "puppetlabs.wash/relative-path"
 	InvalidPaths       = "puppetlabs.wash/invalid-paths"
 	OutOfBounds        = "puppetlabs.wash/out-of-bounds"
+	NonWashPath        = "puppetlabs.wash/non-wash-path"
 )
diff --git a/datastore/cache.go b/datastore/cache.go
index 1110e72..112727f 100644
--- a/datastore/cache.go
+++ b/datastore/cache.go
@@ -115,6 +115,7 @@ func (cache *MemCache) Flush() {
 
 // Delete removes entries from the cache that match the provided regexp.
 func (cache *MemCache) Delete(matcher *regexp.Regexp) []string {
+	log.Debugf("Deleting matches for %v", matcher)
 	items := cache.instance.Items()
 	deleted := make([]string, 0, len(items))
 	for k := range items {
diff --git a/plugin/docker/volume.go b/plugin/docker/volume.go
index 0eb7589..1da16f3 100644
--- a/plugin/docker/volume.go
+++ b/plugin/docker/volume.go
@@ -35,7 +35,6 @@ func newVolume(c *client.Client, v *types.Volume) (*volume, error) {
 		EntryBase: plugin.NewEntry(v.Name),
 		client:    c,
 	}
-	vol.SetTTLOf(plugin.ListOp, 60*time.Second)
 
 	attr := plugin.EntryAttributes{}
 	attr.
@@ -57,6 +56,32 @@ func (v *volume) Metadata(ctx context.Context) (plugin.EntryMetadata, error) {
 }
 
 func (v *volume) List(ctx context.Context) ([]plugin.Entry, error) {
+	return vol.List(ctx, v, "")
+}
+
+// Create a container that mounts a volume to a default mountpoint and runs a command.
+func (v *volume) createContainer(ctx context.Context, cmd []string) (string, error) {
+	// Use tty to avoid messing with the extra log formatting.
+	cfg := docontainer.Config{Image: "busybox", Cmd: cmd, Tty: true}
+	mounts := []mount.Mount{{
+		Type:     mount.TypeVolume,
+		Source:   v.Name(),
+		Target:   mountpoint,
+		ReadOnly: true,
+	}}
+	hostcfg := docontainer.HostConfig{Mounts: mounts}
+	netcfg := network.NetworkingConfig{}
+	created, err := v.client.ContainerCreate(ctx, &cfg, &hostcfg, &netcfg, "")
+	if err != nil {
+		return "", err
+	}
+	for _, warn := range created.Warnings {
+		activity.Record(ctx, "Warning creating %v: %v", created.ID, warn)
+	}
+	return created.ID, nil
+}
+
+func (v *volume) VolumeList(ctx context.Context) (vol.DirMap, error) {
 	// Create a container that mounts a volume and inspects it. Run it and capture the output.
 	cid, err := v.createContainer(ctx, vol.StatCmd(mountpoint))
 	if err != nil {
@@ -105,84 +130,46 @@ func (v *volume) List(ctx context.Context) ([]plugin.Entry, error) {
 		return nil, errors.New(string(bytes))
 	}
 
-	dirs, err := vol.StatParseAll(output, mountpoint)
+	return vol.StatParseAll(output, mountpoint)
+}
+
+func (v *volume) VolumeRead(ctx context.Context, path string) (plugin.SizedReader, error) {
+	// Create a container that mounts a volume and waits. Use it to download a file.
+	cid, err := v.createContainer(ctx, []string{"sleep", "60"})
 	if err != nil {
 		return nil, err
 	}
+	defer func() {
+		err := v.client.ContainerRemove(ctx, cid, types.ContainerRemoveOptions{})
+		activity.Record(ctx, "Deleted temporary container %v: %v", cid, err)
+	}()
 
-	root := dirs[""]
-	entries := make([]plugin.Entry, 0, len(root))
-	for name, attr := range root {
-		if attr.Mode().IsDir() {
-			entries = append(entries, vol.NewDir(name, attr, v.getContentCB(), "/"+name, dirs))
-		} else {
-			entries = append(entries, vol.NewFile(name, attr, v.getContentCB(), "/"+name))
-		}
+	activity.Record(ctx, "Starting container %v", cid)
+	if err := v.client.ContainerStart(ctx, cid, types.ContainerStartOptions{}); err != nil {
+		return nil, err
 	}
-	return entries, nil
-}
+	defer func() {
+		err := v.client.ContainerKill(ctx, cid, "SIGKILL")
+		activity.Record(ctx, "Stopped temporary container %v: %v", cid, err)
+	}()
 
-// Create a container that mounts a volume to a default mountpoint and runs a command.
-func (v *volume) createContainer(ctx context.Context, cmd []string) (string, error) {
-	// Use tty to avoid messing with the extra log formatting.
-	cfg := docontainer.Config{Image: "busybox", Cmd: cmd, Tty: true}
-	mounts := []mount.Mount{{
-		Type:     mount.TypeVolume,
-		Source:   v.Name(),
-		Target:   mountpoint,
-		ReadOnly: true,
-	}}
-	hostcfg := docontainer.HostConfig{Mounts: mounts}
-	netcfg := network.NetworkingConfig{}
-	created, err := v.client.ContainerCreate(ctx, &cfg, &hostcfg, &netcfg, "")
+	// Download file, then kill container.
+	rdr, _, err := v.client.CopyFromContainer(ctx, cid, mountpoint+path)
 	if err != nil {
-		return "", err
-	}
-	for _, warn := range created.Warnings {
-		activity.Record(ctx, "Warning creating %v: %v", created.ID, warn)
+		return nil, err
 	}
-	return created.ID, nil
-}
-
-func (v *volume) getContentCB() vol.ContentCB {
-	return func(ctx context.Context, path string) (plugin.SizedReader, error) {
-		// Create a container that mounts a volume and waits. Use it to download a file.
-		cid, err := v.createContainer(ctx, []string{"sleep", "60"})
-		if err != nil {
-			return nil, err
-		}
-		defer func() {
-			err := v.client.ContainerRemove(ctx, cid, types.ContainerRemoveOptions{})
-			activity.Record(ctx, "Deleted temporary container %v: %v", cid, err)
-		}()
-
-		activity.Record(ctx, "Starting container %v", cid)
-		if err := v.client.ContainerStart(ctx, cid, types.ContainerStartOptions{}); err != nil {
-			return nil, err
-		}
-		defer func() {
-			err := v.client.ContainerKill(ctx, cid, "SIGKILL")
-			activity.Record(ctx, "Stopped temporary container %v: %v", cid, err)
-		}()
-
-		// Download file, then kill container.
-		rdr, _, err := v.client.CopyFromContainer(ctx, cid, mountpoint+path)
-		if err != nil {
-			return nil, err
-		}
-		defer func() {
-			activity.Record(ctx, "Closed file %v on %v: %v", mountpoint+path, cid, rdr.Close())
-		}()
+	defer func() {
+		activity.Record(ctx, "Closed file %v on %v: %v", mountpoint+path, cid, rdr.Close())
+	}()
 
-		tarReader := tar.NewReader(rdr)
-		// Only expect one file.
-		if _, err := tarReader.Next(); err != nil {
-			return nil, err
-		}
-		bits, err := ioutil.ReadAll(tarReader)
-		if err != nil {
-			return nil, err
-		}
-		return bytes.NewReader(bits), nil
+	tarReader := tar.NewReader(rdr)
+	// Only expect one file.
+	if _, err := tarReader.Next(); err != nil {
+		return nil, err
+	}
+	bits, err := ioutil.ReadAll(tarReader)
+	if err != nil {
+		return nil, err
 	}
+	return bytes.NewReader(bits), nil
 }
diff --git a/plugin/kubernetes/pvc.go b/plugin/kubernetes/pvc.go
index a309ec2..1330539 100644
--- a/plugin/kubernetes/pvc.go
+++ b/plugin/kubernetes/pvc.go
@@ -33,7 +33,6 @@ func newPVC(pi typedv1.PersistentVolumeClaimInterface, pd typedv1.PodInterface,
 		pvci:      pi,
 		podi:      pd,
 	}
-	vol.SetTTLOf(plugin.ListOp, 60*time.Second)
 
 	attr := plugin.EntryAttributes{}
 	attr.
@@ -56,54 +55,7 @@ func (v *pvc) Metadata(ctx context.Context) (plugin.EntryMetadata, error) {
 }
 
 func (v *pvc) List(ctx context.Context) ([]plugin.Entry, error) {
-	// Create a container that mounts a pvc and inspects it. Run it and capture the output.
-	pid, err := v.createPod(volume.StatCmd(mountpoint))
-	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 to start", 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
-	}
-
-	activity.Record(ctx, "Gathering log for %v", pid)
-	output, lerr := v.podi.GetLogs(pid, &corev1.PodLogOptions{}).Stream()
-	if lerr != nil {
-		return nil, lerr
-	}
-	defer func() {
-		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
-	}()
-
-	if err == errPodTerminated {
-		bytes, err := ioutil.ReadAll(output)
-		if err != nil {
-			return nil, err
-		}
-		return nil, errors.New(string(bytes))
-	}
-
-	dirs, err := volume.StatParseAll(output, mountpoint)
-	if err != nil {
-		return nil, err
-	}
-	activity.Record(ctx, "Files found in persistent volume claim %v: %+v", v.Name(), dirs)
-
-	root := dirs[""]
-	entries := make([]plugin.Entry, 0, len(root))
-	for name, attr := range root {
-		if attr.Mode().IsDir() {
-			entries = append(entries, volume.NewDir(name, attr, v.getContentCB(), "/"+name, dirs))
-		} else {
-			entries = append(entries, volume.NewFile(name, attr, v.getContentCB(), "/"+name))
-		}
-	}
-	return entries, nil
+	return volume.List(ctx, v, "")
 }
 
 // Create a container that mounts a pvc to a default mountpoint and runs a command.
@@ -182,43 +134,77 @@ func (v *pvc) waitForPod(ctx context.Context, pid string) error {
 	}
 }
 
-func (v *pvc) getContentCB() volume.ContentCB {
-	return func(ctx context.Context, path string) (plugin.SizedReader, error) {
-		// Create a container that mounts a pvc and waits. Use it to download a 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{}))
-		}()
+func (v *pvc) VolumeList(ctx context.Context) (volume.DirMap, error) {
+	// Create a container that mounts a pvc and inspects it. Run it and capture the output.
+	pid, err := v.createPod(volume.StatCmd(mountpoint))
+	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, "Waiting for pod %v to start", 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
+	}
 
-		activity.Record(ctx, "Gathering log for %v", pid)
-		output, err := v.podi.GetLogs(pid, &corev1.PodLogOptions{}).Stream()
-		if err != nil {
-			return nil, err
-		}
-		defer func() {
-			activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
-		}()
+	activity.Record(ctx, "Gathering log for %v", pid)
+	output, lerr := v.podi.GetLogs(pid, &corev1.PodLogOptions{}).Stream()
+	if lerr != nil {
+		return nil, lerr
+	}
+	defer func() {
+		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
+	}()
 
-		bits, err := ioutil.ReadAll(output)
+	if err == errPodTerminated {
+		bytes, err := ioutil.ReadAll(output)
 		if err != nil {
 			return nil, err
 		}
-		activity.Record(ctx, "Read: %v", bits)
+		return nil, errors.New(string(bytes))
+	}
 
-		if podErr == errPodTerminated {
-			return nil, errors.New(string(bits))
-		}
-		return bytes.NewReader(bits), nil
+	return volume.StatParseAll(output, mountpoint)
+}
+
+func (v *pvc) VolumeRead(ctx context.Context, path string) (plugin.SizedReader, error) {
+	// Create a container that mounts a pvc and waits. Use it to download a 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()
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
+	}()
+
+	bits, err := ioutil.ReadAll(output)
+	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
 }
diff --git a/volume/helpers.go b/volume/core.go
similarity index 73%
rename from volume/helpers.go
rename to volume/core.go
index 8167e6c..5b284f1 100644
--- a/volume/helpers.go
+++ b/volume/core.go
@@ -8,6 +8,7 @@ package volume
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"io"
 	"path"
@@ -19,6 +20,16 @@ import (
 	"github.com/puppetlabs/wash/plugin"
 )
 
+// Interface presents methods to access the volume.
+type Interface interface {
+	plugin.Entry
+
+	// Returns a map of volume nodes to their stats, such as that returned by StatParseAll.
+	VolumeList(context.Context) (DirMap, error)
+	// Accepts a path and returns the content associated with that path.
+	VolumeRead(context.Context, string) (plugin.SizedReader, error)
+}
+
 // StatCmd returns the command required to stat all the files in a directory.
 func StatCmd(path string) []string {
 	// size, atime, mtime, ctime, mode, name
@@ -28,10 +39,10 @@ func StatCmd(path string) []string {
 	// %Z - Time of last status change as seconds since Epoch
 	// %f - Raw mode in hex
 	// %n - File name
-	return []string{"find", path, "-mindepth", "1", "-exec", "stat", "-c", "%s %X %Y %Z %f %n", "{}", ";"}
+	return []string{"find", path, "-mindepth", "1", "-exec", "stat", "-c", "%s %X %Y %Z %f %n", "{}", "+"}
 }
 
-// TODO: Should move this over to plugin as a helper at some point
+// Keep as its own specialized function as it will be faster than munge.ToTime.
 func parseTime(t string) (time.Time, error) {
 	epoch, err := strconv.ParseInt(t, 10, 64)
 	if err != nil {
@@ -117,3 +128,25 @@ func StatParseAll(output io.Reader, base string) (DirMap, error) {
 	}
 	return attrs, nil
 }
+
+// List constructs an array of entries for the given path from a DirMap.
+// The root path is an empty string.
+func List(ctx context.Context, impl Interface, path string) ([]plugin.Entry, error) {
+	result, err := plugin.CachedOp(ctx, "VolumeListCB", impl, 30*time.Second, func() (interface{}, error) {
+		return impl.VolumeList(ctx)
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	root := result.(DirMap)[path]
+	entries := make([]plugin.Entry, 0, len(root))
+	for name, attr := range root {
+		if attr.Mode().IsDir() {
+			entries = append(entries, newDir(name, attr, impl, path+"/"+name))
+		} else {
+			entries = append(entries, newFile(name, attr, impl.VolumeRead, path+"/"+name))
+		}
+	}
+	return entries, nil
+}
diff --git a/volume/helpers_test.go b/volume/core_test.go
similarity index 100%
rename from volume/helpers_test.go
rename to volume/core_test.go
diff --git a/volume/dir.go b/volume/dir.go
index c8739c0..50cd994 100644
--- a/volume/dir.go
+++ b/volume/dir.go
@@ -7,44 +7,29 @@ import (
 	"github.com/puppetlabs/wash/plugin"
 )
 
-// Dir represents a directory in a volume. It retains access to a map of directories
-// to their children and attribute data to populate subdirectories.
-//
-// TODO: by tying DirMap to the Node, it doesn't refresh until something triggers listing the resource
-// that produced the DirMap. Need to explore making this a callback that retrieves fresh data. How
-// should that interact with caching? We pass in DirMap so that we only have to retrieve it once
-// when constructing an entire hierarchy.
+// Dir represents a directory in a volume. It populates a subtree with listcb as needed.
 type Dir struct {
 	plugin.EntryBase
-	contentcb ContentCB
-	path      string
-	dirs      DirMap
+	impl Interface
+	path string
 }
 
-// NewDir creates a Dir populated from dirs.
-func NewDir(name string, attr plugin.EntryAttributes, cb ContentCB, path string, dirs DirMap) *Dir {
+// newDir creates a Dir populated from dirs.
+func newDir(name string, attr plugin.EntryAttributes, impl Interface, path string) *Dir {
 	vd := &Dir{
 		EntryBase: plugin.NewEntry(name),
-		contentcb: cb,
+		impl:      impl,
 		path:      path,
-		dirs:      dirs,
 	}
 	vd.SetAttributes(attr)
 	vd.SetTTLOf(plugin.OpenOp, 60*time.Second)
+	// Caching handled in List based on 'impl'.
+	vd.DisableCachingFor(plugin.ListOp)
 
 	return vd
 }
 
 // List lists the children of the directory.
 func (v *Dir) List(ctx context.Context) ([]plugin.Entry, error) {
-	root := v.dirs[v.path]
-	entries := make([]plugin.Entry, 0, len(root))
-	for name, attr := range root {
-		if attr.Mode().IsDir() {
-			entries = append(entries, NewDir(name, attr, v.contentcb, v.path+"/"+name, v.dirs))
-		} else {
-			entries = append(entries, NewFile(name, attr, v.contentcb, v.path+"/"+name))
-		}
-	}
-	return entries, nil
+	return List(ctx, v.impl, v.path)
 }
diff --git a/volume/file.go b/volume/file.go
index f063701..d164df9 100644
--- a/volume/file.go
+++ b/volume/file.go
@@ -7,18 +7,18 @@ import (
 	"github.com/puppetlabs/wash/plugin"
 )
 
-// ContentCB accepts a path and returns the content associated with that path.
-type ContentCB = func(context.Context, string) (plugin.SizedReader, error)
+// contentCB accepts a path and returns the content associated with that path.
+type contentCB = func(context.Context, string) (plugin.SizedReader, error)
 
 // File represents a file in a volume that has content we can access.
 type File struct {
 	plugin.EntryBase
-	contentcb ContentCB
+	contentcb contentCB
 	path      string
 }
 
-// NewFile creates a VolumeFile.
-func NewFile(name string, attr plugin.EntryAttributes, cb ContentCB, path string) *File {
+// newFile creates a VolumeFile.
+func newFile(name string, attr plugin.EntryAttributes, cb contentCB, path string) *File {
 	vf := &File{
 		EntryBase: plugin.NewEntry(name),
 		contentcb: cb,


Test patch:
diff --git a/api/cache_test.go b/api/cache_test.go
index 8131c16..21bf949 100644
--- a/api/cache_test.go
+++ b/api/cache_test.go
@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
 	"regexp"
@@ -10,6 +11,7 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
+	apitypes "github.com/puppetlabs/wash/api/types"
 	"github.com/puppetlabs/wash/plugin"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
@@ -86,13 +88,15 @@ func (suite *CacheHandlerTestSuite) TestClearCache() {
 	group.SetTestID("/dir")
 	group.On("List", mock.Anything).Return([]plugin.Entry{}, nil)
 
+	reqCtx := context.WithValue(context.Background(), mountpointKey, "/")
+
 	expectedChildren := make(map[string]plugin.Entry)
-	if children, err := plugin.CachedList(context.Background(), group); suite.Nil(err) {
+	if children, err := plugin.CachedList(reqCtx, group); suite.Nil(err) {
 		suite.Equal(expectedChildren, children)
 	}
 
 	// Test clearing a different cache
-	req := httptest.NewRequest(http.MethodDelete, "http://example.com/cache?path=file", nil)
+	req := httptest.NewRequest(http.MethodDelete, "http://example.com/cache?path=/file", nil).WithContext(reqCtx)
 	w := httptest.NewRecorder()
 	suite.router.ServeHTTP(w, req)
 	suite.Equal(http.StatusOK, w.Code)
@@ -103,7 +107,7 @@ func (suite *CacheHandlerTestSuite) TestClearCache() {
 	}
 
 	// Test clearing the cache
-	req = httptest.NewRequest(http.MethodDelete, "http://example.com/cache?path=dir", nil)
+	req = httptest.NewRequest(http.MethodDelete, "http://example.com/cache?path=/dir", nil).WithContext(reqCtx)
 	w = httptest.NewRecorder()
 	suite.router.ServeHTTP(w, req)
 	suite.Equal(http.StatusOK, w.Code)
@@ -116,6 +120,27 @@ func (suite *CacheHandlerTestSuite) TestClearCache() {
 	group.AssertNumberOfCalls(suite.T(), "List", 2)
 }
 
+func (suite *CacheHandlerTestSuite) TestClearCacheErrors() {
+	reqCtx := context.WithValue(context.Background(), mountpointKey, "/mnt")
+
+	// Test clearing cache by a relative path
+	req := httptest.NewRequest(http.MethodDelete, "http://example.com/cache?path=mnt/file", nil).WithContext(reqCtx)
+	w := httptest.NewRecorder()
+	suite.router.ServeHTTP(w, req)
+	suite.Equal(http.StatusBadRequest, w.Code)
+	var errResp apitypes.ErrorObj
+	suite.NoError(json.Unmarshal(w.Body.Bytes(), &errResp))
+	suite.Equal(apitypes.RelativePath, errResp.Kind)
+
+	// Test clearing cache outside the mountpoint
+	req = httptest.NewRequest(http.MethodDelete, "http://example.com/cache?path=/a/file", nil).WithContext(reqCtx)
+	w = httptest.NewRecorder()
+	suite.router.ServeHTTP(w, req)
+	suite.Equal(http.StatusBadRequest, w.Code)
+	suite.NoError(json.Unmarshal(w.Body.Bytes(), &errResp))
+	suite.Equal(apitypes.NonWashPath, errResp.Kind)
+}
+
 func TestCacheHandler(t *testing.T) {
 	suite.Run(t, new(CacheHandlerTestSuite))
 }
diff --git a/api/helpers_test.go b/api/helpers_test.go
index b5c5b19..a2ed7a9 100644
--- a/api/helpers_test.go
+++ b/api/helpers_test.go
@@ -107,8 +107,8 @@ func (m *mockRoot) List(ctx context.Context) ([]plugin.Entry, error) {
 	return args.Get(0).([]plugin.Entry), args.Error(1)
 }
 
-func getRequest(path string) *http.Request {
-	return &http.Request{URL: &url.URL{RawQuery: url.Values{"path": []string{path}}.Encode()}}
+func getRequest(ctx context.Context, path string) *http.Request {
+	return (&http.Request{URL: &url.URL{RawQuery: url.Values{"path": []string{path}}.Encode()}}).WithContext(ctx)
 }
 
 func (suite *HelpersTestSuite) TestGetEntryFromPath() {
@@ -121,37 +121,37 @@ func (suite *HelpersTestSuite) TestGetEntryFromPath() {
 	mountpoint := "/mountpoint"
 	ctx = context.WithValue(ctx, mountpointKey, mountpoint)
 
-	_, _, err := getEntryFromRequest(ctx, getRequest("relative"))
+	_, _, err := getEntryFromRequest(getRequest(ctx, "relative"))
 	suite.Error(relativePathResponse("relative"), err)
 
 	// TODO: Add tests for non-Wash entries (i.e. for apifs)
 
-	entry, path, err := getEntryFromRequest(ctx, getRequest(mountpoint))
+	entry, path, err := getEntryFromRequest(getRequest(ctx, mountpoint))
 	if suite.Nil(err) {
 		suite.Equal(mountpoint, path)
 		suite.Equal(reg.Name(), plugin.Name(entry))
 	}
 
-	entry, path, err = getEntryFromRequest(ctx, getRequest(mountpoint+"/"))
+	entry, path, err = getEntryFromRequest(getRequest(ctx, mountpoint+"/"))
 	if suite.Nil(err) {
 		suite.Equal(mountpoint+"/", path)
 		suite.Equal(reg.Name(), plugin.Name(entry))
 	}
 
-	entry, path, err = getEntryFromRequest(ctx, getRequest(mountpoint+"/mine"))
+	entry, path, err = getEntryFromRequest(getRequest(ctx, mountpoint+"/mine"))
 	if suite.Nil(err) {
 		suite.Equal(mountpoint+"/mine", path)
 		suite.Equal(plug.Name(), plugin.Name(entry))
 	}
 
-	_, _, err = getEntryFromRequest(ctx, getRequest(mountpoint+"/yours"))
+	_, _, err = getEntryFromRequest(getRequest(ctx, mountpoint+"/yours"))
 	suite.Error(err)
 
 	file := plugin.NewEntry("a file")
 	file.SetTestID("/mine/a file")
 	plug.On("List", mock.Anything).Return([]plugin.Entry{&file}, nil)
 
-	entry, path, err = getEntryFromRequest(ctx, getRequest(mountpoint+"/mine/a file"))
+	entry, path, err = getEntryFromRequest(getRequest(ctx, mountpoint+"/mine/a file"))
 	if suite.Nil(err) {
 		suite.Equal(mountpoint+"/mine/a file", path)
 		suite.Equal(file.Name(), plugin.Name(entry))
@@ -159,7 +159,7 @@ func (suite *HelpersTestSuite) TestGetEntryFromPath() {
 	plug.AssertExpectations(suite.T())
 
 	plug.On("List", mock.Anything).Return([]plugin.Entry{&file}, nil)
-	_, _, err = getEntryFromRequest(ctx, getRequest(mountpoint+"/mine/a dir"))
+	_, _, err = getEntryFromRequest(getRequest(ctx, mountpoint+"/mine/a dir"))
 	suite.Error(err)
 	plug.AssertExpectations(suite.T())
 }
diff --git a/volume/dir_test.go b/volume/dir_test.go
index f026d4f..bd542cc 100644
--- a/volume/dir_test.go
+++ b/volume/dir_test.go
@@ -6,24 +6,39 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/puppetlabs/wash/datastore"
 	"github.com/puppetlabs/wash/plugin"
 	"github.com/stretchr/testify/assert"
 )
 
+type mockEntry struct {
+	plugin.EntryBase
+	dmap DirMap
+}
+
+func (m *mockEntry) VolumeList(context.Context) (DirMap, error) {
+	return m.dmap, nil
+}
+
+func (m *mockEntry) VolumeRead(context.Context, string) (plugin.SizedReader, error) {
+	return nil, nil
+}
+
 func TestVolumeDir(t *testing.T) {
 	dmap, err := StatParseAll(strings.NewReader(fixture), mountpoint)
 	assert.Nil(t, err)
-	contentcb := func(ctx context.Context, path string) (plugin.SizedReader, error) {
-		return nil, nil
-	}
+
+	plugin.SetTestCache(datastore.NewMemCache())
+	entry := mockEntry{EntryBase: plugin.NewEntry("mine"), dmap: dmap}
+	entry.SetTestID("/mine")
 
 	assert.NotNil(t, dmap[""]["path"])
-	vd := NewDir("path", dmap[""]["path"], contentcb, "/path", dmap)
+	vd := newDir("path", dmap[""]["path"], &entry, "/path")
 	attr := plugin.Attributes(vd)
 	assert.Equal(t, 0755|os.ModeDir, attr.Mode())
 
 	assert.NotNil(t, dmap[""]["path1"])
-	vd = NewDir("path", dmap[""]["path1"], contentcb, "/path1", dmap)
+	vd = newDir("path", dmap[""]["path1"], &entry, "/path1")
 	entries, err := vd.List(context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(entries))
@@ -33,7 +48,7 @@ func TestVolumeDir(t *testing.T) {
 	}
 
 	assert.NotNil(t, dmap[""]["path2"])
-	vd = NewDir("path", dmap[""]["path2"], contentcb, "/path2", dmap)
+	vd = newDir("path", dmap[""]["path2"], &entry, "/path2")
 	entries, err = vd.List(context.Background())
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(entries))
@@ -41,4 +56,6 @@ func TestVolumeDir(t *testing.T) {
 	if entry, ok := entries[0].(*Dir); assert.Equal(t, true, ok) {
 		assert.Equal(t, "/path2/dir", entry.path)
 	}
+
+	plugin.UnsetTestCache()
 }
diff --git a/volume/file_test.go b/volume/file_test.go
index c33a40b..0ce1177 100644
--- a/volume/file_test.go
+++ b/volume/file_test.go
@@ -15,7 +15,7 @@ func TestVolumeFile(t *testing.T) {
 	now := time.Now()
 	initialAttr := plugin.EntryAttributes{}
 	initialAttr.SetCtime(now)
-	vf := NewFile("mine", initialAttr, func(ctx context.Context, path string) (plugin.SizedReader, error) {
+	vf := newFile("mine", initialAttr, func(ctx context.Context, path string) (plugin.SizedReader, error) {
 		assert.Equal(t, "my path", path)
 		return strings.NewReader("hello"), nil
 	}, "my path")
@@ -36,7 +36,7 @@ func TestVolumeFile(t *testing.T) {
 }
 
 func TestVolumeFileErr(t *testing.T) {
-	vf := NewFile("mine", plugin.EntryAttributes{}, func(ctx context.Context, path string) (plugin.SizedReader, error) {
+	vf := newFile("mine", plugin.EntryAttributes{}, func(ctx context.Context, path string) (plugin.SizedReader, error) {
 		return nil, errors.New("fail")
 	}, "my path")