tiered storage: can copy to s3, read from s3

master not aware tiered volume yet, file assigning is not working yet

weed/storage/backend/backend.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/spf13/viper"
 )
 
@@ -23,7 +24,8 @@ type BackendStorageFile interface {
 
 type BackendStorage interface {
 	ToProperties() map[string]string
-	NewStorageFile(key string) BackendStorageFile
+	NewStorageFile(key string, tierInfo *volume_server_pb.VolumeTierInfo) BackendStorageFile
+	CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error)
 }
 
 type StringProperties interface {
@@ -46,13 +48,13 @@ func LoadConfiguration(config *viper.Viper) {
 
 	backendSub := config.Sub(StorageBackendPrefix)
 
-	for backendTypeName, _ := range config.GetStringMap(StorageBackendPrefix) {
+	for backendTypeName := range config.GetStringMap(StorageBackendPrefix) {
 		backendStorageFactory, found := BackendStorageFactories[StorageType(backendTypeName)]
 		if !found {
 			glog.Fatalf("backend storage type %s not found", backendTypeName)
 		}
 		backendTypeSub := backendSub.Sub(backendTypeName)
-		for backendStorageId, _ := range backendSub.GetStringMap(backendTypeName) {
+		for backendStorageId := range backendSub.GetStringMap(backendTypeName) {
 			if !backendTypeSub.GetBool(backendStorageId + ".enabled") {
 				continue
 			}
@@ -105,12 +107,10 @@ func (p *Properties) GetString(key string) string {
 
 func ToPbStorageBackends() (backends []*master_pb.StorageBackend) {
 	for sName, s := range BackendStorages {
-		parts := strings.Split(sName, ".")
-		if len(parts) != 2 {
+		sType, sId := BackendNameToTypeId(sName)
+		if sType == "" {
 			continue
 		}
-
-		sType, sId := parts[0], parts[1]
 		backends = append(backends, &master_pb.StorageBackend{
 			Type:       sType,
 			Id:         sId,
@@ -119,3 +119,16 @@ func ToPbStorageBackends() (backends []*master_pb.StorageBackend) {
 	}
 	return
 }
+
+func BackendNameToTypeId(backendName string) (backendType, backendId string) {
+	parts := strings.Split(backendName, ".")
+	if len(parts) == 1 {
+		return backendName, "default"
+	}
+	if len(parts) != 2 {
+		return
+	}
+
+	backendType, backendId = parts[0], parts[1]
+	return
+}

weed/storage/backend/s3_backend/s3_backend.go

@@ -2,6 +2,7 @@ package s3_backend
 
 import (
 	"fmt"
+	"io"
 	"os"
 	"strings"
 	"time"
@@ -9,7 +10,9 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/storage/backend"
+	"github.com/google/uuid"
 )
 
 func init() {
@@ -57,7 +60,7 @@ func (s *S3BackendStorage) ToProperties() map[string]string {
 	return m
 }
 
-func (s *S3BackendStorage) NewStorageFile(key string) backend.BackendStorageFile {
+func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeTierInfo) backend.BackendStorageFile {
 	if strings.HasPrefix(key, "/") {
 		key = key[1:]
 	}
@@ -65,18 +68,35 @@ func (s *S3BackendStorage) NewStorageFile(key string) backend.BackendStorageFile
 	f := &S3BackendStorageFile{
 		backendStorage: s,
 		key:            key,
+		tierInfo:       tierInfo,
 	}
 
 	return f
 }
 
+func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+	randomUuid, _ := uuid.NewRandom()
+	key = randomUuid.String()
+
+	glog.V(1).Infof("copying dat file of", f.Name(), "to remote s3", s.id, "as", key)
+
+	size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, fn)
+
+	return
+}
+
 type S3BackendStorageFile struct {
 	backendStorage *S3BackendStorage
 	key            string
+	tierInfo       *volume_server_pb.VolumeTierInfo
 }
 
 func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n int, err error) {
+
 	bytesRange := fmt.Sprintf("bytes=%d-%d", off, off+int64(len(p))-1)
+
+	// glog.V(0).Infof("read %s %s", s3backendStorageFile.key, bytesRange)
+
 	getObjectOutput, getObjectErr := s3backendStorageFile.backendStorage.conn.GetObject(&s3.GetObjectInput{
 		Bucket: &s3backendStorageFile.backendStorage.bucket,
 		Key:    &s3backendStorageFile.key,
@@ -84,13 +104,26 @@ func (s3backendStorageFile S3BackendStorageFile) ReadAt(p []byte, off int64) (n
 	})
 
 	if getObjectErr != nil {
-		return 0, fmt.Errorf("bucket %s GetObject %s: %v",
-			s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, getObjectErr)
+		return 0, fmt.Errorf("bucket %s GetObject %s: %v", s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, getObjectErr)
 	}
 	defer getObjectOutput.Body.Close()
 
-	return getObjectOutput.Body.Read(p)
+	glog.V(4).Infof("read %s %s", s3backendStorageFile.key, bytesRange)
+	glog.V(4).Infof("content range: %s, contentLength: %d", *getObjectOutput.ContentRange, *getObjectOutput.ContentLength)
 
+	for {
+		if n, err = getObjectOutput.Body.Read(p); err == nil && n < len(p) {
+			p = p[n:]
+		} else {
+			break
+		}
+	}
+
+	if err == io.EOF {
+		err = nil
+	}
+
+	return
 }
 
 func (s3backendStorageFile S3BackendStorageFile) WriteAt(p []byte, off int64) (n int, err error) {
@@ -107,18 +140,15 @@ func (s3backendStorageFile S3BackendStorageFile) Close() error {
 
 func (s3backendStorageFile S3BackendStorageFile) GetStat() (datSize int64, modTime time.Time, err error) {
 
-	headObjectOutput, headObjectErr := s3backendStorageFile.backendStorage.conn.HeadObject(&s3.HeadObjectInput{
-		Bucket: &s3backendStorageFile.backendStorage.bucket,
-		Key:    &s3backendStorageFile.key,
-	})
+	files := s3backendStorageFile.tierInfo.GetFiles()
 
-	if headObjectErr != nil {
-		return 0, time.Now(), fmt.Errorf("bucket %s HeadObject %s: %v",
-			s3backendStorageFile.backendStorage.bucket, s3backendStorageFile.key, headObjectErr)
+	if len(files)==0 {
+		err = fmt.Errorf("remote file info not found")
+		return
 	}
 
-	datSize = int64(*headObjectOutput.ContentLength)
-	modTime = *headObjectOutput.LastModified
+	datSize = int64(files[0].FileSize)
+	modTime = time.Unix(int64(files[0].ModifiedTime),0)
 
 	return
 }

weed/storage/backend/s3_backend/s3_upload.go

@@ -3,27 +3,30 @@ package s3_backend
 import (
 	"fmt"
 	"os"
+	"sync/atomic"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+	"github.com/chrislusf/seaweedfs/weed/glog"
 )
 
-func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string) error {
+func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string,
+	fn func(progressed int64, percentage float32)error) (fileSize int64, err error) {
 
 	//open the file
 	f, err := os.Open(filename)
 	if err != nil {
-		return fmt.Errorf("failed to open file %q, %v", filename, err)
+		return 0, fmt.Errorf("failed to open file %q, %v", filename, err)
 	}
 	defer f.Close()
 
 	info, err := f.Stat()
 	if err != nil {
-		return fmt.Errorf("failed to stat file %q, %v", filename, err)
+		return 0, fmt.Errorf("failed to stat file %q, %v", filename, err)
 	}
 
-	fileSize := info.Size()
+	fileSize = info.Size()
 
 	partSize := int64(64 * 1024 * 1024) // The minimum/default allowed part size is 5MB
 	for partSize*1000 < fileSize {
@@ -33,14 +36,22 @@ func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey
 	// Create an uploader with the session and custom options
 	uploader := s3manager.NewUploaderWithClient(sess, func(u *s3manager.Uploader) {
 		u.PartSize = partSize
-		u.Concurrency = 15 // default is 15
+		u.Concurrency = 5
 	})
 
+	fileReader := &s3UploadProgressedReader{
+		fp: f,
+		size:fileSize,
+		read:-fileSize,
+		fn:fn,
+	}
+
 	// Upload the file to S3.
-	result, err := uploader.Upload(&s3manager.UploadInput{
+	var result *s3manager.UploadOutput
+	result, err = uploader.Upload(&s3manager.UploadInput{
 		Bucket:               aws.String(destBucket),
 		Key:                  aws.String(destKey),
-		Body:                 f,
+		Body:                 fileReader,
 		ACL:                  aws.String("private"),
 		ServerSideEncryption: aws.String("AES256"),
 		StorageClass:         aws.String("STANDARD_IA"),
@@ -48,9 +59,44 @@ func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey
 
 	//in case it fails to upload
 	if err != nil {
-		return fmt.Errorf("failed to upload file, %v", err)
+		return 0, fmt.Errorf("failed to upload file %s: %v", filename, err)
 	}
-	fmt.Printf("file %s uploaded to %s\n", filename, result.Location)
+	glog.V(1).Infof("file %s uploaded to %s\n", filename, result.Location)
 
-	return nil
+	return
+}
+
+// adapted from https://github.com/aws/aws-sdk-go/pull/1868
+type s3UploadProgressedReader struct {
+	fp   *os.File
+	size int64
+	read int64
+	fn func(progressed int64, percentage float32)error
+}
+func (r *s3UploadProgressedReader) Read(p []byte) (int, error) {
+	return r.fp.Read(p)
+}
+
+func (r *s3UploadProgressedReader) ReadAt(p []byte, off int64) (int, error) {
+	n, err := r.fp.ReadAt(p, off)
+	if err != nil {
+		return n, err
+	}
+
+	// Got the length have read( or means has uploaded), and you can construct your message
+	atomic.AddInt64(&r.read, int64(n))
+
+	if r.fn != nil {
+		read := r.read
+		if err := r.fn(read, float32(read*100)/float32(r.size)); err != nil {
+			return n, err
+		}
+	}
+
+	return n, err
+}
+
+
+func (r *s3UploadProgressedReader) Seek(offset int64, whence int) (int64, error) {
+	return r.fp.Seek(offset, whence)
 }

weed/storage/disk_location.go

@@ -33,8 +33,8 @@ func NewDiskLocation(dir string, maxVolumeCount int) *DiskLocation {
 
 func (l *DiskLocation) volumeIdFromPath(dir os.FileInfo) (needle.VolumeId, string, error) {
 	name := dir.Name()
-	if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
-		base := name[:len(name)-len(".dat")]
+	if !dir.IsDir() && strings.HasSuffix(name, ".idx") {
+		base := name[:len(name)-len(".idx")]
 		collection, volumeId, err := parseCollectionVolumeId(base)
 		return volumeId, collection, err
 	}
@@ -53,7 +53,7 @@ func parseCollectionVolumeId(base string) (collection string, vid needle.VolumeI
 
 func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind NeedleMapType) {
 	name := fileInfo.Name()
-	if !fileInfo.IsDir() && strings.HasSuffix(name, ".dat") {
+	if !fileInfo.IsDir() && strings.HasSuffix(name, ".idx") {
 		vid, collection, err := l.volumeIdFromPath(fileInfo)
 		if err == nil {
 			l.RLock()

weed/storage/volume.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/storage/backend"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
@@ -37,6 +38,8 @@ type Volume struct {
 	lastCompactRevision    uint16
 
 	isCompacting bool
+
+	volumeTierInfo *volume_server_pb.VolumeTierInfo
 }
 
 func NewVolume(dirname string, collection string, id needle.VolumeId, needleMapKind NeedleMapType, replicaPlacement *ReplicaPlacement, ttl *needle.TTL, preallocate int64, memoryMapMaxSizeMb uint32) (v *Volume, e error) {

weed/storage/volume_loading.go

@@ -26,8 +26,11 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 	fileName := v.FileName()
 	alreadyHasSuperBlock := false
 
-	// open dat file
-	if exists, canRead, canWrite, modifiedTime, fileSize := checkFile(fileName + ".dat"); exists {
+	if v.maybeLoadVolumeTierInfo() {
+		// open remote file
+		alreadyHasSuperBlock = true
+	} else if exists, canRead, canWrite, modifiedTime, fileSize := checkFile(fileName + ".dat"); exists {
+		// open dat file
 		if !canRead {
 			return fmt.Errorf("cannot read Volume Data file %s.dat", fileName)
 		}

weed/storage/volume_tier.go

@@ -0,0 +1,104 @@
+package storage
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+
+	_ "github.com/chrislusf/seaweedfs/weed/storage/backend/s3_backend"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/chrislusf/seaweedfs/weed/storage/backend"
+	"github.com/golang/protobuf/jsonpb"
+)
+
+func (v *Volume) GetVolumeTierInfo() *volume_server_pb.VolumeTierInfo {
+	return v.volumeTierInfo
+}
+
+func (v *Volume) maybeLoadVolumeTierInfo() bool {
+
+	v.volumeTierInfo = &volume_server_pb.VolumeTierInfo{}
+
+	tierFileName := v.FileName() + ".tier"
+
+	if exists, canRead, _, _, _ := checkFile(tierFileName); !exists || !canRead {
+		if !exists {
+			return false
+		}
+		if !canRead {
+			glog.Warningf("can not read %s", tierFileName)
+		}
+		return false
+	}
+
+	glog.V(0).Infof("maybeLoadVolumeTierInfo loading volume %d check file", v.Id)
+
+	tierData, readErr := ioutil.ReadFile(tierFileName)
+	if readErr != nil {
+		glog.Warningf("fail to read %s : %v", tierFileName, readErr)
+		return false
+	}
+
+	glog.V(0).Infof("maybeLoadVolumeTierInfo loading volume %d ReadFile", v.Id)
+
+	if err := jsonpb.Unmarshal(bytes.NewReader(tierData), v.volumeTierInfo); err != nil {
+		glog.Warningf("unmarshal error: %v", err)
+		return false
+	}
+
+	glog.V(0).Infof("maybeLoadVolumeTierInfo loading volume %d Unmarshal tierInfo %v", v.Id, v.volumeTierInfo)
+
+	if len(v.volumeTierInfo.GetFiles()) == 0 {
+		return false
+	}
+
+	glog.V(0).Infof("volume %d is tiered to %s as %s and read only", v.Id,
+		v.volumeTierInfo.Files[0].BackendName(), v.volumeTierInfo.Files[0].Key)
+
+	v.readOnly = true
+
+	glog.V(0).Infof("loading volume %d from remote %v", v.Id, v.volumeTierInfo.Files)
+	v.LoadRemoteFile()
+
+	return true
+}
+
+func (v *Volume) LoadRemoteFile() error {
+	tierFile := v.volumeTierInfo.GetFiles()[0]
+	backendStorage := backend.BackendStorages[tierFile.BackendName()]
+
+	if v.DataBackend != nil {
+		v.DataBackend.Close()
+	}
+
+	v.DataBackend = backendStorage.NewStorageFile(tierFile.Key, v.volumeTierInfo)
+	return nil
+}
+
+func (v *Volume) SaveVolumeTierInfo() error {
+
+	tierFileName := v.FileName() + ".tier"
+
+	if exists, _, canWrite, _, _ := checkFile(tierFileName); exists && !canWrite {
+		return fmt.Errorf("%s not writable", tierFileName)
+	}
+
+	m := jsonpb.Marshaler{
+		EmitDefaults: true,
+		Indent:       "  ",
+	}
+
+	text, marshalErr := m.MarshalToString(v.GetVolumeTierInfo())
+	if marshalErr != nil {
+		return fmt.Errorf("marshal volume %d tier info: %v", v.Id, marshalErr)
+	}
+
+	writeErr := ioutil.WriteFile(tierFileName, []byte(text), 0755)
+	if writeErr != nil {
+		return fmt.Errorf("fail to write %s : %v", tierFileName, writeErr)
+	}
+
+	return nil
+}