X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0e9443678fc7fd4862b76a10184dc7357a75155d..45f10d80d1b584808a6e375214b5be6bc7d2a730:/sdk/go/keepclient/support.go diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go index c03578c5d3..b467d06b21 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,8 +8,8 @@ import ( "io" "io/ioutil" "log" + "net" "net/http" - "os" "strings" "time" ) @@ -21,27 +20,63 @@ type keepDisk struct { Port int `json:"service_port"` SSL bool `json:"service_ssl_flag"` SvcType string `json:"service_type"` + ReadOnly bool `json:"read_only"` } 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) setClientSettingsStore() { 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, + } } +} +func (this *KeepClient) DiscoverKeepServers() error { type svcList struct { Items []keepDisk `json:"items"` } @@ -56,30 +91,50 @@ func (this *KeepClient) DiscoverKeepServers() error { } 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) + + 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 + + switch service.SvcType { + case "disk": + localRoots[service.Uuid] = url + case "proxy": + localRoots[service.Uuid] = url this.Using_proxy = true } + + if service.ReadOnly == false { + writableLocalRoots[service.Uuid] = url + } + + // 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.setClientSettingsStore() + } + this.SetServiceRoots(localRoots, writableLocalRoots, gatewayRoots) return 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