From 660972ba5d87dd2a62e352fb0365a6aa65b5b74f Mon Sep 17 00:00:00 2001 From: Hoang Pham Date: Thu, 25 Aug 2022 18:50:35 -0500 Subject: [PATCH] FB-1627 - Added backup/restore tar. Purpose: for cloud team to backup and restore directly through stdout, stdin --- cmd/backup_tar.go | 36 ++++ cmd/restore_tar.go | 40 ++++ cmd/root.go | 2 + ctl/backup_tar.go | 409 ++++++++++++++++++++++++++++++++++++++++ ctl/backup_tar_test.go | 40 ++++ ctl/restore_tar.go | 270 ++++++++++++++++++++++++++ ctl/restore_tar_test.go | 39 ++++ 7 files changed, 836 insertions(+) create mode 100644 cmd/backup_tar.go create mode 100644 cmd/restore_tar.go create mode 100644 ctl/backup_tar.go create mode 100644 ctl/backup_tar_test.go create mode 100644 ctl/restore_tar.go create mode 100644 ctl/restore_tar_test.go diff --git a/cmd/backup_tar.go b/cmd/backup_tar.go new file mode 100644 index 000000000..e480a487c --- /dev/null +++ b/cmd/backup_tar.go @@ -0,0 +1,36 @@ +// Copyright 2022 Molecula Corp. All rights reserved. +package cmd + +import ( + "context" + "io" + + "github.com/molecula/featurebase/v3/ctl" + "github.com/spf13/cobra" +) + +func newBackupTarCommand(stdin io.Reader, stdout io.Writer, stderr io.Writer) *cobra.Command { + cmd := ctl.NewBackupTarCommand(stdin, stdout, stderr) + ccmd := &cobra.Command{ + Use: "backuptar", + Short: "Back up FeatureBase server in tar format", + Long: ` +Backs up a FeatureBase server to a local, tar-formatted snapshot file. +`, + RunE: func(c *cobra.Command, args []string) error { + return cmd.Run(context.Background()) + }, + } + + flags := ccmd.Flags() + flags.StringVarP(&cmd.OutputPath, "output", "o", "", "Output directory to write to.") + flags.StringVar(&cmd.Host, "host", "localhost:10101", "The address (host:port) of FeatureBase (HTTP).") + flags.StringVar(&cmd.Index, "index", "", "Index to backup, default backs up all indexes. ") + flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.") + flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.") + ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification) + flags.StringVar(&cmd.AuthToken, "auth-token", "", "Authentication token") + flags.StringVar(&cmd.HeaderTimeoutStr, "header-timeout", cmd.HeaderTimeoutStr, "Length of time to wait for initial HTTP response before giving up.") + + return ccmd +} diff --git a/cmd/restore_tar.go b/cmd/restore_tar.go new file mode 100644 index 000000000..563487429 --- /dev/null +++ b/cmd/restore_tar.go @@ -0,0 +1,40 @@ +// Copyright 2022 Molecula Corp. All rights reserved. +package cmd + +import ( + "context" + "io" + + "github.com/molecula/featurebase/v3/ctl" + "github.com/spf13/cobra" +) + +func newRestoreTarCommand(stdin io.Reader, stdout, stderr io.Writer) *cobra.Command { + cmd := ctl.NewRestoreTarCommand(stdin, stdout, stderr) + restoreCmd := &cobra.Command{ + Use: "restoretar", + Short: "Restore from a backup in tar format", + Long: ` +The Restore command will take a tar-formatted backup archive and restore it to a new, clean cluster. +`, + RunE: func(c *cobra.Command, args []string) error { + return cmd.Run(context.Background()) + }, + } + flags := restoreCmd.Flags() + flags.StringVarP(&cmd.Path, "source", "s", "", "backup file; specify '-' to restore from stdin tar stream") + flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.") + flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.") + flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.") + flags.StringVar(&cmd.AuthToken, "auth-token", "", "Authentication token") + ctl.SetTLSConfig( + flags, "", + &cmd.TLS.CertificatePath, + &cmd.TLS.CertificateKeyPath, + &cmd.TLS.CACertPath, + &cmd.TLS.SkipVerify, + &cmd.TLS.EnableClientVerification, + ) + + return restoreCmd +} diff --git a/cmd/root.go b/cmd/root.go index b40f7562a..e69987737 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -54,6 +54,8 @@ at https://docs.featurebase.com/. rc.AddCommand(newChkSumCommand(stdin, stdout, stderr)) rc.AddCommand(newBackupCommand(stdin, stdout, stderr)) rc.AddCommand(newRestoreCommand(stdin, stdout, stderr)) + rc.AddCommand(newBackupTarCommand(stdin, stdout, stderr)) + rc.AddCommand(newRestoreTarCommand(stdin, stdout, stderr)) rc.AddCommand(newConfigCommand(stdin, stdout, stderr)) rc.AddCommand(newExportCommand(stdin, stdout, stderr)) rc.AddCommand(newGenerateConfigCommand(stdin, stdout, stderr)) diff --git a/ctl/backup_tar.go b/ctl/backup_tar.go new file mode 100644 index 000000000..86d68edb6 --- /dev/null +++ b/ctl/backup_tar.go @@ -0,0 +1,409 @@ +// Copyright 2022 Molecula Corp. All rights reserved. +package ctl + +import ( + "archive/tar" + "bytes" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "os" + "path" + "time" + + pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/authn" + "github.com/molecula/featurebase/v3/disco" + "github.com/molecula/featurebase/v3/encoding/proto" + "github.com/molecula/featurebase/v3/server" + "github.com/pkg/errors" +) + +// BackupTarCommand represents a command for backing up a Pilosa node. +type BackupTarCommand struct { // nolint: maligned + tlsConfig *tls.Config + + // Destination host and port. + Host string `json:"host"` + + // Optional Index filter + Index string `json:"index"` + + // Path to write the backup to. + OutputPath string + + // Amount of time after first failed request to continue retrying. + RetryPeriod time.Duration `json:"retry-period"` + + // Response Header Timeout for HTTP Requests + HeaderTimeoutStr string + HeaderTimeout time.Duration `json:"header-timeout"` + + // Host:port on which to listen for pprof. + Pprof string `json:"pprof"` + + // Reusable client. + client *pilosa.InternalClient + + // Standard input/output + *pilosa.CmdIO + + TLS server.TLSConfig + + AuthToken string +} + +// NewBackupTarCommand returns a new instance of BackupCommand. +func NewBackupTarCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupTarCommand { + return &BackupTarCommand{ + CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), + RetryPeriod: time.Minute, + HeaderTimeout: time.Second * 3, + Pprof: "localhost:0", + } +} + +// Run executes the main program execution. +func (cmd *BackupTarCommand) Run(ctx context.Context) (err error) { + logger := cmd.Logger() + close, err := startProfilingServer(cmd.Pprof, logger) + if err != nil { + return errors.Wrap(err, "starting profiling server") + } + defer close() + + // Validate arguments. + if cmd.OutputPath == "" { + return fmt.Errorf("-o flag required") + } + useStdout := cmd.OutputPath == "-" + + if cmd.HeaderTimeoutStr != "" { + if dur, err := time.ParseDuration(cmd.HeaderTimeoutStr); err != nil { + return fmt.Errorf("could not parse '%s' as a duration: %v", cmd.HeaderTimeoutStr, err) + } else { + cmd.HeaderTimeout = dur + } + } + + // Parse TLS configuration for node-specific clients. + tls := cmd.TLSConfiguration() + if cmd.tlsConfig, err = server.GetTLSConfig(&tls, cmd.Logger()); err != nil { + return fmt.Errorf("parsing tls config: %w", err) + } + + // Create a client to the server. + client, err := commandClient(cmd, pilosa.WithClientRetryPeriod(cmd.RetryPeriod), pilosa.ClientResponseHeaderTimeoutOption(cmd.HeaderTimeout)) + if err != nil { + return fmt.Errorf("creating client: %w", err) + } + cmd.client = client + + if cmd.AuthToken != "" { + ctx = context.WithValue( + ctx, + authn.ContextValueAccessToken, + "Bearer "+cmd.AuthToken, + ) + } + + // Determine the field type in order to correctly handle the input data. + indexes, err := cmd.client.Schema(ctx) + if err != nil { + return fmt.Errorf("getting schema: %w", err) + } + if cmd.Index != "" { + for _, idx := range indexes { + if idx.Name == cmd.Index { + indexes = make([]*pilosa.IndexInfo, 0) + indexes = append(indexes, idx) + break + } + } + if len(indexes) <= 0 { + return fmt.Errorf("index not found to back up") + } + } + 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 + } + + // Open a tar writer to the temporary file. + tw := tar.NewWriter(w) + defer tw.Close() + + // Backup schema. + if err := cmd.backupTarSchema(ctx, tw, schema); err != nil { + return fmt.Errorf("cannot back up schema: %w", err) + } else if err := cmd.backupTarIDAllocData(ctx, tw); 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.backupTarIndex(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 { + return err + } + } + + return nil +} + +// backupTarSchema writes the schema to the archive. +func (cmd *BackupTarCommand) backupTarSchema(ctx context.Context, tw *tar.Writer, schema *pilosa.Schema) error { + logger := cmd.Logger() + logger.Printf("backing up schema") + + buf, err := json.MarshalIndent(schema, "", "\t") + if err != nil { + 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) + } + + return nil +} + +func (cmd *BackupTarCommand) backupTarIDAllocData(ctx context.Context, tw *tar.Writer) error { + logger := cmd.Logger() + logger.Printf("backing up id alloc data") + + rc, err := cmd.client.IDAllocDataReader(ctx) + if err != nil { + return fmt.Errorf("fetching id alloc data reader: %w", err) + } + 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 { + return err + } else if _, err := io.Copy(tw, &buf); err != nil { + return fmt.Errorf("copying id alloc data to archive: %w", err) + } + + return nil +} + +// backupTarIndex backs up all shards for a given index. +func (cmd *BackupTarCommand) backupTarIndex(ctx context.Context, tw *tar.Writer, ii *pilosa.IndexInfo) error { + logger := cmd.Logger() + logger.Printf("backing up index: %q", ii.Name) + + shards, err := cmd.client.AvailableShards(ctx, ii.Name) + if err != nil { + return fmt.Errorf("cannot find available shards for index %q: %w", ii.Name, err) + } + + // Back up all bitmap data for the index. + for _, shard := range shards { + if err := cmd.backupTarShard(ctx, tw, ii.Name, shard); err != nil { + return fmt.Errorf("cannot backup shard %d on index %q: %w", shard, ii.Name, err) + } + } + + if ii.Options.Keys { + // Back up translation data after bitmap data so we ensure we can translate all data. + if err := cmd.backupTarIndexTranslateData(ctx, tw, ii.Name); err != nil { + return err + } + + } + + // Back up field translation data. + for _, fi := range ii.Fields { + if !fi.Options.Keys { + continue + } + if err := cmd.backupTarFieldTranslateData(ctx, tw, 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) + } + } + + return nil +} + +// backupTarShard backs up a single shard from a single index. +func (cmd *BackupTarCommand) backupTarShard(ctx context.Context, tw *tar.Writer, 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) + } else if len(nodes) == 0 { + return fmt.Errorf("no nodes available") + } + + for _, node := range nodes { + if e := cmd.backupTarShardNode(ctx, tw, indexName, shard, node); e == nil { + return nil // backup ok, exit + } else if err == nil { + err = e // save first error, try next node + } + } + return err +} + +// backupTarShardNode backs up a single shard from a single index on a specific node. +func (cmd *BackupTarCommand) backupTarShardNode(ctx context.Context, tw *tar.Writer, indexName string, shard uint64, node *disco.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 := pilosa.NewInternalClientFromURI(&node.URI, + pilosa.GetHTTPClient(cmd.tlsConfig, pilosa.ClientResponseHeaderTimeoutOption(cmd.HeaderTimeout)), + pilosa.WithClientRetryPeriod(cmd.RetryPeriod), + pilosa.WithSerializer(proto.Serializer{})) + rc, err := client.ShardReader(ctx, indexName, shard) + + if err != nil { + return fmt.Errorf("fetching shard reader: %w", err) + } + 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 { + return err + } else if _, err := io.Copy(tw, &buf); err != nil { + return fmt.Errorf("copying shard data to archive: %w", err) + } + + return nil +} + +func (cmd *BackupTarCommand) backupTarIndexTranslateData(ctx context.Context, tw *tar.Writer, name string) error { + // TODO: Fetch holder partition count. + partitionN := disco.DefaultPartitionN + for partitionID := 0; partitionID < partitionN; partitionID++ { + if err := cmd.backupTarIndexPartitionTranslateData(ctx, tw, 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 *BackupTarCommand) backupTarIndexPartitionTranslateData(ctx context.Context, tw *tar.Writer, name string, partitionID int) error { + logger := cmd.Logger() + logger.Printf("backing up index translation data: %s/%d", name, partitionID) + + rc, err := cmd.client.IndexTranslateDataReader(ctx, name, partitionID) + if err == pilosa.ErrTranslateStoreNotFound { + return nil + } else if err != nil { + return fmt.Errorf("fetching translate data reader: %w", err) + } + 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 { + return err + } else if _, err := io.Copy(tw, &buf); err != nil { + return fmt.Errorf("copying translate data to archive: %w", err) + } + + return nil +} + +func (cmd *BackupTarCommand) backupTarFieldTranslateData(ctx context.Context, tw *tar.Writer, indexName, fieldName string) error { + logger := cmd.Logger() + logger.Printf("backing up field translation data: %s/%s", indexName, fieldName) + + rc, err := cmd.client.FieldTranslateDataReader(ctx, indexName, fieldName) + if err == pilosa.ErrTranslateStoreNotFound { + return nil + } else if err != nil { + return fmt.Errorf("fetching translate data reader: %w", err) + } + 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", indexName, "fields", fieldName, "translate"), + Mode: 0666, + Size: int64(buf.Len()), + ModTime: time.Now(), + }); 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 +} + +func (cmd *BackupTarCommand) TLSHost() string { return cmd.Host } + +func (cmd *BackupTarCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/ctl/backup_tar_test.go b/ctl/backup_tar_test.go new file mode 100644 index 000000000..2c996d965 --- /dev/null +++ b/ctl/backup_tar_test.go @@ -0,0 +1,40 @@ +package ctl + +import ( + "bytes" + "context" + "net/http" + "strings" + "testing" + + "github.com/molecula/featurebase/v3/test" +) + +func TestBackupTarCommand_Run(t *testing.T) { + cluster := test.MustRunCluster(t, 1) + defer cluster.Close() + cmd := cluster.GetNode(0) + + buf := bytes.Buffer{} + stdin, stdout, stderr := GetIO(buf) + cm := NewBackupTarCommand(stdin, stdout, stderr) + hostport := cmd.API.Node().URI.HostPort() + cm.Host = hostport + cm.OutputPath = "-" + + resp, err := http.DefaultClient.Do(test.MustNewHTTPRequest("POST", "http://"+hostport+"/index/i", strings.NewReader(""))) + if err != nil { + t.Fatalf("making http request: %v", err) + } + resp.Body.Close() + resp, err = http.DefaultClient.Do(test.MustNewHTTPRequest("POST", "http://"+hostport+"/index/i/field/f", strings.NewReader(""))) + if err != nil { + t.Fatalf("making http request: %v", err) + } + resp.Body.Close() + + cm.Index = "i" + if err := cm.Run(context.Background()); err != nil { + t.Fatalf("BackupTarCommand Run error: %s", err) + } +} diff --git a/ctl/restore_tar.go b/ctl/restore_tar.go new file mode 100644 index 000000000..33b581906 --- /dev/null +++ b/ctl/restore_tar.go @@ -0,0 +1,270 @@ +// Copyright 2022 Molecula Corp. All rights reserved. +package ctl + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "context" + "crypto/tls" + "fmt" + "io" + "io/ioutil" + gohttp "net/http" + "os" + "strconv" + "strings" + "time" + + pilosa "github.com/molecula/featurebase/v3" + "github.com/molecula/featurebase/v3/authn" + "github.com/molecula/featurebase/v3/disco" + "github.com/molecula/featurebase/v3/server" + "github.com/pkg/errors" + "golang.org/x/sync/errgroup" +) + +// RestoreTarCommand represents a command for restoring a backup to +type RestoreTarCommand struct { + tlsConfig *tls.Config + + Host string + + // Filepath to the backup file. + Path string + + // Amount of time after first failed request to continue retrying. + RetryPeriod time.Duration `json:"retry-period"` + + // Host:port on which to listen for pprof. + Pprof string `json:"pprof"` + + // Reusable client. + client *pilosa.InternalClient + + // Standard input/output + *pilosa.CmdIO + + TLS server.TLSConfig + + AuthToken string +} + +// NewRestoreTarCommand returns a new instance of RestoreTarCommand. +func NewRestoreTarCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreTarCommand { + return &RestoreTarCommand{ + CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), + RetryPeriod: time.Second * 30, + Pprof: "localhost:0", + } +} + +// Run executes the restore. +func (cmd *RestoreTarCommand) Run(ctx context.Context) (err error) { + logger := cmd.Logger() + close, err := startProfilingServer(cmd.Pprof, logger) + if err != nil { + return errors.Wrap(err, "starting profiling server") + } + defer close() + + // Validate arguments. + 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() + if cmd.tlsConfig, err = server.GetTLSConfig(&tls, logger); err != nil { + return fmt.Errorf("parsing tls config: %w", err) + } + // Create a client to the server. + client, err := commandClient(cmd, pilosa.WithClientRetryPeriod(cmd.RetryPeriod)) + if err != nil { + return fmt.Errorf("creating client: %w", err) + } + cmd.client = client + + if cmd.AuthToken != "" { + ctx = context.WithValue(ctx, authn.ContextValueAccessToken, "Bearer "+cmd.AuthToken) + } + + 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 + } + + var primary *disco.Node + for _, node := range nodes { + if node.IsPrimary { + primary = node + break + } + + } + 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 + } + fragmentNodes, err := cmd.client.FragmentNodes(ctx, indexName, shard) + if err != nil { + return fmt.Errorf("cannot determine fragmentNodes: %w", err) + } else if len(fragmentNodes) == 0 { + return fmt.Errorf("no fragmentNodes 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 fragmentNodes { + 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) + rd := func() (io.Reader, error) { + return bytes.NewReader(shardBytes), nil + } + + 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) + rd := func() (io.Reader, error) { + return bytes.NewReader(shardBytes), nil + } + + 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) + } + + } + + } + /* 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. + Upload the index & field translation BoltDB snapshots to each node. + If possible, trigger the node to reload itself. Otherwise a restart would be required. + */ + + return nil +} +func (cmd *RestoreTarCommand) TLSHost() string { return cmd.Host } + +func (cmd *RestoreTarCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/ctl/restore_tar_test.go b/ctl/restore_tar_test.go new file mode 100644 index 000000000..ebe4868ae --- /dev/null +++ b/ctl/restore_tar_test.go @@ -0,0 +1,39 @@ +package ctl + +import ( + "bytes" + "context" + "net/http" + "strings" + "testing" + + "github.com/molecula/featurebase/v3/test" +) + +func TestRestoreTarCommand_Run(t *testing.T) { + cluster := test.MustRunCluster(t, 1) + defer cluster.Close() + cmd := cluster.GetNode(0) + + buf := bytes.Buffer{} + stdin, stdout, stderr := GetIO(buf) + cm := NewRestoreTarCommand(stdin, stdout, stderr) + hostport := cmd.API.Node().URI.HostPort() + cm.Host = hostport + cm.Path = "-" + + resp, err := http.DefaultClient.Do(test.MustNewHTTPRequest("POST", "http://"+hostport+"/index/i", strings.NewReader(""))) + if err != nil { + t.Fatalf("making http request: %v", err) + } + resp.Body.Close() + resp, err = http.DefaultClient.Do(test.MustNewHTTPRequest("POST", "http://"+hostport+"/index/i/field/f", strings.NewReader(""))) + if err != nil { + t.Fatalf("making http request: %v", err) + } + resp.Body.Close() + + if err := cm.Run(context.Background()); err != nil { + t.Fatalf("RestoreTarCommand Run error: %s", err) + } +}