Merge pull request #52 from sinbad/resume-upload-tus

Resumable uploads via tusd

Author: sinbad

README.md

@@ -49,6 +49,8 @@ There are few things that can be configured via environment variables:
 	LFS_CERT        # Certificate file for tls
 	LFS_KEY         # tls key
 	LFS_SCHEME      # set to 'https' to override default http
+    LFS_USETUS      # set to 'true' to enable tusd (tus.io) resumable upload server; tusd must be on PATH, installed separately
+    LFS_TUSHOST     # The host used to start the tusd upload server, default "localhost:1080"
 
 If the `LFS_ADMINUSER` and `LFS_ADMINPASS` variables are set, a
 rudimentary admin interface can be accessed via

config.go

@@ -21,6 +21,8 @@ type Configuration struct {
 	Key         string `config:""`
 	Scheme      string `config:"http"`
 	Public      string `config:"public"`
+	UseTus      string `config:"false"`
+	TusHost     string `config:"localhost:1080"`
 }
 
 func (c *Configuration) IsHTTPS() bool {
@@ -35,6 +37,14 @@ func (c *Configuration) IsPublic() bool {
 	return false
 }
 
+func (c *Configuration) IsUsingTus() bool {
+	switch Config.UseTus {
+	case "1", "true", "TRUE":
+		return true
+	}
+	return false
+}
+
 // Config is the global app configuration
 var Config = &Configuration{}
 

main.go

@@ -107,6 +107,12 @@ func main() {
 	logger.Log(kv{"fn": "main", "msg": "listening", "pid": os.Getpid(), "addr": Config.Listen, "version": version})
 
 	app := NewApp(contentStore, metaStore)
+	if Config.IsUsingTus() {
+		tusServer.Start()
+	}
 	app.Serve(listener)
 	tl.WaitForChildren()
+	if Config.IsUsingTus() {
+		tusServer.Stop()
+	}
 }

server.go

@@ -58,8 +58,20 @@ type ObjectError struct {
 	Message string `json:"message"`
 }
 
-// ObjectLink builds a URL linking to the object.
-func (v *RequestVars) ObjectLink() string {
+// DownloadLink builds a URL to download the object.
+func (v *RequestVars) DownloadLink() string {
+	return v.internalLink("objects")
+}
+
+// UploadLink builds a URL to upload the object.
+func (v *RequestVars) UploadLink(useTus bool) string {
+	if useTus {
+		return v.tusLink()
+	}
+	return v.internalLink("objects")
+}
+
+func (v *RequestVars) internalLink(subpath string) string {
 	path := ""
 
 	if len(v.User) > 0 {
@@ -70,7 +82,25 @@ func (v *RequestVars) ObjectLink() string {
 		path += fmt.Sprintf("/%s", v.Repo)
 	}
 
-	path += fmt.Sprintf("/objects/%s", v.Oid)
+	path += fmt.Sprintf("/%s/%s", subpath, v.Oid)
+
+	if Config.IsHTTPS() {
+		return fmt.Sprintf("%s://%s%s", Config.Scheme, Config.Host, path)
+	}
+
+	return fmt.Sprintf("http://%s%s", Config.Host, path)
+}
+
+func (v *RequestVars) tusLink() string {
+	link, err := tusServer.Create(v.Oid, v.Size)
+	if err != nil {
+		logger.Fatal(kv{"fn": fmt.Sprintf("Unable to create tus link for %s: %v", v.Oid, err)})
+	}
+	return link
+}
+
+func (v *RequestVars) VerifyLink() string {
+	path := fmt.Sprintf("/verify/%s", v.Oid)
 
 	if Config.IsHTTPS() {
 		return fmt.Sprintf("%s://%s%s", Config.Scheme, Config.Host, path)
@@ -115,6 +145,8 @@ func NewApp(content *ContentStore, meta *MetaStore) *App {
 
 	r.HandleFunc("/objects", app.PostHandler).Methods("POST").MatcherFunc(MetaMatcher)
 
+	r.HandleFunc("/verify/{oid}", app.VerifyHandler).Methods("POST")
+
 	app.addMgmt(r)
 
 	app.router = r
@@ -191,7 +223,7 @@ func (a *App) GetMetaHandler(w http.ResponseWriter, r *http.Request) {
 
 	if r.Method == "GET" {
 		enc := json.NewEncoder(w)
-		enc.Encode(a.Represent(rv, meta, true, false))
+		enc.Encode(a.Represent(rv, meta, true, false, false))
 	}
 
 	logRequest(r, 200)
@@ -219,7 +251,7 @@ func (a *App) PostHandler(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(sentStatus)
 
 	enc := json.NewEncoder(w)
-	enc.Encode(a.Represent(rv, meta, meta.Existing, true))
+	enc.Encode(a.Represent(rv, meta, meta.Existing, true, false))
 	logRequest(r, sentStatus)
 }
 
@@ -229,11 +261,21 @@ func (a *App) BatchHandler(w http.ResponseWriter, r *http.Request) {
 
 	var responseObjects []*Representation
 
+	var useTus bool
+	if bv.Operation == "upload" && Config.IsUsingTus() {
+		for _, t := range bv.Transfers {
+			if t == "tus" {
+				useTus = true
+				break
+			}
+		}
+	}
+
 	// Create a response object
 	for _, object := range bv.Objects {
 		meta, err := a.metaStore.Get(object)
 		if err == nil && a.contentStore.Exists(meta) { // Object is found and exists
-			responseObjects = append(responseObjects, a.Represent(object, meta, true, false))
+			responseObjects = append(responseObjects, a.Represent(object, meta, true, false, false))
 			continue
 		}
 
@@ -245,13 +287,17 @@ func (a *App) BatchHandler(w http.ResponseWriter, r *http.Request) {
 		// Object is not found
 		meta, err = a.metaStore.Put(object)
 		if err == nil {
-			responseObjects = append(responseObjects, a.Represent(object, meta, meta.Existing, true))
+			responseObjects = append(responseObjects, a.Represent(object, meta, meta.Existing, true, useTus))
 		}
 	}
 
 	w.Header().Set("Content-Type", metaMediaType)
 
 	respobj := &BatchResponse{Objects: responseObjects}
+	// Respond with TUS support if advertised
+	if useTus {
+		respobj.Transfer = "tus"
+	}
 
 	enc := json.NewEncoder(w)
 	enc.Encode(respobj)
@@ -281,9 +327,21 @@ func (a *App) PutHandler(w http.ResponseWriter, r *http.Request) {
 	logRequest(r, 200)
 }
 
+func (a *App) VerifyHandler(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	oid := vars["oid"]
+	err := tusServer.Finish(oid, a.contentStore)
+
+	if err != nil {
+		logger.Fatal(kv{"fn": "VerifyHandler", "err": fmt.Sprintf("Failed to verify %s: %v", oid, err)})
+	}
+
+	logRequest(r, 200)
+}
+
 // Represent takes a RequestVars and Meta and turns it into a Representation suitable
 // for json encoding
-func (a *App) Represent(rv *RequestVars, meta *MetaObject, download, upload bool) *Representation {
+func (a *App) Represent(rv *RequestVars, meta *MetaObject, download, upload, useTus bool) *Representation {
 	rep := &Representation{
 		Oid:     meta.Oid,
 		Size:    meta.Size,
@@ -296,11 +354,14 @@ func (a *App) Represent(rv *RequestVars, meta *MetaObject, download, upload bool
 		header["Authorization"] = rv.Authorization
 	}
 	if download {
-		rep.Actions["download"] = &link{Href: rv.ObjectLink(), Header: header}
+		rep.Actions["download"] = &link{Href: rv.DownloadLink(), Header: header}
 	}
 
 	if upload {
-		rep.Actions["upload"] = &link{Href: rv.ObjectLink(), Header: header}
+		rep.Actions["upload"] = &link{Href: rv.UploadLink(useTus), Header: header}
+		if useTus {
+			rep.Actions["verify"] = &link{Href: rv.VerifyLink(), Header: header}
+		}
 	}
 	return rep
 }

tus.go

@@ -0,0 +1,158 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+)
+
+type TusServer struct {
+	serverMutex sync.Mutex
+	tusProcess  *exec.Cmd
+	dataPath    string
+	tusBaseUrl  string
+	httpClient  *http.Client
+	oidToTusUrl map[string]string
+}
+
+var (
+	tusServer *TusServer = &TusServer{}
+)
+
+// Start launches the tus server & stores uploads in the given contentPath
+func (t *TusServer) Start() {
+	t.serverMutex.Lock()
+	defer t.serverMutex.Unlock()
+
+	if t.tusProcess != nil {
+		return
+	}
+
+	t.dataPath = filepath.Join(os.TempDir(), "lfs_tusserver")
+	hostparts := strings.Split(Config.TusHost, ":")
+	host := "localhost"
+	port := "1080"
+	if len(hostparts) > 0 {
+		host = hostparts[0]
+	}
+	if len(hostparts) > 1 {
+		port = hostparts[1]
+	}
+	t.tusProcess = exec.Command("tusd",
+		"-dir", t.dataPath,
+		"-host", host,
+		"-port", port)
+	// Make sure tus server is started before continuing
+	var procWait sync.WaitGroup
+	procWait.Add(1)
+	go func(p *exec.Cmd) {
+
+		stdout, err := p.StdoutPipe()
+		if err != nil {
+			panic(fmt.Sprintf("Error getting tus server stdout: %v", err))
+		}
+		stderr, err := p.StderrPipe()
+		if err != nil {
+			panic(fmt.Sprintf("Error getting tus server stderr: %v", err))
+		}
+		err = p.Start()
+		if err != nil {
+			panic(fmt.Sprintf("Error starting tus server: %v", err))
+		}
+		go func() {
+			scanner := bufio.NewScanner(stdout)
+			for scanner.Scan() {
+				logger.Log(kv{"fn": "tusout", "msg": scanner.Text()})
+			}
+		}()
+		go func() {
+			scanner := bufio.NewScanner(stderr)
+			for scanner.Scan() {
+				logger.Log(kv{"fn": "tuserr", "msg": scanner.Text()})
+			}
+		}()
+		time.Sleep(2)
+		procWait.Done()
+		defer p.Wait()
+
+	}(t.tusProcess)
+	procWait.Wait()
+	logger.Log(kv{"fn": "Start", "msg": "Tus server started"})
+	t.tusBaseUrl = fmt.Sprintf("http://%s:%s/files/", host, port)
+	t.httpClient = &http.Client{}
+	t.oidToTusUrl = make(map[string]string)
+}
+
+func (t *TusServer) Stop() {
+	t.serverMutex.Lock()
+	defer t.serverMutex.Unlock()
+	if t.tusProcess != nil {
+		t.tusProcess.Process.Kill()
+		t.tusProcess = nil
+	}
+	logger.Log(kv{"fn": "Stop", "msg": "Tus server stopped"})
+}
+
+// Create a new upload URL for the given object
+// Required to call CREATE on the tus API before uploading but not part of LFS API
+func (t *TusServer) Create(oid string, size int64) (string, error) {
+	t.serverMutex.Lock()
+	defer t.serverMutex.Unlock()
+	req, err := http.NewRequest("POST", t.tusBaseUrl, nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Tus-Resumable", "1.0.0")
+	req.Header.Set("Upload-Length", fmt.Sprintf("%d", size))
+	req.Header.Set("Upload-Metadata", fmt.Sprintf("oid %s", oid))
+
+	res, err := t.httpClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	if res.StatusCode != 201 {
+		return "", fmt.Errorf("Expected tus status code 201, got %d", res.StatusCode)
+	}
+	loc := res.Header.Get("Location")
+	if len(loc) == 0 {
+		return "", fmt.Errorf("Missing Location header in tus response")
+	}
+	t.oidToTusUrl[oid] = loc
+	return loc, nil
+}
+
+// Move the finished uploaded data from TUS to the content store (called by verify)
+func (t *TusServer) Finish(oid string, store *ContentStore) error {
+	t.serverMutex.Lock()
+	defer t.serverMutex.Unlock()
+
+	loc, ok := t.oidToTusUrl[oid]
+	if !ok {
+		return fmt.Errorf("Unable to find upload for %s", oid)
+	}
+	parts := strings.Split(loc, "/")
+	filename := filepath.Join(t.dataPath, fmt.Sprintf("%s.bin", parts[len(parts)-1]))
+	stat, err := os.Stat(filename)
+	if err != nil {
+		return err
+	}
+	meta := &MetaObject{Oid: oid, Size: stat.Size(), Existing: false}
+	f, err := os.Open(filename)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	err = store.Put(meta, f)
+	if err == nil {
+		os.Remove(filename)
+		// tus also stores a .info file, remove that
+		os.Remove(filepath.Join(t.dataPath, fmt.Sprintf("%s.info", parts[len(parts)-1])))
+	}
+	return err
+}