Merge d2ffdeda9c9e9bf917259c6da292d1c34ca70e4e into c86c93ca2951338115159dcdd20711603044e1f1

CHANGELOG/CHANGELOG-3.6.md

@@ -28,6 +28,7 @@ See [code changes](https://github.com/etcd-io/etcd/compare/v3.5.0...v3.6.0).
 ### etcdctl v3
 
 - Add command to generate [shell completion](https://github.com/etcd-io/etcd/pull/13133).
+- Add command [snapshot pipe](https://github.com/etcd-io/etcd/pull/16243).
 - When print endpoint status, [show db size in use](https://github.com/etcd-io/etcd/pull/13639)
 - [Always print the raft_term in decimal](https://github.com/etcd-io/etcd/pull/13711) when displaying member list in json.
 - [Add one more field `storageVersion`](https://github.com/etcd-io/etcd/pull/13773) into the response of command `etcdctl endpoint status`.

client/v3/snapshot/v3_snapshot.go

@@ -37,15 +37,14 @@ func hasChecksum(n int64) bool {
 	return (n % 512) == sha256.Size
 }
 
-// SaveWithVersion fetches snapshot from remote etcd server, saves data
+// WriteSnapshotWithVersion fetches snapshot from remote etcd server, saves data
 // to target path and returns server version. If the context "ctx" is canceled or timed out,
 // snapshot save stream will error out (e.g. context.Canceled,
 // context.DeadlineExceeded). Make sure to specify only one endpoint
 // in client configuration. Snapshot API must be requested to a
 // selected node, and saved snapshot is the point-in-time state of
-// the selected node.
+// the selected node. Nota bene: etcd < v3.6 will return "" as version.
-// Etcd <v3.6 will return "" as version.
+func WriteSnapshotWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config, f *os.File) (string, error) {
-func SaveWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config, dbPath string) (string, error) {
 	cfg.Logger = lg.Named("client")
 	if len(cfg.Endpoints) != 1 {
 		return "", fmt.Errorf("snapshot must be requested to one selected node, not multiple %v", cfg.Endpoints)
@@ -56,21 +55,12 @@ func SaveWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config, d
 	}
 	defer cli.Close()
 
-	partpath := dbPath + ".part"
-	defer os.RemoveAll(partpath)
-
-	var f *os.File
-	f, err = os.OpenFile(partpath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, fileutil.PrivateFileMode)
-	if err != nil {
-		return "", fmt.Errorf("could not open %s (%v)", partpath, err)
-	}
-	lg.Info("created temporary db file", zap.String("path", partpath))
-
 	start := time.Now()
 	resp, err := cli.SnapshotWithVersion(ctx)
 	if err != nil {
 		return "", err
 	}
+
 	defer resp.Snapshot.Close()
 	lg.Info("fetching snapshot", zap.String("endpoint", cfg.Endpoints[0]))
 	var size int64
@@ -93,10 +83,32 @@ func SaveWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config, d
 		zap.Duration("took", time.Since(start)),
 		zap.String("etcd-version", resp.Version),
 	)
-
-	if err = os.Rename(partpath, dbPath); err != nil {
-		return resp.Version, fmt.Errorf("could not rename %s to %s (%v)", partpath, dbPath, err)
-	}
-	lg.Info("saved", zap.String("path", dbPath))
 	return resp.Version, nil
 }
+
+func SaveWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config, dbPath string) (string, error) {
+	partPath := dbPath + ".part"
+	f, err := os.OpenFile(partPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, fileutil.PrivateFileMode)
+	if err != nil {
+		return "", fmt.Errorf("could not open %s (%v)", partPath, err)
+	}
+	lg.Info("created temporary db file", zap.String("path", partPath))
+
+	defer os.RemoveAll(partPath)
+	defer f.Close()
+
+	var version string
+	version, err = WriteSnapshotWithVersion(ctx, lg, cfg, f)
+	if err != nil {
+		return version, err
+	}
+	if err = os.Rename(partPath, dbPath); err != nil {
+		return version, fmt.Errorf("could not rename %s to %s (%v)", partPath, dbPath, err)
+	}
+	lg.Info("saved", zap.String("path", dbPath))
+	return version, nil
+}
+
+func PipeWithVersion(ctx context.Context, lg *zap.Logger, cfg clientv3.Config) (string, error) {
+	return WriteSnapshotWithVersion(ctx, lg, cfg, os.Stdout)
+}

etcdctl/README.md

@@ -1020,6 +1020,21 @@ DEFRAG returns a zero exit code only if it succeeded defragmenting all given end
 
 SNAPSHOT provides commands to restore a snapshot of a running etcd server into a fresh cluster.
 
+### SNAPSHOT PIPE
+
+SNAPSHOT PIPE writes a point-in-time snapshot of the etcd backend database to stdout.
+
+#### Output
+
+The backend snapshot is written to stdout.
+
+#### Example
+
+Write a snapshot to stdout:
+```
+./etcdctl snapshot pipe > snapshot.db
+```
+
 ### SNAPSHOT SAVE \<filename\>
 
 SNAPSHOT SAVE writes a point-in-time snapshot of the etcd backend database to a file.

etcdctl/ctlv3/command/snapshot_command.go

@@ -56,6 +56,7 @@ func NewSnapshotCommand() *cobra.Command {
 		Example: snapshotExample,
 	}
 	cmd.AddCommand(NewSnapshotSaveCommand())
+	cmd.AddCommand(NewSnapshotPipeCommand())
 	return cmd
 }
 
@@ -68,6 +69,14 @@ func NewSnapshotSaveCommand() *cobra.Command {
 	}
 }
 
+func NewSnapshotPipeCommand() *cobra.Command {
+	return &cobra.Command{
+		Use:   "pipe",
+		Short: "Streams an etcd node backend snapshot to stdout",
+		Run:   snapshotPipeCommandFunc,
+	}
+}
+
 func snapshotSaveCommandFunc(cmd *cobra.Command, args []string) {
 	if len(args) != 1 {
 		err := fmt.Errorf("snapshot save expects one argument <filename>")
@@ -97,3 +106,24 @@ func snapshotSaveCommandFunc(cmd *cobra.Command, args []string) {
 		fmt.Printf("Server version %s\n", version)
 	}
 }
+
+func snapshotPipeCommandFunc(cmd *cobra.Command, args []string) {
+
+	lg, err := logutil.CreateDefaultZapLogger(zap.InfoLevel)
+	if err != nil {
+		cobrautl.ExitWithError(cobrautl.ExitError, err)
+	}
+	cfg := mustClientCfgFromCmd(cmd)
+
+	// if user does not specify "--command-timeout" flag, there will be no timeout for snapshot pipe command
+	ctx, cancel := context.WithCancel(context.Background())
+	if isCommandTimeoutFlagSet(cmd) {
+		ctx, cancel = commandCtx(cmd)
+	}
+	defer cancel()
+
+	_, err = snapshot.PipeWithVersion(ctx, lg, *cfg)
+	if err != nil {
+		cobrautl.ExitWithError(cobrautl.ExitInterrupted, err)
+	}
+}