diff --git a/api.go b/api.go index 37d385791..46cf6980b 100644 --- a/api.go +++ b/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"` diff --git a/ctl/restore.go b/ctl/restore.go index c4e7a4974..9329f7303 100644 --- a/ctl/restore.go +++ b/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() diff --git a/http/handler.go b/http/handler.go index d5d14b1d6..6f60b18df 100644 --- a/http/handler.go +++ b/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 +}