X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d005c38bea17107e7be3c5338ad31ba54085df61..0d2a561091b777dfb7735d842a88f8529f4f982a:/services/datamanager/keep/keep.go diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go index c49a0211a8..39d2d5bd5d 100644 --- a/services/datamanager/keep/keep.go +++ b/services/datamanager/keep/keep.go @@ -6,6 +6,7 @@ import ( "bufio" "encoding/json" "errors" + "flag" "fmt" "git.curoverse.com/arvados.git/sdk/go/arvadosclient" "git.curoverse.com/arvados.git/sdk/go/blockdigest" @@ -65,7 +66,7 @@ type ReadServers struct { // GetKeepServersParams struct type GetKeepServersParams struct { - Client arvadosclient.ArvadosClient + Client *arvadosclient.ArvadosClient Logger *logger.Logger Limit int } @@ -76,6 +77,15 @@ type ServiceList struct { KeepServers []ServerAddress `json:"items"` } +var serviceType string + +func init() { + flag.StringVar(&serviceType, + "service-type", + "disk", + "Operate only on keep_services with the specified service_type, ignoring all others.") +} + // String // TODO(misha): Change this to include the UUID as well. func (s ServerAddress) String() string { @@ -108,7 +118,7 @@ func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServer // GetKeepServers from api server func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error) { sdkParams := arvadosclient.Dict{ - "filters": [][]string{[]string{"service_type", "!=", "proxy"}}, + "filters": [][]string{{"service_type", "!=", "proxy"}}, } if params.Limit > 0 { sdkParams["limit"] = params.Limit @@ -121,19 +131,26 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error return } - // Currently, only "disk" types are supported. Stop if any other service types are found. + var keepServers []ServerAddress for _, server := range sdkResponse.KeepServers { - if server.ServiceType != "disk" { - return results, fmt.Errorf("Unsupported service type %q found for: %v", server.ServiceType, server) + if server.ServiceType == serviceType { + keepServers = append(keepServers, server) + } else { + log.Printf("Skipping keep_service %q because its service_type %q does not match -service-type=%q", server, server.ServiceType, serviceType) } } + if len(keepServers) == 0 { + return results, fmt.Errorf("Found no keepservices with the service type %v", serviceType) + } + if params.Logger != nil { params.Logger.Update(func(p map[string]interface{}, e 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 + keepInfo["indexable_keep_servers"] = keepServers }) } @@ -143,7 +160,7 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error return results, fmt.Errorf("Did not receive all available keep servers: %+v", sdkResponse) } - results.KeepServerIndexToAddress = sdkResponse.KeepServers + results.KeepServerIndexToAddress = keepServers results.KeepServerAddressToIndex = make(map[ServerAddress]int) for i, address := range results.KeepServerIndexToAddress { results.KeepServerAddressToIndex[address] = i @@ -153,7 +170,7 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error // Send off all the index requests concurrently responseChan := make(chan ServerResponse) - for _, keepServer := range sdkResponse.KeepServers { + for _, keepServer := range results.KeepServerIndexToAddress { // The above keepsServer variable is reused for each iteration, so // it would be shared across all goroutines. This would result in // us querying one server n times instead of n different servers @@ -171,7 +188,7 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error results.BlockToServers = make(map[blockdigest.DigestWithSize][]BlockServerInfo) // Read all the responses - for i := range sdkResponse.KeepServers { + for i := range results.KeepServerIndexToAddress { _ = i // Here to prevent go from complaining. response := <-responseChan @@ -198,7 +215,7 @@ func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error // GetServerContents of the keep server func GetServerContents(arvLogger *logger.Logger, keepServer ServerAddress, - arv arvadosclient.ArvadosClient) (response ServerResponse) { + arv *arvadosclient.ArvadosClient) (response ServerResponse) { err := GetServerStatus(arvLogger, keepServer, arv) if err != nil { @@ -230,7 +247,7 @@ func GetServerContents(arvLogger *logger.Logger, // GetServerStatus get keep server status by invoking /status.json func GetServerStatus(arvLogger *logger.Logger, keepServer ServerAddress, - arv arvadosclient.ArvadosClient) error { + arv *arvadosclient.ArvadosClient) error { url := fmt.Sprintf("http://%s:%d/status.json", keepServer.Host, keepServer.Port) @@ -281,7 +298,7 @@ func GetServerStatus(arvLogger *logger.Logger, // CreateIndexRequest to the keep server func CreateIndexRequest(arvLogger *logger.Logger, keepServer ServerAddress, - arv arvadosclient.ArvadosClient) (req *http.Request, err error) { + arv *arvadosclient.ArvadosClient) (req *http.Request, err error) { url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port) log.Println("About to fetch keep server contents from " + url) @@ -413,13 +430,23 @@ func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err err return } - blockInfo.Mtime, err = strconv.ParseInt(tokens[1], 10, 64) + var ns int64 + ns, err = strconv.ParseInt(tokens[1], 10, 64) if err != nil { return } - blockInfo.Digest = - blockdigest.DigestWithSize{Digest: locator.Digest, - Size: uint32(locator.Size)} + if ns < 1e12 { + // An old version of keepstore is giving us timestamps + // in seconds instead of nanoseconds. (This threshold + // correctly handles all times between 1970-01-02 and + // 33658-09-27.) + ns = ns * 1e9 + } + blockInfo.Mtime = ns + blockInfo.Digest = blockdigest.DigestWithSize{ + Digest: locator.Digest, + Size: uint32(locator.Size), + } return } @@ -456,57 +483,61 @@ func SendTrashLists(arvLogger *logger.Logger, kc *keepclient.KeepClient, spl map client := kc.Client for url, v := range spl { + if arvLogger != nil { + // We need a local variable because Update doesn't call our mutator func until later, + // when our list variable might have been reused by the next loop iteration. + url := url + trashLen := len(v) + arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { + trashListInfo := logger.GetOrCreateMap(p, "trash_list_len") + trashListInfo[url] = trashLen + }) + } + if dryRun { - if arvLogger != nil { - for url, v := range spl { - arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) { - trashListInfo := logger.GetOrCreateMap(p, "trash_list") - trashListInfo["started_at"] = time.Now() - trashListInfo[url] = v - }) - } - } - } else { - count++ - 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) + log.Printf("dry run, not sending trash list to service %s with %d blocks", url, len(v)) + continue } + + count++ + 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++ {