first commit

Author: daixba

.gitignore

@@ -0,0 +1,20 @@
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+config
+
+.DS_Store
+.idea
+dthcli
+
+

Dockerfile

@@ -0,0 +1,45 @@
+FROM amazonlinux:latest as builder
+
+RUN yum update && yum install -y tar gzip
+
+RUN cd /tmp && \
+    curl -LO https://golang.org/dl/go1.16.4.linux-amd64.tar.gz && \
+    rm -rf /usr/local/go && \
+    tar -C /usr/local -xzf /tmp/go1.16.4.linux-amd64.tar.gz
+
+WORKDIR /build
+COPY . .
+RUN PATH=$PATH:/usr/local/go/bin GOPROXY=https://goproxy.io,direct GOOS=linux go build -o dthcli .
+
+FROM amazonlinux:latest
+
+ENV SOURCE_TYPE Amazon_S3
+
+ENV JOB_TABLE_NAME ''
+ENV JOB_QUEUE_NAME ''
+
+ENV SRC_BUCKET ''
+ENV SRC_PREFIX ''
+ENV SRC_REGION ''
+ENV SRC_ENDPOINT ''
+ENV SRC_CREDENTIALS ''
+ENV SRC_IN_CURRENT_ACCOUNT false
+
+ENV DEST_BUCKET ''
+ENV DEST_PREFIX ''
+ENV DEST_REGION ''
+ENV DEST_CREDENTIALS ''
+ENV DEST_IN_CURRENT_ACCOUNT false
+
+ENV MAX_KEYS 1000
+ENV CHUNK_SIZE 5
+ENV MULTIPART_THRESHOLD 10
+ENV MESSAGE_BATCH_SIZE 10
+ENV FINDER_DEPTH 0
+ENV FINDER_NUMBER 1
+ENV WORKER_NUMBER 4
+
+WORKDIR /app
+RUN touch config.yaml
+COPY --from=builder /build/dthcli .
+ENTRYPOINT ["/app/dthcli", "run"]

README.md

@@ -1,17 +1,83 @@
-## My Project
+# data-transfer-hub-cli
 
-TODO: Fill this README out!
+dthcli (short for data-transfer-hub-cli) is a distributed tool to transfer data to Amazon S3 from other cloud storage services (including Aliyun OSS, Tencent COS, Qiniu Kodo, etc.) or between AWS regions (cross partition).
 
-Be sure to:
+## Introduction
 
-* Change the title in this README
-* Edit your repository description on GitHub
+This tool leverages Amazon SQS to distribute the transfer processes in many worker nodes. You can run as many worker nodes as required concurrently for large volume of objects. Each worker node will consume the messages from Amazon SQS and start transferring from the source to destination. Each message contains information that represents a object in cloud storage service to be transfered.
 
-## Security
 
