From 73f0668fe17aac536002b9e5463ea570645d6d1f Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Mon, 14 Jun 2021 09:29:05 -0600 Subject: [PATCH] Refactor backup/restore to use directory archive --- cmd/backup.go | 3 +- ctl/backup.go | 220 ++++++++++++------------------ ctl/restore.go | 355 +++++++++++++++++++++++++++++-------------------- 3 files changed, 297 insertions(+), 281 deletions(-) diff --git a/cmd/backup.go b/cmd/backup.go index d91975aa5..4fe129f1b 100644 --- a/cmd/backup.go +++ b/cmd/backup.go @@ -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 diff --git a/ctl/backup.go b/ctl/backup.go index d193eb139..31abae8ad 100644 --- a/ctl/backup.go +++ b/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 } diff --git a/ctl/restore.go b/ctl/restore.go index 6b149768c..b29a41780 100644 --- a/ctl/restore.go +++ b/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 }