Now reading default replication level from api server.

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

diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go
index e1c9c29214114c2a816c09d2f117139da07f01da..c666337d0dd98da68b1a21e24dd06304b80f4930 100644 (file)
--- a/services/datamanager/keep/keep.go
+++ b/services/datamanager/keep/keep.go
@@ -10,7 +10,6 @@ import (
        "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
        "git.curoverse.com/arvados.git/sdk/go/blockdigest"
        "git.curoverse.com/arvados.git/sdk/go/logger"
-       "git.curoverse.com/arvados.git/sdk/go/manifest"
        "git.curoverse.com/arvados.git/services/datamanager/loggerutil"
        "io/ioutil"
        "log"
@@ -29,20 +28,18 @@ type ServerAddress struct {
 
 // Info about a particular block returned by the server
 type BlockInfo struct {
-       Digest blockdigest.BlockDigest
-       Size   int
+       Digest blockdigest.DigestWithSize
        Mtime  int64 // TODO(misha): Replace this with a timestamp.
 }
 
 // Info about a specified block given by a server
 type BlockServerInfo struct {
        ServerIndex int
-       Size        int
        Mtime       int64 // TODO(misha): Replace this with a timestamp.
 }
 
 type ServerContents struct {
-       BlockDigestToInfo map[blockdigest.BlockDigest]BlockInfo
+       BlockDigestToInfo map[blockdigest.DigestWithSize]BlockInfo
 }
 
 type ServerResponse struct {
@@ -55,7 +52,7 @@ type ReadServers struct {
        KeepServerIndexToAddress []ServerAddress
        KeepServerAddressToIndex map[ServerAddress]int
        ServerToContents         map[ServerAddress]ServerContents
-       BlockToServers           map[blockdigest.BlockDigest][]BlockServerInfo
+       BlockToServers           map[blockdigest.DigestWithSize][]BlockServerInfo
        BlockReplicationCounts   map[int]int
 }
 
@@ -192,7 +189,7 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers) {
        }
 
        results.ServerToContents = make(map[ServerAddress]ServerContents)
-       results.BlockToServers = make(map[blockdigest.BlockDigest][]BlockServerInfo)
+       results.BlockToServers = make(map[blockdigest.DigestWithSize][]BlockServerInfo)
 
        // Read all the responses
        for i := range sdkResponse.KeepServers {
@@ -207,7 +204,6 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers) {
                        results.BlockToServers[blockInfo.Digest] = append(
                                results.BlockToServers[blockInfo.Digest],
                                BlockServerInfo{ServerIndex: serverIndex,
-                                       Size:  blockInfo.Size,
                                        Mtime: blockInfo.Mtime})
                }
        }
@@ -331,7 +327,7 @@ func ReadServerResponse(arvLogger *logger.Logger,
 
        response.Address = keepServer
        response.Contents.BlockDigestToInfo =
-               make(map[blockdigest.BlockDigest]BlockInfo)
+               make(map[blockdigest.DigestWithSize]BlockInfo)
        scanner := bufio.NewScanner(resp.Body)
        numLines, numDuplicates, numSizeDisagreements := 0, 0, 0
        for scanner.Scan() {
@@ -348,33 +344,8 @@ func ReadServerResponse(arvLogger *logger.Logger,
                if storedBlock, ok := response.Contents.BlockDigestToInfo[blockInfo.Digest]; ok {
                        // This server returned multiple lines containing the same block digest.
                        numDuplicates += 1
-                       if storedBlock.Size != blockInfo.Size {
-                               numSizeDisagreements += 1
-                               // TODO(misha): Consider failing here.
-                               message := fmt.Sprintf("Saw different sizes for the same block "+
-                                       "on %s: %+v %+v",
-                                       keepServer.String(),
-                                       storedBlock,
-                                       blockInfo)
-                               log.Println(message)
-                               if arvLogger != nil {
-                                       arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
-                                               keepInfo := logger.GetOrCreateMap(p, "keep_info")
-                                               serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
-                                               var error_list []string
-                                               read_error_list, has_list := serverInfo["error_list"]
-                                               if has_list {
-                                                       error_list = read_error_list.([]string)
-                                               } // If we didn't have the list, error_list is already an empty list
-                                               serverInfo["error_list"] = append(error_list, message)
-                                       })
-                               }
-                       }
-                       // Keep the block that is bigger, or the block that's newer in
-                       // the case of a size tie.
-                       if storedBlock.Size < blockInfo.Size ||
-                               (storedBlock.Size == blockInfo.Size &&
-                                       storedBlock.Mtime < blockInfo.Mtime) {
+                       // Keep the block that's newer.
+                       if storedBlock.Mtime < blockInfo.Mtime {
                                response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
                        }
                } else {
@@ -419,8 +390,8 @@ func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err err
                        tokens)
        }
 
-       var locator manifest.BlockLocator
-       if locator, err = manifest.ParseBlockLocator(tokens[0]); err != nil {
+       var locator blockdigest.BlockLocator
+       if locator, err = blockdigest.ParseBlockLocator(tokens[0]); err != nil {
                err = fmt.Errorf("%v Received error while parsing line \"%s\"",
                        err, indexLine)
                return
@@ -436,8 +407,9 @@ func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err err
        if err != nil {
                return
        }
-       blockInfo.Digest = locator.Digest
-       blockInfo.Size = locator.Size
+       blockInfo.Digest =
+               blockdigest.DigestWithSize{Digest: locator.Digest,
+                       Size: uint32(locator.Size)}
        return
 }