GOSDK-18: Add object error and completed listeners to Strategies in helpers. Adding in a listener strategy for the helpers. This allows a callback to be specified for all errors that occur on a per-file basis. Removed the aggregate error that used to collect all errors that happened per-file. Now the PutObjects and GetObjects call return a simple error which affects the entire job only. Individual file errors must be retrieved via the new error callback.

Author: RachelTucker

helpers/getProducer.go

@@ -1,13 +1,13 @@
 package helpers
 
 import (
-    ds3Models "github.com/SpectraLogic/ds3_go_sdk/ds3/models"
     "github.com/SpectraLogic/ds3_go_sdk/ds3"
-    "sync"
-    "io"
-    "github.com/SpectraLogic/ds3_go_sdk/helpers/ranges"
+    ds3Models "github.com/SpectraLogic/ds3_go_sdk/ds3/models"
     helperModels "github.com/SpectraLogic/ds3_go_sdk/helpers/models"
+    "github.com/SpectraLogic/ds3_go_sdk/helpers/ranges"
     "github.com/SpectraLogic/ds3_go_sdk/sdk_log"
+    "io"
+    "sync"
 )
 
 type getProducer struct {
@@ -56,14 +56,14 @@ func toReadObjectMap(getObjects *[]helperModels.GetObject) map[string]helperMode
 }
 
 // Processes all the blobs in a chunk that are ready for transfer from BP
-func (producer *getProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string, aggErr *ds3Models.AggregateError) {
+func (producer *getProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) {
     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)
-        producer.queueBlobForTransfer(&blob, bucketName, jobId, aggErr)
+        producer.queueBlobForTransfer(&blob, bucketName, jobId)
     }
 }
 
@@ -76,7 +76,7 @@ type getObjectInfo struct {
 }
 
 // Creates the transfer operation that will perform the data retrieval of the specified blob from BP
