Feature: full backups

dgraph/cmd/alpha/admin.go

@@ -32,8 +32,8 @@ import (
 )
 
 // handlerInit does some standard checks. Returns false if something is wrong.
-func handlerInit(w http.ResponseWriter, r *http.Request) bool {
-	if r.Method != http.MethodGet {
+func handlerInit(w http.ResponseWriter, r *http.Request, method string) bool {
+	if r.Method != method {
 		x.SetStatus(w, x.ErrorInvalidMethod, "Invalid method")
 		return false
 	}
@@ -47,7 +47,7 @@ func handlerInit(w http.ResponseWriter, r *http.Request) bool {
 }
 
 func shutDownHandler(w http.ResponseWriter, r *http.Request) {
-	if !handlerInit(w, r) {
+	if !handlerInit(w, r, http.MethodGet) {
 		return
 	}
 
@@ -57,19 +57,37 @@ func shutDownHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func exportHandler(w http.ResponseWriter, r *http.Request) {
-	if !handlerInit(w, r) {
+	if !handlerInit(w, r, http.MethodGet) {
 		return
 	}
-	ctx := context.Background()
 	// Export logic can be moved to dgraphzero.
-	if err := worker.ExportOverNetwork(ctx); err != nil {
+	if err := worker.ExportOverNetwork(context.Background()); err != nil {
 		x.SetStatus(w, err.Error(), "Export failed.")
 		return
 	}
 	w.Header().Set("Content-Type", "application/json")
 	w.Write([]byte(`{"code": "Success", "message": "Export completed."}`))
 }
 
+func backupHandler(w http.ResponseWriter, r *http.Request) {
+	if !handlerInit(w, r, http.MethodPost) {
+		return
+	}
+	target := r.FormValue("destination")
+	if target == "" {
+		err := x.Errorf("You must specify a 'destination' value")
+		x.SetStatus(w, err.Error(), "Backup failed.")
+		return
+	}
+	if err := worker.BackupOverNetwork(context.Background(), target); err != nil {
+		x.SetStatus(w, err.Error(), "Backup failed.")
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.Write([]byte(`{"code": "Success", "message": "Backup completed."}`))
+
+}
+
 func memoryLimitHandler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case http.MethodGet:

dgraph/cmd/alpha/run.go

@@ -272,6 +272,7 @@ func setupServer() {
 	http.HandleFunc("/share", shareHandler)
 	http.HandleFunc("/debug/store", storeStatsHandler)
 	http.HandleFunc("/admin/shutdown", shutDownHandler)
+	http.HandleFunc("/admin/backup", backupHandler)
 	http.HandleFunc("/admin/export", exportHandler)
 	http.HandleFunc("/admin/config/lru_mb", memoryLimitHandler)
 

ee/README.md

@@ -0,0 +1,4 @@
+# Dgraph Enterprise Edition (EE)
+
+The files stored here correspond to the Dgraph Enterprise Edition features, which are _not_ under the Apache 2 License.
+

ee/backup/backup.go

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2018 Dgraph Labs, Inc. All rights reserved.
+ *
+ */
+
+package backup
+
+import (
+	"context"
+
+	"github.com/dgraph-io/badger"
+	"github.com/dgraph-io/dgraph/posting"
+	"github.com/dgraph-io/dgraph/protos/pb"
+	"github.com/dgraph-io/dgraph/stream"
+	"github.com/dgraph-io/dgraph/x"
+
+	"github.com/golang/glog"
+)
+
+// Request has all the information needed to perform a backup.
+type Request struct {
+	DB     *badger.DB // Badger pstore managed by this node.
+	Sizex  int64      // approximate upload size
+	Backup *pb.BackupRequest
+}
+
+// Process uses the request values to create a stream writer then hand off the data
+// retrieval to stream.Orchestrate. The writer will create all the fd's needed to
+// collect the data and later move to the target.
+// Returns errors on failure, nil on success.
+func (r *Request) Process(ctx context.Context) error {
+	w, err := r.newWriter()
+	if err != nil {
+		return err
+	}
+
+	sl := stream.Lists{Stream: w, DB: r.DB}
+	sl.ChooseKeyFunc = nil
+	// sl.ChooseKeyFunc = func(_ *badger.Item) bool { return true }
+	sl.ItemToKVFunc = func(key []byte, itr *badger.Iterator) (*pb.KV, error) {
+		item := itr.Item()
+		pk := x.Parse(key)
+		if pk.IsSchema() {
+			val, err := item.ValueCopy(nil)
+			if err != nil {
+				return nil, err
+			}
+			kv := &pb.KV{
+				Key:      key,
+				Val:      val,
+				UserMeta: []byte{item.UserMeta()},
+				Version:  item.Version(),
+			}
+			return kv, nil
+		}
+		l, err := posting.ReadPostingList(key, itr)
+		if err != nil {
+			return nil, err
+		}
+		return l.MarshalToKv()
+	}
+
+	br := r.Backup
+	glog.V(2).Infof("Backup started ...")
+	if err = sl.Orchestrate(ctx, "Backup", br.ReadTs); err != nil {
+		return err
+	}
+	if err = w.cleanup(); err != nil {
+		return err
+	}
+	glog.Infof("Backup complete: group %d at %d", br.GroupId, br.ReadTs)
+
+	return nil
+}

ee/backup/handler_file.go

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2018 Dgraph Labs, Inc. All rights reserved.
+ *
+ */
+
+package backup
+
+import (
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+
+	"github.com/dgraph-io/dgraph/x"
+
+	"github.com/golang/glog"
+)
+
+// fileHandler is used for 'file:' URI scheme.
+type fileHandler struct {
+	fp *os.File
+}
+
+// Open authenticates or prepares a handler session.
+// Returns error on failure, nil on success.
+func (h *fileHandler) Open(uri *url.URL, req *Request) error {
+	// check that this path exists and we can access it.
+	if !h.exists(uri.Path) {
+		return x.Errorf("The path %q does not exist or it is inaccessible.", uri.Path)
+	}
+
+	dir := filepath.Join(uri.Path, fmt.Sprintf("dgraph.%s", req.Backup.UnixTs))
+	if err := os.Mkdir(dir, 0700); err != nil {
+		return err
+	}
+
+	path := filepath.Join(dir,
+		fmt.Sprintf("r%d-g%d.backup", req.Backup.ReadTs, req.Backup.GroupId))
+	fp, err := os.Create(path)
+	if err != nil {
+		return err
+	}
+	glog.V(2).Infof("Using file path: %q", path)
+	h.fp = fp
+	return nil
+}
+
+func (h *fileHandler) Close() error {
+	if err := h.fp.Sync(); err != nil {
+		return err
+	}
+	return h.fp.Close()
+}
+
+func (h *fileHandler) Write(b []byte) (int, error) {
+	return h.fp.Write(b)
+}
+
+// Exists checks if a path (file or dir) is found at target.
+// Returns true if found, false otherwise.
+func (h *fileHandler) exists(path string) bool {
+	_, err := os.Stat(path)
+	if err == nil {
+		return true
+	}
+	return !os.IsNotExist(err) && !os.IsPermission(err)
+}

ee/backup/handler_s3.go

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2018 Dgraph Labs, Inc. All rights reserved.
+ *
+ */
+
+package backup
+
+import (
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+
+	"github.com/dgraph-io/dgraph/x"
+
+	"github.com/golang/glog"
+	minio "github.com/minio/minio-go"
+)
+
+const (
+	s3defaultEndpoint = "s3.amazonaws.com"
+	s3accelerateHost  = "s3-accelerate"
+)
+
+// s3Handler is used for 's3:' URI scheme.
+type s3Handler struct {
+	bucket string
+	path   string
+	w      io.Writer
+	n      uint64
+}
+
+// Open creates an AWS session and sends our data stream to an S3 blob.
+// URI formats:
+//   s3://<s3 region endpoint>/bucket/folder1.../folderN?secure=true|false
+//   s3:///bucket/folder1.../folderN?secure=true|false (use default S3 endpoint)
+func (h *s3Handler) Open(uri *url.URL, req *Request) error {
+	accessKeyID := os.Getenv("AWS_ACCESS_KEY_ID")
+	secretAccessKey := os.Getenv("AWS_SECRET_ACCESS_KEY")
+	if accessKeyID == "" || secretAccessKey == "" {
+		return x.Errorf("Env vars AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY not set.")
+	}
+
+	glog.V(2).Infof("S3Handler got uri: %+v\n", uri)
+	var host, path string
+	// s3:///bucket/folder
+	if !strings.Contains(uri.Host, ".") {
+		path, host = uri.Host, s3defaultEndpoint
+	}
+	glog.V(2).Infof("Backup using S3 host: %s, path: %s", host, path)
+
+	if len(path) < 1 {
+		return x.Errorf("The S3 bucket %q is invalid", path)
+	}
+
+	parts := strings.Split(path[1:], "/")
+	h.bucket = parts[0] // bucket
+
+	// split path into bucket and blob
+	h.path = filepath.Join(parts[1:]...)
+	h.path = filepath.Join(h.path, fmt.Sprintf("dgraph.%s", req.Backup.UnixTs),
+		fmt.Sprintf("r%d-g%d.backup", req.Backup.ReadTs, req.Backup.GroupId))
+
+	// if len(parts) > 1 {
+	// 	parts = append(parts, s.file)
+	// 	// blob: folder1/...folderN/file1
+	// 	s.file = filepath.Join(parts[1:]...)
+	// }
+	glog.V(3).Infof("Sending to S3. Path=%s", h.path)
+
+	// secure by default
+	// uri.Query()
+	// TODO: Do the arg query here.
+	// secure := s.Getarg("secure") != "false"
+	secure := true
+
+	mc, err := minio.New(host, accessKeyID, secretAccessKey, secure)
+	if err != nil {
+		return err
+	}
+	loc, err := mc.GetBucketLocation(h.bucket)
+	if err != nil {
+		return err
+	}
+	if err := mc.MakeBucket(path, loc); err != nil {
+		return err
+	}
+
+	// S3 transfer acceleration support.
+	if strings.Contains(host, "s3-accelerate") {
+		mc.SetS3TransferAccelerate(host)
+	}
+	mc.TraceOn(os.Stderr)
+
+	// TODO: Fix this up.
+	// go h.send(c, s)
+
+	return nil
+}
+
+// func (h *s3Handler) send(c *minio.Client, s *session) {
+// 	glog.Infof("Backup streaming in background, est. %d bytes", s.size)
+// 	start := time.Now()
+// 	pr, pw := io.Pipe()
+// 	buf := bufio.NewWriterSize(pw, 5<<20)
+// 	h.w = buf
+
+// 	// block until done or pipe closed.
+// 	for {
+// 		time.Sleep(10 * time.Millisecond) // wait for some buffered data
+// 		n, err := c.PutObject(s.path, s.file, pr, s.size,
+// 			minio.PutObjectOptions{})
+// 		if err != nil {
+// 			glog.Errorf("Failure while uploading backup: %s", err)
+// 		}
+// 		glog.V(3).Infof("sent %d bytes, actual %d bytes, time elapsed %s", n, h.n, time.Since(start))
+// 		break
+// 	}
+// 	pr.CloseWithError(nil) // EOF
+// }
+
+func (h *s3Handler) Close() error {
+	return nil
+}
+
+func (h *s3Handler) Write(b []byte) (int, error) {
+	n, err := h.w.Write(b)
+	atomic.AddUint64(&h.n, uint64(n))
+	return n, err
+}