From c883ba18d90e2e044409060540db413936d07738 Mon Sep 17 00:00:00 2001 From: Yi Zheng Date: Fri, 15 Apr 2022 16:38:13 +0800 Subject: [PATCH] support archive file migration for aliyun --- dth/client.go | 4 ++- dth/common.go | 1 + dth/job.go | 82 +++++++++++++++++++++++++++++++++++++++++++++++++-- go.mod | 1 + go.sum | 3 ++ 5 files changed, 88 insertions(+), 3 deletions(-) diff --git a/dth/client.go b/dth/client.go index 12eaed3..02f37d5 100644 --- a/dth/client.go +++ b/dth/client.go @@ -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" { + 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 }) } diff --git a/dth/common.go b/dth/common.go index 7729448..7def8c1 100644 --- a/dth/common.go +++ b/dth/common.go @@ -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 diff --git a/dth/job.go b/dth/job.go index ec410ec..06af3f1 100644 --- a/dth/job.go +++ b/dth/job.go @@ -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 ( @@ -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) @@ -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" + + // 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) + 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...") @@ -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) @@ -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) + //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 { @@ -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{ @@ -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 diff --git a/go.mod b/go.mod index 715dce9..ae65871 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index dd32d58..171677b 100644 --- a/go.sum +++ b/go.sum @@ -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= @@ -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=