Merge branch 'master' into importSpeedups

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

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,13 +66,20 @@ 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.
 	if cmd.OutputPath == "" {
 		return fmt.Errorf("-o flag required")
 	}
+	useStdout := cmd.OutputPath == "-"
+
+	// 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)
@@ -85,17 +95,20 @@ func (cmd *BackupCommand) Run(ctx context.Context) 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/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,12 +125,19 @@ func (cmd *BackupCommand) Run(ctx context.Context) 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 {
+	// 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
 }
 
@@ -210,13 +230,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)
 	}

executor.go

@@ -5705,7 +5705,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")
 	}
 	done := ctx.Done()