Skip to content

Added a feature to support migrating archive files from Aliyun OSS #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
successzy wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Added a feature to support migrating archive files from Aliyun OSS #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion dth/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,12 +277,14 @@ func (c *S3Client) ListObjects(ctx context.Context, continuationToken, prefix *s

for _, obj := range output.Contents {
// log.Printf("key=%s size=%d", *obj.Key, obj.Size)
if obj.StorageClass == "GLACIER" || obj.StorageClass == "DEEP_ARCHIVE" {

@daixba daixba Apr 24, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这个改动会影响现有s3 的吗? 本来过滤了Glacier,给取消了

if obj.StorageClass == "DEEP_ARCHIVE" {
continue
}
//fmt.Printf("%T", string(obj.StorageClass))
result = append(result, &Object{
Key: *obj.Key,
Size: obj.Size,
StorageClass: string(obj.StorageClass), // Henry
})
}

Expand Down
1 change: 1 addition & 0 deletions dth/common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ type Object struct {
Key string `json:"key"`
Size int64 `json:"size"`
Sequencer string `json:"sequencer,omitempty"`
StorageClass string `json:"storageclass,omitempty"` // Henry
}

// S3Event represents a basic structure of a S3 Event Message
Expand Down
82 changes: 80 additions & 2 deletions dth/job.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,10 @@ import (
"time"

"github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/aliyun/aliyun-oss-go-sdk/oss"

"fmt"
//"github.com/aws/aws-sdk-go-v2/aws"
)

const (
Expand Down Expand Up @@ -409,6 +413,8 @@ func NewWorker(ctx context.Context, cfg *JobConfig) (w *Worker) {
srcCred := getCredentials(ctx, cfg.SrcCredential, cfg.SrcInCurrentAccount, sm)
desCred := getCredentials(ctx, cfg.DestCredential, cfg.DestInCurrentAccount, sm)

//fmt.Println(srcCred)

srcClient := NewS3Client(ctx, cfg.SrcBucket, cfg.SrcPrefix, cfg.SrcPrefixList, cfg.SrcEndpoint, cfg.SrcRegion, cfg.SrcType, srcCred)
desClient := NewS3Client(ctx, cfg.DestBucket, cfg.DestPrefix, "", "", cfg.DestRegion, "Amazon_S3", desCred)

Expand All @@ -421,6 +427,57 @@ func NewWorker(ctx context.Context, cfg *JobConfig) (w *Worker) {
}
}

//++++++++++Customization Start++++++++++
// Initialize aliyun oss client and oss bucket
func InitializaOss(ctx context.Context, cfg *JobConfig) (*oss.Bucket, error) {
sm, err := NewSecretService(ctx)
if err != nil {
log.Printf("Unable to load credentials - %s\n", err.Error())
}
srcCred := getCredentials(ctx, cfg.SrcCredential, cfg.SrcInCurrentAccount, sm)
aliyun_endpoint := "https://oss-" + cfg.SrcRegion + ".aliyuncs.com"

@daixba daixba Apr 24, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

golang 不是这种 命名规范, 驼峰, 不带下划线


// Initialize oss_client and oss_bucket
oss_client, err := oss.New(aliyun_endpoint, srcCred.accessKey, srcCred.secretKey)
if err != nil {
log.Printf("Unable to create oss_client - %s\n", err.Error())
}

oss_bucket, err := oss_client.Bucket(cfg.SrcBucket)
if err != nil {
log.Printf("Unable to create oss_bucket - %s\n", err.Error())
}

return oss_bucket, nil
}

// Restore Aliyun Glacier Object
func RestoreObject(obj *Object, oss_bucket *oss.Bucket) error {
err := oss_bucket.RestoreObject(obj.Key)

@daixba daixba Apr 24, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这个功能可以事先做吗? 意思就是说, 可以先批量恢复后再 调跑dth 吗

if err != nil {
log.Printf("Unable to restore oss_key - %s\n", err.Error())
}

// Wait for restore completed
for {
meta, err := oss_bucket.GetObjectDetailedMeta(obj.Key)
if err != nil {
log.Printf("Unable to get object detailed meta - %s\n", err.Error())
}

// ongoing-request="false" means restoring object has completed.
if meta.Get("X-Oss-Restore") != "ongoing-request=\"true\"" {
fmt.Println("----------Restoring object completed.----------")
break
}

time.Sleep(10 * time.Second)
}

return nil
}
//++++++++++Customization End++++++++++

// Run a Worker job
func (w *Worker) Run(ctx context.Context) {
// log.Println("Start Worker Job...")
Expand All @@ -441,6 +498,12 @@ func (w *Worker) Run(ctx context.Context) {
// Buffer size is cfg.WorkerNumber * 2 - 1 (More buffer for multipart upload)
transferCh := make(chan struct{}, buffer*2-1)

// Initialize oss_bucket (Henry)
oss_bucket, err := InitializaOss(ctx, w.cfg)
if err != nil {
log.Printf("Unable to initialize oss_bucket - %s\n", err.Error())
}

for {
msg, rh := w.sqs.ReceiveMessages(ctx)

Expand All @@ -455,6 +518,17 @@ func (w *Worker) Run(ctx context.Context) {
continue
}

// Restore object (Henry)
//fmt.Println("======restore object======")
if obj.StorageClass == "GLACIER" {
fmt.Println("----------Start restoring object.----------")
err := RestoreObject(obj, oss_bucket)

@daixba daixba Apr 24, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

所有glacier 都能调用吗?

@successzy successzy May 23, 2022 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

不好意思，回复晚了。同时，非常感谢答复！

  1. 这个改动会影响现有s3 的吗? 本来过滤了Glacier,给取消了
    //会影响，目前这个只是针对特定的客户和场景。
  2. 这个功能可以事先做吗? 意思就是说, 可以先批量恢复后再 调跑dth 吗
  • 因为程序里会过滤掉Glacier和Deep Archive，如果不注释过滤Glacier和Deep Archive的代码，则需要修改Storage Class，把Glacier改成Standard或者IA
  • 如果是批量恢复，则需要把过滤Glacier和Deep Archive的代码注释掉。因为已经做了批量恢复，不再需要过滤Glacier
  • 只是如果迁移周期比较长，那么第一种方式会增加很多成本，第二种方式需要反复解冻，也会增加成本。
  • 上述两种方式都有额外步骤需要客户操作且迁移到S3后都是Standard存储类型
  • 参考1：归档类型Object：对于归档类型的Object，解冻状态默认持续24小时，24小时内再次调用RestoreObject接口则解冻状态会自动延长24小时，一次解冻流程内可有效调用7次RestoreObject接口达到最长7天的解冻持续时间。您也可以通过传入解冻天数，一次调用RestoreObject接口指定最长7天的解冻持续时间。
  • 参考2： 数据取回费用是0.06元/GB
  1. 所有glacier 都能调用吗?
    //目前只是测试了阿里云，且我理解是需要对dth支持的每个公有云对象存储额外增加RestoreObject函数。

//fmt.Printf("%T", obj.Key)
if err != nil {
log.Printf("Unable to restore obj - %s\n", err.Error())
}
}

destKey := appendPrefix(&obj.Key, &w.cfg.DestPrefix)

if action == Transfer {
Expand Down Expand Up @@ -667,7 +741,9 @@ func (w *Worker) migrateBigFile(ctx context.Context, obj *Object, destKey *strin
meta = w.srcClient.HeadObject(ctx, &obj.Key)
}

uploadID, err = w.desClient.CreateMultipartUpload(ctx, destKey, &w.cfg.DestStorageClass, &w.cfg.DestAcl, meta)
//uploadID, err = w.desClient.CreateMultipartUpload(ctx, destKey, &w.cfg.DestStorageClass, &w.cfg.DestAcl, meta)
string_storageclass := string(obj.StorageClass) // Henry
uploadID, err = w.desClient.CreateMultipartUpload(ctx, destKey, &string_storageclass, &w.cfg.DestAcl, meta) // Henry
if err != nil {
log.Printf("Failed to create upload ID - %s for %s\n", err.Error(), *destKey)
return &TransferResult{
Expand Down Expand Up @@ -818,7 +894,9 @@ func (w *Worker) transfer(ctx context.Context, obj *Object, destKey *string, sta

} else {
log.Printf("----->Uploading %d Bytes to %s/%s\n", chunkSize, w.cfg.DestBucket, *destKey)
etag, err = w.desClient.PutObject(ctx, destKey, body, &w.cfg.DestStorageClass, &w.cfg.DestAcl, meta)
//etag, err = w.desClient.PutObject(ctx, destKey, body, &w.cfg.DestStorageClass, &w.cfg.DestAcl, meta)
string_storageclass := string(obj.StorageClass) // Henry
etag, err = w.desClient.PutObject(ctx, destKey, body, &string_storageclass, &w.cfg.DestAcl, meta) // Henry
}

body = nil // release memory
Expand Down
1 change: 1 addition & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ module golang.a2z.com/dthcli
go 1.16

require (
github.com/aliyun/aliyun-oss-go-sdk v2.2.2+incompatible // indirect
github.com/aws/aws-sdk-go-v2 v1.8.1
github.com/aws/aws-sdk-go-v2/config v1.6.1
github.com/aws/aws-sdk-go-v2/credentials v1.3.3
Expand Down
3 changes: 3 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/aliyun/aliyun-oss-go-sdk v2.2.2+incompatible h1:9gWa46nstkJ9miBReJcN8Gq34cBFbzSpQZVVT9N09TM=
github.com/aliyun/aliyun-oss-go-sdk v2.2.2+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8=
github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
Expand Down Expand Up @@ -319,6 +321,7 @@ golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
Expand Down