-See [CONTRIBUTING](CONTRIBUTING.md#security-issue-notifications) for more information.
+## Installation
 
-## License
+You can clone this repo and build by yourself (Go Version >= v1.16)
 
-This project is licensed under the Apache-2.0 License.
 
+## Prerequisites
+
+You can run this tool in any places, even in local. However, you will need to have a SQS Queue and a DynamoDB table created before using the tool. DynamoDB is used to store the replication status of each objects, the partition key for DynamoDB must be `ObjectKey`
+
+If you need to provide AK/SK to accessing cloud service, you will need to set up a credential secure string in Amazon Secrets Manager with a format as below
+
+```
+{
+  "access_key_id": "<Your Access Key ID>",
+  "secret_access_key": "<Your Access Key Secret>"
+}
+```
+
+## Configuration
+
+The job information such as Source Bucket, Destination Bucket etc. are provided in either a config file or via environment variables. 
+
+An example of full config file can be found [here](./config-example.yaml) 
+
+You must provide at least the minimum information as below:
+```yaml
+srcBucket: src-bucket
+srcRegion: us-west-2
+
+destBucket: dest-bucket
+destRegion: cn-north-1
+
+jobTableName: test-table
+jobQueueName: test-queue
+```
+
+By default, this tool will try to read a `config.yaml` in the same folder, if you create the configuration file in a different folder or with a different file name, please use extra option `--config xxx.yaml` to load your config file.
+
+
+## Usage
+
+Run `dthcli help` for more details.
+```
+$ ./dthcli help
+A distributed CLI to replicate data to Amazon S3 from other cloud storage services.
+
+Usage:
+  dthcli [command]
+
+Available Commands:
+  help        Help about any command
+  run         Start running a job
+  version     Print the version number
+
+Flags:
+      --config string   config file (default is ./config.yaml)
+  -h, --help            help for dthcli
+
+Use "dthcli [command] --help" for more information about a command.
+```
+
+To actually start the job, use `dthcli run` command.
+
+- Start Finder Job
+
+```
+./dthcli run -t Finder
+```
+
+- Start Worker Job
+
+```
+./dthcli run -t Worker
+```

cmd/root.go

@@ -0,0 +1,211 @@
+/*
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+	"golang.a2z.com/dthcli/dth"
+)
+
+// Version Number
+const Version = "v1.0.0"
+
+var (
+	cfgFile, jobType string
+	cfg              *dth.JobConfig
+)
+
+// rootCmd represents the base command when called without any subcommands
+var rootCmd = &cobra.Command{
+	Use:   "dthcli",
+	Short: "A distributed CLI to transfer data to Amazon S3",
+	Long:  `A distributed CLI to transfer data to Amazon S3 from other cloud storage services.`,
+
+	// Uncomment the following line if your bare application
+	// has an action associated with it:
+	//	Run: func(cmd *cobra.Command, args []string) { },
+}
+
+// Execute adds all child commands to the root command and sets flags appropriately.
+// This is called by main.main(). It only needs to happen once to the rootCmd.
+func Execute() {
+	if err := rootCmd.Execute(); err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+}
+
+func init() {
+	cobra.OnInitialize(initConfig)
+
+	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is ./config.yaml)")
+
+	runCmd.Flags().StringVarP(&jobType, "type", "t", "Finder", "Job Type, choose either Finder or Worker")
+
+	rootCmd.AddCommand(versionCmd)
+	rootCmd.AddCommand(runCmd)
+}
+
+// initConfig reads in config file and ENV variables if set.
+func initConfig() {
+	viper.SetDefault("srcType", "Amazon_S3")
+	viper.SetDefault("destStorageClass", "STANDARD")
+	viper.SetDefault("srcPrefix", "")
+	viper.SetDefault("srcCredential", "")
+	viper.SetDefault("srcEndpoint", "")
+	viper.SetDefault("destPrefix", "")
+	viper.SetDefault("destCredential", "")
+	viper.SetDefault("destAcl", "bucket-owner-full-control")
+
+	viper.SetDefault("options.chunkSize", dth.DefaultChunkSize)
+	viper.SetDefault("options.multipartThreshold", dth.DefaultMultipartThreshold)
+	viper.SetDefault("options.maxKeys", dth.DefaultMaxKeys)
+	viper.SetDefault("options.messageBatchSize", dth.DefaultMessageBatchSize)
+	viper.SetDefault("options.finderDepth", dth.DefaultFinderDepth)
+	viper.SetDefault("options.finderNumber", dth.DefaultFinderNumber)
+	viper.SetDefault("options.workerNumber", dth.DefaultWorkerNumber)
+	viper.SetDefault("options.includeMetadata", false)
+
+	viper.BindEnv("srcType", "SOURCE_TYPE")
+	viper.BindEnv("srcBucket", "SRC_BUCKET")
+	viper.BindEnv("srcPrefix", "SRC_PREFIX")
+	viper.BindEnv("srcRegion", "SRC_REGION")
+	viper.BindEnv("srcEndpoint", "SRC_ENDPOINT")
+	viper.BindEnv("srcCredential", "SRC_CREDENTIALS")
+	viper.BindEnv("SrcInCurrentAccount", "SRC_IN_CURRENT_ACCOUNT")
+
+	viper.BindEnv("destBucket", "DEST_BUCKET")
+	viper.BindEnv("destPrefix", "DEST_PREFIX")
+	viper.BindEnv("destRegion", "DEST_REGION")
+	viper.BindEnv("destCredential", "DEST_CREDENTIALS")
+	viper.BindEnv("destInCurrentAccount", "DEST_IN_CURRENT_ACCOUNT")
+	viper.BindEnv("destStorageClass", "DEST_STORAGE_CLASS")
+	viper.BindEnv("destAcl", "DEST_ACL")
+
+	viper.BindEnv("jobTableName", "JOB_TABLE_NAME")
+	viper.BindEnv("jobQueueName", "JOB_QUEUE_NAME")
+
+	viper.BindEnv("options.maxKeys", "MAX_KEYS")
+	viper.BindEnv("options.chunkSize", "CHUNK_SIZE")
+	viper.BindEnv("options.multipartThreshold", "MULTIPART_THRESHOLD")
+	viper.BindEnv("options.messageBatchSize", "MESSAGE_BATCH_SIZE")
+	viper.BindEnv("options.finderDepth", "FINDER_DEPTH")
+	viper.BindEnv("options.finderNumber", "FINDER_NUMBER")
+	viper.BindEnv("options.workerNumber", "WORKER_NUMBER")
+	viper.BindEnv("options.includeMetadata", "INCLUDE_METADATA")
+
+	if cfgFile != "" {
+		// Use config file from the flag.
+		viper.SetConfigFile(cfgFile)
+	} else {
+		// Default config file is "./config.yaml"
+		viper.AddConfigPath(".")
+		viper.SetConfigName("config")
+	}
+
+	viper.AutomaticEnv() // read in environment variables that match
+
+	// If a config file is found, read it in.
+	viper.ReadInConfig()
+	if err := viper.ReadInConfig(); err == nil {
+		log.Println("Using config file:", viper.ConfigFileUsed())
+	}
+
+	options := &dth.JobOptions{
+		ChunkSize:          viper.GetInt("options.chunkSize"),
+		MultipartThreshold: viper.GetInt("options.multipartThreshold"),
+		MaxKeys:            viper.GetInt32("options.maxKeys"),
+		MessageBatchSize:   viper.GetInt("options.messageBatchSize"),
+		FinderDepth:        viper.GetInt("options.finderDepth"),
+		FinderNumber:       viper.GetInt("options.finderNumber"),
+		WorkerNumber:       viper.GetInt("options.workerNumber"),
+		IncludeMetadata:    viper.GetBool("options.includeMetadata"),
+	}
+
+	cfg = &dth.JobConfig{
+		SrcType:              viper.GetString("srcType"),
+		SrcBucket:            viper.GetString("srcBucket"),
+		SrcPrefix:            viper.GetString("srcPrefix"),
+		SrcRegion:            viper.GetString("srcRegion"),
+		SrcEndpoint:          viper.GetString("srcEndpoint"),
+		SrcCredential:        viper.GetString("srcCredential"),
+		SrcInCurrentAccount:  viper.GetBool("srcInCurrentAccount"),
+		DestBucket:           viper.GetString("destBucket"),
+		DestPrefix:           viper.GetString("destPrefix"),
+		DestRegion:           viper.GetString("destRegion"),
+		DestCredential:       viper.GetString("destCredential"),
+		DestStorageClass:     viper.GetString("destStorageClass"),
+		DestAcl:              viper.GetString("destAcl"),
+		DestInCurrentAccount: viper.GetBool("destInCurrentAccount"),
+		JobTableName:         viper.GetString("jobTableName"),
+		JobQueueName:         viper.GetString("jobQueueName"),
+		JobOptions:           options,
+	}
+
+}
+
+var versionCmd = &cobra.Command{
+	Use:   "version",
+	Short: "Print the version number",
+	Long:  "Print the version number",
+	Run: func(cmd *cobra.Command, args []string) {
+		fmt.Printf("dthcli version %s\n", Version)
+	},
+}
+
+var runCmd = &cobra.Command{
+	Use:   "run",
+	Short: "Start running a job",
+	Long: `Start running a job.
+
+For example: dthcli run -t Finder
+
+Supported types:
+	- Finder: Finder is a job that lists and compares objects in the source and target buckets, and sends the delta list to SQS Queue.
+	- Worker: Worker is a job that consumes the messages from SQS Queue and start the migration
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+
+		if cfg.SrcBucket == "" || cfg.DestBucket == "" {
+			log.Fatalf("Cannot find source or destination bucket name, please check if you have run with a config file or environment variables. Run `dthcli help` for more details")
+		}
+
+		log.Printf("Start running %s job", jobType)
+		ctx := context.Background()
+
+		var job dth.Job
+
+		switch jobType {
+		case "Finder":
+			job = dth.NewFinder(ctx, cfg)
+
+		case "Worker":
+			job = dth.NewWorker(ctx, cfg)
+
+		default:
+			log.Fatalf("Unknown Job Type - %s. Type must be either Finder or Worker\n, please start again", jobType)
+
+		}
+		job.Run(ctx)
+	},
+}