PBM-1506 GCS support concurrent downloading (#1104)

* add gcs download
* add download for shared logic
* move common logic to storage/download.go
* update storage
* update gcs download
* add NewDownloadStat
* add gcs to physical restore

Author: veceraj

pbm/restore/phys/phys.go

@@ -8,15 +8,15 @@ import (
 	"go.mongodb.org/mongo-driver/bson/primitive"
 
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
-	"github.com/percona/percona-backup-mongodb/pbm/storage/s3"
+	"github.com/percona/percona-backup-mongodb/pbm/storage"
 )
 
 type RestoreStat struct {
 	RS map[string]map[string]RestoreRSMetrics `bson:"rs,omitempty" json:"rs,omitempty"`
 }
 type RestoreRSMetrics struct {
-	DistTxn  DistTxnStat     `bson:"txn,omitempty" json:"txn,omitempty"`
-	Download s3.DownloadStat `bson:"download,omitempty" json:"download,omitempty"`
+	DistTxn  DistTxnStat          `bson:"txn,omitempty" json:"txn,omitempty"`
+	Download storage.DownloadStat `bson:"download,omitempty" json:"download,omitempty"`
 }
 
 type DistTxnStat struct {
@@ -35,8 +35,8 @@ type DistTxnStat struct {
 }
 
 type RestoreShardStat struct {
-	Txn DistTxnStat      `json:"txn"`
-	D   *s3.DownloadStat `json:"d"`
+	Txn DistTxnStat           `json:"txn"`
+	D   *storage.DownloadStat `json:"d"`
 }
 
 type TxnState string

pbm/restore/physical.go

@@ -39,6 +39,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/restore/phys"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
+	"github.com/percona/percona-backup-mongodb/pbm/storage/gcs"
 	"github.com/percona/percona-backup-mongodb/pbm/storage/s3"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -1271,17 +1272,25 @@ func (r *PhysRestore) dumpMeta(meta *RestoreMeta, s defs.Status, msg string) err
 	return nil
 }
 
-func (r *PhysRestore) copyFiles() (*s3.DownloadStat, error) {
-	var stat *s3.DownloadStat
+func (r *PhysRestore) copyFiles() (*storage.DownloadStat, error) {
+	var stat *storage.DownloadStat
 	readFn := r.bcpStg.SourceReader
-	if t, ok := r.bcpStg.(*s3.S3); ok {
+
+	switch t := r.bcpStg.(type) {
+	case *s3.S3:
+		d := t.NewDownload(r.confOpts.NumDownloadWorkers, r.confOpts.MaxDownloadBufferMb, r.confOpts.DownloadChunkMb)
+		readFn = d.SourceReader
+		defer func() {
+			s := d.Stat()
+			stat = &s
+			r.log.Debug("download stat: %s", s)
+		}()
+	case *gcs.GCS:
 		d := t.NewDownload(r.confOpts.NumDownloadWorkers, r.confOpts.MaxDownloadBufferMb, r.confOpts.DownloadChunkMb)
 		readFn = d.SourceReader
-
 		defer func() {
 			s := d.Stat()
 			stat = &s
-
 			r.log.Debug("download stat: %s", s)
 		}()
 	}