Merge pull request #669 from hashicorp/f-fs-api

Introduce FS API

client/allocdir/alloc_dir.go

@@ -38,6 +38,20 @@ type AllocDir struct {
 	mounted []string
 }
 
+// AllocFileInfo holds information about a file inside the AllocDir
+type AllocFileInfo struct {
+	Name  string
+	IsDir bool
+	Size  int64
+}
+
+// AllocDirFS returns methods which exposes file operations on the alloc dir
+type AllocDirFS interface {
+	List(path string) ([]*AllocFileInfo, error)
+	Stat(path string) (*AllocFileInfo, error)
+	ReadAt(path string, offset int64, limit int64) (io.ReadCloser, error)
+}
+
 func NewAllocDir(allocDir string) *AllocDir {
 	d := &AllocDir{AllocDir: allocDir, TaskDirs: make(map[string]string)}
 	d.SharedDir = filepath.Join(d.AllocDir, SharedAllocName)
@@ -217,6 +231,57 @@ func (d *AllocDir) MountSharedDir(task string) error {
 	return nil
 }
 
+// List returns the list of files at a path relative to the alloc dir
+func (d *AllocDir) List(path string) ([]*AllocFileInfo, error) {
+	p := filepath.Join(d.AllocDir, path)
+	finfos, err := ioutil.ReadDir(p)
+	if err != nil {
+		return []*AllocFileInfo{}, err
+	}
+	files := make([]*AllocFileInfo, len(finfos))
+	for idx, info := range finfos {
+		files[idx] = &AllocFileInfo{
+			Name:  info.Name(),
+			IsDir: info.IsDir(),
+			Size:  info.Size(),
+		}
+	}
+	return files, err
+}
+
+// Stat returns information about the file at path relative to the alloc dir
+func (d *AllocDir) Stat(path string) (*AllocFileInfo, error) {
+	p := filepath.Join(d.AllocDir, path)
+	info, err := os.Stat(p)
+	if err != nil {
+		return nil, err
+	}
+
+	return &AllocFileInfo{
+		Size:  info.Size(),
+		Name:  info.Name(),
+		IsDir: info.IsDir(),
+	}, nil
+}
+
+// ReadAt returns a reader  for a file at the path relative to the alloc dir
+//which will read a chunk of bytes at a particular offset
+func (d *AllocDir) ReadAt(path string, offset int64, limit int64) (io.ReadCloser, error) {
+	p := filepath.Join(d.AllocDir, path)
+	f, err := os.Open(p)
+	if err != nil {
+		return nil, err
+	}
+	return &FileReadCloser{Reader: io.LimitReader(f, limit), Closer: f}, nil
+}
+
+// FileReadCloser wraps a LimitReader so that a file is closed once it has been
+// read
+type FileReadCloser struct {
+	io.Reader
+	io.Closer
+}
+
 func fileCopy(src, dst string, perm os.FileMode) error {
 	// Do a simple copy.
 	srcFile, err := os.Open(src)

client/client.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/nomad/client/allocdir"
 	"github.com/hashicorp/nomad/client/config"
 	"github.com/hashicorp/nomad/client/driver"
 	"github.com/hashicorp/nomad/client/fingerprint"
@@ -353,6 +354,16 @@ func (c *Client) Node() *structs.Node {
 	return c.config.Node
 }
 
+// GetAllocFS returns the AllocFS interface for the alloc dir of an allocation
+func (c *Client) GetAllocFS(allocID string) (allocdir.AllocDirFS, error) {
+	ar, ok := c.allocs[allocID]
+	if !ok {
+		return nil, fmt.Errorf("alloc not found")
+	}
+	return ar.ctx.AllocDir, nil
+
+}
+
 // restoreState is used to restore our state from the data dir
 func (c *Client) restoreState() error {
 	if c.config.DevMode {

command/agent/fs_endpoint.go

@@ -0,0 +1,77 @@
+package agent
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+var (
+	allocIDNotPresentErr  = fmt.Errorf("must provide a valid alloc id")
+	fileNameNotPresentErr = fmt.Errorf("must provide a file name")
+)
+
+func (s *HTTPServer) DirectoryListRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	var allocID, path string
+
+	if allocID = strings.TrimPrefix(req.URL.Path, "/v1/client/fs/ls/"); allocID == "" {
+		return nil, allocIDNotPresentErr
+	}
+	if path = req.URL.Query().Get("path"); path == "" {
+		path = "/"
+	}
+	fs, err := s.agent.client.GetAllocFS(allocID)
+	if err != nil {
+		return nil, err
+	}
+	return fs.List(path)
+}
+
+func (s *HTTPServer) FileStatRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	var allocID, path string
+	if allocID = strings.TrimPrefix(req.URL.Path, "/v1/client/fs/stat/"); allocID == "" {
+		return nil, allocIDNotPresentErr
+	}
+	if path = req.URL.Query().Get("path"); path == "" {
+		return nil, fileNameNotPresentErr
+	}
+	fs, err := s.agent.client.GetAllocFS(allocID)
+	if err != nil {
+		return nil, err
+	}
+	return fs.Stat(path)
+}
+
+func (s *HTTPServer) FileReadAtRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+	var allocID, path string
+	var offset, limit int64
+	var err error
+
+	q := req.URL.Query()
+
+	if allocID = strings.TrimPrefix(req.URL.Path, "/v1/client/fs/readat/"); allocID == "" {
+		return nil, allocIDNotPresentErr
+	}
+	if path = q.Get("path"); path == "" {
+		return nil, fileNameNotPresentErr
+	}
+
+	if offset, err = strconv.ParseInt(q.Get("offset"), 10, 64); err != nil {
+		return nil, fmt.Errorf("error parsing offset: %v", err)
+	}
+	if limit, err = strconv.ParseInt(q.Get("limit"), 10, 64); err != nil {
+		return nil, fmt.Errorf("error parsing limit: %v", err)
+	}
+	fs, err := s.agent.client.GetAllocFS(allocID)
+	if err != nil {
+		return nil, err
+	}
+	r, err := fs.ReadAt(path, offset, limit)
+	if err != nil {
+		return nil, err
+	}
+	io.Copy(resp, r)
+	return nil, nil
+}

command/agent/fs_endpoint_test.go

@@ -0,0 +1,86 @@
+package agent
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestAllocDirFS_List(t *testing.T) {
+	httpTest(t, nil, func(s *TestServer) {
+		req, err := http.NewRequest("GET", "/v1/client/fs/ls/", nil)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		respW := httptest.NewRecorder()
+
+		_, err = s.Server.DirectoryListRequest(respW, req)
+		if err != allocIDNotPresentErr {
+			t.Fatalf("expected err: %v, actual: %v", allocIDNotPresentErr, err)
+		}
+	})
+}
+
+func TestAllocDirFS_Stat(t *testing.T) {
+	httpTest(t, nil, func(s *TestServer) {
+		req, err := http.NewRequest("GET", "/v1/client/fs/stat/", nil)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		respW := httptest.NewRecorder()
+
+		_, err = s.Server.FileStatRequest(respW, req)
+		if err != allocIDNotPresentErr {
+			t.Fatalf("expected err: %v, actual: %v", allocIDNotPresentErr, err)
+		}
+
+		req, err = http.NewRequest("GET", "/v1/client/fs/stat/foo", nil)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		respW = httptest.NewRecorder()
+
+		_, err = s.Server.FileStatRequest(respW, req)
+		if err != fileNameNotPresentErr {
+			t.Fatalf("expected err: %v, actual: %v", allocIDNotPresentErr, err)
+		}
+
+	})
+}
+
+func TestAllocDirFS_ReadAt(t *testing.T) {
+	httpTest(t, nil, func(s *TestServer) {
+		req, err := http.NewRequest("GET", "/v1/client/fs/readat/", nil)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		respW := httptest.NewRecorder()
+
+		_, err = s.Server.FileReadAtRequest(respW, req)
+		if err == nil {
+			t.Fatal("expected error")
+		}
+
+		req, err = http.NewRequest("GET", "/v1/client/fs/readat/foo", nil)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		respW = httptest.NewRecorder()
+
+		_, err = s.Server.FileReadAtRequest(respW, req)
+		if err == nil {
+			t.Fatal("expected error")
+		}
+
+		req, err = http.NewRequest("GET", "/v1/client/fs/readat/foo?path=/path/to/file", nil)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		respW = httptest.NewRecorder()
+
+		_, err = s.Server.FileReadAtRequest(respW, req)
+		if err == nil {
+			t.Fatal("expected error")
+		}
+	})
+}

command/agent/http.go

@@ -103,6 +103,10 @@ func (s *HTTPServer) registerHandlers(enableDebug bool) {
 	s.mux.HandleFunc("/v1/evaluations", s.wrap(s.EvalsRequest))
 	s.mux.HandleFunc("/v1/evaluation/", s.wrap(s.EvalSpecificRequest))
 
+	s.mux.HandleFunc("/v1/client/fs/ls/", s.wrap(s.DirectoryListRequest))
+	s.mux.HandleFunc("/v1/client/fs/stat/", s.wrap(s.FileStatRequest))
+	s.mux.HandleFunc("/v1/client/fs/readat/", s.wrap(s.FileReadAtRequest))
+
 	s.mux.HandleFunc("/v1/agent/self", s.wrap(s.AgentSelfRequest))
 	s.mux.HandleFunc("/v1/agent/join", s.wrap(s.AgentJoinRequest))
 	s.mux.HandleFunc("/v1/agent/members", s.wrap(s.AgentMembersRequest))