FileManager

go/cmd/paddlecloud/paddlecloud.go

@@ -16,6 +16,10 @@ func main() {
 	subcommands.Register(&paddlecloud.LogsCommand{}, "")
 	subcommands.Register(&paddlecloud.GetCommand{}, "")
 	subcommands.Register(&paddlecloud.KillCommand{}, "")
+	subcommands.Register(&paddlecloud.LsCommand{}, "")
+	subcommands.Register(&paddlecloud.CpCommand{}, "")
+	subcommands.Register(&paddlecloud.RmCommand{}, "")
+	subcommands.Register(&paddlecloud.MkdirCommand{}, "")
 
 	flag.Parse()
 	ctx := context.Background()

go/cmd/pfsserver/main.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"github.com/PaddlePaddle/cloud/go/filemanager/pfsserver"
+	"net/http"
+
+	log "github.com/golang/glog"
+)
+
+func main() {
+
+	router := pfsserver.NewRouter()
+
+	portPtr := flag.Int("port", 8080, "listen port")
+	flag.Parse()
+
+	addr := fmt.Sprintf("0.0.0.0:%d", *portPtr)
+
+	log.Infof("server on:%s\n", addr)
+	log.Fatal(http.ListenAndServe(addr, router))
+}

go/filemanager/pfsmod/chunk.go

@@ -0,0 +1,118 @@
+package pfsmod
+
+import (
+	"errors"
+	"fmt"
+	log "github.com/golang/glog"
+	"io"
+	"net/url"
+	"os"
+	"strconv"
+)
+
+//ChunkCmd structure
+type ChunkCmd struct {
+	Path      string
+	Offset    int64
+	ChunkSize int64
+	Data      []byte
+}
+
+//NewChunkCmd get a new ChunkCmd
+func NewChunkCmd(path string, offset, chunkSize int64) *ChunkCmd {
+	return &ChunkCmd{
+		Path:      path,
+		Offset:    offset,
+		ChunkSize: chunkSize,
+	}
+}
+
+//ToURLParam encodes variables to url encoding parameters
+func (p *ChunkCmd) ToURLParam() string {
+	parameters := url.Values{}
+	parameters.Add("path", p.Path)
+
+	str := fmt.Sprint(p.Offset)
+	parameters.Add("offset", str)
+
+	str = fmt.Sprint(p.ChunkSize)
+	parameters.Add("chunksize", str)
+
+	return parameters.Encode()
+}
+
+//ToJSON encodes chunkcmd to json string
+func (p *ChunkCmd) ToJSON() ([]byte, error) {
+	return nil, nil
+}
+
+//Run function runs a ChunkCmd
+func (p *ChunkCmd) Run() (interface{}, error) {
+	return nil, nil
+}
+
+//NewChunkCmdFromURLParam get a ChunkCmd structure
+// path example:
+// 	  path=/pfs/datacenter1/1.txt&offset=4096&chunksize=4096
+func NewChunkCmdFromURLParam(path string) (*ChunkCmd, error) {
+	cmd := ChunkCmd{}
+
+	m, err := url.ParseQuery(path)
+	if err != nil ||
+		len(m["path"]) == 0 ||
+		len(m["offset"]) == 0 ||
+		len(m["chunksize"]) == 0 {
+		return nil, errors.New(StatusText(StatusJSONErr))
+	}
+
+	//var err error
+	cmd.Path = m["path"][0]
+	cmd.Offset, err = strconv.ParseInt(m["offset"][0], 10, 64)
+	if err != nil {
+		return nil, errors.New(StatusText(StatusJSONErr))
+	}
+
+	chunkSize, err := strconv.ParseInt(m["chunksize"][0], 10, 64)
+	if err != nil {
+		return nil, errors.New(StatusText(StatusBadChunkSize))
+	}
+	cmd.ChunkSize = chunkSize
+
+	return &cmd, nil
+}
+
+//LoadChunkData loads a specified chunk to w
+func (p *ChunkCmd) LoadChunkData(w io.Writer) error {
+	f, err := os.Open(p.Path)
+	if err != nil {
+		return err
+	}
+	defer Close(f)
+
+	_, err = f.Seek(p.Offset, 0)
+	if err != nil {
+		return err
+	}
+
+	writen, err := io.CopyN(w, f, p.ChunkSize)
+	log.V(2).Infof("writen:%d\n", writen)
+	return err
+}
+
+//SaveChunkData save data from r
+func (p *ChunkCmd) SaveChunkData(r io.Reader) error {
+	f, err := os.OpenFile(p.Path, os.O_WRONLY, 0600)
+	if err != nil {
+		return err
+	}
+	defer Close(f)
+
+	_, err = f.Seek(p.Offset, 0)
+	if err != nil {
+		return err
+	}
+
+	writen, err := io.CopyN(f, r, p.ChunkSize)
+	log.V(2).Infof("chunksize:%d writen:%d\n", p.ChunkSize, writen)
+	return err
+}

