// GetKeepServersParams struct
type GetKeepServersParams struct {
- Client arvadosclient.ArvadosClient
+ Client *arvadosclient.ArvadosClient
Logger *logger.Logger
Limit int
}
KeepServers []ServerAddress `json:"items"`
}
-var supportedServiceType string
+var serviceType string
func init() {
- flag.StringVar(&supportedServiceType,
+ flag.StringVar(&serviceType,
"service-type",
"disk",
- "Supported keepservice type. Default is disk.")
+ "Operate only on keep_services with the specified service_type, ignoring all others.")
}
// String
// 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
return
}
- // Ignore any services that are of type other than the "supportedServiceType".
- // If no services of the "supportedServiceType" are found, raise an error.
- var indexableKeepServers []ServerAddress
+ var keepServers []ServerAddress
for _, server := range sdkResponse.KeepServers {
- if server.ServiceType == supportedServiceType {
- indexableKeepServers = append(indexableKeepServers, server)
+ if server.ServiceType == serviceType {
+ keepServers = append(keepServers, server)
} else {
- log.Printf("Ignore unsupported service type: %v", server.ServiceType)
+ log.Printf("Skipping keep_service %q because its service_type %q does not match -service-type=%q", server, server.ServiceType, serviceType)
}
}
- if len(indexableKeepServers) == 0 {
- return results, fmt.Errorf("Found no keepservices with the supported type %v", supportedServiceType)
+ if len(keepServers) == 0 {
+ return results, fmt.Errorf("Found no keepservices with the service type %v", serviceType)
}
if params.Logger != nil {
keepInfo["num_keep_servers_available"] = sdkResponse.ItemsAvailable
keepInfo["num_keep_servers_received"] = len(sdkResponse.KeepServers)
keepInfo["keep_servers"] = sdkResponse.KeepServers
- keepInfo["indexable_keep_servers"] = indexableKeepServers
+ keepInfo["indexable_keep_servers"] = keepServers
})
}
return results, fmt.Errorf("Did not receive all available keep servers: %+v", sdkResponse)
}
- results.KeepServerIndexToAddress = indexableKeepServers
+ results.KeepServerIndexToAddress = keepServers
results.KeepServerAddressToIndex = make(map[ServerAddress]int)
for i, address := range results.KeepServerIndexToAddress {
results.KeepServerAddressToIndex[address] = i
// 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 {
// 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)
// 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)
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
}
type TrashList []TrashRequest
// SendTrashLists to trash queue
-func SendTrashLists(kc *keepclient.KeepClient, spl map[string]TrashList) (errs []error) {
+func SendTrashLists(arvLogger *logger.Logger, kc *keepclient.KeepClient, spl map[string]TrashList, dryRun bool) (errs []error) {
count := 0
barrier := make(chan error)
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 {
+ 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)
barrier <- nil
}
})(url, v)
-
}
for i := 0; i < count; i++ {
projects / arvados.git / blobdiff