README
¶
drhlib
This repo is the Golang implementaion of data replication migration lib
Getting Started
Prerequisites
Prepare environment variables with job info as below
export JOB_TABLE_NAME=test-table
export SQS_QUEUE_NAME=test-queue
export SRC_BUCKET_NAME=test-src
export SRC_BUCKET_PREFIX=
export SRC_REGION=us-west-2
export DEST_BUCKET_NAME=test-dest
export DEST_BUCKET_PREFIX=
export DEST_REGION=us-west-1
export STORAGE_CLASS=STANDARD
...
export SOURCE_TYPE=Amazon_S3
export MULTIPART_THRESHOLD=10
export CHUNK_SIZE=5
export MAX_THREADS=10
Then run go get
go get github.com/daixba/drhlib
Start Finder
Finder lists and compares objects in source and target and get the delta and send that to SQS Queue.
For example:
import (
"context"
"github.com/daixba/drhlib"
)
func main() {
ctx := context.TODO()
f := drhlib.InitFinder(ctx)
f.CompareAndSend()
}
Start Migrator
Migrator receives messages from SQS Queue and replicate the object from source to target (Get from Source, Put to Target)
For example:
import (
"context"
"github.com/daixba/drhlib"
)
func main() {
ctx := context.TODO()
// ... To be updated.
}
Documentation
¶
Index ¶
- Constants
- Variables
- type Client
- type DBService
- type Finder
- type Item
- type JobConfig
- type JobInfo
- type Message
- type Metadata
- type Migrator
- type S3Client
- func (c *S3Client) CompleteMultipartUpload(key string)
- func (c *S3Client) CreateMultipartUpload(key string)
- func (c *S3Client) GetObject(key string, size, start, chunkSize int, version string) ([]byte, error)
- func (c *S3Client) HeadObject(key string)
- func (c *S3Client) ListMultipartUploads(key string)
- func (c *S3Client) ListObjects(continuationToken *string) ([]JobInfo, error)
- func (c *S3Client) ListParts(key string)
- func (c *S3Client) PutObject(key string, body []byte, storageClass string) string
- func (c *S3Client) UploadPart(key string)
- type SqsService
Constants ¶
const ( // MaxKeys is the maximum number of keys returned per listing request, default is 1000 MaxKeys int32 = 1000 // MultipartThreshold is the threshold size (in Bytes) to determine to use multipart upload or not. // When object size is greater or equals to MultipartThreshold, multipart upload will be used. MultipartThreshold int = 10 * 1024 * 1024 // ChunkSize is the chunk size (in Bytes) for each part when using multipart upload ChunkSize int = 5 * 1024 * 1024 // MaxParts the maximum number of parts is 10000 for multipart upload MaxParts int = 10000 )
Variables ¶
var DefaultConfig = &JobConfig{ ChunkSize: ChunkSize, MultipartThreshold: MultipartThreshold, MaxKeys: MaxKeys, MaxParts: MaxParts, }
DefaultConfig is default config used for replication
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client interface {
// GET
ListObjects(continuationToken *string) ([]JobInfo, error)
HeadObject(key string)
GetObject(key string, size, start, chunkSize int, version string) ([]byte, error)
// PUT
PutObject(key string, body []byte, storageClass string) string
CreateMultipartUpload(key string)
CompleteMultipartUpload(key string)
UploadPart(key string)
ListParts(key string)
ListMultipartUploads(key string)
}
Client is an interface used to contact with Cloud Storage Services
type DBService ¶
type DBService struct {
// contains filtered or unexported fields
}
DBService is a wrapper service used to interact with Amazon DynamoDB
type Finder ¶
type Finder struct {
// contains filtered or unexported fields
}
Finder struct
func (*Finder) CompareAndSend ¶
func (f *Finder) CompareAndSend()
CompareAndSend is main execution function. This function will compare source and target and get a list of delta, and then send delta to SQS Queue.
type Item ¶
type Item struct {
ObjectKey, StorageClass, DesBucket, DesKey, Status, Version string
Size int32
StartTime, EndTime int
ExtraInfo Metadata
}
Item holds info about the items to be stored in DynamoDB
type Message ¶
type Message struct {
// contains filtered or unexported fields
}
Message is Default message format
type Migrator ¶
type Migrator struct {
// contains filtered or unexported fields
}
Migrator is a struct ..
func (*Migrator) StartMigration ¶
func (m *Migrator) StartMigration()
StartMigration is a function to
type S3Client ¶
type S3Client struct {
// contains filtered or unexported fields
}
S3Client is an implementation of Client interface for Amazon S3
func NewS3Client ¶
NewS3Client create a S3Client instance
func (*S3Client) CompleteMultipartUpload ¶
CompleteMultipartUpload is
func (*S3Client) CreateMultipartUpload ¶
CreateMultipartUpload is
func (*S3Client) GetObject ¶
func (c *S3Client) GetObject(key string, size, start, chunkSize int, version string) ([]byte, error)
GetObject is a function to get (download) object from Amazon S3
func (*S3Client) HeadObject ¶
HeadObject is a function to get extra metadata from Amazon S3
func (*S3Client) ListMultipartUploads ¶
ListMultipartUploads is
func (*S3Client) ListObjects ¶
ListObjects is a function to list objects from Amazon S3
type SqsService ¶
type SqsService struct {
// contains filtered or unexported fields
}
SqsService is a wrapper service used to interact with Amazon SQS
func NewSqsService ¶
func NewSqsService(ctx context.Context, queueName string) (*SqsService, error)
NewSqsService is helper function used to create a SqsService instance
func (*SqsService) DeleteMessage ¶
func (ss *SqsService) DeleteMessage() (ok bool)
DeleteMessage function is used to delete message from the Queue Returns True if message is deleted successfully
func (*SqsService) IsQueueEmpty ¶
func (ss *SqsService) IsQueueEmpty() bool
IsQueueEmpty is a function to check if the Queue is empty or not
func (*SqsService) ReceiveMessages ¶
func (ss *SqsService) ReceiveMessages()
ReceiveMessages function receives many messages in batch from the Queue
func (*SqsService) SendMessage ¶
func (ss *SqsService) SendMessage(msg Message)
SendMessage function sends 1 message at a time to the Queue
func (*SqsService) SendMessageInBatch ¶
func (ss *SqsService) SendMessageInBatch(msgs []Message)
SendMessageInBatch function sends messages to the Queue in batch. Each batch can only contains up to 10 messages
Source Files