-func (producer *getProducer) transferOperationBuilder(info getObjectInfo, aggErr *ds3Models.AggregateError) TransferOperation {
+func (producer *getProducer) transferOperationBuilder(info getObjectInfo) TransferOperation {
     return func() {
         // has this file fatally errored while transferring a different blob?
         if info.channelBuilder.HasFatalError() {
@@ -98,7 +98,7 @@ func (producer *getProducer) transferOperationBuilder(info getObjectInfo, aggErr
 
         getObjResponse, err := producer.client.GetObject(getObjRequest)
         if err != nil {
-            aggErr.Append(err)
+            producer.strategy.Listeners.Errored(info.blob.Name(), err)
             info.channelBuilder.SetFatalError(err)
             producer.Errorf("unable to retrieve object '%s' at offset %d: %s", info.blob.Name(), info.blob.Offset(), err.Error())
             return
@@ -111,15 +111,15 @@ func (producer *getProducer) transferOperationBuilder(info getObjectInfo, aggErr
         if len(blobRanges) == 0 {
             writer, err := info.channelBuilder.GetChannel(info.blob.Offset())
             if err != nil {
-                aggErr.Append(err)
+                producer.strategy.Listeners.Errored(info.blob.Name(), err)
                 info.channelBuilder.SetFatalError(err)
                 producer.Errorf("unable to read contents of object '%s' at offset '%d': %s", info.blob.Name(), info.blob.Offset(), err.Error())
                 return
             }
             defer info.channelBuilder.OnDone(writer)
             _, err = io.Copy(writer, getObjResponse.Content) //copy all content from response reader to destination writer
             if err != nil {
-                aggErr.Append(err)
+                producer.strategy.Listeners.Errored(info.blob.Name(), err)
                 info.channelBuilder.SetFatalError(err)
                 producer.Errorf("unable to copy content of object '%s' at offset '%d' from source to destination: %s", info.blob.Name(), info.blob.Offset(), err.Error())
             }
@@ -130,7 +130,7 @@ func (producer *getProducer) transferOperationBuilder(info getObjectInfo, aggErr
         for _, r := range blobRanges {
             err := writeRangeToDestination(info.channelBuilder, r, getObjResponse.Content)
             if err != nil {
-                aggErr.Append(err)
+                producer.strategy.Listeners.Errored(info.blob.Name(), err)
                 info.channelBuilder.SetFatalError(err)
                 producer.Errorf("unable to write to destination channel for object '%s' with range '%v': %s", info.blob.Name(), r, err.Error())
             }
@@ -153,7 +153,7 @@ 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
-func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string, aggErr *ds3Models.AggregateError) {
+func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) {
     if producer.processedBlobTracker.IsProcessed(*blob) {
         return
     }
@@ -183,7 +183,7 @@ func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
         jobId:          jobId,
     }
 
-    var transfer TransferOperation = producer.transferOperationBuilder(objInfo, aggErr)
+    var transfer TransferOperation = producer.transferOperationBuilder(objInfo)
 
     // Increment wait group, and enqueue transfer operation
     producer.waitGroup.Add(1)
@@ -195,7 +195,7 @@ func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
 
 // 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.
-func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string, aggErr *ds3Models.AggregateError) {
+func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string) {
     // attempt to process all blobs in waiting to be transferred
     waitingBlobs := producer.deferredBlobQueue.Size()
     for i := 0; i < waitingBlobs; i++ {
@@ -204,18 +204,17 @@ func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string
         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
-            aggErr.Append(err)
+            producer.strategy.Listeners.Errored(curBlob.Name(), err)
             producer.Errorf("failure during blob transfer '%s' at offset %d: %s", curBlob.Name(), curBlob.Offset(), err.Error())
         }
-        producer.queueBlobForTransfer(curBlob, bucketName, jobId, aggErr)
+        producer.queueBlobForTransfer(curBlob, bucketName, jobId)
     }
 }
 
 // This initiates the production of the transfer operations which will be consumed by a consumer running in a separate go routine.
 // Each transfer operation will retrieve one blob of content from the BP.
 // Once all blobs have been queued to be transferred, the producer will finish, even if all operations have not been consumed yet.
-func (producer *getProducer) run(aggErr *ds3Models.AggregateError) {
-    defer producer.waitGroup.Done()
+func (producer *getProducer) run() error {
     defer close(*producer.queue)
 
     // determine number of blobs to be processed
@@ -230,9 +229,8 @@ func (producer *getProducer) run(aggErr *ds3Models.AggregateError) {
         chunksReady := ds3Models.NewGetJobChunksReadyForClientProcessingSpectraS3Request(producer.JobMasterObjectList.JobId)
         chunksReadyResponse, err := producer.client.GetJobChunksReadyForClientProcessingSpectraS3(chunksReady)
         if err != nil {
-            aggErr.Append(err)
             producer.Errorf("unrecoverable error: %v", err)
-            return
+            return err
         }
 
         // Check to see if any chunks can be processed
@@ -241,17 +239,18 @@ func (producer *getProducer) run(aggErr *ds3Models.AggregateError) {
             // 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 {
-                producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId, aggErr)
+                producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
             }
 
             // Attempt to transfer waiting blobs
-            producer.processWaitingBlobs(*chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId, aggErr)
+            producer.processWaitingBlobs(*chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
         } else {
             // When no chunks are returned we need to sleep to allow for cache space to
             // be freed.
             producer.strategy.BlobStrategy.delay()
         }
     }
+    return nil
 }
 
 // Determines the number of blobs to be transferred.

helpers/getTransfernator.go

@@ -81,9 +81,9 @@ func (transceiver *getTransceiver) transfer() (string, error) {
 
     // Wait for completion of producer-consumer goroutines
     var aggErr ds3Models.AggregateError
-    waitGroup.Add(2)  // adding producer and consumer goroutines to wait group
-    go producer.run(&aggErr) // producer will add to waitGroup for every blob retrieval added to queue, and each transfer performed will decrement from waitGroup
+    waitGroup.Add(1)  // adding producer and consumer goroutines to wait group
     go consumer.run()
+    err = producer.run() // producer will add to waitGroup for every blob retrieval added to queue, and each transfer performed will decrement from waitGroup
     waitGroup.Wait()
 
     return bulkGetResponse.MasterObjectList.JobId, aggErr.GetErrors()

helpers/listeners.go

@@ -0,0 +1,13 @@
+package helpers
+
+type ListenerStrategy struct {
+    // Called when an error occurred during transfer of an object.
+    // This must be a thread safe function.
+    ErrorCallback func(objectName string, err error)
+}
+
+func (listener *ListenerStrategy) Errored(objectName string, err error) {
+    if listener.ErrorCallback != nil {
+        listener.ErrorCallback(objectName, err)
+    }
+}

helpers/putProducer.go

@@ -60,7 +60,7 @@ type putObjectInfo struct {
 }
 
 // Creates the transfer operation that will perform the data upload of the specified blob to BP
-func (producer *putProducer) transferOperationBuilder(info putObjectInfo, aggErr *ds3Models.AggregateError) TransferOperation {
+func (producer *putProducer) transferOperationBuilder(info putObjectInfo) TransferOperation {
     return func() {
         // has this file fatally errored while transferring a different blob?
         if info.channelBuilder.HasFatalError() {
@@ -70,7 +70,8 @@ func (producer *putProducer) transferOperationBuilder(info putObjectInfo, aggErr
         }
         reader, err := info.channelBuilder.GetChannel(info.blob.Offset())
         if err != nil {
-            aggErr.Append(err)
+            producer.strategy.Listeners.Errored(info.blob.Name(), err)
+
             info.channelBuilder.SetFatalError(err)
             producer.Errorf("could not get reader for object with name='%s' offset=%d length=%d: %v", info.blob.Name(), info.blob.Offset(), info.blob.Length(), err)
             return
@@ -86,7 +87,8 @@ func (producer *putProducer) transferOperationBuilder(info putObjectInfo, aggErr
 
         _, err = producer.client.PutObject(putObjRequest)
         if err != nil {
-            aggErr.Append(err)
+            producer.strategy.Listeners.Errored(info.blob.Name(), err)
+
             info.channelBuilder.SetFatalError(err)
             producer.Errorf("problem during transfer of %s: %s", info.blob.Name(), err.Error())
         }
@@ -120,38 +122,38 @@ 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.
-func (producer *putProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string, aggErr *ds3Models.AggregateError) {
+func (producer *putProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) {
     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)
-        producer.queueBlobForTransfer(&blob, bucketName, jobId, aggErr)
+        producer.queueBlobForTransfer(&blob, bucketName, jobId)
     }
 }
 
 // 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.
-func (producer *putProducer) processWaitingBlobs(bucketName string, jobId string, aggErr *ds3Models.AggregateError) {
+func (producer *putProducer) processWaitingBlobs(bucketName string, jobId string) {
     // 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()
         if err != nil {
-            aggErr.Append(err)
+            producer.strategy.Listeners.Errored(curBlob.Name(), err)
             producer.Errorf("problem when getting next blob to be transferred: %s", err.Error())
             continue
         }
         producer.Debugf("attempting to process %s offset=%d length=%d", curBlob.Name(), curBlob.Offset(), curBlob.Length())
-        producer.queueBlobForTransfer(curBlob, bucketName, jobId, aggErr)
+        producer.queueBlobForTransfer(curBlob, bucketName, jobId)
     }
 }
 
 // Attempts to transfer a single blob. If the blob is not ready for transfer,
 // it is added to the waiting to transfer queue.
-func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string, aggErr *ds3Models.AggregateError) {
+func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) {
     if producer.processedBlobTracker.IsProcessed(*blob) {
         return
     }
@@ -183,7 +185,7 @@ func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
         jobId:          jobId,
     }
 
-    var transfer TransferOperation = producer.transferOperationBuilder(objInfo, aggErr)
+    var transfer TransferOperation = producer.transferOperationBuilder(objInfo)
 
     // Increment wait group, and enqueue transfer operation
     producer.waitGroup.Add(1)
@@ -196,8 +198,7 @@ func (producer *putProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
 // This initiates the production of the transfer operations which will be consumed by a consumer running in a separate go routine.
 // Each transfer operation will put one blob of content to the BP.
 // Once all blobs have been queued to be transferred, the producer will finish, even if all operations have not been consumed yet.
-func (producer *putProducer) run(aggErr *ds3Models.AggregateError) {
-    defer producer.waitGroup.Done()
+func (producer *putProducer) run() error {
     defer close(*producer.queue)
 
     // determine number of blobs to be processed
@@ -212,9 +213,8 @@ func (producer *putProducer) run(aggErr *ds3Models.AggregateError) {
         chunksReady := ds3Models.NewGetJobChunksReadyForClientProcessingSpectraS3Request(producer.JobMasterObjectList.JobId)
         chunksReadyResponse, err := producer.client.GetJobChunksReadyForClientProcessingSpectraS3(chunksReady)
         if err != nil {
-            aggErr.Append(err)
             producer.Errorf("unrecoverable error: %v", err)
-            return
+            return err
         }
 
         // Check to see if any chunks can be processed
@@ -223,17 +223,18 @@ func (producer *putProducer) run(aggErr *ds3Models.AggregateError) {
             // 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 {
-                producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId, aggErr)
+                producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
             }
 
             // Attempt to transfer waiting blobs
-            producer.processWaitingBlobs(*chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId, aggErr)
+            producer.processWaitingBlobs(*chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
         } else {
             // When no chunks are returned we need to sleep to allow for cache space to
             // be freed.
             producer.strategy.BlobStrategy.delay()
         }
     }
+    return nil
 }
 
 // Determines the number of blobs to be transferred.

helpers/putTransceiver.go

@@ -78,14 +78,13 @@ func (transceiver *putTransceiver) transfer() (string, error) {
     consumer := newConsumer(&queue, &waitGroup, transceiver.Strategy.BlobStrategy.maxConcurrentTransfers())
 
     // Wait for completion of producer-consumer goroutines
-    waitGroup.Add(2)  // adding producer and consumer goroutines to wait group
+    waitGroup.Add(1)  // adding producer and consumer goroutines to wait group
 
-    var aggErr ds3Models.AggregateError
-    go producer.run(&aggErr) // producer will add to waitGroup for every blob added to queue, and each transfer performed will decrement from waitGroup
     go consumer.run()
+    err = producer.run() // producer will add to waitGroup for every blob added to queue, and each transfer performed will decrement from waitGroup
     waitGroup.Wait()
 
-    return bulkPutResponse.MasterObjectList.JobId, aggErr.GetErrors()
+    return bulkPutResponse.MasterObjectList.JobId, err
 }
 
 /*

helpers/readTransferStrategy.go

@@ -5,6 +5,7 @@ import "github.com/SpectraLogic/ds3_go_sdk/ds3/models"
 type ReadTransferStrategy struct {
     BlobStrategy ReadBlobStrategy
     Options      ReadBulkJobOptions
+    Listeners    ListenerStrategy
 }
 
 // Defines the options to use on the get bulk job

helpers/writeTransferStrategy.go

@@ -10,6 +10,7 @@ var MinUploadSize int64 = 10485760
 type WriteTransferStrategy struct {
     BlobStrategy WriteBlobStrategy
     Options      WriteBulkJobOptions
+    Listeners    ListenerStrategy
 }
 
 // Defines the options to use on the put bulk job