X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c4023fb9b27ba31926faf530a4e59b9aee0989b3..f9e3d32c92920a806d50548dbb9b6c0aab7d8c54:/services/datamanager/keep/keep.go

diff --git a/services/datamanager/keep/keep.go b/services/datamanager/keep/keep.go
index 7da74b1680..e7843ea02f 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"
@@ -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
 
@@ -449,13 +466,29 @@ type TrashRequest struct {
 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)
 
@@ -495,7 +528,6 @@ func SendTrashLists(kc *keepclient.KeepClient, spl map[string]TrashList) (errs [
 				barrier <- nil
 			}
 		})(url, v)
-
 	}
 
 	for i := 0; i < count; i++ {