From 9d24fb07b7d52b7f683ad11d6c7db7e43b760deb Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 13 May 2021 11:28:52 -0500 Subject: [PATCH] wired in restore command --- api.go | 21 +++++++++++--- bolt.go | 17 +++++++++++ cmd/root.go | 1 + ctl/restore.go | 77 ++++++++++++++++++++++++++++++++++++-------------- dbshard.go | 3 ++ fragment.go | 3 ++ lattice | 2 +- rbf.go | 20 +++++++++++++ rrtx.go | 7 +++++ 9 files changed, 125 insertions(+), 26 deletions(-) diff --git a/api.go b/api.go index 46cf6980b..a452bf8d5 100644 --- a/api.go +++ b/api.go @@ -2235,10 +2235,10 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64 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() + db := dbs.W[0] + finalPath := db.Path() + "/data" tempPath := finalPath + ".tmp" vprint.VV("restore to %v", tempPath) o, err := os.OpenFile(tempPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0666) @@ -2248,15 +2248,28 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64 w := bufio.NewWriter(o) //close db if open vprint.VV("restore index:%v shard:%v", indexName, shard) - _, err = io.Copy(w, rd) + n, err := io.Copy(w, rd) + vprint.VV("Written:%v", n) w.Flush() o.Close() if err != nil { _ = os.Remove(tempPath) return err } + err = db.CloseDB() + if err != nil { + return err + } vprint.VV("Rename %v to %v", tempPath, finalPath) - return os.Rename(tempPath, finalPath) + err = os.Rename(tempPath, finalPath) + if err != nil { + _ = os.Remove(tempPath) + return err + } + api.holder.recalculateCaches() + + return db.OpenDB() + } type serverInfo struct { diff --git a/bolt.go b/bolt.go index 0ce42b0f5..c0dbb600f 100644 --- a/bolt.go +++ b/bolt.go @@ -217,6 +217,23 @@ func (w *BoltWrapper) HasData() (has bool, err error) { func (w *BoltWrapper) CleanupTx(tx Tx) { // inlined into Rollback and Commit, so this is a no-op, just here to satisfy the interface. } +func (w *BoltWrapper) CloseDB() error { + w.muDb.Lock() + defer w.muDb.Unlock() + w.closed = true + return w.db.Close() +} +func (w *BoltWrapper) OpenDB() error { + w.muDb.Lock() + defer w.muDb.Unlock() + db, err := bolt.Open(w.path, 0666, &bolt.Options{Timeout: 5 * time.Second, InitialMmapSize: TxInitialMmapSize}) + if err != nil { + return err + } + w.db = db + w.closed = false + return nil +} func (tx *BoltTx) IsDone() (done bool) { return atomic.LoadInt64(&tx.unlocked) == 1 diff --git a/cmd/root.go b/cmd/root.go index a6de7527e..6fd307a04 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -64,6 +64,7 @@ at https://www.pilosa.com/docs/. rc.PersistentFlags().StringP("config", "c", "", "Configuration file to read from.") rc.AddCommand(newBackupCommand(stdin, stdout, stderr)) + rc.AddCommand(newRestoreCommand(stdin, stdout, stderr)) rc.AddCommand(newCheckCommand(stdin, stdout, stderr)) rc.AddCommand(newConfigCommand(stdin, stdout, stderr)) rc.AddCommand(newExportCommand(stdin, stdout, stderr)) diff --git a/ctl/restore.go b/ctl/restore.go index 9329f7303..2c270f00f 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -16,17 +16,15 @@ package ctl import ( "archive/tar" - "bytes" "compress/gzip" "context" "io" "os" "strings" - gohttp "net/http" - "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/http" + "github.com/pilosa/pilosa/v2/vprint" ) // RestoreCommand represents a command for restoring a backup to @@ -41,16 +39,18 @@ type RestoreCommand struct { // NewRestoreCommand returns a new instance of RestoreCommand. func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand { - h := &gohttp.Client{} - host := "SOMETHING" - c, err := http.NewInternalClient(host, h) - if err != nil { - panic(err) - } + /* + h := &gohttp.Client{} + host := "SOMETHING" + c, err := http.NewInternalClient(host, h) + if err != nil { + panic(err) + } + */ return &RestoreCommand{ - CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), - client: c, + CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), + // client: c, } } @@ -95,14 +95,16 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error { 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{} - url := "FIXME" - _, err = client.Post(url+"/schema", "application/json", bytes.NewBufferString(schemaJson)) - if err != nil { - return err - } + /* + schemaJson := readSchema(cmd.Path) + //Push the schema from the archive into the cluster belonging to the host. + client := &gohttp.Client{} + url := "FIXME" + _, err = client.Post(url+"/schema", "application/json", bytes.NewBufferString(schemaJson)) + if err != nil { + return err + } + */ //TODO (twg) load schema //TODO (twg) load rbf shard //TODO (twg) load row keys @@ -113,11 +115,44 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error { for { header, err := tarReader.Next() - //fmt.Println("What %v", header.Name) - _ = header if err == io.EOF { break } + record := strings.Split(header.Name, "/") + if len(record) == 1 { + switch record[0] { + case "schema": + vprint.VV("Load Schema") + case "idalloc": + vprint.VV("Load ids") + default: + panic("UNKNOWN " + record[0]) + + } + continue + } + indexName := record[1] + switch record[2] { + case "shards": + shard := record[3] + vprint.VV("shard %v %v", shard, indexName) + case "translate": + vprint.VV("column keys %v", indexName) + case "attributes": + vprint.VV("column attributes %v", indexName) + case "fields": + fieldName := record[3] + switch action := record[4]; action { + case "translate": + vprint.VV("field keys %v %v", indexName, fieldName) + case "attributes": + vprint.VV("field attributes %v %v", indexName, fieldName) + default: + panic("unknown:" + action) + } + + } + } /* Fetch the cluster nodes from the target host. For each index: diff --git a/dbshard.go b/dbshard.go index 8e6c92fcb..573f9b6ff 100644 --- a/dbshard.go +++ b/dbshard.go @@ -66,6 +66,9 @@ type DBWrapper interface { Path() string HasData() (has bool, err error) SetHolder(h *Holder) + //needed for restore + CloseDB() error + OpenDB() error } type DBRegistry interface { diff --git a/fragment.go b/fragment.go index 592701331..915a29556 100644 --- a/fragment.go +++ b/fragment.go @@ -2909,6 +2909,9 @@ func (f *fragment) flushCache() error { return errors.Wrap(err, "marshalling") } + if err := os.MkdirAll(filepath.Dir(f.cachePath()), 0777); err != nil { + return errors.Wrap(err, "mkdir") + } // Write to disk. if err := ioutil.WriteFile(f.cachePath(), buf, 0666); err != nil { return errors.Wrap(err, "writing") diff --git a/lattice b/lattice index 7ea3d77f8..d4bada428 160000 --- a/lattice +++ b/lattice @@ -1 +1 @@ -Subproject commit 7ea3d77f89771a06cbe59867f9135436fc8ea3b6 +Subproject commit d4bada428e45823a70432321843c38fcf2a384a0 diff --git a/rbf.go b/rbf.go index f1b4a98c6..73e8f92d1 100644 --- a/rbf.go +++ b/rbf.go @@ -571,6 +571,26 @@ func (w *RbfDBWrapper) Close() error { return w.db.Close() } +// needed to handle the special case on reload, the close method unregisters the wrapper and all that is +// required is the backing file get reloaded + +func (w *RbfDBWrapper) CloseDB() error { + w.muDb.Lock() + defer w.muDb.Unlock() + w.closed = true + return w.db.Close() +} +func (w *RbfDBWrapper) OpenDB() error { + w.muDb.Lock() + defer w.muDb.Unlock() + err := w.db.Open() + if err != nil { + return err + } + w.closed = false + return nil +} + var globalNextTxSnRBFTx int64 func (w *RbfDBWrapper) NewTx(write bool, initialIndex string, o Txo) (_ Tx, err error) { diff --git a/rrtx.go b/rrtx.go index 288b4f361..9cce080c3 100644 --- a/rrtx.go +++ b/rrtx.go @@ -663,6 +663,13 @@ func (w *RoaringWrapper) OpenSnList() (slc []int64) { return nil } +func (w *RoaringWrapper) CloseDB() error { + return errors.New("CloseDB not supported in roaring") +} +func (w *RoaringWrapper) OpenDB() error { + return errors.New("OpenDB not supported in roaring") +} + // statically confirm that RoaringTx satisfies the Tx interface. var _ Tx = (*RoaringTx)(nil)