command/cp: add progress bar (#590)

Resolves #51

Enables usage of cp command with show-progress flag to show progress bar for a copy task.

Example progress bar:
78.00%  ━━━━━━━━━━━━━━━━━━━━────────  780 MB / 1.00 GB (189.37 MB/s) 5.9s (28/29)

CHANGELOG.md

@@ -8,6 +8,7 @@
 - Added `--content-disposition` flag to `cp` command. ([#569](https://github.com/peak/s5cmd/issues/569))
 - Added `--show-fullpath` flag to `ls`. ([#596](https://github.com/peak/s5cmd/issues/596))
 - Added `pipe` command. ([#182](https://github.com/peak/s5cmd/issues/182))
+- Added `--show-progress` flag to `cp` to show a progress bar. ([#51](https://github.com/peak/s5cmd/issues/51))
 
 #### Improvements
 - Implemented concurrent multipart download support for `cat`. ([#245](https://github.com/peak/s5cmd/issues/245)) 

command/cp.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/hashicorp/go-multierror"
 	"github.com/urfave/cli/v2"
@@ -18,6 +19,7 @@ import (
 	"github.com/peak/s5cmd/v2/log"
 	"github.com/peak/s5cmd/v2/log/stat"
 	"github.com/peak/s5cmd/v2/parallel"
+	"github.com/peak/s5cmd/v2/progressbar"
 	"github.com/peak/s5cmd/v2/storage"
 	"github.com/peak/s5cmd/v2/storage/url"
 )
@@ -218,6 +220,11 @@ func NewCopyCommandFlags() []cli.Flag {
 			Name:  "version-id",
 			Usage: "use the specified version of an object",
 		},
+		&cli.BoolFlag{
+			Name:    "show-progress",
+			Aliases: []string{"sp"},
+			Usage:   "show a progress bar",
+		},
 	}
 	sharedFlags := NewSharedFlags()
 	return append(copyFlags, sharedFlags...)
@@ -280,6 +287,9 @@ type Copy struct {
 	contentType           string
 	contentEncoding       string
 	contentDisposition    string
+	showProgress          bool
+	progressbar           progressbar.ProgressBar
+
 	// region settings
 	srcRegion string
 	dstRegion string
@@ -307,6 +317,14 @@ func NewCopy(c *cli.Context, deleteSource bool) (*Copy, error) {
 		return nil, err
 	}
 
+	var commandProgressBar progressbar.ProgressBar
+
+	if c.Bool("show-progress") && !(src.Type == dst.Type) {
+		commandProgressBar = progressbar.New()
+	} else {
+		commandProgressBar = &progressbar.NoOp{}
+	}
+
 	return &Copy{
 		src:          src,
 		dst:          dst,
@@ -333,6 +351,9 @@ func NewCopy(c *cli.Context, deleteSource bool) (*Copy, error) {
 		contentType:           c.String("content-type"),
 		contentEncoding:       c.String("content-encoding"),
 		contentDisposition:    c.String("content-disposition"),
+		showProgress:          c.Bool("show-progress"),
+		progressbar:           commandProgressBar,
+
 		// region settings
 		srcRegion: c.String("source-region"),
 		dstRegion: c.String("destination-region"),
@@ -361,12 +382,13 @@ func (c Copy) Run(ctx context.Context) error {
 	}
 
 	objch, err := expandSource(ctx, client, c.followSymlinks, c.src)
-
 	if err != nil {
 		printError(c.fullCommand, c.op, err)
 		return err
 	}
 
+	c.progressbar.Start()
+	defer c.progressbar.Finish()
 	waiter := parallel.NewWaiter()
 
 	var (
@@ -433,6 +455,15 @@ func (c Copy) Run(ctx context.Context) error {
 		srcurl := object.URL
 		var task parallel.Task
 
+		if object.Size == 0 && !(srcurl.Type == c.dst.Type) {
+			obj, err := client.Stat(ctx, srcurl)
+			if err == nil {
+				object.Size = obj.Size
+			}
+		}
+		c.progressbar.AddTotalBytes(object.Size)
+		c.progressbar.IncrementTotalObjects()
+
 		switch {
 		case srcurl.Type == c.dst.Type: // local->local or remote->remote
 			task = c.prepareCopyTask(ctx, srcurl, c.dst, isBatch)
@@ -443,10 +474,8 @@ func (c Copy) Run(ctx context.Context) error {
 		default:
 			panic("unexpected src-dst pair")
 		}
-
 		parallel.Run(task, waiter)
 	}
-
 	waiter.Wait()
 	<-errDoneCh
 
@@ -470,6 +499,7 @@ func (c Copy) prepareCopyTask(
 				Err: err,
 			}
 		}
+		c.progressbar.IncrementCompletedObjects()
 		return nil
 	}
 }
@@ -494,6 +524,7 @@ func (c Copy) prepareDownloadTask(
 				Err: err,
 			}
 		}
+		c.progressbar.IncrementCompletedObjects()
 		return nil
 	}
 }
@@ -515,6 +546,7 @@ func (c Copy) prepareUploadTask(
 				Err: err,
 			}
 		}
+		c.progressbar.IncrementCompletedObjects()
 		return nil
 	}
 }
@@ -545,8 +577,10 @@ func (c Copy) doDownload(ctx context.Context, srcurl *url.URL, dsturl *url.URL)
 		return err
 	}
 
-	size, err := srcClient.Get(ctx, srcurl, file, c.concurrency, c.partSize)
+	writer := newCountingReaderWriter(file, c.progressbar)
+	size, err := srcClient.Get(ctx, srcurl, writer, c.concurrency, c.partSize)
 	file.Close()
+
 	if err != nil {
 		dErr := dstClient.Delete(ctx, &url.URL{Path: file.Name(), Type: dsturl.Type})
 		if dErr != nil {
@@ -564,15 +598,17 @@ func (c Copy) doDownload(ctx context.Context, srcurl *url.URL, dsturl *url.URL)
 		return err
 	}
 
-	msg := log.InfoMessage{
-		Operation:   c.op,
-		Source:      srcurl,
-		Destination: dsturl,
-		Object: &storage.Object{
-			Size: size,
-		},
+	if !c.showProgress {
+		msg := log.InfoMessage{
+			Operation:   c.op,
+			Source:      srcurl,
+			Destination: dsturl,
+			Object: &storage.Object{
+				Size: size,
+			},
+		}
+		log.Info(msg)
 	}
-	log.Info(msg)
 
 	return nil
 }
@@ -624,7 +660,8 @@ func (c Copy) doUpload(ctx context.Context, srcurl *url.URL, dsturl *url.URL) er
 	if c.contentDisposition != "" {
 		metadata.SetContentDisposition(c.contentDisposition)
 	}
-	err = dstClient.Put(ctx, file, dsturl, metadata, c.concurrency, c.partSize)
+	reader := newCountingReaderWriter(file, c.progressbar)
+	err = dstClient.Put(ctx, reader, dsturl, metadata, c.concurrency, c.partSize)
 	if err != nil {
 		return err
 	}
@@ -642,16 +679,18 @@ func (c Copy) doUpload(ctx context.Context, srcurl *url.URL, dsturl *url.URL) er
 		}
 	}
 
-	msg := log.InfoMessage{
-		Operation:   c.op,
-		Source:      srcurl,
-		Destination: dsturl,
-		Object: &storage.Object{
-			Size:         obj.Size,
-			StorageClass: c.storageClass,
-		},
+	if !c.showProgress {
+		msg := log.InfoMessage{
+			Operation:   c.op,
+			Source:      srcurl,
+			Destination: dsturl,
+			Object: &storage.Object{
+				Size:         obj.Size,
+				StorageClass: c.storageClass,
+			},
+		}
+		log.Info(msg)
 	}
-	log.Info(msg)
 
 	return nil
 }
@@ -972,3 +1011,48 @@ func guessContentType(file *os.File) string {
 	}
 	return contentType
 }
+
+type countingReaderWriter struct {
+	pb      progressbar.ProgressBar
+	fp      *os.File
+	signMap map[int64]struct{}
+	mu      sync.Mutex
+}
+
+func newCountingReaderWriter(file *os.File, pb progressbar.ProgressBar) *countingReaderWriter {
+	return &countingReaderWriter{
+		pb:      pb,
+		fp:      file,
+		signMap: map[int64]struct{}{},
+	}
+}
+
+func (r *countingReaderWriter) WriteAt(p []byte, off int64) (int, error) {
+	n, err := r.fp.WriteAt(p, off)
+	r.pb.AddCompletedBytes(int64(n))
+	return n, err
+}
+
+func (r *countingReaderWriter) Read(p []byte) (int, error) {
+	n, err := r.fp.Read(p)
+	r.pb.AddCompletedBytes(int64(n))
+	return n, err
+}
+
+func (r *countingReaderWriter) ReadAt(p []byte, off int64) (int, error) {
+	n, err := r.fp.ReadAt(p, off)
+	r.mu.Lock()
+	// Ignore the first signature call
+	if _, ok := r.signMap[off]; ok {
+		// Got the length have read (or means has uploaded)
+		r.pb.AddCompletedBytes(int64(n))
+	} else {
+		r.signMap[off] = struct{}{}
+	}
+	r.mu.Unlock()
+	return n, err
+}
+
+func (r *countingReaderWriter) Seek(offset int64, whence int) (int64, error) {
+	return r.fp.Seek(offset, whence)
+}

command/pipe.go

@@ -256,7 +256,7 @@ func (c Pipe) shouldOverride(ctx context.Context, dsturl *url.URL) error {
 		return err
 	}
 
-	obj, err := getObject(ctx, dsturl, client)
+	obj, err := statObject(ctx, dsturl, client)
 	if err != nil {
 		return err
 	}

e2e/cp_test.go

@@ -4171,3 +4171,28 @@ func TestLocalFileOverridenWhenDownloadFailed(t *testing.T) {
 	expected := fs.Expected(t, fs.WithFile(filename, content))
 	assert.Assert(t, fs.Equal(workdir.Path(), expected))
 }
+
+// Test that counting writer does not corrupt objects during a download process
+func TestCountingWriter(t *testing.T) {
+	t.Parallel()
+
+	const (
+		filename = "log.txt"
+	)
+
+	content := randomString(3_000_000)
+
+	s3client, s5cmd := setup(t)
+	bucket := s3BucketFromTestName(t)
+	createBucket(t, s3client, bucket)
+	putFile(t, s3client, bucket, filename, content)
+
+	cmd := s5cmd("cp", "--show-progress", "--concurrency", "3", "--part-size", "1", "s3://"+bucket+"/"+filename, ".")
+	result := icmd.RunCmd(cmd)
+
+	result.Assert(t, icmd.Success)
+
+	// assert the downloaded file has the same content with the remote object
+	expected := fs.Expected(t, fs.WithFile(filename, content, fs.WithMode(0644)))
+	assert.Assert(t, fs.Equal(cmd.Dir, expected))
+}

go.mod

@@ -4,6 +4,7 @@ go 1.17
 
 require (
 	github.com/aws/aws-sdk-go v1.44.256
+	github.com/cheggaaa/pb/v3 v3.1.4
 	github.com/golang/mock v1.6.0
 	github.com/google/go-cmp v0.4.0
 	github.com/hashicorp/go-multierror v1.0.0
@@ -18,12 +19,18 @@ require (
 )
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/fatih/color v1.15.0 // indirect
 	github.com/hashicorp/errwrap v1.0.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
+	github.com/mattn/go-runewidth v0.0.14 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 // indirect
 	github.com/shabbyrobe/gocovmerge v0.0.0-20190829150210-3e036491d500 // indirect

go.sum

@@ -1,12 +1,18 @@
 github.com/BurntSushi/toml v1.1.0/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
+github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow=
+github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4=
 github.com/aws/aws-sdk-go v1.44.256 h1:O8VH+bJqgLDguqkH/xQBFz5o/YheeZqgcOYIgsTVWY4=
 github.com/aws/aws-sdk-go v1.44.256/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
+github.com/cheggaaa/pb/v3 v3.1.4 h1:DN8j4TVVdKu3WxVwcRKu0sG00IIU6FewoABZzXbRQeo=
+github.com/cheggaaa/pb/v3 v3.1.4/go.mod h1:6wVjILNBaXMs8c21qRiaUM8BR82erfgau1DQ4iUXmSA=
 github.com/cpuguy83/go-md2man/v2 v2.0.2 h1:p1EgwI/C7NhT0JmVkwCD2ZBK8j4aeHQX2pMHHBfMQ6w=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/fatih/color v1.15.0 h1:kOqh6YHBtK8aywxGerMG2Eq3H6Qgoqeo13Bk2Mv/nBs=
+github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -37,11 +43,21 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lanrat/extsort v1.0.0 h1:JjvkCUbD55+gs5s64FHmCU93kWjegEAM5n10XN6GB3c=
 github.com/lanrat/extsort v1.0.0/go.mod h1:bkDEvem4UnD1h87yKICydXs63mKrIGW3W9OGPMg93Ww=
+github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
+github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
+github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-isatty v0.0.17/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
+github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-runewidth v0.0.14 h1:+xnbZSEeDbOIg5/mE6JF0w6n9duR1l3/WmbinWVwUuU=
+github.com/mattn/go-runewidth v0.0.14/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
+github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
@@ -94,8 +110,10 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
 golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=