Repository: puppetlabs/wash (go)

Issue:
Stream logs from VMs
Expanding on #216, we should be able to stream logs from VMs rather than just open them.

Implement the Streamable interface for volumes, probably with tail -f as the backend. Tricky part is (maybe) stopping the command on the target.

Pull request description:
Stream logs from VMs
Add Stream support to the volume package. Add Tty support to Exec on Docker and Kubernetes to support correctly cleaning up processes after terminating streaming.

Resolves #227.

Patch:
diff --git a/api/exec.go b/api/exec.go
index 044074c..9fc32ad 100644
--- a/api/exec.go
+++ b/api/exec.go
@@ -29,19 +29,15 @@ func sendPacket(ctx context.Context, w *json.Encoder, p *apitypes.ExecPacket) {
 	}
 }
 
-var outputStreamNames = [2]string{apitypes.Stdout, apitypes.Stderr}
-
 func streamOutput(ctx context.Context, w *json.Encoder, outputCh <-chan plugin.ExecOutputChunk) {
 	if outputCh == nil {
 		return
 	}
 
 	for chunk := range outputCh {
-		stream := outputStreamNames[chunk.StreamID]
-
-		packet := apitypes.ExecPacket{TypeField: stream, Timestamp: chunk.Timestamp}
+		packet := apitypes.ExecPacket{TypeField: chunk.StreamID, Timestamp: chunk.Timestamp}
 		if err := chunk.Err; err != nil {
-			packet.Err = newStreamingErrorObj(stream, err.Error())
+			packet.Err = newStreamingErrorObj(chunk.StreamID, err.Error())
 		} else {
 			packet.Data = chunk.Data
 		}
diff --git a/api/stream.go b/api/stream.go
index 79d0596..9ccb6b1 100644
--- a/api/stream.go
+++ b/api/stream.go
@@ -57,13 +57,12 @@ var streamHandler handler = func(w http.ResponseWriter, r *http.Request) *errorR
 	wf := &streamableResponseWriter{f}
 	f.Flush()
 
-	if closer, ok := rdr.(io.Closer); ok {
-		// If a ReadCloser, ensure it's closed when the context is cancelled.
-		go func() {
-			<-r.Context().Done()
-			activity.Record(ctx, "API: Stream %v closed by completed context: %v", path, closer.Close())
-		}()
-	}
+	// Ensure it's closed when the context is cancelled.
+	go func() {
+		<-r.Context().Done()
+		activity.Record(ctx, "API: Stream %v closed by completed context: %v", path, rdr.Close())
+	}()
+
 	if _, err := io.Copy(wf, rdr); err != nil {
 		// Common for copy to error when the caller closes the connection.
 		log.Debugf("Errored streaming response for entry %v: %v", path, err)
diff --git a/api/types/exec.go b/api/types/exec.go
index 06d69ae..58e44c0 100644
--- a/api/types/exec.go
+++ b/api/types/exec.go
@@ -1,6 +1,7 @@
 package apitypes
 
 import (
+	"github.com/puppetlabs/wash/plugin"
 	"time"
 )
 
@@ -30,14 +31,11 @@ type ExecBody struct {
 	Opts ExecOptions `json:"opts"`
 }
 
-// ExecPacketType identifies the packet type.
-type ExecPacketType = string
-
-// Enumerates packet types.
+// Enumerates packet types used by the API.
 const (
-	Stdout   ExecPacketType = "stdout"
-	Stderr   ExecPacketType = "stderr"
-	Exitcode ExecPacketType = "exitcode"
+	Stdout   plugin.ExecPacketType = plugin.Stdout
+	Stderr   plugin.ExecPacketType = plugin.Stderr
+	Exitcode plugin.ExecPacketType = "exitcode"
 )
 
 // ExecPacket is a single packet of results from an exec.
@@ -46,8 +44,8 @@ const (
 //
 // swagger:response
 type ExecPacket struct {
-	TypeField ExecPacketType `json:"type"`
-	Timestamp time.Time      `json:"timestamp"`
-	Data      interface{}    `json:"data"`
-	Err       *ErrorObj      `json:"error"`
+	TypeField plugin.ExecPacketType `json:"type"`
+	Timestamp time.Time             `json:"timestamp"`
+	Data      interface{}           `json:"data"`
+	Err       *ErrorObj             `json:"error"`
 }
diff --git a/plugin/aws/s3Object.go b/plugin/aws/s3Object.go
index 8568c0d..b960f14 100644
--- a/plugin/aws/s3Object.go
+++ b/plugin/aws/s3Object.go
@@ -95,7 +95,7 @@ func (o *s3Object) Open(ctx context.Context) (plugin.SizedReader, error) {
 	return &s3ObjectReader{o: o}, nil
 }
 
-func (o *s3Object) Stream(context.Context) (io.Reader, error) {
+func (o *s3Object) Stream(context.Context) (io.ReadCloser, error) {
 	return o.fetchContent(0)
 }
 
diff --git a/plugin/cleanupReader.go b/plugin/cleanupReader.go
new file mode 100644
index 0000000..b38a78f
--- /dev/null
+++ b/plugin/cleanupReader.go
@@ -0,0 +1,16 @@
+package plugin
+
+import "io"
+
+// CleanupReader is a wrapper for an io.ReadCloser that performs cleanup when closed.
+type CleanupReader struct {
+	io.ReadCloser
+	Cleanup func()
+}
+
+// Close closes the reader it wraps, then calls the Cleanup function and returns any errors.
+func (c CleanupReader) Close() error {
+	err := c.ReadCloser.Close()
+	c.Cleanup()
+	return err
+}
diff --git a/plugin/docker/container-log.go b/plugin/docker/container-log.go
index 6bc9ea4..ae436c9 100644
--- a/plugin/docker/container-log.go
+++ b/plugin/docker/container-log.go
@@ -52,7 +52,7 @@ func (clf *containerLogFile) Open(ctx context.Context) (plugin.SizedReader, erro
 	return bytes.NewReader(buf.Bytes()), nil
 }
 
-func (clf *containerLogFile) Stream(ctx context.Context) (io.Reader, error) {
+func (clf *containerLogFile) Stream(ctx context.Context) (io.ReadCloser, error) {
 	opts := types.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true, Tail: "10"}
 	rdr, err := clf.client.ContainerLogs(ctx, clf.containerName, opts)
 	if err != nil {
diff --git a/plugin/docker/container.go b/plugin/docker/container.go
index 0a1793a..6c5336b 100644
--- a/plugin/docker/container.go
+++ b/plugin/docker/container.go
@@ -60,8 +60,10 @@ func (c *container) Exec(ctx context.Context, cmd string, args []string, opts pl
 	execResult := plugin.ExecResult{}
 
 	command := append([]string{cmd}, args...)
-	cfg := types.ExecConfig{Cmd: command, AttachStdout: true, AttachStderr: true}
-	if opts.Stdin != nil {
+	activity.Record(ctx, "Exec %v on %v", command, c.Name())
+
+	cfg := types.ExecConfig{Cmd: command, AttachStdout: true, AttachStderr: true, Tty: opts.Tty}
+	if opts.Stdin != nil || opts.Tty {
 		cfg.AttachStdin = true
 	}
 	created, err := c.client.ContainerExecCreate(ctx, c.id, cfg)
@@ -76,7 +78,20 @@ func (c *container) Exec(ctx context.Context, cmd string, args []string, opts pl
 
 	outputCh, stdout, stderr := plugin.CreateExecOutputStreams(ctx)
 	go func() {
-		defer resp.Close()
+		go func() {
+			// Close the response on context cancellation. Copying will block until there's more to read
+			// from the exec output. For an action with no more output it may never return.
+			// Asynchronously watch for context cancellation, and when it happens close the response so
+			// the parent goroutine can complete. We expect all contexts to complete eventually.
+			<-ctx.Done()
+
+			if opts.Tty {
+				// If resp.Conn is still open, Send Ctrl-C over resp.Conn before closing it.
+				_, err := resp.Conn.Write([]byte{0x03})
+				activity.Record(ctx, "Sent ETX on context termination: %v", err)
+			}
+			resp.Close()
+		}()
 
 		_, err := stdcopy.StdCopy(stdout, stderr, resp.Reader)
 		activity.Record(ctx, "Exec on %v complete: %v", c.Name(), err)
@@ -88,8 +103,13 @@ func (c *container) Exec(ctx context.Context, cmd string, args []string, opts pl
 	if opts.Stdin != nil {
 		go func() {
 			_, writeErr = io.Copy(resp.Conn, opts.Stdin)
-			respErr := resp.CloseWrite()
-			activity.Record(ctx, "Closed execution input stream for %v: %v, %v", c.Name(), writeErr, respErr)
+			// If using a Tty, wait until done reading in case we need to send Ctrl-C; when a Tty is
+			// allocated commands expect user input and will respond to control signals. Otherwise close
+			// input now to ensure commands that depend on EOF execute correctly.
+			if !opts.Tty {
+				respErr := resp.CloseWrite()
+				activity.Record(ctx, "Closed execution input stream for %v: %v, %v", c.Name(), writeErr, respErr)
+			}
 		}()
 	}
 
diff --git a/plugin/docker/volume.go b/plugin/docker/volume.go
index d8319e6..887a6a7 100644
--- a/plugin/docker/volume.go
+++ b/plugin/docker/volume.go
@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"io"
 	"io/ioutil"
 	"time"
 
@@ -173,3 +174,57 @@ func (v *volume) VolumeOpen(ctx context.Context, path string) (plugin.SizedReade
 	}
 	return bytes.NewReader(bits), nil
 }
+
+func (v *volume) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
+	// Create a container that mounts a volume and tails a file. Run it and capture the output.
+	cid, err := v.createContainer(ctx, []string{"tail", "-f", mountpoint + path})
+	if err != nil {
+		return nil, err
+	}
+
+	// Manually use this in case of errors. On success, the returned Closer will need to call instead.
+	delete := func() {
+		err := v.client.ContainerRemove(ctx, cid, types.ContainerRemoveOptions{})
+		activity.Record(ctx, "Deleted container %v: %v", cid, err)
+	}
+
+	activity.Record(ctx, "Starting container %v", cid)
+	if err := v.client.ContainerStart(ctx, cid, types.ContainerStartOptions{}); err != nil {
+		delete()
+		return nil, err
+	}
+
+	// Manually use this in case of errors. On success, the returned Closer will need to call instead.
+	killAndDelete := func() {
+		err := v.client.ContainerKill(ctx, cid, "SIGKILL")
+		activity.Record(ctx, "Stopped temporary container %v: %v", cid, err)
+		delete()
+	}
+
+	activity.Record(ctx, "Waiting for container %v", cid)
+	waitC, errC := v.client.ContainerWait(ctx, cid, docontainer.WaitConditionNotRunning)
+	var statusCode int64
+	select {
+	case err := <-errC:
+		killAndDelete()
+		return nil, err
+	case result := <-waitC:
+		statusCode = result.StatusCode
+		activity.Record(ctx, "Container %v finished[%v]: %v", cid, result.StatusCode, result.Error)
+	}
+
+	opts := types.ContainerLogsOptions{ShowStdout: true}
+	if statusCode != 0 {
+		opts.ShowStderr = true
+	}
+
+	activity.Record(ctx, "Gathering log for %v", cid)
+	output, err := v.client.ContainerLogs(ctx, cid, opts)
+	if err != nil {
+		killAndDelete()
+		return nil, err
+	}
+
+	// Wrap the log output in a ReadCloser that stops and kills the container on Close.
+	return plugin.CleanupReader{ReadCloser: output, Cleanup: killAndDelete}, nil
+}
diff --git a/plugin/externalPluginEntry.go b/plugin/externalPluginEntry.go
index 54fa6c3..897d104 100644
--- a/plugin/externalPluginEntry.go
+++ b/plugin/externalPluginEntry.go
@@ -179,7 +179,7 @@ func (s *stdoutStreamer) Close() error {
 }
 
 // Stream streams the entry's content
-func (e *externalPluginEntry) Stream(ctx context.Context) (io.Reader, error) {
+func (e *externalPluginEntry) Stream(ctx context.Context) (io.ReadCloser, error) {
 	cmd, stdoutR, stderrR, err := CreateCommand(
 		e.script.Path(),
 		"stream",
diff --git a/plugin/kubernetes/pod.go b/plugin/kubernetes/pod.go
index 2d28c0f..9cbb54f 100644
--- a/plugin/kubernetes/pod.go
+++ b/plugin/kubernetes/pod.go
@@ -113,7 +113,7 @@ func (p *pod) Open(ctx context.Context) (plugin.SizedReader, error) {
 	return bytes.NewReader(pdInfo.logContent), nil
 }
 
-func (p *pod) Stream(ctx context.Context) (io.Reader, error) {
+func (p *pod) Stream(ctx context.Context) (io.ReadCloser, error) {
 	var tailLines int64 = 10
 	req := p.client.CoreV1().Pods(p.ns).GetLogs(p.Name(), &corev1.PodLogOptions{Follow: true, TailLines: &tailLines})
 	return req.Stream()
@@ -147,7 +147,33 @@ func (p *pod) Exec(ctx context.Context, cmd string, args []string, opts plugin.E
 	outputCh, stdout, stderr := plugin.CreateExecOutputStreams(ctx)
 	exitcode := 0
 	go func() {
-		streamOpts := remotecommand.StreamOptions{Stdout: stdout, Stderr: stderr, Stdin: opts.Stdin}
+		stdin := opts.Stdin
+		if opts.Tty {
+			// If using a Tty, create an input stream that allows us to send Ctrl-C to end execution;
+			// when a Tty is allocated commands expect user input and will respond to control signals.
+			// Otherwise close input now to ensure commands that depend on EOF execute correctly.
+			r, w := io.Pipe()
+			if stdin != nil {
+				stdin = io.MultiReader(stdin, r)
+			} else {
+				stdin = r
+			}
+
+			go func() {
+				// Close the response on context cancellation. Copying will block until there's more to
+				// read from the exec output. For an action with no more output it may never return.
+				// Asynchronously watch for context cancellation, and when it happens close the response
+				// so the parent goroutine can complete. We expect all contexts to complete eventually.
+				<-ctx.Done()
+
+				// Append Ctrl-C to input to signal end of execution.
+				_, err := w.Write([]byte{0x03})
+				activity.Record(ctx, "Sent ETX on context termination: %v", err)
+				w.Close()
+			}()
+		}
+
+		streamOpts := remotecommand.StreamOptions{Stdout: stdout, Stderr: stderr, Stdin: stdin, Tty: opts.Tty}
 		err = executor.Stream(streamOpts)
 		activity.Record(ctx, "Exec on %v complete: %v", p.Name(), err)
 		if exerr, ok := err.(k8exec.ExitError); ok {
diff --git a/plugin/kubernetes/pvc.go b/plugin/kubernetes/pvc.go
index d2b18d8..147650c 100644
--- a/plugin/kubernetes/pvc.go
+++ b/plugin/kubernetes/pvc.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"time"
 
@@ -171,7 +172,7 @@ func (v *pvc) VolumeList(ctx context.Context) (volume.DirMap, error) {
 }
 
 func (v *pvc) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, error) {
-	// Create a container that mounts a pvc and waits. Use it to download a file.
+	// 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 {
@@ -208,3 +209,47 @@ func (v *pvc) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader,
 	}
 	return bytes.NewReader(bits), nil
 }
+
+func (v *pvc) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
+	// Create a container that mounts a pvc and tail the file.
+	pid, err := v.createPod([]string{"tail", "-f", mountpoint + path})
+	activity.Record(ctx, "Streaming from: %v", mountpoint+path)
+	if err != nil {
+		return nil, err
+	}
+
+	// Manually use this in case of errors. On success, the returned Closer will need to call instead.
+	delete := 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 {
+		delete()
+		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 {
+		delete()
+		return nil, err
+	}
+
+	if podErr == errPodTerminated {
+		bits, err := ioutil.ReadAll(output)
+		activity.Record(ctx, "Closed log for %v: %v", pid, output.Close())
+		delete()
+		if err != nil {
+			return nil, err
+		}
+		activity.Record(ctx, "Read: %v", bits)
+
+		return nil, errors.New(string(bits))
+	}
+
+	// Wrap the log output in a ReadCloser that stops and kills the container on Close.
+	return plugin.CleanupReader{ReadCloser: output, Cleanup: delete}, nil
+}
diff --git a/plugin/outputStream.go b/plugin/outputStream.go
index 3963c5b..b3eb847 100644
--- a/plugin/outputStream.go
+++ b/plugin/outputStream.go
@@ -6,18 +6,11 @@ import (
 	"time"
 )
 
-const (
-	// StdoutID represents Stdout
-	StdoutID = iota
-	// StderrID represents Stderr
-	StderrID
-)
-
 // OutputStream represents stdout/stderr.
 type OutputStream struct {
 	ctx        context.Context
 	sentCtxErr bool
-	id         int8
+	id         ExecPacketType
 	ch         chan ExecOutputChunk
 	closer     *multiCloser
 }
@@ -91,8 +84,8 @@ func CreateExecOutputStreams(ctx context.Context) (<-chan ExecOutputChunk, *Outp
 	outputCh := make(chan ExecOutputChunk)
 	closer := &multiCloser{ch: outputCh, countdown: 2}
 
-	stdout := &OutputStream{ctx: ctx, id: StdoutID, ch: outputCh, closer: closer}
-	stderr := &OutputStream{ctx: ctx, id: StderrID, ch: outputCh, closer: closer}
+	stdout := &OutputStream{ctx: ctx, id: Stdout, ch: outputCh, closer: closer}
+	stderr := &OutputStream{ctx: ctx, id: Stderr, ch: outputCh, closer: closer}
 
 	return outputCh, stdout, stderr
 }
diff --git a/plugin/types.go b/plugin/types.go
index c487231..b3266fa 100644
--- a/plugin/types.go
+++ b/plugin/types.go
@@ -63,15 +63,39 @@ type Root interface {
 }
 
 // ExecOptions is a struct we can add new features to that must be serializable to JSON.
-// Examples of potential features: user, privileged, tty, map of environment variables, string of stdin, timeout.
+// Examples of potential features: user, privileged, map of environment variables, timeout.
 type ExecOptions struct {
+	// Stdin can be used to pass a stream of input to write to stdin when executing the command.
 	Stdin io.Reader
+
+	// Tty instructs the executor to allocate a TTY (pseudo-terminal), which lets Wash communicate
+	// with the running process via its Stdin. The TTY is used to send a process termination signal
+	// (Ctrl+C) via Stdin when the passed-in Exec context is cancelled.
+	//
+	// NOTE TO PLUGIN AUTHORS: The Tty option is only relevant for executors that do not have an API
+	// endpoint to stop a running command (e.g. Docker, Kubernetes). If your executor does have an
+	// API endpoint to stop a running command, then ignore the Tty option. Note that the reason we
+	// make Tty an option instead of having the relevant executors always attach a TTY is because
+	// attaching a TTY can change the behavior of the command that's being executed.
+	//
+	// NOTE TO CALLERS: The Tty option is useful for executing your own stream-like commands (e.g.
+	// tail -f), because it ensures that there are no orphaned processes after the request is
+	// cancelled/finished.
+	Tty bool
 }
 
+// ExecPacketType identifies the packet type.
+type ExecPacketType = string
+
+// Enumerates packet types.
+const (
+	Stdout ExecPacketType = "stdout"
+	Stderr ExecPacketType = "stderr"
+)
+
 // ExecOutputChunk is a struct containing a chunk of the Exec'ed cmd's output.
-// For StreamID, 0 = stdout and 1 = stderr.
 type ExecOutputChunk struct {
-	StreamID  int8
+	StreamID  ExecPacketType
 	Timestamp time.Time
 	Data      string
 	Err       error
@@ -93,7 +117,7 @@ type Execable interface {
 // Streamable is an entry that returns a stream of updates.
 type Streamable interface {
 	Entry
-	Stream(context.Context) (io.Reader, error)
+	Stream(context.Context) (io.ReadCloser, error)
 }
 
 // SizedReader returns a ReaderAt that can report its Size.
diff --git a/volume/core.go b/volume/core.go
index 4d0a10f..6513173 100644
--- a/volume/core.go
+++ b/volume/core.go
@@ -8,6 +8,7 @@ package volume
 
 import (
 	"context"
+	"io"
 	"sort"
 	"time"
 
@@ -25,7 +26,8 @@ type Interface interface {
 	VolumeList(context.Context) (DirMap, error)
 	// Accepts a path and returns the content associated with that path.
 	VolumeOpen(context.Context, string) (plugin.SizedReader, error)
-	// TODO: add VolumeStream
+	// Accepts a path and streams updates to the content associated with that path.
+	VolumeStream(context.Context, string) (io.ReadCloser, error)
 }
 
 // A Dir is a map of files in a directory to their attributes.
@@ -51,7 +53,7 @@ func List(ctx context.Context, impl Interface, path string) ([]plugin.Entry, err
 		if attr.Mode().IsDir() {
 			entries = append(entries, newDir(name, attr, impl, path+"/"+name))
 		} else {
-			entries = append(entries, newFile(name, attr, impl.VolumeOpen, path+"/"+name))
+			entries = append(entries, newFile(name, attr, impl, path+"/"+name))
 		}
 	}
 	// Sort entries so they have a deterministic order.
diff --git a/volume/file.go b/volume/file.go
index b0b7cb4..80a9ad8 100644
--- a/volume/file.go
+++ b/volume/file.go
@@ -2,26 +2,24 @@ package volume
 
 import (
 	"context"
+	"io"
 	"time"
 
 	"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)
-
 // file represents a file in a volume that has content we can access.
 type file struct {
 	plugin.EntryBase
-	contentcb contentCB
-	path      string
+	impl Interface
+	path string
 }
 
 // newFile creates a VolumeFile.
-func newFile(name string, attr plugin.EntryAttributes, cb contentCB, path string) *file {
+func newFile(name string, attr plugin.EntryAttributes, impl Interface, path string) *file {
 	vf := &file{
 		EntryBase: plugin.NewEntry(name),
-		contentcb: cb,
+		impl:      impl,
 		path:      path,
 	}
 	vf.SetAttributes(attr)
@@ -32,5 +30,9 @@ func newFile(name string, attr plugin.EntryAttributes, cb contentCB, path string
 
 // Open returns the content of the file as a SizedReader.
 func (v *file) Open(ctx context.Context) (plugin.SizedReader, error) {
-	return v.contentcb(ctx, v.path)
+	return v.impl.VolumeOpen(ctx, v.path)
+}
+
+func (v *file) Stream(ctx context.Context) (io.ReadCloser, error) {
+	return v.impl.VolumeStream(ctx, v.path)
 }
diff --git a/volume/fs.go b/volume/fs.go
index 3ea5373..01c733d 100644
--- a/volume/fs.go
+++ b/volume/fs.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/puppetlabs/wash/activity"
 	"github.com/puppetlabs/wash/plugin"
@@ -41,11 +42,11 @@ func exec(ctx context.Context, executor plugin.Execable, cmdline []string) (*byt
 	for chunk := range result.OutputCh {
 		if chunk.Err != nil {
 			errs = append(errs, chunk.Err)
-		} else if chunk.StreamID == 0 {
-			activity.Record(ctx, "Stdout: %v", chunk.Data)
-			fmt.Fprint(&buf, chunk.Data)
 		} else {
-			activity.Record(ctx, "Stderr: %v", chunk.Data)
+			activity.Record(ctx, "%v: %v", chunk.StreamID, chunk.Data)
+			if chunk.StreamID == plugin.Stdout {
+				fmt.Fprint(&buf, chunk.Data)
+			}
 		}
 	}
 
@@ -85,3 +86,42 @@ func (d *FS) VolumeOpen(ctx context.Context, path string) (plugin.SizedReader, e
 	}
 	return bytes.NewReader(buf.Bytes()), nil
 }
+
+// VolumeStream satisfies the Interface required by List to stream file contents.
+func (d *FS) VolumeStream(ctx context.Context, path string) (io.ReadCloser, error) {
+	activity.Record(ctx, "Streaming %v on %v", path, plugin.ID(d.executor))
+	result, err := d.executor.Exec(ctx, "tail", []string{"-f", path}, plugin.ExecOptions{Tty: true})
+	if err != nil {
+		activity.Record(ctx, "Exec error in VolumeRead: %v", err)
+		return nil, err
+	}
+
+	r, w := io.Pipe()
+	go func() {
+		// Exec uses context; if it's canceled, the OutputCh will close. Close the writer.
+		var errs []error
+		for chunk := range result.OutputCh {
+			if chunk.Err != nil {
+				activity.Record(ctx, "Error on exec: %v", chunk.Err)
+				errs = append(errs, chunk.Err)
+				continue
+			}
+
+			activity.Record(ctx, "%v: %v", chunk.StreamID, chunk.Data)
+			if len(errs) == 0 {
+				if _, err := w.Write([]byte(chunk.Data)); err != nil {
+					activity.Record(ctx, "Error copying exec result: %v", err)
+					errs = append(errs, err)
+				}
+			}
+		}
+
+		if len(errs) > 0 {
+			err = w.CloseWithError(fmt.Errorf("Multiple errors from exec output: %v", errs))
+		} else {
+			err = w.Close()
+		}
+		activity.Record(ctx, "Closing write pipe: %v", err)
+	}()
+	return r, nil
+}


Test patch:
diff --git a/plugin/cleanupReader_test.go b/plugin/cleanupReader_test.go
new file mode 100644
index 0000000..35faf83
--- /dev/null
+++ b/plugin/cleanupReader_test.go
@@ -0,0 +1,35 @@
+package plugin
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCleanupReader(t *testing.T) {
+	called := false
+	cleanup := func() {
+		called = true
+	}
+
+	r, w := io.Pipe()
+	defer w.Close()
+	rdr := CleanupReader{ReadCloser: r, Cleanup: cleanup}
+	go func() {
+		_, err := w.Write([]byte("hello"))
+		assert.NoError(t, err)
+	}()
+
+	buf := make([]byte, 5)
+	n, err := rdr.Read(buf)
+	assert.NoError(t, err)
+	assert.Equal(t, 5, n)
+	assert.Equal(t, "hello", string(buf))
+	assert.False(t, called)
+
+	rdr.Close()
+	assert.True(t, called)
+	_, err = w.Write([]byte("goodbye"))
+	assert.Error(t, err)
+}
diff --git a/plugin/outputStream_test.go b/plugin/outputStream_test.go
index c56f7af..877064c 100644
--- a/plugin/outputStream_test.go
+++ b/plugin/outputStream_test.go
@@ -15,31 +15,10 @@ type OutputStreamTestSuite struct {
 }
 
 func (suite *OutputStreamTestSuite) EqualChunk(expected ExecOutputChunk, actual ExecOutputChunk) bool {
-	streamName := func(id int8) (string, error) {
-		if id == StdoutID {
-			return "stdout", nil
-		}
-
-		if id == StderrID {
-			return "stderr", nil
-		}
-
-		return "", fmt.Errorf("unknown ID: %v", id)
-	}
-
-	expectedStreamName, err := streamName(expected.StreamID)
-	if err != nil {
-		suite.Fail("could not get the stream name of the expected chunk: %v", err)
-	}
-
-	actualStreamName, err := streamName(actual.StreamID)
-	if err != nil {
-		suite.Fail("actual chunk has an invalid stream ID: %v", err)
-	}
-
+	expectedStreamName := expected.StreamID
 	eqlStreamName := suite.Equal(
 		expectedStreamName,
-		actualStreamName,
+		actual.StreamID,
 		fmt.Sprintf("The sent ExecOutputChunk should have come from the %v stream", expectedStreamName),
 	)
 
@@ -83,7 +62,7 @@ func (suite *OutputStreamTestSuite) TestWrite() {
 	ch := make(chan ExecOutputChunk, 1)
 	defer close(ch)
 
-	stream := OutputStream{ctx: ctx, id: StdoutID, ch: ch}
+	stream := OutputStream{ctx: ctx, id: Stdout, ch: ch}
 
 	// Test a successful write
 	data := []byte("data")
@@ -94,7 +73,7 @@ func (suite *OutputStreamTestSuite) TestWrite() {
 	select {
 	case chunk := <-stream.ch:
 		suite.EqualChunk(
-			ExecOutputChunk{StreamID: StdoutID, Data: string(data)},
+			ExecOutputChunk{StreamID: Stdout, Data: string(data)},
 			chunk,
 		)
 	default:
@@ -108,7 +87,7 @@ func (suite *OutputStreamTestSuite) TestWrite() {
 	select {
 	case chunk := <-stream.ch:
 		suite.EqualChunk(
-			ExecOutputChunk{StreamID: StdoutID, Err: ctx.Err()},
+			ExecOutputChunk{StreamID: Stdout, Err: ctx.Err()},
 			chunk,
 		)
 		suite.Equal(stream.sentCtxErr, true, "The stream should mark that the context's error was sent")
@@ -142,7 +121,7 @@ func (suite *OutputStreamTestSuite) TestClose() {
 func (suite *OutputStreamTestSuite) TestCloseWithError() {
 	newOutputStream := func(ctx context.Context) OutputStream {
 		ch := make(chan ExecOutputChunk, 1)
-		return OutputStream{ctx: ctx, id: StdoutID, ch: ch, closer: &multiCloser{ch: ch, countdown: 1}}
+		return OutputStream{ctx: ctx, id: Stdout, ch: ch, closer: &multiCloser{ch: ch, countdown: 1}}
 	}
 
 	// Test that if err == nil, then nothing was sent to the channel
@@ -197,15 +176,6 @@ func (suite *OutputStreamTestSuite) TestCloseWithError() {
 func (suite *OutputStreamTestSuite) TestCreateExecOutputStreams() {
 	outputCh, stdout, stderr := CreateExecOutputStreams(context.Background())
 
-	assertStreamID := func(streamName string, expectedID int8, stream *OutputStream) {
-		if !suite.Equal(expectedID, stream.id) {
-			msg := fmt.Sprintf("expected %v stream to have stream ID %v, got %v instead", streamName, expectedID, stream.id)
-			suite.FailNow(msg)
-		}
-	}
-	assertStreamID("stdout", StdoutID, stdout)
-	assertStreamID("stderr", StderrID, stderr)
-
 	// Our simulated command alternates writing to stdout + stderr
 	expectedChunksCh := make(chan ExecOutputChunk, 1)
 	go func() {
diff --git a/volume/dir_test.go b/volume/dir_test.go
index f9a7adc..371dd49 100644
--- a/volume/dir_test.go
+++ b/volume/dir_test.go
@@ -2,6 +2,7 @@ package volume
 
 import (
 	"context"
+	"io"
 	"os"
 	"strings"
 	"testing"
@@ -11,16 +12,20 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-type mockEntry struct {
+type mockDirEntry struct {
 	plugin.EntryBase
 	dmap DirMap
 }
 
-func (m *mockEntry) VolumeList(context.Context) (DirMap, error) {
+func (m *mockDirEntry) VolumeList(context.Context) (DirMap, error) {
 	return m.dmap, nil
 }
 
-func (m *mockEntry) VolumeOpen(context.Context, string) (plugin.SizedReader, error) {
+func (m *mockDirEntry) VolumeOpen(context.Context, string) (plugin.SizedReader, error) {
+	return nil, nil
+}
+
+func (m *mockDirEntry) VolumeStream(context.Context, string) (io.ReadCloser, error) {
 	return nil, nil
 }
 
@@ -29,7 +34,7 @@ func TestVolumeDir(t *testing.T) {
 	assert.Nil(t, err)
 
 	plugin.SetTestCache(datastore.NewMemCache())
-	entry := mockEntry{EntryBase: plugin.NewEntry("mine"), dmap: dmap}
+	entry := mockDirEntry{EntryBase: plugin.NewEntry("mine"), dmap: dmap}
 	entry.SetTestID("/mine")
 
 	assert.NotNil(t, dmap[""]["path"])
diff --git a/volume/file_test.go b/volume/file_test.go
index 0ce1177..c32a2ab 100644
--- a/volume/file_test.go
+++ b/volume/file_test.go
@@ -3,6 +3,8 @@ package volume
 import (
 	"context"
 	"errors"
+	"io"
+	"io/ioutil"
 	"strings"
 	"testing"
 	"time"
@@ -11,19 +13,43 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+type mockFileEntry struct {
+	plugin.EntryBase
+	content string
+	err     error
+}
+
+func (m *mockFileEntry) VolumeList(context.Context) (DirMap, error) {
+	return nil, nil
+}
+
+func (m *mockFileEntry) VolumeOpen(context.Context, string) (plugin.SizedReader, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+	return strings.NewReader(m.content), nil
+}
+
+func (m *mockFileEntry) VolumeStream(context.Context, string) (io.ReadCloser, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+	return ioutil.NopCloser(strings.NewReader(m.content)), nil
+}
+
 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) {
-		assert.Equal(t, "my path", path)
-		return strings.NewReader("hello"), nil
-	}, "my path")
+
+	impl := &mockFileEntry{EntryBase: plugin.NewEntry("parent"), content: "hello"}
+	vf := newFile("mine", initialAttr, impl, "my path")
 
 	attr := plugin.Attributes(vf)
 	expectedAttr := plugin.EntryAttributes{}
 	expectedAttr.SetCtime(now)
 	assert.Equal(t, expectedAttr, attr)
+
 	rdr, err := vf.Open(context.Background())
 	assert.Nil(t, err)
 	if assert.NotNil(t, rdr) {
@@ -33,14 +59,26 @@ func TestVolumeFile(t *testing.T) {
 		assert.Equal(t, int64(n), rdr.Size())
 		assert.Equal(t, "hello", string(buf))
 	}
+
+	rdr2, err := vf.Stream(context.Background())
+	assert.Nil(t, err)
+	if assert.NotNil(t, rdr2) {
+		buf, err := ioutil.ReadAll(rdr2)
+		if assert.NoError(t, err) {
+			assert.Equal(t, "hello", string(buf))
+		}
+	}
 }
 
 func TestVolumeFileErr(t *testing.T) {
-	vf := newFile("mine", plugin.EntryAttributes{}, func(ctx context.Context, path string) (plugin.SizedReader, error) {
-		return nil, errors.New("fail")
-	}, "my path")
+	impl := &mockFileEntry{EntryBase: plugin.NewEntry("parent"), err: errors.New("fail")}
+	vf := newFile("mine", plugin.EntryAttributes{}, impl, "my path")
 
 	rdr, err := vf.Open(context.Background())
 	assert.Nil(t, rdr)
 	assert.Equal(t, errors.New("fail"), err)
+
+	rdr2, err := vf.Stream(context.Background())
+	assert.Nil(t, rdr2)
+	assert.Equal(t, errors.New("fail"), err)
 }
diff --git a/volume/fs_test.go b/volume/fs_test.go
index 3bad252..257ae5f 100644
--- a/volume/fs_test.go
+++ b/volume/fs_test.go
@@ -42,7 +42,7 @@ func createResult(data string) plugin.ExecResult {
 		OutputCh:   outputch,
 		ExitCodeCB: func() (int, error) { return 0, nil },
 	}
-	outputch <- plugin.ExecOutputChunk{Data: data}
+	outputch <- plugin.ExecOutputChunk{StreamID: plugin.Stdout, Data: data}
 	close(outputch)
 	return execResult
 }