X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e5840cee519c5ff8b88e37e14160f9e4e12908ec..53a97213dad02d1fe684ad9b9e3717cd193a3058:/services/datamanager/keep/keep.go diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go index 20a59316b7..0e3cc1d44e 100644 --- a/services/datamanager/keep/keep.go +++ b/services/datamanager/keep/keep.go @@ -5,13 +5,15 @@ package keep import ( "bufio" "encoding/json" + "errors" "flag" "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,26 +24,26 @@ import ( ) type ServerAddress struct { + SSL bool `json:service_ssl_flag` Host string `json:"service_host"` Port int `json:"service_port"` + Uuid string `json:"uuid"` } // 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 { @@ -54,7 +56,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 } @@ -69,15 +71,6 @@ type KeepServiceList struct { KeepServers []ServerAddress `json:"items"` } -// Methods to implement util.SdkListResponse Interface -func (k KeepServiceList) NumItemsAvailable() (numAvailable int, err error) { - return k.ItemsAvailable, nil -} - -func (k KeepServiceList) NumItemsContained() (numContained int, err error) { - return len(k.KeepServers), nil -} - var ( // Don't access the token directly, use getDataManagerToken() to // make sure it's been read. @@ -93,11 +86,20 @@ func init() { "File with the API token we should use to contact keep servers.") } +// 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 { +func GetDataManagerToken(arvLogger *logger.Logger) string { readDataManagerToken := func() { if dataManagerTokenFile == "" { flag.Usage() @@ -123,7 +125,7 @@ func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServer results = GetKeepServers(params) log.Printf("Returned %d keep disks", len(results.ServerToContents)) - ComputeBlockReplicationCounts(&results) + results.Summarize(params.Logger) log.Printf("Replication level distribution: %v", results.BlockReplicationCounts) @@ -153,13 +155,10 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers) { if params.Logger != nil { params.Logger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := make(map[string]interface{}) - + keepInfo := logger.GetOrCreateMap(p, "keep_info") keepInfo["num_keep_servers_available"] = sdkResponse.ItemsAvailable keepInfo["num_keep_servers_received"] = len(sdkResponse.KeepServers) keepInfo["keep_servers"] = sdkResponse.KeepServers - - p["keep_info"] = keepInfo }) } @@ -198,7 +197,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 { @@ -213,7 +212,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}) } } @@ -230,7 +228,10 @@ func GetServerContents(arvLogger *logger.Logger, resp, err := client.Do(req) if err != nil { loggerutil.FatalWithMessage(arvLogger, - fmt.Sprintf("Error fetching %s: %v", req.URL.String(), err)) + fmt.Sprintf("Error fetching %s: %v. Response was %+v", + req.URL.String(), + err, + resp)) } return ReadServerResponse(arvLogger, keepServer, resp) @@ -244,12 +245,15 @@ func GetServerStatus(arvLogger *logger.Logger, keepServer.Port) if arvLogger != nil { + now := time.Now() arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := p["keep_info"].(map[string]interface{}) + keepInfo := logger.GetOrCreateMap(p, "keep_info") serverInfo := make(map[string]interface{}) - serverInfo["time_status_request_sent"] = time.Now() + serverInfo["status_request_sent_at"] = now + serverInfo["host"] = keepServer.Host + serverInfo["port"] = keepServer.Port - keepInfo[keepServer.String()] = serverInfo + keepInfo[keepServer.Uuid] = serverInfo }) } @@ -274,10 +278,11 @@ func GetServerStatus(arvLogger *logger.Logger, } if arvLogger != nil { + now := time.Now() arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := p["keep_info"].(map[string]interface{}) - serverInfo := keepInfo[keepServer.String()].(map[string]interface{}) - serverInfo["time_status_response_processed"] = time.Now() + keepInfo := logger.GetOrCreateMap(p, "keep_info") + serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{}) + serverInfo["status_response_processed_at"] = now serverInfo["status"] = keepStatus }) } @@ -289,10 +294,11 @@ func CreateIndexRequest(arvLogger *logger.Logger, log.Println("About to fetch keep server contents from " + url) if arvLogger != nil { + now := time.Now() arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := p["keep_info"].(map[string]interface{}) - serverInfo := keepInfo[keepServer.String()].(map[string]interface{}) - serverInfo["time_index_request_sent"] = time.Now() + keepInfo := logger.GetOrCreateMap(p, "keep_info") + serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{}) + serverInfo["index_request_sent_at"] = now }) } @@ -303,7 +309,7 @@ func CreateIndexRequest(arvLogger *logger.Logger, } req.Header.Add("Authorization", - fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger))) + fmt.Sprintf("OAuth2 %s", GetDataManagerToken(arvLogger))) return } @@ -319,22 +325,46 @@ func ReadServerResponse(arvLogger *logger.Logger, } if arvLogger != nil { + now := time.Now() arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := p["keep_info"].(map[string]interface{}) - serverInfo := keepInfo[keepServer.String()].(map[string]interface{}) - - serverInfo["time_index_response_received"] = time.Now() + keepInfo := logger.GetOrCreateMap(p, "keep_info") + serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{}) + serverInfo["index_response_received_at"] = now }) } 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 "+ @@ -346,63 +376,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 := p["keep_info"].(map[string]interface{}) - serverInfo := keepInfo[keepServer.String()].(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 { - arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - keepInfo := p["keep_info"].(map[string]interface{}) - serverInfo := keepInfo[keepServer.String()].(map[string]interface{}) - - serverInfo["time_processing_finished"] = time.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 @@ -416,8 +416,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 { @@ -431,15 +433,91 @@ 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 } -func ComputeBlockReplicationCounts(readServers *ReadServers) { +func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) { readServers.BlockReplicationCounts = make(map[int]int) for _, infos := range readServers.BlockToServers { replication := len(infos) readServers.BlockReplicationCounts[replication] += 1 } + + if arvLogger != nil { + arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { + keepInfo := logger.GetOrCreateMap(p, "keep_info") + keepInfo["distinct_blocks_stored"] = len(readServers.BlockToServers) + }) + } + +} + +type TrashRequest struct { + Locator string `json:"locator"` + BlockMtime int64 `json:"block_mtime"` +} + +type TrashList []TrashRequest + +func SendTrashLists(dataManagerToken string, kc *keepclient.KeepClient, spl map[string]TrashList) (errs []error) { + count := 0 + barrier := make(chan error) + + client := kc.Client + + for url, v := range spl { + count += 1 + log.Printf("Sending trash list to %v", url) + + go (func(url string, v TrashList) { + 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()) + barrier <- err + return + } + + // Add api token header + req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", dataManagerToken)) + + // Make the request + var resp *http.Response + if resp, err = client.Do(req); err != nil { + log.Printf("Error sending trash list to %v error: %v", url, err.Error()) + barrier <- err + return + } + + log.Printf("Sent trash list to %v: response was HTTP %v", url, resp.Status) + + io.Copy(ioutil.Discard, resp.Body) + resp.Body.Close() + + if resp.StatusCode != 200 { + barrier <- errors.New(fmt.Sprintf("Got HTTP code %v", resp.StatusCode)) + } else { + barrier <- nil + } + })(url, v) + + } + + for i := 0; i < count; i += 1 { + b := <-barrier + if b != nil { + errs = append(errs, b) + } + } + + return errs }