Merge pull request #121 from RachelTucker/improved_error_handling

Improving error handling in helpers in case of malformed objects

Author: LogicalChaos

helpers/getProducer.go

@@ -71,19 +71,24 @@ func toReadObjectMap(getObjects *[]helperModels.GetObject) map[string]helperMode
 
 // Processes all the blobs in a chunk that are ready for transfer from BP
 // Returns the number of blobs queued for process
-func (producer *getProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) int {
+func (producer *getProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) (int, error) {
     producer.Debugf("begin chunk processing %s", curChunk.ChunkId)
 
     processedCount := 0
     // transfer blobs that are ready, and queue those that are waiting for channel
     for _, curObj := range curChunk.Objects {
         producer.Debugf("queuing object in waiting to be processed %s offset=%d length=%d", *curObj.Name, curObj.Offset, curObj.Length)
         blob := helperModels.NewBlobDescription(*curObj.Name, curObj.Offset, curObj.Length)
-        if producer.queueBlobForTransfer(&blob, bucketName, jobId) {
+
+        blobQueued, err := producer.queueBlobForTransfer(&blob, bucketName, jobId)
+        if err != nil {
+            return 0, err
+        }
+        if blobQueued {
             processedCount++
         }
     }
-    return processedCount
+    return processedCount, nil
 }
 
 // Information required to perform a get operation of a blob with BP as data source and channelBuilder as destination
@@ -231,24 +236,37 @@ func writeRangeToDestination(channelBuilder helperModels.WriteChannelBuilder, bl
 // Attempts to transfer a single blob from the BP to the client. If the blob is not ready for transfer,
 // then it is added to the waiting to transfer queue
 // Returns whether or not the blob was queued for transfer
-func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) bool {
+func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) (bool, error) {
     if producer.processedBlobTracker.IsProcessed(*blob) {
-        return false // already been processed
+        return false, nil // already been processed
     }
 
-    curReadObj := producer.readObjectMap[blob.Name()]
+    curReadObj, ok := producer.readObjectMap[blob.Name()]
+    if !ok {
+        err := fmt.Errorf("failed to find object associated with blob in object map: %s offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
+        producer.Errorf("unrecoverable error: %v", err)
+        producer.processedBlobTracker.MarkProcessed(*blob)
+        return false, err // fatal error occurred
+    }
+
+    if curReadObj.ChannelBuilder == nil {
+        err := fmt.Errorf("failed to transfer object, it does not have a channel builder: %s", curReadObj.Name)
+        producer.Errorf("unrecoverable error: %v", err)
+        producer.processedBlobTracker.MarkProcessed(*blob)
+        return false, err // fatal error occurred
+    }
 
     if curReadObj.ChannelBuilder.HasFatalError() {
         // a fatal error happened on a previous blob for this file, skip processing
         producer.Warningf("fatal error occurred while transferring previous blob on this file, skipping blob '%s' offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
         producer.processedBlobTracker.MarkProcessed(*blob)
-        return false // not going to process
+        return false, nil // not going to process
     }
 
     if !curReadObj.ChannelBuilder.IsChannelAvailable(blob.Offset()) {
         producer.Debugf("channel is not currently available for getting blob '%s' offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
         producer.deferredBlobQueue.Push(blob)
-        return false // not ready to be processed
+        return false, nil // not ready to be processed
     }
 
     producer.Debugf("channel is available for getting blob '%s' offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
@@ -261,7 +279,7 @@ func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
         jobId:          jobId,
     }
 
-    var transfer TransferOperation = producer.transferOperationBuilder(objInfo)
+    transfer := producer.transferOperationBuilder(objInfo)
 
     // Increment wait group, and enqueue transfer operation
     producer.waitGroup.Add(1)
@@ -270,31 +288,35 @@ func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
     // Mark blob as processed
     producer.processedBlobTracker.MarkProcessed(*blob)
 
-    return true
+    return true, nil
 }
 
 // Attempts to process all blobs whose channels were not available for transfer.
 // Blobs whose channels are still not available are placed back on the queue.
 // Returns the number of blobs queued for processing.
-func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string) int {
+func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string) (int, error) {
     processedCount := 0
 
     // attempt to process all blobs in waiting to be transferred
     waitingBlobs := producer.deferredBlobQueue.Size()
     for i := 0; i < waitingBlobs; i++ {
         //attempt transfer
         curBlob, err := producer.deferredBlobQueue.Pop()
-        producer.Debugf("attempting to process '%s' offset=%d length=%d", curBlob.Name(), curBlob.Offset(), curBlob.Length())
         if err != nil {
             //should not be possible to get here
-            producer.Errorf("failure during blob transfer '%s' at offset %d: %s", curBlob.Name(), curBlob.Offset(), err.Error())
+            producer.Errorf(err.Error())
             break
         }
-        if producer.queueBlobForTransfer(curBlob, bucketName, jobId) {
+        producer.Debugf("attempting to process '%s' offset=%d length=%d", curBlob.Name(), curBlob.Offset(), curBlob.Length())
+        blobQueued, err := producer.queueBlobForTransfer(curBlob, bucketName, jobId)
+        if err != nil {
+            return 0, err
+        }
+        if blobQueued {
             processedCount++
         }
     }
-    return processedCount
+    return processedCount, nil
 }
 
 // This initiates the production of the transfer operations which will be consumed by a consumer running in a separate go routine.
@@ -304,7 +326,7 @@ func (producer *getProducer) run() error {
     defer close(*producer.queue)
 
     // determine number of blobs to be processed
-    var totalBlobCount int64 = producer.totalBlobCount()
+    var totalBlobCount = producer.totalBlobCount()
     producer.Debugf("job status totalBlobs=%d processedBlobs=%d", totalBlobCount, producer.processedBlobTracker.NumberOfProcessedBlobs())
 
     // process all chunks and make sure all blobs are queued for transfer
@@ -332,7 +354,10 @@ func (producer *getProducer) hasMoreToProcess(totalBlobCount int64) bool {
 // Returns the number of blobs that have been queued for transfer
 func (producer *getProducer) queueBlobsReadyForTransfer(totalBlobCount int64) (int, error) {
     // Attempt to transfer waiting blobs
-    processedCount := producer.processWaitingBlobs(*producer.JobMasterObjectList.BucketName, producer.JobMasterObjectList.JobId)
+    processedCount, err := producer.processWaitingBlobs(*producer.JobMasterObjectList.BucketName, producer.JobMasterObjectList.JobId)
+    if err != nil {
+        return 0, err
+    }
 
     // Check if we need to query the BP for allocated blobs, or if we already know everything is allocated.
     if int64(producer.deferredBlobQueue.Size()) + producer.processedBlobTracker.NumberOfProcessedBlobs() >= totalBlobCount {
@@ -356,7 +381,11 @@ func (producer *getProducer) queueBlobsReadyForTransfer(totalBlobCount int64) (i
         // Loop through all the chunks that are available for processing, and send
         // the files that are contained within them.
         for _, curChunk := range chunksReadyResponse.MasterObjectList.Objects {
-            processedCount += producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
+            justProcessedCount, err := producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
+            if err != nil {
+                return 0, err
+            }
+            processedCount += justProcessedCount
         }
     }
     return processedCount, nil

helpers/putProducer.go

@@ -1,6 +1,7 @@
 package helpers
 
 import (
+    "fmt"
     "github.com/SpectraLogic/ds3_go_sdk/ds3"
     ds3Models "github.com/SpectraLogic/ds3_go_sdk/ds3/models"
     helperModels "github.com/SpectraLogic/ds3_go_sdk/helpers/models"
@@ -136,25 +137,29 @@ func (producer *putProducer) metadataFrom(info putObjectInfo) map[string]string
 // Processes all the blobs in a chunk and attempts to add them to the transfer queue.
 // If a blob is not ready for transfer, then it is added to the waiting to be transferred queue.
 // Returns the number of  blobs added to queue.
-func (producer *putProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) int {
+func (producer *putProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) (int, error) {
     processedCount := 0
     producer.Debugf("begin chunk processing %s", curChunk.ChunkId)
 
     // transfer blobs that are ready, and queue those that are waiting for channel
     for _, curObj := range curChunk.Objects {
         producer.Debugf("queuing object in waiting to be processed %s offset=%d length=%d", *curObj.Name, curObj.Offset, curObj.Length)
         blob := helperModels.NewBlobDescription(*curObj.Name, curObj.Offset, curObj.Length)
-        if producer.queueBlobForTransfer(&blob, bucketName, jobId) {
+        blobQueued, err := producer.queueBlobForTransfer(&blob, bucketName, jobId)
+        if err != nil {
+            return 0, err
+        }
+        if blobQueued {
             processedCount++
         }
     }
-    return processedCount
+    return processedCount, nil
 }
 
 // Iterates through blobs that are waiting to be transferred and attempts to transfer.
 // If successful, blob is removed from queue. Else, it is re-queued.
 // Returns the number of blobs added to queue.
-func (producer *putProducer) processWaitingBlobs(bucketName string, jobId string) int {
+func (producer *putProducer) processWaitingBlobs(bucketName string, jobId string) (int, error) {
     processedCount := 0
 
     // attempt to process all blobs in waiting to be transferred
@@ -168,36 +173,53 @@ func (producer *putProducer) processWaitingBlobs(bucketName string, jobId string
             break
         }
         producer.Debugf("attempting to process %s offset=%d length=%d", curBlob.Name(), curBlob.Offset(), curBlob.Length())
-        if producer.queueBlobForTransfer(curBlob, bucketName, jobId) {
+        blobQueued, err := producer.queueBlobForTransfer(curBlob, bucketName, jobId)
+        if err != nil {
+            return 0, err
+        }
+        if blobQueued {
             processedCount++
         }
     }
 
-    return processedCount
+    return processedCount, nil
 }
 
 // Attempts to transfer a single blob. If the blob is not ready for transfer,
 // it is added to the waiting to transfer queue.
 // Returns whether or not the blob was queued for transfer.
-func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) bool {
+func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) (bool, error) {
     if producer.processedBlobTracker.IsProcessed(*blob) {
-        return false // this was already processed
+        return false, nil // this was already processed
     }
 
-    curWriteObj := producer.writeObjectMap[blob.Name()]
+    curWriteObj, ok := producer.writeObjectMap[blob.Name()]
+    if !ok {
+        err := fmt.Errorf("failed to find object associated with blob in object map: %s offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
+        producer.Errorf("unrecoverable error: %v", err)
+        producer.processedBlobTracker.MarkProcessed(*blob)
+        return false, err // fatal error occurred
+    }
+
+    if curWriteObj.ChannelBuilder == nil {
+        err := fmt.Errorf("failed to transfer object, it does not have a channel builder: %s", curWriteObj.PutObject.Name)
+        producer.Errorf("unrecoverable error: %v", err)
+        producer.processedBlobTracker.MarkProcessed(*blob)
+        return false, err // fatal error occurred
+    }
 
     if curWriteObj.ChannelBuilder.HasFatalError() {
         // a fatal error happened on a previous blob for this file, skip processing
         producer.Warningf("fatal error occurred while transferring previous blob on this file, skipping blob %s offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
         producer.processedBlobTracker.MarkProcessed(*blob)
-        return false // not actually transferring this blob
+        return false, nil // not actually transferring this blob
     }
 
     if !curWriteObj.ChannelBuilder.IsChannelAvailable(blob.Offset()) {
         producer.Debugf("channel is not currently available for blob %s offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
         // Not ready to be transferred
         producer.deferredBlobQueue.Push(blob)
-        return false // not ready to be sent
+        return false, nil // not ready to be sent
     }
 
     producer.Debugf("channel is available for blob %s offset=%d length=%d", curWriteObj.PutObject.Name, blob.Offset(), blob.Length())
@@ -211,7 +233,7 @@ func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
         jobId:          jobId,
     }
 
-    var transfer TransferOperation = producer.transferOperationBuilder(objInfo)
+    transfer := producer.transferOperationBuilder(objInfo)
 
     // Increment wait group, and enqueue transfer operation
     producer.waitGroup.Add(1)
@@ -220,7 +242,7 @@ func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
     // Mark blob as processed
     producer.processedBlobTracker.MarkProcessed(*blob)
 
-    return true
+    return true, nil
 }
 
 // This initiates the production of the transfer operations which will be consumed by a consumer running in a separate go routine.
@@ -230,7 +252,7 @@ func (producer *putProducer) run() error {
     defer close(*producer.queue)
 
     // determine number of blobs to be processed
-    var totalBlobCount int64 = producer.totalBlobCount()
+    totalBlobCount := producer.totalBlobCount()
     producer.Debugf("job status totalBlobs=%d processedBlobs=%d", totalBlobCount, producer.processedBlobTracker.NumberOfProcessedBlobs())
 
     // process all chunks and make sure all blobs are queued for transfer
@@ -259,7 +281,10 @@ func (producer *putProducer) hasMoreToProcess(totalBlobCount int64) bool {
 // Returns the number of items queued for work.
 func (producer *putProducer) queueBlobsReadyForTransfer(totalBlobCount int64) (int, error) {
     // Attempt to transfer waiting blobs
-    processedCount := producer.processWaitingBlobs(*producer.JobMasterObjectList.BucketName, producer.JobMasterObjectList.JobId)
+    processedCount, err := producer.processWaitingBlobs(*producer.JobMasterObjectList.BucketName, producer.JobMasterObjectList.JobId)
+    if err != nil {
+        return 0, err
+    }
 
     // Check if we need to query the BP for allocated blobs, or if we already know everything is allocated.
     if int64(producer.deferredBlobQueue.Size()) + producer.processedBlobTracker.NumberOfProcessedBlobs() >= totalBlobCount {
@@ -283,7 +308,11 @@ func (producer *putProducer) queueBlobsReadyForTransfer(totalBlobCount int64) (i
         // Loop through all the chunks that are available for processing, and send
         // the files that are contained within them.
         for _, curChunk := range chunksReadyResponse.MasterObjectList.Objects {
-            processedCount += producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
+            justProcessedCount, err := producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
+            if err != nil {
+                return 0, err
+            }
+            processedCount += justProcessedCount
         }
     }
     return processedCount, nil