Modifies sanity check to query for the total expected number of collections

[arvados.git] / services / datamanager / collection / collection.go

diff --git a/services/datamanager/collection/collection.go b/services/datamanager/collection/collection.go
index 7102ec3c054bb24df5e3e0979784b9c08b3949ed..42e0cc429054fbb8da0831f117881682f4be225f 100644 (file)
--- 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,29 @@ 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
+       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 +267,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 +280,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 +297,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 +310,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