X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/74e4c93a1f333352914314b3a53172a0164a20e9..cdad40f38151c5d678f13787853f361566fbcd43:/services/datamanager/collection/collection.go

diff --git a/services/datamanager/collection/collection.go b/services/datamanager/collection/collection.go
index 7102ec3c05..7cc3ff7204 100644
--- a/services/datamanager/collection/collection.go
+++ b/services/datamanager/collection/collection.go
@@ -18,9 +18,6 @@ import (
 
 var (
 	HeapProfileFilename string
-	// globals for debugging
-	totalManifestSize uint64
-	maxManifestSize   uint64
 )
 
 // Collection representation
@@ -41,7 +38,6 @@ type ReadCollections struct {
 	CollectionUUIDToIndex     map[string]int
 	CollectionIndexToUUID     []string
 	BlockToCollectionIndices  map[blockdigest.DigestWithSize][]int
-	Err                       error
 }
 
 // GetCollectionsParams params
@@ -96,10 +92,9 @@ func WriteHeapProfile() error {
 }
 
 // GetCollectionsAndSummarize gets collections from api and summarizes
-func GetCollectionsAndSummarize(params GetCollectionsParams) (results ReadCollections) {
-	results, err := GetCollections(params)
+func GetCollectionsAndSummarize(params GetCollectionsParams) (results ReadCollections, err error) {
+	results, err = GetCollections(params)
 	if err != nil {
-		results.Err = err
 		return
 	}
 
@@ -200,13 +195,12 @@ func GetCollections(params GetCollectionsParams) (results ReadCollections, err e
 		}
 
 		// Process collection and update our date filter.
-		var latestModificationDate time.Time
-		latestModificationDate, err = ProcessCollections(params.Logger,
+		latestModificationDate, maxManifestSize, totalManifestSize, err := ProcessCollections(params.Logger,
 			collections.Items,
 			defaultReplicationLevel,
 			results.UUIDToCollection)
 		if err != nil {
-			return
+			return results, err
 		}
 		sdkParams["filters"].([][]string)[0][2] = latestModificationDate.Format(time.RFC3339)
 
@@ -233,6 +227,30 @@ func GetCollections(params GetCollectionsParams) (results ReadCollections, err e
 		}
 	}
 
+	// Make one final API request to verify that we have processed all collections available up to the latest modification date
+	var collections SdkCollectionList
+	sdkParams["filters"].([][]string)[0][1] = "<="
+	sdkParams["limit"] = 0
+	err = params.Client.List("collections", sdkParams, &collections)
+	if err != nil {
+		return
+	}
+	finalNumberOfCollectionsAvailable, err :=
+		util.NumberItemsAvailable(params.Client, "collections")
+	if err != nil {
+		return
+	}
+	if totalCollections < finalNumberOfCollectionsAvailable {
+		err = fmt.Errorf("API server indicates a total of %d collections "+
+				"available up to %v, but we only retrieved %d. "+
+				"Refusing to continue as this could indicate an "+
+				"otherwise undetected failure.",
+				finalNumberOfCollectionsAvailable, 
+				sdkParams["filters"].([][]string)[0][2],
+				totalCollections)
+		return
+	}
+
 	// Write the heap profile for examining memory usage
 	err = WriteHeapProfile()
 
@@ -250,7 +268,12 @@ func StrCopy(s string) string {
 func ProcessCollections(arvLogger *logger.Logger,
 	receivedCollections []SdkCollectionInfo,
 	defaultReplicationLevel int,
-	UUIDToCollection map[string]Collection) (latestModificationDate time.Time, err error) {
+	UUIDToCollection map[string]Collection,
+) (
+	latestModificationDate time.Time,
+	maxManifestSize, totalManifestSize uint64,
+	err error,
+) {
 	for _, sdkCollection := range receivedCollections {
 		collection := Collection{UUID: StrCopy(sdkCollection.UUID),
 			OwnerUUID:         StrCopy(sdkCollection.OwnerUUID),
@@ -258,12 +281,13 @@ func ProcessCollections(arvLogger *logger.Logger,
 			BlockDigestToSize: make(map[blockdigest.BlockDigest]int)}
 
 		if sdkCollection.ModifiedAt.IsZero() {
-			return latestModificationDate, fmt.Errorf(
+			err = fmt.Errorf(
 				"Arvados SDK collection returned with unexpected zero "+
 					"modification date. This probably means that either we failed to "+
 					"parse the modification date or the API server has changed how "+
 					"it returns modification dates: %+v",
 				collection)
+			return
 		}
 
 		if sdkCollection.ModifiedAt.After(latestModificationDate) {
@@ -274,7 +298,7 @@ func ProcessCollections(arvLogger *logger.Logger,
 			collection.ReplicationLevel = defaultReplicationLevel
 		}
 
-		manifest := manifest.Manifest{sdkCollection.ManifestText}
+		manifest := manifest.Manifest{Text: sdkCollection.ManifestText}
 		manifestSize := uint64(len(sdkCollection.ManifestText))
 
 		if _, alreadySeen := UUIDToCollection[collection.UUID]; !alreadySeen {
@@ -287,16 +311,20 @@ func ProcessCollections(arvLogger *logger.Logger,
 		blockChannel := manifest.BlockIterWithDuplicates()
 		for block := range blockChannel {
 			if storedSize, stored := collection.BlockDigestToSize[block.Digest]; stored && storedSize != block.Size {
-				err = fmt.Errorf(
+				log.Printf(
 					"Collection %s contains multiple sizes (%d and %d) for block %s",
 					collection.UUID,
 					storedSize,
 					block.Size,
 					block.Digest)
-				return
 			}
 			collection.BlockDigestToSize[block.Digest] = block.Size
 		}
+		if manifest.Err != nil {
+			err = manifest.Err
+			return
+		}
+
 		collection.TotalSize = 0
 		for _, size := range collection.BlockDigestToSize {
 			collection.TotalSize += size