go/filemanager/pfsmod/chunkmeta.go

@@ -0,0 +1,221 @@
+package pfsmod
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"sort"
+	"strconv"
+)
+
+const (
+	defaultMaxChunkSize = 4 * 1024 * 1024
+	defaultMinChunkSize = 4 * 1024
+	//DefaultChunkSize is the default chunk's size
+	DefaultChunkSize = 2 * 1024 * 1024
+)
+const (
+	chunkMetaCmdName = "GetChunkMeta"
+)
+
+//ChunkMeta holds the chunk meta's info
+type ChunkMeta struct {
+	Offset   int64  `json:"offset"`
+	Checksum string `json:"checksum"`
+	Len      int64  `json:"len"`
+}
+
+//ChunkMetaCmd is a command
+type ChunkMetaCmd struct {
+	Method    string `json:"method"`
+	FilePath  string `json:"path"`
+	ChunkSize int64  `json:"chunksize"`
+}
+
+//ToURLParam encodes ChunkMetaCmd to URL encoding string
+func (p *ChunkMetaCmd) ToURLParam() string {
+	parameters := url.Values{}
+	parameters.Add("method", p.Method)
+	parameters.Add("path", p.FilePath)
+
+	str := fmt.Sprint(p.ChunkSize)
+	parameters.Add("chunksize", str)
+
+	return parameters.Encode()
+}
+
+//ToJSON encodes ChunkMetaCmd to JSON string
+func (p *ChunkMetaCmd) ToJSON() ([]byte, error) {
+	return json.Marshal(p)
+}
+
+//Run is a functions which run ChunkMetaCmd
+func (p *ChunkMetaCmd) Run() (interface{}, error) {
+	return GetChunkMeta(p.FilePath, p.ChunkSize)
+}
+
+func (p *ChunkMetaCmd) checkChunkSize() error {
+	if p.ChunkSize < defaultMinChunkSize ||
+		p.ChunkSize > defaultMaxChunkSize {
+		return errors.New(StatusText(StatusBadChunkSize))
+	}
+
+	return nil
+}
+
+//CloudCheck checks the conditions when running on cloud
+func (p *ChunkMetaCmd) CloudCheck() error {
+	if !IsCloudPath(p.FilePath) {
+		return errors.New(StatusText(StatusShouldBePfsPath) + ": p.FilePath")
+	}
+	if !CheckUser(p.FilePath) {
+		return errors.New(StatusText(StatusUnAuthorized) + ":" + p.FilePath)
+	}
+	return p.checkChunkSize()
+}
+
+//LocalCheck checks the conditions when running local
+func (p *ChunkMetaCmd) LocalCheck() error {
+	return p.checkChunkSize()
+}
+
+//NewChunkMetaCmdFromURLParam get a new ChunkMetaCmd
+func NewChunkMetaCmdFromURLParam(r *http.Request) (*ChunkMetaCmd, error) {
+	method := r.URL.Query().Get("method")
+	path := r.URL.Query().Get("path")
+	chunkStr := r.URL.Query().Get("chunksize")
+
+	if len(method) == 0 ||
+		method != chunkMetaCmdName ||
+		len(path) < 4 ||
+		len(chunkStr) == 0 {
+		return nil, errors.New(http.StatusText(http.StatusBadRequest))
+	}
+
+	chunkSize, err := strconv.ParseInt(chunkStr, 10, 64)
+	if err != nil {
+		return nil, errors.New(StatusText(StatusBadChunkSize))
+	}
+
+	return &ChunkMetaCmd{
+		Method:    method,
+		FilePath:  path,
+		ChunkSize: chunkSize,
+	}, nil
+}
+
+//NewChunkMetaCmd get a new ChunkMetaCmd with path and chunksize
+func NewChunkMetaCmd(path string, chunkSize int64) *ChunkMetaCmd {
+	return &ChunkMetaCmd{
+		Method:    chunkMetaCmdName,
+		FilePath:  path,
+		ChunkSize: chunkSize,
+	}
+}
+
+type metaSlice []ChunkMeta
+
+func (a metaSlice) Len() int           { return len(a) }
+func (a metaSlice) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+func (a metaSlice) Less(i, j int) bool { return a[i].Offset < a[j].Offset }
+
+//GetDiffChunkMeta gets difference between srcMeta and dstMeta
+func GetDiffChunkMeta(srcMeta []ChunkMeta, dstMeta []ChunkMeta) ([]ChunkMeta, error) {
+	if len(dstMeta) == 0 || len(srcMeta) == 0 {
+		return srcMeta, nil
+	}
+
+	if !sort.IsSorted(metaSlice(srcMeta)) {
+		sort.Sort(metaSlice(srcMeta))
+	}
+
+	if !sort.IsSorted(metaSlice(dstMeta)) {
+		sort.Sort(metaSlice(dstMeta))
+	}
+
+	dstIdx := 0
+	srcIdx := 0
+	diff := make([]ChunkMeta, 0, len(srcMeta))
+
+	for {
+		if srcMeta[srcIdx].Offset < dstMeta[dstIdx].Offset {
+			diff = append(diff, srcMeta[srcIdx])
+			srcIdx++
+		} else if srcMeta[srcIdx].Offset > dstMeta[dstIdx].Offset {
+			dstIdx++
+		} else {
+			if srcMeta[srcIdx].Checksum != dstMeta[dstIdx].Checksum {
+				diff = append(diff, srcMeta[srcIdx])
+			}
+
+			dstIdx++
+			srcIdx++
+		}
+
+		if dstIdx >= len(dstMeta) {
+			break
+		}
+
+		if srcIdx >= len(srcMeta) {
+			break
+		}
+	}
+
+	if srcIdx < len(srcMeta) {
+		diff = append(diff, srcMeta[srcIdx:]...)
+	}
+
+	return diff, nil
+}
+
+//GetChunkMeta gets chunk metas from path of file
+func GetChunkMeta(path string, len int64) ([]ChunkMeta, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer Close(f)
+
+	if len > defaultMaxChunkSize || len < defaultMinChunkSize {
+		return nil, errors.New(StatusText(StatusBadChunkSize))
+	}
+
+	fi, err := f.Stat()
+	if err != nil {
+		return nil, err
+	}
+
+	metas := make([]ChunkMeta, 0, fi.Size()/len+1)
+	data := make([]byte, len)
+	offset := int64(0)
+
+	for {
+		n, err := f.Read(data)
+		if err != nil && err != io.EOF {
+			return metas, err
+		}
+
+		if err == io.EOF {
+			break
+		}
+
+		//log.Println(n)
+		m := ChunkMeta{}
+		m.Offset = offset
+		sum := md5.Sum(data[:n])
+		m.Checksum = hex.EncodeToString(sum[:])
+		m.Len = int64(n)
+
+		metas = append(metas, m)
+
+		offset += int64(n)
+	}
+
+	return metas, nil
+}