api compiles

api.go

@@ -17,6 +17,7 @@
 package pilosa
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/binary"
@@ -26,6 +27,7 @@ import (
 	"io/ioutil"
 	"math"
 	"net/url"
+	"os"
 	"sort"
 	"strconv"
 	"strings"
@@ -38,6 +40,7 @@ import (
 	"github.com/pilosa/pilosa/v2/stats"
 	"github.com/pilosa/pilosa/v2/topology"
 	"github.com/pilosa/pilosa/v2/tracing"
+	"github.com/pilosa/pilosa/v2/vprint"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
@@ -2219,6 +2222,43 @@ func (api *API) TranslateFieldDB(ctx context.Context, indexName, fieldName strin
 	return err
 }
 
+// RestoreShard
+func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64, rd io.Reader) error {
+	snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
+	if !snap.OwnsShard(api.server.nodeID, indexName, shard) {
+		return ErrClusterDoesNotOwnShard // TODO (twg)really just node doesn't own shard but leave for now
+	}
+
+	idx := api.holder.Index(indexName)
+	//need to get a dbShard
+	dbs, err := idx.Txf().dbPerShard.GetDBShard(indexName, shard, idx)
+	if err != nil {
+		return err
+	}
+	dbs.Close()
+	//need to find the path to the db
+	//will not work on blue green
+	finalPath := dbs.W[0].Path()
+	tempPath := finalPath + ".tmp"
+	vprint.VV("restore to %v", tempPath)
+	o, err := os.OpenFile(tempPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666)
+	if err != nil {
+		return err
+	}
+	w := bufio.NewWriter(o)
+	//close db if open
+	vprint.VV("restore index:%v shard:%v", indexName, shard)
+	_, err = io.Copy(w, rd)
+	w.Flush()
+	o.Close()
+	if err != nil {
+		_ = os.Remove(tempPath)
+		return err
+	}
+	vprint.VV("Rename %v to %v", tempPath, finalPath)
+	return os.Rename(tempPath, finalPath)
+}
+
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
 	ReplicaN         int    `json:"replicaN"`

ctl/restore.go

@@ -94,6 +94,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error {
 	} else {
 		tarReader = tar.NewReader(f)
 	}
+	//maybe begin transaction?
 	schemaJson := readSchema(cmd.Path)
 	//Push the schema from the archive into the cluster belonging to the host.
 	client := &gohttp.Client{}
@@ -102,6 +103,13 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	//TODO (twg) load schema
+	//TODO (twg) load rbf shard
+	//TODO (twg) load row keys
+	//TODO (twg) load column keys
+	//TODO (twg) load row attributes keys
+	//TODO (twg) load col attributes keys
+	//TODO (twg) load idalloc
 
 	for {
 		header, err := tarReader.Next()

http/handler.go

@@ -441,6 +441,7 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/internal/idalloc/reset/{index}", handler.handleResetIDAlloc).Methods("POST").Name("ResetIDAlloc")
 	router.HandleFunc("/internal/idalloc/data", handler.handleIDAllocData).Methods("GET").Name("IDAllocData")
 
+	router.HandleFunc("/internal/restore/{index}/{shardID}", handler.handlePostRestore).Methods("POST").Name("Restore")
 	// endpoints for collecting cpu profiles from a chosen begin point to
 	// when the client wants to stop. Used for profiling imports that
 	// could be long or short.
@@ -2911,3 +2912,37 @@ func (h *Handler) handleIDAllocData(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 }
+
+func (h *Handler) handlePostRestore(w http.ResponseWriter, r *http.Request) {
+	/*
+		if !validHeaderAcceptType(r.Header, "text", "plain") {
+			http.Error(w, "text/plain is not an acceptable response type", http.StatusNotAcceptable)
+		}
+	*/
+	indexName, ok := mux.Vars(r)["index"]
+	if !ok {
+		http.Error(w, "index name is required", http.StatusBadRequest)
+		return
+	}
+	shardID, ok := mux.Vars(r)["shardID"]
+	if !ok {
+		http.Error(w, "shardID is required", http.StatusBadRequest)
+		return
+	}
+	shard, err := strconv.ParseUint(shardID, 10, 64)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("failed to parse shard %v %v err:%v", indexName, shardID, err), http.StatusBadRequest)
+		return
+	}
+	ctx := context.Background()
+	//validate shard for this node
+	err = h.api.RestoreShard(ctx, indexName, shard, r.Body)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("failed to restore shared %v %v err:%v", indexName, shard, err), http.StatusBadRequest)
+		return
+	}
+
+	w.Header().Add("Content-Type", "text/plain")
+	w.WriteHeader(http.StatusOK)
+	w.Write([]byte("OK")) //nolint:errcheck
+}