X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/4df1175e30c21850af394fcd60c9bb7ca3d981a5..20b7c375715bed99ccfe2dd12ed816b545e94324:/services/datamanager/keep/keep.go?ds=inline diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go index cf6803ac96..1a75bb1e33 100644 --- a/services/datamanager/keep/keep.go +++ b/services/datamanager/keep/keep.go @@ -4,43 +4,44 @@ package keep import ( "bufio" - "flag" + "encoding/json" + "errors" "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" "strconv" "strings" - "sync" "time" ) 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 { @@ -53,7 +54,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 } @@ -68,61 +69,24 @@ 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. - dataManagerToken string - dataManagerTokenFile string - dataManagerTokenFileReadOnce sync.Once -) - -func init() { - flag.StringVar(&dataManagerTokenFile, - "data-manager-token-file", - "", - "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 getDataManagerToken(arvLogger *logger.Logger) string { - readDataManagerToken := func() { - if dataManagerTokenFile == "" { - flag.Usage() - loggerutil.FatalWithMessage(arvLogger, - "Data Manager Token needed, but data manager token file not specified.") - } else { - rawRead, err := ioutil.ReadFile(dataManagerTokenFile) - if err != nil { - loggerutil.FatalWithMessage(arvLogger, - fmt.Sprintf("Unexpected error reading token file %s: %v", - dataManagerTokenFile, - err)) - } - dataManagerToken = strings.TrimSpace(string(rawRead)) - } +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) } - - dataManagerTokenFileReadOnce.Do(readDataManagerToken) - return dataManagerToken } func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServers) { 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) @@ -130,11 +94,6 @@ func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServer } func GetKeepServers(params GetKeepServersParams) (results ReadServers) { - if ¶ms.Client == nil { - log.Fatalf("params.Client passed to GetKeepServers() should " + - "contain a valid ArvadosClient, but instead it is nil.") - } - sdkParams := arvadosclient.Dict{ "filters": [][]string{[]string{"service_type", "=", "disk"}}, } @@ -152,13 +111,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 }) } @@ -177,9 +133,6 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers) { log.Printf("Got Server Addresses: %v", results) - // This is safe for concurrent use - client := http.Client{} - // Send off all the index requests concurrently responseChan := make(chan ServerResponse) for _, keepServer := range sdkResponse.KeepServers { @@ -192,12 +145,12 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers) { go func(keepServer ServerAddress) { responseChan <- GetServerContents(params.Logger, keepServer, - client) + params.Client) }(keepServer) } 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 { @@ -212,7 +165,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}) } } @@ -221,30 +173,86 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers) { func GetServerContents(arvLogger *logger.Logger, keepServer ServerAddress, - client http.Client) (response ServerResponse) { + arv arvadosclient.ArvadosClient) (response ServerResponse) { + + GetServerStatus(arvLogger, keepServer, arv) - req := CreateIndexRequest(arvLogger, keepServer) - resp, err := client.Do(req) + req := CreateIndexRequest(arvLogger, keepServer, arv) + resp, err := arv.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) } +func GetServerStatus(arvLogger *logger.Logger, + keepServer ServerAddress, + arv arvadosclient.ArvadosClient) { + url := fmt.Sprintf("http://%s:%d/status.json", + keepServer.Host, + keepServer.Port) + + if arvLogger != nil { + now := time.Now() + arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { + keepInfo := logger.GetOrCreateMap(p, "keep_info") + serverInfo := make(map[string]interface{}) + serverInfo["status_request_sent_at"] = now + serverInfo["host"] = keepServer.Host + serverInfo["port"] = keepServer.Port + + keepInfo[keepServer.Uuid] = serverInfo + }) + } + + resp, err := arv.Client.Get(url) + if err != nil { + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Error getting keep status from %s: %v", url, err)) + } else if resp.StatusCode != 200 { + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Received error code %d in response to request "+ + "for %s status: %s", + resp.StatusCode, url, resp.Status)) + } + + var keepStatus map[string]interface{} + decoder := json.NewDecoder(resp.Body) + decoder.UseNumber() + err = decoder.Decode(&keepStatus) + if err != nil { + loggerutil.FatalWithMessage(arvLogger, + fmt.Sprintf("Error decoding keep status from %s: %v", url, err)) + } + + 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["status_response_processed_at"] = now + serverInfo["status"] = keepStatus + }) + } +} + func CreateIndexRequest(arvLogger *logger.Logger, - keepServer ServerAddress) (req *http.Request) { + keepServer ServerAddress, + arv arvadosclient.ArvadosClient) (req *http.Request) { url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port) 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 := make(map[string]interface{}) - serverInfo["request_sent"] = time.Now() - - keepInfo[keepServer.String()] = serverInfo + keepInfo := logger.GetOrCreateMap(p, "keep_info") + serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{}) + serverInfo["index_request_sent_at"] = now }) } @@ -254,8 +262,7 @@ func CreateIndexRequest(arvLogger *logger.Logger, fmt.Sprintf("Error building http request for %s: %v", url, err)) } - req.Header.Add("Authorization", - fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger))) + req.Header.Add("Authorization", "OAuth2 " + arv.ApiToken) return } @@ -271,22 +278,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["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 "+ @@ -298,63 +329,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["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 @@ -368,8 +369,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 { @@ -383,15 +386,90 @@ 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(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 + } + + req.Header.Add("Authorization", "OAuth2 " + kc.Arvados.ApiToken) + + // 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 }