@@ -113,8 +113,8 @@ func NewFinder(ctx context.Context, cfg *JobConfig) (f *Finder) {
113113 srcCred := getCredentials (ctx , cfg .SrcCredential , cfg .SrcInCurrentAccount , sm )
114114 desCred := getCredentials (ctx , cfg .DestCredential , cfg .DestInCurrentAccount , sm )
115115
116- srcClient := NewS3Client (ctx , cfg .SrcBucket , cfg .SrcPrefix , cfg .SrcEndpoint , cfg .SrcRegion , cfg .SrcType , srcCred )
117- desClient := NewS3Client (ctx , cfg .DestBucket , cfg .DestPrefix , "" , cfg .DestRegion , "Amazon_S3" , desCred )
116+ srcClient := NewS3Client (ctx , cfg .SrcBucket , cfg .SrcPrefix , cfg .SrcPrefixList , cfg . SrcEndpoint , cfg .SrcRegion , cfg .SrcType , srcCred )
117+ desClient := NewS3Client (ctx , cfg .DestBucket , cfg .DestPrefix , "" , "" , cfg .DestRegion , "Amazon_S3" , desCred )
118118
119119 f = & Finder {
120120 srcClient : srcClient ,
@@ -147,16 +147,27 @@ func (f *Finder) Run(ctx context.Context) {
147147 // Note that bigger number needs more memory
148148 compareCh := make (chan struct {}, f .cfg .FinderNumber )
149149
150- prefixes := f .srcClient .ListCommonPrefixes (ctx , f .cfg .FinderDepth , f .cfg .MaxKeys )
150+ var prefixes []* string
151+ log .Printf ("Prefix List File: %s" , f .cfg .SrcPrefixList )
151152
153+ if len (f .cfg .SrcPrefixList ) > 0 {
154+ prefixes = f .srcClient .ListSelectedPrefixes (ctx , & f .cfg .SrcPrefixList )
155+ } else {
156+ prefixes = f .srcClient .ListCommonPrefixes (ctx , f .cfg .FinderDepth , f .cfg .MaxKeys )
157+ }
152158 var wg sync.WaitGroup
153159
154160 start := time .Now ()
155161
156162 for _ , p := range prefixes {
157163 compareCh <- struct {}{}
164+ log .Printf ("prefix: %s" , * p )
158165 wg .Add (1 )
159- go f .compareAndSend (ctx , p , batchCh , msgCh , compareCh , & wg )
166+ if f .cfg .SkipCompare {
167+ go f .directSend (ctx , p , batchCh , msgCh , compareCh , & wg )
168+ } else {
169+ go f .compareAndSend (ctx , p , batchCh , msgCh , compareCh , & wg )
170+ }
160171 }
161172 wg .Wait ()
162173
@@ -295,6 +306,94 @@ func (f *Finder) compareAndSend(ctx context.Context, prefix *string, batchCh cha
295306 <- compareCh
296307}
297308
309+ // This function will send the task to SQS Queue directly, without comparison.
310+ func (f * Finder ) directSend (ctx context.Context , prefix * string , batchCh chan struct {}, msgCh chan * string , compareCh chan struct {}, wg * sync.WaitGroup ) {
311+ defer wg .Done ()
312+
313+ log .Printf ("Scanning prefix /%s\n " , * prefix )
314+
315+ token := ""
316+ i , j := 0 , 0
317+ retry := 0
318+ // batch := make([]*string, f.cfg.MessageBatchSize)
319+
320+ log .Printf ("Start sending without comparison ...\n " )
321+ // start := time.Now()
322+
323+ for token != "End" {
324+ // source := f.getSourceObjects(ctx, &token, prefix)
325+ source , err := f .srcClient .ListObjects (ctx , & token , prefix , f .cfg .MaxKeys )
326+ if err != nil {
327+ log .Printf ("Fail to get source list - %s\n " , err .Error ())
328+ //
329+ log .Printf ("Sleep for 1 minute and try again..." )
330+ retry ++
331+
332+ if retry <= MaxRetries {
333+ time .Sleep (time .Minute * 1 )
334+ continue
335+ } else {
336+ log .Printf ("Still unable to list source list after %d retries\n " , MaxRetries )
337+ // Log the last token and exit
338+ log .Fatalf ("The last token is %s\n " , token )
339+ }
340+
341+ }
342+
343+ // if a successful list, reset to 0
344+ retry = 0
345+
346+ for _ , obj := range source {
347+ // TODO: Check if there is another way to compare
348+ // Currently, map is used to search if such object exists in target
349+ msgCh <- obj .toString ()
350+ i ++
351+ if i % f .cfg .MessageBatchSize == 0 {
352+ wg .Add (1 )
353+ j ++
354+ if j % 100 == 0 {
355+ log .Printf ("Found %d batches in prefix /%s\n " , j , * prefix )
356+ }
357+ batchCh <- struct {}{}
358+
359+ // start a go routine to send messages in batch
360+ go func (i int ) {
361+ defer wg .Done ()
362+ batch := make ([]* string , i )
363+ for a := 0 ; a < i ; a ++ {
364+ batch [a ] = <- msgCh
365+ }
366+
367+ f .sqs .SendMessageInBatch (ctx , batch )
368+ <- batchCh
369+ }(f .cfg .MessageBatchSize )
370+ i = 0
371+ }
372+ }
373+ }
374+ // For remainning objects.
375+ if i != 0 {
376+ j ++
377+ wg .Add (1 )
378+ batchCh <- struct {}{}
379+ go func (i int ) {
380+ defer wg .Done ()
381+ batch := make ([]* string , i )
382+ for a := 0 ; a < i ; a ++ {
383+ batch [a ] = <- msgCh
384+ }
385+
386+ f .sqs .SendMessageInBatch (ctx , batch )
387+ <- batchCh
388+ }(i )
389+ }
390+
391+ // end := time.Since(start)
392+ // log.Printf("Compared and Sent %d batches in %v", j, end)
393+ log .Printf ("Completed in prefix /%s, found %d batches in total" , * prefix , j )
394+ <- compareCh
395+ }
396+
298397// NewWorker creates a new Worker instance
299398func NewWorker (ctx context.Context , cfg * JobConfig ) (w * Worker ) {
300399 log .Printf ("Source Type is %s\n " , cfg .SrcType )
@@ -310,8 +409,8 @@ func NewWorker(ctx context.Context, cfg *JobConfig) (w *Worker) {
310409 srcCred := getCredentials (ctx , cfg .SrcCredential , cfg .SrcInCurrentAccount , sm )
311410 desCred := getCredentials (ctx , cfg .DestCredential , cfg .DestInCurrentAccount , sm )
312411
313- srcClient := NewS3Client (ctx , cfg .SrcBucket , cfg .SrcPrefix , cfg .SrcEndpoint , cfg .SrcRegion , cfg .SrcType , srcCred )
314- desClient := NewS3Client (ctx , cfg .DestBucket , cfg .DestPrefix , "" , cfg .DestRegion , "Amazon_S3" , desCred )
412+ srcClient := NewS3Client (ctx , cfg .SrcBucket , cfg .SrcPrefix , cfg .SrcPrefixList , cfg . SrcEndpoint , cfg .SrcRegion , cfg .SrcType , srcCred )
413+ desClient := NewS3Client (ctx , cfg .DestBucket , cfg .DestPrefix , "" , "" , cfg .DestRegion , "Amazon_S3" , desCred )
315414
316415 return & Worker {
317416 srcClient : srcClient ,
0 commit comments