From 63196e7981da6ec518e537352e8c711d956d393c Mon Sep 17 00:00:00 2001 From: Maxton Huff Date: Thu, 13 May 2021 11:39:23 -0500 Subject: [PATCH 1/6] add String() to shard slices to reduce error length --- executor.go | 54 ++++++++++++++++++++++++++--------------------------- 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/executor.go b/executor.go index f37c23172..97e72c2e2 100644 --- a/executor.go +++ b/executor.go @@ -743,47 +743,47 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p case "Sum": statFn() res, err := e.executeSum(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeSum %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeSum %v", shardSlice(shards).String()) case "Min": statFn() res, err := e.executeMin(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMin %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeMin %v", shardSlice(shards).String()) case "Max": statFn() res, err := e.executeMax(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMax %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeMax %v", shardSlice(shards).String()) case "MinRow": statFn() res, err := e.executeMinRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMinRow %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeMinRow %v", shardSlice(shards).String()) case "MaxRow": statFn() res, err := e.executeMaxRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMaxRow %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeMaxRow %v", shardSlice(shards).String()) case "Clear": statFn() res, err := e.executeClearBit(ctx, qcx, index, c, opt) - return res, errors.Wrapf(err, "executeClearBit %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeClearBit %v", shardSlice(shards).String()) case "ClearRow": statFn() res, err := e.executeClearRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeClearRow %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeClearRow %v", shardSlice(shards).String()) case "Distinct": statFn() res, err := e.executeDistinct(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeDistinct %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeDistinct %v", shardSlice(shards).String()) case "Store": statFn() res, err := e.executeSetRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeSetRow %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeSetRow %v", shardSlice(shards).String()) case "Count": statFn() res, err := e.executeCount(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeCount %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeCount %v", shardSlice(shards).String()) case "Set": statFn() res, err := e.executeSet(ctx, qcx, index, c, opt) - return res, errors.Wrapf(err, "executeSet %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeSet %v", shardSlice(shards).String()) case "SetRowAttrs": statFn() return nil, errors.Wrap(e.executeSetRowAttrs(ctx, qcx, index, c, opt), "executeSetRowAttrs") @@ -793,61 +793,61 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p case "TopK": statFn() res, err := e.executeTopK(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeTopK %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeTopK %v", shardSlice(shards).String()) case "TopN": statFn() res, err := e.executeTopN(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeTopN %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeTopN %v", shardSlice(shards).String()) case "Rows": statFn() res, err := e.executeRows(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeRows %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeRows %v", shardSlice(shards).String()) case "ExternalLookup": statFn() res, err := e.executeExternalLookup(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeExternalLookup %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeExternalLookup %v", shardSlice(shards).String()) case "Extract": statFn() res, err := e.executeExtract(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeExtract %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeExtract %v", shardSlice(shards).String()) case "GroupBy": statFn() res, err := e.executeGroupBy(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeGroupBy %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeGroupBy %v", shardSlice(shards).String()) case "Options": statFn() res, err := e.executeOptionsCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeOptionsCall %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeOptionsCall %v", shardSlice(shards).String()) case "IncludesColumn": res, err := e.executeIncludesColumnCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeIncludesColumnCall %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeIncludesColumnCall %v", shardSlice(shards).String()) case "FieldValue": statFn() res, err := e.executeFieldValueCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeFieldValueCall %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeFieldValueCall %v", shardSlice(shards).String()) case "Precomputed": res, err := e.executePrecomputedCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executePrecomputedCall %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executePrecomputedCall %v", shardSlice(shards).String()) case "UnionRows": res, err := e.executeUnionRows(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeUnionRows %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeUnionRows %v", shardSlice(shards).String()) case "ConstRow": res, err := e.executeConstRow(ctx, index, c) - return res, errors.Wrapf(err, "executeConstRow %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeConstRow %v", shardSlice(shards).String()) case "Limit": res, err := e.executeLimitCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeLimitCall %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeLimitCall %v", shardSlice(shards).String()) case "Percentile": res, err := e.executePercentile(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executePercentile %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executePercentile %v", shardSlice(shards).String()) case "Delete": statFn() //TODO(twg) need this? res, err := e.executeDeleteRecords(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeDelete %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeDelete %v", shardSlice(shards).String()) default: // e.g. "Row", "Union", "Intersect" or anything that returns a bitmap. statFn() res, err := e.executeBitmapCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeBitmapCall %v", shardSlice(shards)) + return res, errors.Wrapf(err, "executeBitmapCall %v", shardSlice(shards).String()) } } From 53da35eb00f228c6403a8ded1811743231863046 Mon Sep 17 00:00:00 2001 From: Maxton Huff Date: Thu, 13 May 2021 12:01:12 -0500 Subject: [PATCH 2/6] add missing String() calls in error messages --- executor.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/executor.go b/executor.go index 97e72c2e2..d93e92bbe 100644 --- a/executor.go +++ b/executor.go @@ -3052,7 +3052,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c // Get full result set. other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) if err != nil { - return nil, errors.Wrapf(err, "mapReduce shards: %v", shardSlice(shards)) + return nil, errors.Wrapf(err, "mapReduce shards: %v", shardSlice(shards).String()) } results, _ := other.([]GroupCount) @@ -6066,7 +6066,7 @@ func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapRe // Group shards together by nodes. m, err := e.shardsByNode(nodes, index, shards) if err != nil { - return errors.Wrapf(err, "shards by node %v", shardSlice(shards)) + return errors.Wrapf(err, "shards by node %v", shardSlice(shards).String()) } done := ctx.Done() From eb79c35cbd1c34ce67b124d0bae44c38893196b1 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Tue, 18 May 2021 15:17:28 -0600 Subject: [PATCH 3/6] Add support for clustered backups --- ctl/backup.go | 48 ++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 40 insertions(+), 8 deletions(-) diff --git a/ctl/backup.go b/ctl/backup.go index 9351589b3..de8599020 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -17,8 +17,8 @@ package ctl import ( "archive/tar" "bytes" - "compress/gzip" "context" + "crypto/tls" "encoding/json" "fmt" "io" @@ -28,12 +28,15 @@ import ( "time" "github.com/pilosa/pilosa/v2" + "github.com/pilosa/pilosa/v2/http" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/topology" ) // BackupCommand represents a command for backing up a Pilosa node. type BackupCommand struct { // nolint: maligned + tlsConfig *tls.Config + // Destination host and port. Host string `json:"host"` @@ -63,7 +66,7 @@ func (cmd *BackupCommand) TempPath() string { } // Run executes the main program execution. -func (cmd *BackupCommand) Run(ctx context.Context) error { +func (cmd *BackupCommand) Run(ctx context.Context) (err error) { logger := cmd.Logger() // Validate arguments. @@ -71,6 +74,12 @@ func (cmd *BackupCommand) Run(ctx context.Context) error { return fmt.Errorf("-o flag required") } + // 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) if err != nil { @@ -92,10 +101,8 @@ func (cmd *BackupCommand) Run(ctx context.Context) error { } defer w.Close() - // Open a tar/gzip writer to the temporary file. - gw := gzip.NewWriter(w) - defer gw.Close() - tw := tar.NewWriter(gw) + // Open a tar writer to the temporary file. + tw := tar.NewWriter(w) defer tw.Close() // Backup schema. @@ -112,6 +119,11 @@ func (cmd *BackupCommand) Run(ctx context.Context) error { } } + // Close archive. + if err := tw.Close(); err != nil { + return err + } + // Move data file to final location. logger.Printf("writing backup: %s", cmd.OutputPath) if err := os.Rename(cmd.OutputPath+".tmp", cmd.OutputPath); err != nil { @@ -210,13 +222,33 @@ 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) error { +func (cmd *BackupCommand) backupShard(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.backupShardNode(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 +} + +// 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 { logger := cmd.Logger() logger.Printf("backing up shard: index=%q id=%d", indexName, shard) filename := path.Join("indexes", indexName, "shards", fmt.Sprintf("%04d", shard)) - rc, err := cmd.client.ShardReader(ctx, indexName, shard) + client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig)) + rc, err := client.ShardReader(ctx, indexName, shard) if err != nil { return fmt.Errorf("fetching shard reader: %w", err) } From 04bf214f8111b649767e0eeed09e61ec8c9ac631 Mon Sep 17 00:00:00 2001 From: Maxton Huff Date: Wed, 19 May 2021 12:20:39 -0500 Subject: [PATCH 4/6] remove shard list from mapper error message to avoid duplicate output --- executor.go | 58 ++++++++++++++++++++++++++--------------------------- 1 file changed, 29 insertions(+), 29 deletions(-) diff --git a/executor.go b/executor.go index d93e92bbe..f24a5fdbd 100644 --- a/executor.go +++ b/executor.go @@ -743,47 +743,47 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p case "Sum": statFn() res, err := e.executeSum(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeSum %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeSum %v", shardSlice(shards)) case "Min": statFn() res, err := e.executeMin(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMin %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeMin %v", shardSlice(shards)) case "Max": statFn() res, err := e.executeMax(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMax %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeMax %v", shardSlice(shards)) case "MinRow": statFn() res, err := e.executeMinRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMinRow %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeMinRow %v", shardSlice(shards)) case "MaxRow": statFn() res, err := e.executeMaxRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeMaxRow %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeMaxRow %v", shardSlice(shards)) case "Clear": statFn() res, err := e.executeClearBit(ctx, qcx, index, c, opt) - return res, errors.Wrapf(err, "executeClearBit %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeClearBit %v", shardSlice(shards)) case "ClearRow": statFn() res, err := e.executeClearRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeClearRow %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeClearRow %v", shardSlice(shards)) case "Distinct": statFn() res, err := e.executeDistinct(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeDistinct %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeDistinct %v", shardSlice(shards)) case "Store": statFn() res, err := e.executeSetRow(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeSetRow %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeSetRow %v", shardSlice(shards)) case "Count": statFn() res, err := e.executeCount(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeCount %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeCount %v", shardSlice(shards)) case "Set": statFn() res, err := e.executeSet(ctx, qcx, index, c, opt) - return res, errors.Wrapf(err, "executeSet %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeSet %v", shardSlice(shards)) case "SetRowAttrs": statFn() return nil, errors.Wrap(e.executeSetRowAttrs(ctx, qcx, index, c, opt), "executeSetRowAttrs") @@ -793,61 +793,61 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p case "TopK": statFn() res, err := e.executeTopK(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeTopK %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeTopK %v", shardSlice(shards)) case "TopN": statFn() res, err := e.executeTopN(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeTopN %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeTopN %v", shardSlice(shards)) case "Rows": statFn() res, err := e.executeRows(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeRows %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeRows %v", shardSlice(shards)) case "ExternalLookup": statFn() res, err := e.executeExternalLookup(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeExternalLookup %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeExternalLookup %v", shardSlice(shards)) case "Extract": statFn() res, err := e.executeExtract(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeExtract %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeExtract %v", shardSlice(shards)) case "GroupBy": statFn() res, err := e.executeGroupBy(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeGroupBy %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeGroupBy %v", shardSlice(shards)) case "Options": statFn() res, err := e.executeOptionsCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeOptionsCall %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeOptionsCall %v", shardSlice(shards)) case "IncludesColumn": res, err := e.executeIncludesColumnCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeIncludesColumnCall %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeIncludesColumnCall %v", shardSlice(shards)) case "FieldValue": statFn() res, err := e.executeFieldValueCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeFieldValueCall %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeFieldValueCall %v", shardSlice(shards)) case "Precomputed": res, err := e.executePrecomputedCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executePrecomputedCall %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executePrecomputedCall %v", shardSlice(shards)) case "UnionRows": res, err := e.executeUnionRows(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeUnionRows %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeUnionRows %v", shardSlice(shards)) case "ConstRow": res, err := e.executeConstRow(ctx, index, c) - return res, errors.Wrapf(err, "executeConstRow %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeConstRow %v", shardSlice(shards)) case "Limit": res, err := e.executeLimitCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeLimitCall %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeLimitCall %v", shardSlice(shards)) case "Percentile": res, err := e.executePercentile(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executePercentile %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executePercentile %v", shardSlice(shards)) case "Delete": statFn() //TODO(twg) need this? res, err := e.executeDeleteRecords(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeDelete %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeDelete %v", shardSlice(shards)) default: // e.g. "Row", "Union", "Intersect" or anything that returns a bitmap. statFn() res, err := e.executeBitmapCall(ctx, qcx, index, c, shards, opt) - return res, errors.Wrapf(err, "executeBitmapCall %v", shardSlice(shards).String()) + return res, errors.Wrapf(err, "executeBitmapCall %v", shardSlice(shards)) } } @@ -3052,7 +3052,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c // Get full result set. other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) if err != nil { - return nil, errors.Wrapf(err, "mapReduce shards: %v", shardSlice(shards).String()) + return nil, errors.Wrapf(err, "mapReduce shards: %v", shardSlice(shards)) } results, _ := other.([]GroupCount) @@ -6066,7 +6066,7 @@ func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapRe // Group shards together by nodes. m, err := e.shardsByNode(nodes, index, shards) if err != nil { - return errors.Wrapf(err, "shards by node %v", shardSlice(shards).String()) + return errors.Wrapf(err, "") } done := ctx.Done() From a4f282c8e83a2447f22cdf230d90f0e91df880b6 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Wed, 19 May 2021 15:04:36 -0600 Subject: [PATCH 5/6] Allow backup to stdout --- cmd/backup.go | 4 ++-- ctl/backup.go | 24 ++++++++++++++++-------- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/cmd/backup.go b/cmd/backup.go index 8e272c526..d91975aa5 100644 --- a/cmd/backup.go +++ b/cmd/backup.go @@ -28,7 +28,7 @@ func newBackupCommand(stdin io.Reader, stdout io.Writer, stderr io.Writer) *cobr Use: "backup", Short: "Back up pilosa server", Long: ` -Backs up a pilosa server to a local snapshot file. +Backs up a pilosa server to a local, tar-formatted snapshot file. `, RunE: func(c *cobra.Command, args []string) error { return cmd.Run(context.Background()) @@ -36,7 +36,7 @@ Backs up a pilosa server to a local snapshot file. } flags := ccmd.Flags() - flags.StringVarP(&cmd.OutputPath, "output", "o", "", "output path to write to") + flags.StringVarP(&cmd.OutputPath, "output", "o", "", "output path to write to; specify '-' to send to stdout") 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 de8599020..644b295f6 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -73,6 +73,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) { if cmd.OutputPath == "" { return fmt.Errorf("-o flag required") } + useStdout := cmd.OutputPath == "-" // Parse TLS configuration for node-specific clients. tls := cmd.TLSConfiguration() @@ -94,12 +95,17 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) { } schema := &pilosa.Schema{Indexes: indexes} - // Create output file in temporary location. - w, err := os.Create(cmd.OutputPath + ".tmp") - if err != nil { - return err + // 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() } - defer w.Close() // Open a tar writer to the temporary file. tw := tar.NewWriter(w) @@ -125,9 +131,11 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) { } // Move data file to final location. - logger.Printf("writing backup: %s", cmd.OutputPath) - if err := os.Rename(cmd.OutputPath+".tmp", cmd.OutputPath); err != nil { - return err + if !useStdout { + logger.Printf("writing backup: %s", cmd.OutputPath) + if err := os.Rename(cmd.OutputPath+".tmp", cmd.OutputPath); err != nil { + return err + } } return nil From e646d7ac79548655b71a7f3025e8ea4d9ecd0e01 Mon Sep 17 00:00:00 2001 From: Maxton Huff Date: Thu, 20 May 2021 09:55:42 -0500 Subject: [PATCH 6/6] wrap mapper error with shards by node --- executor.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/executor.go b/executor.go index f24a5fdbd..025b5baf5 100644 --- a/executor.go +++ b/executor.go @@ -6066,7 +6066,7 @@ func (e *executor) mapper(ctx context.Context, eg *errgroup.Group, ch chan mapRe // Group shards together by nodes. m, err := e.shardsByNode(nodes, index, shards) if err != nil { - return errors.Wrapf(err, "") + return errors.Wrapf(err, "shards by node") } done := ctx.Done()