mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-09 22:51:02 +00:00
Merge branch 'master' into reset-cache-on-schema
This commit is contained in:
Samir Patel
GitHub
commit
92ff74c72d
9 changed files with 448 additions and 290 deletions
|
|
@ -36,7 +36,8 @@ Backs up a pilosa server to a local, tar-formatted snapshot file.
|
|||
}
|
||||
|
||||
flags := ccmd.Flags()
|
||||
flags.StringVarP(&cmd.OutputPath, "output", "o", "", "output path to write to; specify '-' to send to stdout")
|
||||
flags.StringVarP(&cmd.OutputDir, "output", "o", "", "output dir to write to")
|
||||
flags.BoolVar(&cmd.NoSync, "no-sync", false, "disable file sync")
|
||||
flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of Pilosa.")
|
||||
ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification)
|
||||
return ccmd
|
||||
|
|
|
|||
220
ctl/backup.go
220
ctl/backup.go
|
|
@ -15,17 +15,14 @@
|
|||
package ctl
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/pilosa/pilosa/v2"
|
||||
"github.com/pilosa/pilosa/v2/http"
|
||||
|
|
@ -41,7 +38,10 @@ type BackupCommand struct { // nolint: maligned
|
|||
Host string `json:"host"`
|
||||
|
||||
// Path to write the backup to.
|
||||
OutputPath string
|
||||
OutputDir string
|
||||
|
||||
// If true, skips file sync.
|
||||
NoSync bool
|
||||
|
||||
// Reusable client.
|
||||
client pilosa.InternalClient
|
||||
|
|
@ -59,21 +59,12 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand
|
|||
}
|
||||
}
|
||||
|
||||
// TempPath returns the path to the temporary file to write the archive to.
|
||||
func (cmd *BackupCommand) TempPath() string {
|
||||
dir, base := filepath.Split(cmd.OutputPath)
|
||||
return filepath.Join(dir, "."+base)
|
||||
}
|
||||
|
||||
// Run executes the main program execution.
|
||||
func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
|
||||
logger := cmd.Logger()
|
||||
|
||||
// Validate arguments.
|
||||
if cmd.OutputPath == "" {
|
||||
if cmd.OutputDir == "" {
|
||||
return fmt.Errorf("-o flag required")
|
||||
}
|
||||
useStdout := cmd.OutputPath == "-"
|
||||
|
||||
// Parse TLS configuration for node-specific clients.
|
||||
tls := cmd.TLSConfiguration()
|
||||
|
|
@ -95,46 +86,23 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
|
|||
}
|
||||
schema := &pilosa.Schema{Indexes: indexes}
|
||||
|
||||
// Create output file in temporary location, or send to stdout if a dash is specified.
|
||||
var w io.Writer
|
||||
if useStdout {
|
||||
w = os.Stdout
|
||||
} else {
|
||||
f, err := os.Create(cmd.OutputPath + ".tmp")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
w = f
|
||||
// Ensure output directory doesn't exist; then create output directory.
|
||||
if _, err := os.Stat(cmd.OutputDir); !os.IsNotExist(err) {
|
||||
return fmt.Errorf("output directory already exists")
|
||||
} else if err := os.MkdirAll(cmd.OutputDir, 0777); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Open a tar writer to the temporary file.
|
||||
tw := tar.NewWriter(w)
|
||||
defer tw.Close()
|
||||
|
||||
// Backup schema.
|
||||
if err := cmd.backupSchema(ctx, tw, schema); err != nil {
|
||||
if err := cmd.backupSchema(ctx, schema); err != nil {
|
||||
return fmt.Errorf("cannot back up schema: %w", err)
|
||||
} else if err := cmd.backupIDAllocData(ctx, tw); err != nil {
|
||||
} else if err := cmd.backupIDAllocData(ctx); err != nil {
|
||||
return fmt.Errorf("cannot back up id alloc data: %w", err)
|
||||
}
|
||||
|
||||
// Backup data for each index.
|
||||
for _, ii := range schema.Indexes {
|
||||
if err := cmd.backupIndex(ctx, tw, ii); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Close archive.
|
||||
if err := tw.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Move data file to final location.
|
||||
if !useStdout {
|
||||
logger.Printf("writing backup: %s", cmd.OutputPath)
|
||||
if err := os.Rename(cmd.OutputPath+".tmp", cmd.OutputPath); err != nil {
|
||||
if err := cmd.backupIndex(ctx, ii); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
@ -143,7 +111,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) {
|
|||
}
|
||||
|
||||
// backupSchema writes the schema to the archive.
|
||||
func (cmd *BackupCommand) backupSchema(ctx context.Context, tw *tar.Writer, schema *pilosa.Schema) error {
|
||||
func (cmd *BackupCommand) backupSchema(ctx context.Context, schema *pilosa.Schema) error {
|
||||
logger := cmd.Logger()
|
||||
logger.Printf("backing up schema")
|
||||
|
||||
|
|
@ -152,22 +120,14 @@ func (cmd *BackupCommand) backupSchema(ctx context.Context, tw *tar.Writer, sche
|
|||
return fmt.Errorf("marshaling schema: %w", err)
|
||||
}
|
||||
|
||||
// Build header & copy data to archive.
|
||||
if err = tw.WriteHeader(&tar.Header{
|
||||
Name: "schema",
|
||||
Mode: 0666,
|
||||
Size: int64(len(buf)),
|
||||
ModTime: time.Now(),
|
||||
}); err != nil {
|
||||
return err
|
||||
} else if _, err := tw.Write(buf); err != nil {
|
||||
return fmt.Errorf("copying schema to archive: %w", err)
|
||||
if err := ioutil.WriteFile(filepath.Join(cmd.OutputDir, "schema"), buf, 0666); err != nil {
|
||||
return fmt.Errorf("writing schema: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *BackupCommand) backupIDAllocData(ctx context.Context, tw *tar.Writer) error {
|
||||
func (cmd *BackupCommand) backupIDAllocData(ctx context.Context) error {
|
||||
logger := cmd.Logger()
|
||||
logger.Printf("backing up id alloc data")
|
||||
|
||||
|
|
@ -177,29 +137,22 @@ func (cmd *BackupCommand) backupIDAllocData(ctx context.Context, tw *tar.Writer)
|
|||
}
|
||||
defer rc.Close()
|
||||
|
||||
// Read to buffer to determine size.
|
||||
var buf bytes.Buffer
|
||||
if _, err := buf.ReadFrom(rc); err != nil {
|
||||
return fmt.Errorf("copying id alloc data to memory: %w", err)
|
||||
}
|
||||
|
||||
// Build header & copy data to archive.
|
||||
if err = tw.WriteHeader(&tar.Header{
|
||||
Name: "idalloc",
|
||||
Mode: 0666,
|
||||
Size: int64(buf.Len()),
|
||||
ModTime: time.Now(),
|
||||
}); err != nil {
|
||||
f, err := os.Create(filepath.Join(cmd.OutputDir, "idalloc"))
|
||||
if err != nil {
|
||||
return err
|
||||
} else if _, err := io.Copy(tw, &buf); err != nil {
|
||||
return fmt.Errorf("copying id alloc data to archive: %w", err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
return nil
|
||||
if _, err := io.Copy(f, rc); err != nil {
|
||||
return err
|
||||
} else if err := cmd.syncFile(f); err != nil {
|
||||
return err
|
||||
}
|
||||
return f.Close()
|
||||
}
|
||||
|
||||
// backupIndex backs up all shards for a given index.
|
||||
func (cmd *BackupCommand) backupIndex(ctx context.Context, tw *tar.Writer, ii *pilosa.IndexInfo) error {
|
||||
func (cmd *BackupCommand) backupIndex(ctx context.Context, ii *pilosa.IndexInfo) error {
|
||||
logger := cmd.Logger()
|
||||
logger.Printf("backing up index: %q", ii.Name)
|
||||
|
||||
|
|
@ -210,19 +163,19 @@ func (cmd *BackupCommand) backupIndex(ctx context.Context, tw *tar.Writer, ii *p
|
|||
|
||||
// Back up all bitmap data for the index.
|
||||
for _, shard := range shards {
|
||||
if err := cmd.backupShard(ctx, tw, ii.Name, shard); err != nil {
|
||||
if err := cmd.backupShard(ctx, ii.Name, shard); err != nil {
|
||||
return fmt.Errorf("cannot backup shard %d on index %q: %w", shard, ii.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Back up translation data after bitmap data so we ensure we can translate all data.
|
||||
if err := cmd.backupIndexTranslateData(ctx, tw, ii.Name); err != nil {
|
||||
if err := cmd.backupIndexTranslateData(ctx, ii.Name); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Back up field translation data.
|
||||
for _, fi := range ii.Fields {
|
||||
if err := cmd.backupFieldTranslateData(ctx, tw, ii.Name, fi.Name); err != nil {
|
||||
if err := cmd.backupFieldTranslateData(ctx, ii.Name, fi.Name); err != nil {
|
||||
return fmt.Errorf("cannot backup field translation data for field %q on index %q: %w", fi.Name, ii.Name, err)
|
||||
}
|
||||
}
|
||||
|
|
@ -231,7 +184,7 @@ func (cmd *BackupCommand) backupIndex(ctx context.Context, tw *tar.Writer, ii *p
|
|||
}
|
||||
|
||||
// backupShard backs up a single shard from a single index.
|
||||
func (cmd *BackupCommand) backupShard(ctx context.Context, tw *tar.Writer, indexName string, shard uint64) (err error) {
|
||||
func (cmd *BackupCommand) backupShard(ctx context.Context, indexName string, shard uint64) (err error) {
|
||||
nodes, err := cmd.client.FragmentNodes(ctx, indexName, shard)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot determine fragment nodes: %w", err)
|
||||
|
|
@ -240,7 +193,7 @@ func (cmd *BackupCommand) backupShard(ctx context.Context, tw *tar.Writer, index
|
|||
}
|
||||
|
||||
for _, node := range nodes {
|
||||
if e := cmd.backupShardNode(ctx, tw, indexName, shard, node); e == nil {
|
||||
if e := cmd.backupShardNode(ctx, indexName, shard, node); e == nil {
|
||||
return nil // backup ok, exit
|
||||
} else if err == nil {
|
||||
err = e // save first error, try next node
|
||||
|
|
@ -250,12 +203,10 @@ func (cmd *BackupCommand) backupShard(ctx context.Context, tw *tar.Writer, index
|
|||
}
|
||||
|
||||
// backupShardNode backs up a single shard from a single index on a specific node.
|
||||
func (cmd *BackupCommand) backupShardNode(ctx context.Context, tw *tar.Writer, indexName string, shard uint64, node *topology.Node) error {
|
||||
func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, shard uint64, node *topology.Node) error {
|
||||
logger := cmd.Logger()
|
||||
logger.Printf("backing up shard: index=%q id=%d", indexName, shard)
|
||||
|
||||
filename := path.Join("indexes", indexName, "shards", fmt.Sprintf("%04d", shard))
|
||||
|
||||
client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig))
|
||||
rc, err := client.ShardReader(ctx, indexName, shard)
|
||||
if err != nil {
|
||||
|
|
@ -263,40 +214,37 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, tw *tar.Writer, i
|
|||
}
|
||||
defer rc.Close()
|
||||
|
||||
// Read to buffer to determine size.
|
||||
// TODO: Provide size via the reader itself.
|
||||
var buf bytes.Buffer
|
||||
if _, err := buf.ReadFrom(rc); err != nil {
|
||||
return fmt.Errorf("copying shard data to memory: %w", err)
|
||||
}
|
||||
|
||||
// Build header & copy data to archive.
|
||||
if err = tw.WriteHeader(&tar.Header{
|
||||
Name: filename,
|
||||
Mode: 0666,
|
||||
Size: int64(buf.Len()),
|
||||
ModTime: time.Now(),
|
||||
}); err != nil {
|
||||
filename := filepath.Join(cmd.OutputDir, "indexes", indexName, "shards", fmt.Sprintf("%04d", shard))
|
||||
if err := os.MkdirAll(filepath.Dir(filename), 0777); err != nil {
|
||||
return err
|
||||
} else if _, err := io.Copy(tw, &buf); err != nil {
|
||||
return fmt.Errorf("copying shard data to archive: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
f, err := os.Create(filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
if _, err := io.Copy(f, rc); err != nil {
|
||||
return err
|
||||
} else if err := cmd.syncFile(f); err != nil {
|
||||
return err
|
||||
}
|
||||
return f.Close()
|
||||
}
|
||||
|
||||
func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, tw *tar.Writer, name string) error {
|
||||
func (cmd *BackupCommand) backupIndexTranslateData(ctx context.Context, name string) error {
|
||||
// TODO: Fetch holder partition count.
|
||||
partitionN := topology.DefaultPartitionN
|
||||
for partitionID := 0; partitionID < partitionN; partitionID++ {
|
||||
if err := cmd.backupIndexPartitionTranslateData(ctx, tw, name, partitionID); err != nil {
|
||||
if err := cmd.backupIndexPartitionTranslateData(ctx, name, partitionID); err != nil {
|
||||
return fmt.Errorf("cannot backup index translation data for partition %d on %q: %w", partitionID, name, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context, tw *tar.Writer, name string, partitionID int) error {
|
||||
func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context, name string, partitionID int) error {
|
||||
logger := cmd.Logger()
|
||||
logger.Printf("backing up index translation data: %s/%d", name, partitionID)
|
||||
|
||||
|
|
@ -308,28 +256,26 @@ func (cmd *BackupCommand) backupIndexPartitionTranslateData(ctx context.Context,
|
|||
}
|
||||
defer rc.Close()
|
||||
|
||||
// Read to buffer to determine size.
|
||||
var buf bytes.Buffer
|
||||
if _, err := buf.ReadFrom(rc); err != nil {
|
||||
return fmt.Errorf("copying translate data to memory: %w", err)
|
||||
}
|
||||
|
||||
// Build header & copy data to archive.
|
||||
if err = tw.WriteHeader(&tar.Header{
|
||||
Name: path.Join("indexes", name, "translate", fmt.Sprintf("%04d", partitionID)),
|
||||
Mode: 0666,
|
||||
Size: int64(buf.Len()),
|
||||
ModTime: time.Now(),
|
||||
}); err != nil {
|
||||
filename := filepath.Join(cmd.OutputDir, "indexes", name, "translate", fmt.Sprintf("%04d", partitionID))
|
||||
if err := os.MkdirAll(filepath.Dir(filename), 0777); err != nil {
|
||||
return err
|
||||
} else if _, err := io.Copy(tw, &buf); err != nil {
|
||||
return fmt.Errorf("copying translate data to archive: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
f, err := os.Create(filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
if _, err := io.Copy(f, rc); err != nil {
|
||||
return err
|
||||
} else if err := cmd.syncFile(f); err != nil {
|
||||
return err
|
||||
}
|
||||
return f.Close()
|
||||
}
|
||||
|
||||
func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, tw *tar.Writer, indexName, fieldName string) error {
|
||||
func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, indexName, fieldName string) error {
|
||||
logger := cmd.Logger()
|
||||
logger.Printf("backing up field translation data: %s/%s", indexName, fieldName)
|
||||
|
||||
|
|
@ -341,24 +287,30 @@ func (cmd *BackupCommand) backupFieldTranslateData(ctx context.Context, tw *tar.
|
|||
}
|
||||
defer rc.Close()
|
||||
|
||||
// Read to buffer to determine size.
|
||||
var buf bytes.Buffer
|
||||
if _, err := buf.ReadFrom(rc); err != nil {
|
||||
return fmt.Errorf("copying translate data to memory: %w", err)
|
||||
filename := filepath.Join(cmd.OutputDir, "indexes", indexName, "fields", fieldName, "translate")
|
||||
if err := os.MkdirAll(filepath.Dir(filename), 0777); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Build header & copy data to archive.
|
||||
if err = tw.WriteHeader(&tar.Header{
|
||||
Name: path.Join("indexes", indexName, "fields", fieldName, "translate"),
|
||||
Mode: 0666,
|
||||
Size: int64(buf.Len()),
|
||||
ModTime: time.Now(),
|
||||
}); err != nil {
|
||||
f, err := os.Create(filename)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if _, err := io.Copy(tw, &buf); err != nil {
|
||||
return fmt.Errorf("copying translate data to archive: %w", err)
|
||||
}
|
||||
return nil
|
||||
defer f.Close()
|
||||
|
||||
if _, err := io.Copy(f, rc); err != nil {
|
||||
return err
|
||||
} else if err := cmd.syncFile(f); err != nil {
|
||||
return err
|
||||
}
|
||||
return f.Close()
|
||||
}
|
||||
|
||||
func (cmd *BackupCommand) syncFile(f *os.File) error {
|
||||
if cmd.NoSync {
|
||||
return nil
|
||||
}
|
||||
return f.Sync()
|
||||
}
|
||||
|
||||
func (cmd *BackupCommand) TLSHost() string { return cmd.Host }
|
||||
|
|
|
|||
355
ctl/restore.go
355
ctl/restore.go
|
|
@ -15,17 +15,14 @@
|
|||
package ctl
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
gohttp "net/http"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
|
|
@ -65,19 +62,6 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
|
|||
if cmd.Path == "" {
|
||||
return fmt.Errorf("-s flag required")
|
||||
}
|
||||
useStdin := cmd.Path == "-"
|
||||
|
||||
var f *os.File
|
||||
// read from Stdin if path specified as -
|
||||
if useStdin {
|
||||
f = os.Stdin
|
||||
} else {
|
||||
f, err = os.Open(cmd.Path)
|
||||
if err != nil {
|
||||
return (err)
|
||||
}
|
||||
defer f.Close()
|
||||
}
|
||||
|
||||
// Parse TLS configuration for node-specific clients.
|
||||
tls := cmd.TLSConfiguration()
|
||||
|
|
@ -90,16 +74,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
|
|||
return fmt.Errorf("creating client: %w", err)
|
||||
}
|
||||
cmd.client = client
|
||||
var tarReader *tar.Reader
|
||||
if strings.HasSuffix(cmd.Path, "gz") {
|
||||
gzf, err := gzip.NewReader(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tarReader = tar.NewReader(gzf)
|
||||
} else {
|
||||
tarReader = tar.NewReader(f)
|
||||
}
|
||||
|
||||
nodes, err := cmd.client.Nodes(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -116,126 +91,21 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
|
|||
if primary == nil {
|
||||
return errors.New("no primary")
|
||||
}
|
||||
c := &gohttp.Client{}
|
||||
for {
|
||||
header, err := tarReader.Next()
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
record := strings.Split(header.Name, "/")
|
||||
if len(record) == 1 {
|
||||
switch record[0] {
|
||||
case "schema":
|
||||
logger.Printf("Load Schema")
|
||||
url := primary.URI.Path("/schema")
|
||||
_, err = c.Post(url, "application/json", tarReader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
case "idalloc":
|
||||
logger.Printf("Load ids")
|
||||
url := primary.URI.Path("/internal/idalloc/restore")
|
||||
_, err = c.Post(url, "application/octet-stream", tarReader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
return err
|
||||
|
||||
}
|
||||
continue
|
||||
}
|
||||
indexName := record[1]
|
||||
switch record[2] {
|
||||
case "shards":
|
||||
shard, err := strconv.ParseUint(record[3], 10, 64)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
nodes, err := cmd.client.FragmentNodes(ctx, indexName, shard)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot determine fragment nodes: %w", err)
|
||||
} else if len(nodes) == 0 {
|
||||
return fmt.Errorf("no nodes available")
|
||||
}
|
||||
|
||||
shardBytes, err := ioutil.ReadAll(tarReader) // this feels wrong but works for now
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g, _ := errgroup.WithContext(ctx)
|
||||
for _, node := range nodes {
|
||||
node := node
|
||||
g.Go(func() error {
|
||||
client := &gohttp.Client{}
|
||||
rd := bytes.NewReader(shardBytes)
|
||||
logger.Printf("shard %v %v", shard, indexName)
|
||||
url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard))
|
||||
_, err = client.Post(url, "application/octet-stream", rd)
|
||||
return err
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
case "translate":
|
||||
partitionID, err := strconv.Atoi(record[3])
|
||||
logger.Printf("column keys %v (%v)", indexName, partitionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
partitionNodes, err := cmd.client.PartitionNodes(ctx, partitionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
shardBytes, err := ioutil.ReadAll(tarReader) // this feels wrong but works for now
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g, _ := errgroup.WithContext(ctx)
|
||||
for _, node := range partitionNodes {
|
||||
node := node
|
||||
g.Go(func() error {
|
||||
rd := bytes.NewReader(shardBytes)
|
||||
return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, rd)
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
case "attributes":
|
||||
//skip
|
||||
case "fields":
|
||||
fieldName := record[3]
|
||||
switch action := record[4]; action {
|
||||
case "translate":
|
||||
logger.Printf("field keys %v %v", indexName, fieldName)
|
||||
//needs to go to all nodes
|
||||
shardBytes, err := ioutil.ReadAll(tarReader) // this feels wrong but works for now
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
g, _ := errgroup.WithContext(ctx)
|
||||
for _, node := range nodes {
|
||||
node := node
|
||||
g.Go(func() error {
|
||||
rd := bytes.NewReader(shardBytes)
|
||||
return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, rd)
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
case "attributes":
|
||||
//skip
|
||||
default:
|
||||
return fmt.Errorf("unknown restore action: %v", action)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if err := cmd.restoreSchema(ctx, primary); err != nil {
|
||||
return fmt.Errorf("cannot restore schema: %w", err)
|
||||
} else if err := cmd.restoreIDAlloc(ctx, primary); err != nil {
|
||||
return fmt.Errorf("cannot restore idalloc: %w", err)
|
||||
}
|
||||
|
||||
if err := cmd.restoreShards(ctx); err != nil {
|
||||
return fmt.Errorf("cannot restore shards: %w", err)
|
||||
} else if err := cmd.restoreIndexTranslation(ctx); err != nil {
|
||||
return fmt.Errorf("cannot restore index translation: %w", err)
|
||||
} else if err := cmd.restoreFieldTranslation(ctx, nodes); err != nil {
|
||||
return fmt.Errorf("cannot restore field translation: %w", err)
|
||||
}
|
||||
|
||||
/* Fetch the cluster nodes from the target host.
|
||||
For each index:
|
||||
Upload the RBF snapshot for each shard to the nodes that own the shard.
|
||||
|
|
@ -245,6 +115,199 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) {
|
|||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology.Node) error {
|
||||
f, err := os.Open(filepath.Join(cmd.Path, "schema"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
cmd.Logger().Printf("Load Schema")
|
||||
url := primary.URI.Path("/schema")
|
||||
|
||||
var client http.Client
|
||||
_, err = client.Post(url, "application/json", f)
|
||||
return err
|
||||
}
|
||||
|
||||
func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error {
|
||||
logger := cmd.Logger()
|
||||
|
||||
f, err := os.Open(filepath.Join(cmd.Path, "idalloc"))
|
||||
if os.IsNotExist(err) {
|
||||
logger.Printf("No idalloc, skipping")
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
logger.Printf("Load idalloc")
|
||||
url := primary.URI.Path("/internal/idalloc/restore")
|
||||
|
||||
var client http.Client
|
||||
_, err = client.Post(url, "application/octet-stream", f)
|
||||
return err
|
||||
}
|
||||
|
||||
func (cmd *RestoreCommand) restoreShards(ctx context.Context) error {
|
||||
logger := cmd.Logger()
|
||||
|
||||
filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "shards", "*"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, filename := range filenames {
|
||||
rel, err := filepath.Rel(cmd.Path, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
record := strings.Split(rel, string(os.PathSeparator))
|
||||
indexName := record[1]
|
||||
|
||||
shard, err := strconv.ParseUint(record[3], 10, 64)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
nodes, err := cmd.client.FragmentNodes(ctx, indexName, shard)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot determine fragment nodes: %w", err)
|
||||
} else if len(nodes) == 0 {
|
||||
return fmt.Errorf("no nodes available")
|
||||
}
|
||||
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
for _, node := range nodes {
|
||||
node := node
|
||||
g.Go(func() error {
|
||||
logger.Printf("shard %v %v", shard, indexName)
|
||||
|
||||
f, err := os.Open(filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard))
|
||||
req, err := http.NewRequest("POST", url, f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req = req.WithContext(ctx)
|
||||
req.Header.Set("Content-Type", "application/octet-stream")
|
||||
|
||||
var client http.Client
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if err := resp.Body.Close(); err != nil {
|
||||
return err
|
||||
} else if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *RestoreCommand) restoreIndexTranslation(ctx context.Context) error {
|
||||
logger := cmd.Logger()
|
||||
|
||||
filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "translate", "*"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, filename := range filenames {
|
||||
rel, err := filepath.Rel(cmd.Path, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
record := strings.Split(rel, string(os.PathSeparator))
|
||||
indexName := record[1]
|
||||
|
||||
partitionID, err := strconv.Atoi(record[3])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logger.Printf("column keys %v (%v)", indexName, partitionID)
|
||||
|
||||
nodes, err := cmd.client.PartitionNodes(ctx, partitionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
for _, node := range nodes {
|
||||
node := node
|
||||
g.Go(func() error {
|
||||
f, err := os.Open(filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, f)
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *RestoreCommand) restoreFieldTranslation(ctx context.Context, nodes []*topology.Node) error {
|
||||
logger := cmd.Logger()
|
||||
|
||||
filenames, err := filepath.Glob(filepath.Join(cmd.Path, "indexes", "*", "fields", "*", "translate"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, filename := range filenames {
|
||||
rel, err := filepath.Rel(cmd.Path, filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
record := strings.Split(rel, string(os.PathSeparator))
|
||||
indexName, fieldName := record[1], record[3]
|
||||
|
||||
logger.Printf("field keys %v %v", indexName, fieldName)
|
||||
|
||||
g, ctx := errgroup.WithContext(ctx)
|
||||
for _, node := range nodes {
|
||||
node := node
|
||||
g.Go(func() error {
|
||||
f, err := os.Open(filename)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, f)
|
||||
})
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *RestoreCommand) TLSHost() string { return cmd.Host }
|
||||
|
||||
func (cmd *RestoreCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS }
|
||||
|
|
|
|||
|
|
@ -126,3 +126,15 @@
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
.infoMessage {
|
||||
padding: 16px;
|
||||
border: 1px solid rgba(var(--primary-rgb), 0.5);
|
||||
background: rgba(var(--primary-rgb), 0.1);
|
||||
border-radius: 4px;
|
||||
margin: 4px 0 16px;
|
||||
|
||||
.infoTooltip {
|
||||
border-bottom: 1px dashed rgba(var(--primary-rgb), 0.7);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,8 +4,10 @@ import Breadcrumbs from '@material-ui/core/Breadcrumbs';
|
|||
import classNames from 'classnames';
|
||||
import Fuse from 'fuse.js';
|
||||
import Highlighter from 'react-highlight-words';
|
||||
import isEmpty from 'lodash/isEmpty';
|
||||
import Link from '@material-ui/core/Link';
|
||||
import map from 'lodash/map';
|
||||
import moment from 'moment';
|
||||
import OrderBy from 'lodash/orderBy';
|
||||
import Reduce from 'lodash/reduce';
|
||||
import Table from '@material-ui/core/Table';
|
||||
|
|
@ -14,6 +16,7 @@ import TableCell from '@material-ui/core/TableCell';
|
|||
import TableHead from '@material-ui/core/TableHead';
|
||||
import TableRow from '@material-ui/core/TableRow';
|
||||
import TextField from '@material-ui/core/TextField';
|
||||
import Tooltip from '@material-ui/core/Tooltip';
|
||||
import Typography from '@material-ui/core/Typography';
|
||||
import { Block } from 'shared/Block';
|
||||
import { Pager } from 'shared/Pager';
|
||||
|
|
@ -23,11 +26,13 @@ import css from './MoleculaTable.module.scss';
|
|||
type MoleculaTableProps = {
|
||||
table: any;
|
||||
dataDistribution: any;
|
||||
lastUpdated: string;
|
||||
};
|
||||
|
||||
export const MoleculaTable: FC<MoleculaTableProps> = ({
|
||||
table,
|
||||
dataDistribution
|
||||
dataDistribution,
|
||||
lastUpdated
|
||||
}) => {
|
||||
const [page, setPage] = useState<number>(1);
|
||||
const [resultsPerPage, setResultsPerPage] = useState<number>(10);
|
||||
|
|
@ -38,9 +43,10 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
|
|||
const [maxFieldSize, setMaxFieldSize] = useState<number>(0);
|
||||
const [sort, setSort] = useState<string>('total');
|
||||
const [sortDir, setSortDir] = useState<'asc' | 'desc'>('desc');
|
||||
const lastUpdatedMoment = lastUpdated ? moment(lastUpdated).utc() : undefined;
|
||||
|
||||
useEffect(() => {
|
||||
if (dataDistribution) {
|
||||
if (dataDistribution && !dataDistribution.uncached) {
|
||||
const aggregatedFieldsData = Reduce(
|
||||
dataDistribution.fields,
|
||||
(result, value) => {
|
||||
|
|
@ -77,7 +83,7 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
|
|||
threshold: 0
|
||||
});
|
||||
const result = fuse.search(searchText);
|
||||
|
||||
|
||||
let resultsArray: any[] = [];
|
||||
result.forEach((r: any) => {
|
||||
resultsArray.push({ ...r?.item, ...fieldsData[r?.item.name] });
|
||||
|
|
@ -125,6 +131,46 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
|
|||
<Typography variant="h5" color="textSecondary">
|
||||
{table.name}
|
||||
</Typography>
|
||||
{lastUpdatedMoment ? (
|
||||
<div className={css.infoMessage}>
|
||||
{dataDistribution && dataDistribution.uncached ? (
|
||||
<Fragment>
|
||||
Disk usage will be calculated at the next{` `}
|
||||
<Tooltip
|
||||
title={
|
||||
<Fragment>
|
||||
Disk and memory information shown here are read from a
|
||||
cache, the behavior of which can be controlled with the{` `}
|
||||
<code style={{ whiteSpace: 'nowrap' }}>
|
||||
--usage-duty-cycle
|
||||
</code>{' '}
|
||||
command line flag.
|
||||
</Fragment>
|
||||
}
|
||||
placement="top"
|
||||
arrow
|
||||
>
|
||||
<span className={css.infoTooltip}>cache refresh</span>
|
||||
</Tooltip>
|
||||
.
|
||||
</Fragment>
|
||||
) : (
|
||||
<Fragment>
|
||||
Disk usage last updated{' '}
|
||||
<Tooltip
|
||||
title={`${lastUpdatedMoment.format('M/D/YYYY hh:mm a')} UTC`}
|
||||
placement="top"
|
||||
arrow
|
||||
>
|
||||
<span className={css.infoTooltip}>
|
||||
{lastUpdatedMoment.fromNow()}
|
||||
</span>
|
||||
</Tooltip>
|
||||
.
|
||||
</Fragment>
|
||||
)}
|
||||
</div>
|
||||
) : null}
|
||||
<div className={css.layout}>
|
||||
<div>
|
||||
<label className={css.label}>keys</label>
|
||||
|
|
@ -251,7 +297,15 @@ export const MoleculaTable: FC<MoleculaTableProps> = ({
|
|||
</TableCell>
|
||||
<TableCell className={css.tableCell}>
|
||||
<UsageBreakdown
|
||||
data={field}
|
||||
data={
|
||||
isEmpty(field)
|
||||
? field
|
||||
: dataDistribution
|
||||
? dataDistribution.uncached
|
||||
? dataDistribution
|
||||
: field
|
||||
: field
|
||||
}
|
||||
width={`${(field.total / maxFieldSize) * 150}px`}
|
||||
showLabel={false}
|
||||
usageValueSize="small"
|
||||
|
|
|
|||
|
|
@ -69,3 +69,15 @@
|
|||
.pilosaError {
|
||||
padding: 8px 16px;
|
||||
}
|
||||
|
||||
.infoMessage {
|
||||
padding: 16px;
|
||||
border: 1px solid rgba(var(--primary-rgb), 0.5);
|
||||
background: rgba(var(--primary-rgb), 0.1);
|
||||
border-radius: 4px;
|
||||
margin: 4px 0 16px;
|
||||
|
||||
.infoTooltip {
|
||||
border-bottom: 1px dashed rgba(var(--primary-rgb), 0.7);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,10 @@
|
|||
import React, { FC, Fragment, useEffect, useState } from 'react';
|
||||
import Card from '@material-ui/core/Card';
|
||||
import CardContent from '@material-ui/core/CardContent';
|
||||
import moment from 'moment';
|
||||
import OrderBy from 'lodash/orderBy';
|
||||
import Paper from '@material-ui/core/Paper';
|
||||
import Tooltip from '@material-ui/core/Tooltip';
|
||||
import Typography from '@material-ui/core/Typography';
|
||||
import { Block } from 'shared/Block';
|
||||
import { SortBy } from 'shared/SortBy';
|
||||
|
|
@ -13,16 +15,19 @@ import css from './MoleculaTables.module.scss';
|
|||
type MoleculaTablesProps = {
|
||||
tables: any;
|
||||
dataDistribution: any;
|
||||
lastUpdated: string;
|
||||
maxSize: number;
|
||||
};
|
||||
|
||||
export const MoleculaTables: FC<MoleculaTablesProps> = ({
|
||||
tables,
|
||||
dataDistribution,
|
||||
lastUpdated,
|
||||
maxSize
|
||||
}) => {
|
||||
const history = useHistory();
|
||||
const [sortedTables, setSortedTables] = useState<any>([]);
|
||||
const lastUpdatedMoment = lastUpdated ? moment(lastUpdated).utc() : undefined;
|
||||
|
||||
useEffect(() => {
|
||||
if (tables && dataDistribution) {
|
||||
|
|
@ -51,6 +56,38 @@ export const MoleculaTables: FC<MoleculaTablesProps> = ({
|
|||
<Typography variant="h5" color="textSecondary">
|
||||
Tables
|
||||
</Typography>
|
||||
{lastUpdatedMoment ? (
|
||||
<div className={css.infoMessage}>
|
||||
Disk usage last updated{' '}
|
||||
<Tooltip
|
||||
title={`${lastUpdatedMoment.format('M/D/YYYY hh:mm a')} UTC`}
|
||||
placement="top"
|
||||
arrow
|
||||
>
|
||||
<span className={css.infoTooltip}>
|
||||
{lastUpdatedMoment.fromNow()}
|
||||
</span>
|
||||
</Tooltip>
|
||||
. Disk usage for new tables will be calculated at the next{` `}
|
||||
<Tooltip
|
||||
title={
|
||||
<Fragment>
|
||||
Disk and memory information shown here are read from a cache,
|
||||
the behavior of which can be controlled with the{` `}
|
||||
<code style={{ whiteSpace: 'nowrap' }}>
|
||||
--usage-duty-cycle
|
||||
</code>{' '}
|
||||
command line flag.
|
||||
</Fragment>
|
||||
}
|
||||
placement="top"
|
||||
arrow
|
||||
>
|
||||
<span className={css.infoTooltip}>cache refresh</span>
|
||||
</Tooltip>
|
||||
.
|
||||
</div>
|
||||
) : null}
|
||||
<div className={css.actions}>
|
||||
<SortBy
|
||||
options={[
|
||||
|
|
@ -77,10 +114,14 @@ export const MoleculaTables: FC<MoleculaTablesProps> = ({
|
|||
<div className={css.section}>
|
||||
<UsageBreakdown
|
||||
data={
|
||||
dataDistribution ? dataDistribution[name] : undefined
|
||||
dataDistribution
|
||||
? dataDistribution[name]
|
||||
? dataDistribution[name]
|
||||
: { uncached: true }
|
||||
: undefined
|
||||
}
|
||||
width={
|
||||
dataDistribution
|
||||
dataDistribution && dataDistribution[name]
|
||||
? `${(dataDistribution[name].total / maxSize) * 100}%`
|
||||
: '0px'
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ export const MoleculaTablesContainer = () => {
|
|||
const [selectedTable, setSelectedTable] = useState<any>();
|
||||
const [dataDistribution, setDataDistribution] = useState<any>();
|
||||
const [maxSize, setMaxSize] = useState<number>(0);
|
||||
const [lastUpdated, setLastUpdated] = useState<string>('');
|
||||
|
||||
useEffectOnce(() => {
|
||||
pilosa.get
|
||||
|
|
@ -25,7 +26,7 @@ export const MoleculaTablesContainer = () => {
|
|||
.then((res) => setTables(res.data.indexes))
|
||||
.catch((err) => console.log(err))
|
||||
);
|
||||
|
||||
|
||||
pilosa.get.usage().then((res) => {
|
||||
const nodes = Object.keys(res.data);
|
||||
let data = {};
|
||||
|
|
@ -54,6 +55,10 @@ export const MoleculaTablesContainer = () => {
|
|||
};
|
||||
}
|
||||
});
|
||||
|
||||
if(!lastUpdated) {
|
||||
setLastUpdated(res.data[node].lastUpdated);
|
||||
}
|
||||
});
|
||||
|
||||
const sorted = OrderBy(data, ['total'], ['desc']);
|
||||
|
|
@ -83,13 +88,19 @@ export const MoleculaTablesContainer = () => {
|
|||
<MoleculaTable
|
||||
table={selectedTable}
|
||||
dataDistribution={
|
||||
dataDistribution ? dataDistribution[selectedTable.name] : undefined
|
||||
dataDistribution
|
||||
? dataDistribution[selectedTable.name]
|
||||
? dataDistribution[selectedTable.name]
|
||||
: { uncached: true }
|
||||
: undefined
|
||||
}
|
||||
lastUpdated={lastUpdated}
|
||||
/>
|
||||
) : (
|
||||
<MoleculaTables
|
||||
tables={tables}
|
||||
dataDistribution={dataDistribution}
|
||||
lastUpdated={lastUpdated}
|
||||
maxSize={maxSize}
|
||||
/>
|
||||
);
|
||||
|
|
|
|||
|
|
@ -18,7 +18,15 @@ export const UsageBreakdown: FC<UsageBreakdownProps> = ({
|
|||
showLabel = true,
|
||||
usageValueSize = 'medium'
|
||||
}) => {
|
||||
const { total, fieldKeysTotal, indexKeys, fragments, metadata, keys } = data;
|
||||
const {
|
||||
total,
|
||||
fieldKeysTotal,
|
||||
indexKeys,
|
||||
fragments,
|
||||
metadata,
|
||||
keys,
|
||||
uncached
|
||||
} = data;
|
||||
const fieldKeysPercentage =
|
||||
fieldKeysTotal && total ? (fieldKeysTotal / total) * 100 : 0;
|
||||
const indexKeysPercentage = indexKeys ? (indexKeys / total) * 100 : 0;
|
||||
|
|
@ -160,6 +168,10 @@ export const UsageBreakdown: FC<UsageBreakdownProps> = ({
|
|||
) : null}
|
||||
</div>
|
||||
</Fragment>
|
||||
) : uncached ? (
|
||||
<Typography variant="caption" component="div">
|
||||
Waiting...
|
||||
</Typography>
|
||||
) : (
|
||||
<Typography variant="caption" component="div">
|
||||
Calculating...
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue