Merge pull request #1118 from percona/PBM-1531-deadlock-physical-restore-file-copy

PBM-1531: Fix deadlock during physical restore file copying

Author: boris-ilijic

pbm/restore/physical.go

@@ -987,7 +987,10 @@ func (r *PhysRestore) Snapshot(
 		buf:   &bytes.Buffer{},
 		path:  fmt.Sprintf("%s/%s/rs.%s/log/%s", defs.PhysRestoresDir, r.name, r.rsConf.ID, r.nodeInfo.Me),
 		limit: 1 << 20, // 1Mb
-		write: func(name string, data io.Reader) error { return r.stg.Save(name, data, -1) },
+		write: func(name string, data io.Reader) error {
+			// Logger should be disabled due to: PBM-1531
+			return r.stg.Save(name, data, -1, storage.UseLogger(false))
+		},
 	})
 	logger.PauseMgo()
 

pbm/storage/azure/azure.go

@@ -128,7 +128,14 @@ func (*Blob) Type() storage.Type {
 	return storage.Azure
 }
 
-func (b *Blob) Save(name string, data io.Reader, sizeb int64) error {
+func (b *Blob) Save(name string, data io.Reader, sizeb int64, options ...storage.Option) error {
+	opts := storage.GetDefaultOpts()
+	for _, opt := range options {
+		if err := opt(opts); err != nil {
+			return errors.Wrap(err, "processing options for save")
+		}
+	}
+
 	bufsz := defaultUploadBuff
 	if sizeb > 0 {
 		ps := int(sizeb / maxBlocks * 11 / 10) // add 10% just in case
@@ -142,7 +149,7 @@ func (b *Blob) Save(name string, data io.Reader, sizeb int64) error {
 		cc = 1
 	}
 
-	if b.log != nil {
+	if b.log != nil && opts.UseLogger {
 		b.log.Debug("BufferSize is set to %d (~%dMb) | %d", bufsz, bufsz>>20, sizeb)
 	}
 

pbm/storage/blackhole/blackhole.go

@@ -18,7 +18,7 @@ func (*Blackhole) Type() storage.Type {
 	return storage.Blackhole
 }
 
-func (*Blackhole) Save(_ string, data io.Reader, _ int64) error {
+func (*Blackhole) Save(_ string, data io.Reader, _ int64, _ ...storage.Option) error {
 	_, err := io.Copy(io.Discard, data)
 	return err
 }

pbm/storage/fs/fs.go

@@ -151,7 +151,7 @@ func writeSync(finalpath string, data io.Reader) (err error) {
 	return nil
 }
 
-func (fs *FS) Save(name string, data io.Reader, _ int64) error {
+func (fs *FS) Save(name string, data io.Reader, _ int64, _ ...storage.Option) error {
 	return writeSync(path.Join(fs.root, name), data)
 }
 

pbm/storage/gcs/gcs.go

@@ -142,7 +142,7 @@ func (*GCS) Type() storage.Type {
 	return storage.GCS
 }
 
-func (g *GCS) Save(name string, data io.Reader, size int64) error {
+func (g *GCS) Save(name string, data io.Reader, size int64, _ ...storage.Option) error {
 	ctx := context.Background()
 
 	w := g.bucketHandle.Object(path.Join(g.opts.Prefix, name)).NewWriter(ctx)

pbm/storage/s3/s3.go

@@ -314,7 +314,13 @@ func (*S3) Type() storage.Type {
 	return storage.S3
 }
 
-func (s *S3) Save(name string, data io.Reader, sizeb int64) error {
+func (s *S3) Save(name string, data io.Reader, sizeb int64, options ...storage.Option) error {
+	opts := storage.GetDefaultOpts()
+	for _, opt := range options {
+		if err := opt(opts); err != nil {
+			return errors.Wrap(err, "processing options for save")
+		}
+	}
 	cc := runtime.NumCPU() / 2
 	if cc == 0 {
 		cc = 1
@@ -367,7 +373,7 @@ func (s *S3) Save(name string, data io.Reader, sizeb int64) error {
 		}
 	}
 
-	if s.log != nil {
+	if s.log != nil && opts.UseLogger {
 		s.log.Debug("uploading %q [size hint: %v (%v); part size: %v (%v)]",
 			name,
 			sizeb,

pbm/storage/storage.go

@@ -37,7 +37,7 @@ type FileInfo struct {
 
 type Storage interface {
 	Type() Type
-	Save(name string, data io.Reader, size int64) error
+	Save(name string, data io.Reader, size int64, options ...Option) error
 	SourceReader(name string) (io.ReadCloser, error)
 	// FileStat returns file info. It returns error if file is empty or not exists.
 	FileStat(name string) (FileInfo, error)
@@ -69,6 +69,30 @@ func ParseType(s string) Type {
 	}
 }
 
+// Opts represents storage options
+type Opts struct {
+	UseLogger bool
+}
+
+// GetDefaultOpts creates default options.
+func GetDefaultOpts() *Opts {
+	return &Opts{
+		UseLogger: true,
+	}
+}
+
+// Option is function for setting the storage option
+type Option func(*Opts) error
+
+// UseLogger option enables/disables logger when working with storage.
+// Logger is enabled by default.
+func UseLogger(useLogger bool) Option {
+	return func(o *Opts) error {
+		o.UseLogger = useLogger
+		return nil
+	}
+}
+
 // IsInitialized checks if there is PBM init file on the storage.
 func IsInitialized(ctx context.Context, stg Storage) (bool, error) {
 	_, err := stg.FileStat(defs.StorInitFile)