X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/53f785c298338645b6880f22f26b0c36a7cfab4d..7da85b85af2dd9b22186bfd48190de7b68f75837:/services/datamanager/keep/keep.go diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go index 857ab86672..871acc8e79 100644 --- a/services/datamanager/keep/keep.go +++ b/services/datamanager/keep/keep.go @@ -9,9 +9,10 @@ import ( "fmt" "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/blockdigest" + "git.curoverse.com/arvados.git/sdk/go/keepclient" "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" "io/ioutil" "log" "net/http" @@ -22,6 +23,7 @@ import ( ) type ServerAddress struct { + SSL bool `json:service_ssl_flag` Host string `json:"service_host"` Port int `json:"service_port"` Uuid string `json:"uuid"` @@ -29,20 +31,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 +55,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 } @@ -87,7 +87,15 @@ func init() { // TODO(misha): Change this to include the UUID as well. func (s ServerAddress) String() string { - return fmt.Sprintf("%s:%d", s.Host, s.Port) + return s.URL() +} + +func (s ServerAddress) URL() string { + if s.SSL { + return fmt.Sprintf("https://%s:%d", s.Host, s.Port) + } else { + return fmt.Sprintf("http://%s:%d", s.Host, s.Port) + } } func getDataManagerToken(arvLogger *logger.Logger) string { @@ -188,7 +196,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 { @@ -203,7 +211,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}) } } @@ -327,12 +334,36 @@ func ReadServerResponse(arvLogger *logger.Logger, response.Address = keepServer response.Contents.BlockDigestToInfo = - make(map[blockdigest.BlockDigest]BlockInfo) - scanner := bufio.NewScanner(resp.Body) + make(map[blockdigest.DigestWithSize]BlockInfo) + reader := bufio.NewReader(resp.Body) numLines, numDuplicates, numSizeDisagreements := 0, 0, 0 - for scanner.Scan() { + for { numLines++ - blockInfo, err := parseBlockInfoFromIndexLine(scanner.Text()) + line, err := reader.ReadString('\n') + if err == io.EOF { + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Index from %s truncated at line %d", + keepServer.String(), numLines)) + } else if err != nil { + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Error reading index response from %s at line %d: %v", + keepServer.String(), numLines, err)) + } + if line == "\n" { + if _, err := reader.Peek(1); err == nil { + extra, _ := reader.ReadString('\n') + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Index from %s had trailing data at line %d after EOF marker: %s", + keepServer.String(), numLines+1, extra)) + } else if err != io.EOF { + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Index from %s had read error after EOF marker at line %d: %v", + keepServer.String(), numLines, err)) + } + numLines-- + break + } + blockInfo, err := parseBlockInfoFromIndexLine(line) if err != nil { loggerutil.FatalWithMessage(arvLogger, fmt.Sprintf("Error parsing BlockInfo from index line "+ @@ -344,64 +375,33 @@ 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 { response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo } } - if err := scanner.Err(); err != nil { - loggerutil.FatalWithMessage(arvLogger, - fmt.Sprintf("Received error scanning index response from %s: %v", - keepServer.String(), - err)) - } else { - log.Printf("%s index contained %d lines with %d duplicates with "+ - "%d size disagreements", - keepServer.String(), - numLines, - numDuplicates, - numSizeDisagreements) - - if arvLogger != nil { - now := time.Now() - arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := logger.GetOrCreateMap(p, "keep_info") - serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{}) - - serverInfo["processing_finished_at"] = now - serverInfo["lines_received"] = numLines - serverInfo["duplicates_seen"] = numDuplicates - serverInfo["size_disagreements_seen"] = numSizeDisagreements - }) - } + + log.Printf("%s index contained %d lines with %d duplicates with "+ + "%d size disagreements", + keepServer.String(), + numLines, + numDuplicates, + numSizeDisagreements) + + if arvLogger != nil { + now := time.Now() + arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { + keepInfo := logger.GetOrCreateMap(p, "keep_info") + serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{}) + + serverInfo["processing_finished_at"] = now + serverInfo["lines_received"] = numLines + serverInfo["duplicates_seen"] = numDuplicates + serverInfo["size_disagreements_seen"] = numSizeDisagreements + }) } resp.Body.Close() return @@ -415,8 +415,10 @@ 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 } if len(locator.Hints) > 0 { @@ -430,8 +432,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 } @@ -450,3 +453,59 @@ func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) { } } + +type TrashRequest struct { + Locator string `json:"locator"` + BlockMtime int64 `json:"block_mtime"` +} + +type TrashList []TrashRequest + +func SendTrashLists(arvLogger *logger.Logger, kc *keepclient.KeepClient, spl map[string]TrashList) { + count := 0 + rendezvous := make(chan bool) + + for url, v := range spl { + count += 1 + log.Printf("Sending trash list to %v", url) + + go (func(url string, v TrashList) { + defer (func() { + rendezvous <- true + })() + + pipeReader, pipeWriter := io.Pipe() + go (func() { + enc := json.NewEncoder(pipeWriter) + enc.Encode(v) + pipeWriter.Close() + })() + + req, err := http.NewRequest("PUT", fmt.Sprintf("%s/trash", url), pipeReader) + if err != nil { + log.Printf("Error creating trash list request for %v error: %v", url, err.Error()) + return + } + + // Add api token header + req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger))) + + // Make the request + var resp *http.Response + if resp, err = kc.Client.Do(req); err != nil { + log.Printf("Error sending trash list to %v error: %v", url, err.Error()) + return + } + + log.Printf("Sent trash list to %v: response was HTTP %d", url, resp.Status) + + io.Copy(ioutil.Discard, resp.Body) + resp.Body.Close() + })(url, v) + + } + + for i := 0; i < count; i += 1 { + <-rendezvous + } +}