X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/5251261a5fbf8ff746f8a2ff2226d8af6ff6c12d..0d2a561091b777dfb7735d842a88f8529f4f982a:/services/datamanager/keep/keep.go diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go index cbc404755e..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,12 +483,14 @@ func SendTrashLists(arvLogger *logger.Logger, kc *keepclient.KeepClient, spl map client := kc.Client for url, v := range spl { - // 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. 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] = len(v) + trashListInfo[url] = trashLen }) }