func (task *BackupTask) uploadFileWithRetry(key string, path string) error {
	var backoff = time.Second
	var expiry = time.Now().UnixNano() + int64(20*time.Minute)
	log.Infof("Uploading file=%v to s3key=%v", path, key)
	for !isQuit(task.quitC) {
		f, err := os.Open(path)
		if err != nil {
			return err
		}
		reader, writer := io.Pipe()
		go func() {
			gw := gzip.NewWriter(writer)
			io.Copy(gw, f)
			f.Close()
			gw.Close()
			writer.Close()
		}()
		err = task.s3.Put(task.s3BucketName, key, reader)
		f.Close()
		if err == nil {
			break
		}
		reqErr, ok := toAWSReqErr(err)
		if !ok || reqErr.StatusCode() != http.StatusServiceUnavailable {
			return err  
		}
		if time.Now().UnixNano()+int64(backoff) > expiry {
			return err  
		}
		log.Infof("Error uploading file to S3, will retry, file=%v, err=%v", path, err)
		sleep(task.quitC, backoff)
		backoff = time.Duration(2 * int64(backoff))
	}
	return nil
}
uploads file with exponential backoff
package cassandra
import (
	"compress/gzip"
	"encoding/hex"
	"encoding/json"
	"io"
	"io/ioutil"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"time"
	"github.com/aws/aws-sdk-go/aws/awserr"
	log "github.com/sirupsen/logrus"
	"github.com/uber/cherami-server/tools/awscloud"
)
type BackupTask struct {
	taskBase
	hostname        string
	keyspace        string
	rootDirPath     string
	schemaBackupDir string
	deployment      string
	s3BucketName    string
	name            string  
	quitC           chan struct{}
	stoppedC        chan struct{}
	nodeTool        NodeTool
	s3              awscloud.SecureS3Client
}
const dirPerm = os.ModeDir | os.FileMode(0755)
const filePerm = os.FileMode(0644)
type taskBase struct{}
const (
	schemaFileName = "schema.txt"  
	peersFileName  = "peers.json"  
)
func (task *taskBase) mkFileName(name string, keyspace string, suffix string) string {
	return name + "_" + keyspace + "_" + suffix
}
func (task *taskBase) s3KeyPrefix(hostname string, name string, keyspace string, deployment string) string {
	return deployment + "/cassandra/backups/" + keyspace + "/" + hostname + "/" + name
}
func NewBackupTask(config *BackupToolConfig, snapshotName string, hostname string, secret *Secret) (*BackupTask, error) {
	aesKey, err := hex.DecodeString(secret.Snapshot.AESMasterKey)
	if err != nil {
		return nil, err
	}
	cp := awscloud.NewCredProvider(secret.AWSAccessKeyID, secret.AWSSecretKey)
	s3 := awscloud.NewSecureS3Client(config.AWS.Region, cp, aesKey)
	addr, err := findEth0Addr()
	if err != nil {
		addr = "127.0.0.1"
	}
	nodeTool := NewNodeTool(addr)
	return &BackupTask{
		s3:              s3,
		nodeTool:        nodeTool,
		hostname:        hostname,
		deployment:      config.Deployment,
		schemaBackupDir: config.Backup.SchemaBackupDir,
		rootDirPath:     config.RootDirPath,
		s3BucketName:    config.AWS.S3Bucket,
		keyspace:        config.Backup.Keyspace,
		name:            snapshotName,
		quitC:           make(chan struct{}),
		stoppedC:        make(chan struct{}),
	}, nil
}
func (task *BackupTask) Run() error {
	objMap, ok := task.isSnapshotOnS3()
	if ok {
		log.Infof("Snapshot already on S3, keyspace=%v,name=%v", task.keyspace, task.name)
		return nil
	}
	var err error
	if !task.isSnapshotOnDisk() {
		err = task.takeSnapshot()
		if err != nil {
			return err
		}
	}
	if isQuit(task.quitC) {
		goto done
	}
	err = task.uploadSnapshot(objMap)
	if err != nil {
		return err
	}
	if isQuit(task.quitC) {
		goto done
	}
	task.writeManifestFile()
	log.Infof("BackupTask succeeded, keyspace=%v,name=%v", task.keyspace, task.name)
done:
	close(task.stoppedC)
	return nil
}
func (task *BackupTask) Stop(d time.Duration) bool {
	close(task.quitC)
	select {
	case <-task.stoppedC:
		return true
	case <-time.After(d):
		return false
	}
}
func (task *BackupTask) uploadSnapshot(objMap map[string]int64) error {
	keyToPath, err := task.findSnapshotFiles()
	if err != nil {
		return newTaskError("Can't find all snapshot files, err=%v", err)
	}
	if len(keyToPath) < 2 {
		log.Errorf("Can't find all snapshot files, keyspace=%v,snapshot=%v", task.keyspace, task.name)
		return newTaskError("Cannot find snapshot files after snapshot", nil)
	}
	log.Infof("Preparing for upload")
	for key, path := range keyToPath {
		s3Key := task.s3KeyPrefix(task.hostname, task.name, task.keyspace, task.deployment) + "/" + key
		if _, ok := objMap[s3Key]; ok {
			continue  
		}
		e := task.uploadFileWithRetry(s3Key, path)
		if e != nil {
			log.Errorf("Giving up snapshot upload, s3 upload error; keyspace=%v, snapshot=%v, err=%v", task.keyspace, task.name, err)
			return newTaskError("s3 upload error", err)
		}
	}
	return nil
}