X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/49030c3161523bc427e0aaa9d3edb517fc4c296c..20c5b73d598463a7dfb4ce711993480d56e23838:/sdk/go/keepclient/support.go diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go index c03578c5d3..63800b1e08 100644 --- a/sdk/go/keepclient/support.go +++ b/sdk/go/keepclient/support.go @@ -1,4 +1,3 @@ -/* Internal methods to support keepclient.go */ package keepclient import ( @@ -9,77 +8,136 @@ import ( "io" "io/ioutil" "log" + "net" "net/http" - "os" "strings" "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))) } -func (this *KeepClient) DiscoverKeepServers() error { - if prx := os.Getenv("ARVADOS_KEEP_PROXY"); prx != "" { - sr := map[string]string{"proxy": prx} - this.SetServiceRoots(sr) - this.Using_proxy = true - if this.Client.Timeout == 0 { - this.Client.Timeout = 10 * time.Minute +// Set timeouts apply when connecting to keepproxy services (assumed to be over +// the Internet). +func (this *KeepClient) setClientSettingsProxy() { + if this.Client.Timeout == 0 { + // Maximum time to wait for a complete response + this.Client.Timeout = 300 * time.Second + + // TCP and TLS connection settings + this.Client.Transport = &http.Transport{ + Dial: (&net.Dialer{ + // The maximum time to wait to set up + // the initial TCP connection. + Timeout: 30 * time.Second, + + // The TCP keep alive heartbeat + // interval. + KeepAlive: 120 * time.Second, + }).Dial, + + TLSHandshakeTimeout: 10 * time.Second, } - return nil } +} +// Set timeouts apply when connecting to keepstore services directly (assumed +// to be on the local network). +func (this *KeepClient) setClientSettingsDisk() { if this.Client.Timeout == 0 { - this.Client.Timeout = 15 * time.Second + // Maximum time to wait for a complete response + this.Client.Timeout = 20 * time.Second + + // TCP and TLS connection timeouts + this.Client.Transport = &http.Transport{ + Dial: (&net.Dialer{ + // The maximum time to wait to set up + // the initial TCP connection. + Timeout: 2 * time.Second, + + // The TCP keep alive heartbeat + // interval. + KeepAlive: 180 * time.Second, + }).Dial, + + TLSHandshakeTimeout: 4 * time.Second, + } } +} +// DiscoverKeepServers gets list of available keep services from api server +func (this *KeepClient) DiscoverKeepServers() error { type svcList struct { - Items []keepDisk `json:"items"` + Items []keepService `json:"items"` } var m svcList + // Get keep services from api server err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &m) - 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) - - for _, element := range m.Items { - n := "" - - if element.SSL { - n = "s" + 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 m.Items { + scheme := "http" + if service.SSL { + scheme = "https" } - - // Construct server URL - url := fmt.Sprintf("http%s://%s:%d", n, element.Hostname, element.Port) + 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 } - this.SetServiceRoots(service_roots) + if this.Using_proxy { + this.setClientSettingsProxy() + } else { + this.setClientSettingsDisk() + } + this.SetServiceRoots(localRoots, writableLocalRoots, gatewayRoots) return nil } @@ -119,10 +177,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 { @@ -163,7 +218,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 @@ -178,8 +233,14 @@ func (this KeepClient) putReplicas( // 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])