maxManifestSize uint64
)
-const (
- DefaultReplicationLevel = 2
-)
-
type Collection struct {
Uuid string
OwnerUuid string
}
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 {
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 {
collectionInfo := logger.GetOrCreateMap(p, "collection_info")
collectionInfo["num_collections_at_start"] = initialNumberOfCollectionsAvailable
collectionInfo["batch_size"] = params.BatchSize
+ collectionInfo["default_replication_level"] = defaultReplicationLevel
})
}
sdkParams["filters"].([][]string)[0][2] =
ProcessCollections(params.Logger,
collections.Items,
+ defaultReplicationLevel,
results.UuidToCollection).Format(time.RFC3339)
// update counts
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),
}
if collection.ReplicationLevel == 0 {
- collection.ReplicationLevel = DefaultReplicationLevel
+ collection.ReplicationLevel = defaultReplicationLevel
}
manifest := manifest.Manifest{sdkCollection.ManifestText}
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)
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
}
}
}
collectionInfo["owner_to_collection_size"] =
readCollections.OwnerToCollectionSize
collectionInfo["distinct_blocks_named"] =
- len(readCollections.BlockToReplication)
+ len(readCollections.BlockToDesiredReplication)
})
}