Method uploadDirectory

upload/s3.go:234–287

()

func (s sem) decr() { <-s }

func (u *S3) uploadDirectory() error {
	wg := sync.WaitGroup{}

	ctx := log.WithFields(log.Fields{"f": "s3upload.uploadDirectory"})
	ctx.Info("Uploading")
	sem := make(sem, u.Cfg.Concurrency)
	ctx.Info("Starting to walk...")
	exitErr := atomic.Value{}
	err := filepath.Walk(u.Cfg.StagingPath, func(fpath string, info os.FileInfo, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		// If a fatal error happened in any of the goroutines, then exit immediately
		e := exitErr.Load()
		if e != nil {
			return e.(error)
		}

		if info.IsDir() {
			return nil
		}
		ctx.WithFields(log.Fields{"fpath": fpath}).Info("Upload scheduled")
		wg.Add(1)
		sem.incr()
		go func(fpath string) {
			defer func() { sem.decr(); wg.Done() }()

			for i := 0; i < u.Cfg.Retries; i++ {
				if exitErr.Load() != nil {
					return
				}
				err := s3UploadFile(u.uploader, u.Cfg.Bucket, u.Cfg.Prefix, u.Cfg.StagingPath, fpath)
				if err == nil {
					break
				}

				atomic.AddInt64(&u.totalerr, int64(1))
				if u.Cfg.ExitOnError {
					exitErr.Store(err)
					return
				}
				log.WithError(err).WithFields(log.Fields{"retry#": i + 1}).Error("failed upload")
			}
			// Decrease the queued elements counter both in case of success and not fatal error.
			atomic.AddInt64(&u.queuedn, int64(-1))
		}(fpath)
		return nil
	})
	ctx.Info("All Scheduling done")
	wg.Wait()

	ctx.Info("All upload done")
	return err
}

func s3UploadFile(uploader *s3manager.Uploader, bucket, prefix, localPath, fpath string) error {
	ctx := log.WithFields(log.Fields{"localPath": localPath, "filepath": fpath})