X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/d1957808f6e3ccece499ac2f4048d4ef850b262c..e3a1bf022cbc081cf7c7ebb441dc1ab1df200f98:/sdk/go/keepclient/support.go diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go index 9db6ebcbaa..f5554618fe 100644 --- a/sdk/go/keepclient/support.go +++ b/sdk/go/keepclient/support.go @@ -1,8 +1,8 @@ -/* Internal methods to support keepclient.go */ package keepclient import ( "crypto/md5" + "encoding/json" "errors" "fmt" "git.curoverse.com/arvados.git/sdk/go/streamer" @@ -15,14 +15,16 @@ import ( "time" ) -type keepDisk struct { +type keepService struct { Uuid string `json:"uuid"` Hostname string `json:"service_host"` Port int `json:"service_port"` SSL bool `json:"service_ssl_flag"` SvcType string `json:"service_type"` + ReadOnly bool `json:"read_only"` } +// Md5String returns md5 hash for the bytes in the given string func Md5String(s string) string { return fmt.Sprintf("%x", md5.Sum([]byte(s))) } @@ -49,12 +51,11 @@ func (this *KeepClient) setClientSettingsProxy() { TLSHandshakeTimeout: 10 * time.Second, } } - } // Set timeouts apply when connecting to keepstore services directly (assumed // to be on the local network). -func (this *KeepClient) setClientSettingsStore() { +func (this *KeepClient) setClientSettingsDisk() { if this.Client.Timeout == 0 { // Maximum time to wait for a complete response this.Client.Timeout = 20 * time.Second @@ -76,51 +77,87 @@ func (this *KeepClient) setClientSettingsStore() { } } -func (this *KeepClient) DiscoverKeepServers() error { - type svcList struct { - Items []keepDisk `json:"items"` - } - var m svcList +type svcList struct { + Items []keepService `json:"items"` +} - err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &m) +// DiscoverKeepServers gets list of available keep services from api server +func (this *KeepClient) DiscoverKeepServers() error { + var list svcList + // Get keep services from api server + err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &list) if err != nil { - if err := this.Arvados.List("keep_disks", nil, &m); err != nil { - return err - } + return err } - listed := make(map[string]bool) - service_roots := make(map[string]string) + return this.loadKeepServers(list) +} - for _, element := range m.Items { - n := "" +// LoadKeepServicesFromJSON gets list of available keep services from given JSON +func (this *KeepClient) LoadKeepServicesFromJSON(services string) error { + var list svcList - if element.SSL { - n = "s" - } + // Load keep services from given json + dec := json.NewDecoder(strings.NewReader(services)) + if err := dec.Decode(&list); err != nil { + return err + } - // Construct server URL - url := fmt.Sprintf("http%s://%s:%d", n, element.Hostname, element.Port) + return this.loadKeepServers(list) +} + +// loadKeepServers +func (this *KeepClient) loadKeepServers(list svcList) error { + listed := make(map[string]bool) + localRoots := make(map[string]string) + gatewayRoots := make(map[string]string) + writableLocalRoots := make(map[string]string) + + // replicasPerService is 1 for disks; unknown or unlimited otherwise + this.replicasPerService = 1 + this.Using_proxy = false + + for _, service := range list.Items { + scheme := "http" + if service.SSL { + scheme = "https" + } + url := fmt.Sprintf("%s://%s:%d", scheme, service.Hostname, service.Port) // Skip duplicates - if !listed[url] { - listed[url] = true - service_roots[element.Uuid] = url + if listed[url] { + continue } - if element.SvcType == "proxy" { + listed[url] = true + + localRoots[service.Uuid] = url + if service.SvcType == "proxy" { this.Using_proxy = true } + + if service.ReadOnly == false { + writableLocalRoots[service.Uuid] = url + if service.SvcType != "disk" { + this.replicasPerService = 0 + } + } + + // Gateway services are only used when specified by + // UUID, so there's nothing to gain by filtering them + // by service type. Including all accessible services + // (gateway and otherwise) merely accommodates more + // service configurations. + gatewayRoots[service.Uuid] = url } if this.Using_proxy { this.setClientSettingsProxy() } else { - this.setClientSettingsStore() + this.setClientSettingsDisk() } - this.SetServiceRoots(service_roots) - + this.SetServiceRoots(localRoots, writableLocalRoots, gatewayRoots) return nil } @@ -132,7 +169,7 @@ type uploadStatus struct { response string } -func (this KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser, +func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser, upload_status chan<- uploadStatus, expectedLength int64, requestId string) { var req *http.Request @@ -160,10 +197,7 @@ func (this KeepClient) uploadToKeepServer(host string, hash string, body io.Read req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.Arvados.ApiToken)) req.Header.Add("Content-Type", "application/octet-stream") - - if this.Using_proxy { - req.Header.Add(X_Keep_Desired_Replicas, fmt.Sprint(this.Want_replicas)) - } + req.Header.Add(X_Keep_Desired_Replicas, fmt.Sprint(this.Want_replicas)) var resp *http.Response if resp, err = this.Client.Do(req); err != nil { @@ -180,7 +214,7 @@ func (this KeepClient) uploadToKeepServer(host string, hash string, body io.Read defer resp.Body.Close() defer io.Copy(ioutil.Discard, resp.Body) - respbody, err2 := ioutil.ReadAll(&io.LimitedReader{resp.Body, 4096}) + respbody, err2 := ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096}) response := strings.TrimSpace(string(respbody)) if err2 != nil && err2 != io.EOF { log.Printf("[%v] Upload %v error: %v response: %v", requestId, url, err2.Error(), response) @@ -194,7 +228,7 @@ func (this KeepClient) uploadToKeepServer(host string, hash string, body io.Read } } -func (this KeepClient) putReplicas( +func (this *KeepClient) putReplicas( hash string, tr *streamer.AsyncStream, expectedLength int64) (locator string, replicas int, err error) { @@ -204,7 +238,7 @@ func (this KeepClient) putReplicas( requestId := fmt.Sprintf("%x", md5.Sum([]byte(locator+time.Now().String())))[0:8] // Calculate the ordering for uploading to servers - sv := NewRootSorter(this.ServiceRoots(), hash).GetSortedRoots() + sv := NewRootSorter(this.WritableLocalRoots(), hash).GetSortedRoots() // The next server to try contacting next_server := 0 @@ -214,13 +248,29 @@ func (this KeepClient) putReplicas( // Used to communicate status from the upload goroutines upload_status := make(chan uploadStatus) - defer close(upload_status) + defer func() { + // Wait for any abandoned uploads (e.g., we started + // two uploads and the first replied with replicas=2) + // to finish before closing the status channel. + go func() { + for active > 0 { + <-upload_status + } + close(upload_status) + }() + }() // Desired number of replicas remaining_replicas := this.Want_replicas + replicasPerThread := this.replicasPerService + if replicasPerThread < 1 { + // unlimited or unknown + replicasPerThread = remaining_replicas + } + for remaining_replicas > 0 { - for active < remaining_replicas { + for active*replicasPerThread < remaining_replicas { // Start some upload requests if next_server < len(sv) { log.Printf("[%v] Begin upload %s to %s", requestId, hash, sv[next_server])