X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/ef6773029872db261f1b3bf3fb2ca86f47dfdcfe..44321d4b5e7677df2ad70d88db6358b2778b58e5:/services/datamanager/collection/collection.go

diff --git a/services/datamanager/collection/collection.go b/services/datamanager/collection/collection.go
index e0929b7757..5519ad8670 100644
--- a/services/datamanager/collection/collection.go
+++ b/services/datamanager/collection/collection.go
@@ -24,10 +24,6 @@ var (
 	maxManifestSize   uint64
 )
 
-const (
-	DefaultReplicationLevel = 2
-)
-
 type Collection struct {
 	Uuid              string
 	OwnerUuid         string
@@ -37,13 +33,13 @@ type Collection struct {
 }
 
 type ReadCollections struct {
-	ReadAllCollections       bool
-	UuidToCollection         map[string]Collection
-	OwnerToCollectionSize    map[string]int
-	BlockToReplication       map[blockdigest.BlockDigest]int
-	CollectionUuidToIndex    map[string]int
-	CollectionIndexToUuid    []string
-	BlockToCollectionIndices map[blockdigest.BlockDigest][]int
+	ReadAllCollections        bool
+	UuidToCollection          map[string]Collection
+	OwnerToCollectionSize     map[string]int
+	BlockToDesiredReplication map[blockdigest.DigestWithSize]int
+	CollectionUuidToIndex     map[string]int
+	CollectionIndexToUuid     []string
+	BlockToCollectionIndices  map[blockdigest.DigestWithSize][]int
 }
 
 type GetCollectionsParams struct {
@@ -134,6 +130,23 @@ func GetCollections(params GetCollectionsParams) (results ReadCollections) {
 		sdkParams["limit"] = params.BatchSize
 	}
 
+	var defaultReplicationLevel int
+	{
+		value, err := params.Client.Discovery("defaultCollectionReplication")
+		if err != nil {
+			loggerutil.FatalWithMessage(params.Logger,
+				fmt.Sprintf("Error querying default collection replication: %v", err))
+		}
+
+		defaultReplicationLevel = int(value.(float64))
+		if defaultReplicationLevel <= 0 {
+			loggerutil.FatalWithMessage(params.Logger,
+				fmt.Sprintf("Default collection replication returned by arvados SDK "+
+					"should be a positive integer but instead it was %d.",
+					defaultReplicationLevel))
+		}
+	}
+
 	initialNumberOfCollectionsAvailable, err :=
 		util.NumberItemsAvailable(params.Client, "collections")
 	if err != nil {
@@ -151,6 +164,7 @@ func GetCollections(params GetCollectionsParams) (results ReadCollections) {
 			collectionInfo := logger.GetOrCreateMap(p, "collection_info")
 			collectionInfo["num_collections_at_start"] = initialNumberOfCollectionsAvailable
 			collectionInfo["batch_size"] = params.BatchSize
+			collectionInfo["default_replication_level"] = defaultReplicationLevel
 		})
 	}
 
@@ -176,6 +190,7 @@ func GetCollections(params GetCollectionsParams) (results ReadCollections) {
 		sdkParams["filters"].([][]string)[0][2] =
 			ProcessCollections(params.Logger,
 				collections.Items,
+				defaultReplicationLevel,
 				results.UuidToCollection).Format(time.RFC3339)
 
 		// update counts
@@ -216,6 +231,7 @@ func StrCopy(s string) string {
 
 func ProcessCollections(arvLogger *logger.Logger,
 	receivedCollections []SdkCollectionInfo,
+	defaultReplicationLevel int,
 	uuidToCollection map[string]Collection) (latestModificationDate time.Time) {
 	for _, sdkCollection := range receivedCollections {
 		collection := Collection{Uuid: StrCopy(sdkCollection.Uuid),
@@ -238,7 +254,7 @@ func ProcessCollections(arvLogger *logger.Logger,
 		}
 
 		if collection.ReplicationLevel == 0 {
-			collection.ReplicationLevel = DefaultReplicationLevel
+			collection.ReplicationLevel = defaultReplicationLevel
 		}
 
 		manifest := manifest.Manifest{sdkCollection.ManifestText}
@@ -281,11 +297,11 @@ func ProcessCollections(arvLogger *logger.Logger,
 
 func (readCollections *ReadCollections) Summarize(arvLogger *logger.Logger) {
 	readCollections.OwnerToCollectionSize = make(map[string]int)
-	readCollections.BlockToReplication = make(map[blockdigest.BlockDigest]int)
+	readCollections.BlockToDesiredReplication = make(map[blockdigest.DigestWithSize]int)
 	numCollections := len(readCollections.UuidToCollection)
 	readCollections.CollectionUuidToIndex = make(map[string]int, numCollections)
 	readCollections.CollectionIndexToUuid = make([]string, 0, numCollections)
-	readCollections.BlockToCollectionIndices = make(map[blockdigest.BlockDigest][]int)
+	readCollections.BlockToCollectionIndices = make(map[blockdigest.DigestWithSize][]int)
 
 	for _, coll := range readCollections.UuidToCollection {
 		collectionIndex := len(readCollections.CollectionIndexToUuid)
@@ -296,12 +312,15 @@ func (readCollections *ReadCollections) Summarize(arvLogger *logger.Logger) {
 		readCollections.OwnerToCollectionSize[coll.OwnerUuid] =
 			readCollections.OwnerToCollectionSize[coll.OwnerUuid] + coll.TotalSize
 
-		for block, _ := range coll.BlockDigestToSize {
-			readCollections.BlockToCollectionIndices[block] =
-				append(readCollections.BlockToCollectionIndices[block], collectionIndex)
-			storedReplication := readCollections.BlockToReplication[block]
+		for block, size := range coll.BlockDigestToSize {
+			locator := blockdigest.DigestWithSize{Digest: block, Size: uint32(size)}
+			readCollections.BlockToCollectionIndices[locator] =
+				append(readCollections.BlockToCollectionIndices[locator],
+					collectionIndex)
+			storedReplication := readCollections.BlockToDesiredReplication[locator]
 			if coll.ReplicationLevel > storedReplication {
-				readCollections.BlockToReplication[block] = coll.ReplicationLevel
+				readCollections.BlockToDesiredReplication[locator] =
+					coll.ReplicationLevel
 			}
 		}
 	}
@@ -315,7 +334,7 @@ func (readCollections *ReadCollections) Summarize(arvLogger *logger.Logger) {
 			collectionInfo["owner_to_collection_size"] =
 				readCollections.OwnerToCollectionSize
 			collectionInfo["distinct_blocks_named"] =
-				len(readCollections.BlockToReplication)
+				len(readCollections.BlockToDesiredReplication)
 		})
 	}