From 48a88ab18de9f583d3ca7c7ad7a3d9a84c49d7d8 Mon Sep 17 00:00:00 2001
From: Oliver Seitz <seitz-oliver@gmx.de>
Date: Thu, 23 Sep 2021 12:23:27 +0200
Subject: [PATCH] feat: Added IgnoreMissingObjects to ignore objects that were
 not found in the source bucket

---
 cmd/root.go   | 18 ++++++++++--------
 dth/config.go |  2 +-
 dth/job.go    | 17 +++++++++++++++++
 3 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/cmd/root.go b/cmd/root.go
index a9999d9..902731a 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -86,6 +86,7 @@ func initConfig() {
 	viper.SetDefault("options.finderNumber", dth.DefaultFinderNumber)
 	viper.SetDefault("options.workerNumber", dth.DefaultWorkerNumber)
 	viper.SetDefault("options.includeMetadata", false)
+	viper.SetDefault("options.ignoreMissingObjects", false)
 
 	viper.BindEnv("srcType", "SOURCE_TYPE")
 	viper.BindEnv("srcBucket", "SRC_BUCKET")
@@ -135,14 +136,15 @@ func initConfig() {
 	}
 
 	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"),
+		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.worker-Number"),
+		IncludeMetadata:      viper.GetBool("options.includeMetadata"),
+		IgnoreMissingObjects: viper.GetBool("options.ignoreMissingObjects"),
 	}
 
 	cfg = &dth.JobConfig{
diff --git a/dth/config.go b/dth/config.go
index 31b7135..2c0be03 100644
--- a/dth/config.go
+++ b/dth/config.go
@@ -61,7 +61,7 @@ const (
 type JobOptions struct {
 	ChunkSize, MultipartThreshold, MessageBatchSize, FinderDepth, FinderNumber, WorkerNumber int
 	MaxKeys                                                                                  int32
-	IncludeMetadata                                                                          bool
+	IncludeMetadata, IgnoreMissingObjects                                                    bool
 }
 
 // JobConfig is General Job Info
diff --git a/dth/job.go b/dth/job.go
index ec410ec..457c3d7 100644
--- a/dth/job.go
+++ b/dth/job.go
@@ -455,6 +455,23 @@ func (w *Worker) Run(ctx context.Context) {
 			continue
 		}
 
+		if w.cfg.IgnoreMissingObjects {
+			// Delete message if object does not exist.
+			// This might happen if the object was deleted in the meantime.
+			meta := w.srcClient.HeadObject(ctx, &obj.Key)
+			if meta == nil {
+				log.Printf("Object not found %s/%s\n", w.cfg.SrcBucket, obj.Key)
+				res := &TransferResult{
+					status: "NOT_FOUND",
+					etag:   nil,
+					err:    nil,
+				}
+				w.db.UpdateItem(ctx, &obj.Key, res)
+				w.sqs.DeleteMessage(ctx, rh)
+				continue
+			}
+		}
+
 		destKey := appendPrefix(&obj.Key, &w.cfg.DestPrefix)
 
 		if action == Transfer {