X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/bbf13d5bc532c5222a933317461f187a0e118c9a..e59c1d365d9b6e1eff9b5cb030a8b1a3aaf14353:/sdk/go/keepclient/keepclient.go diff --git a/sdk/go/keepclient/keepclient.go b/sdk/go/keepclient/keepclient.go index f15a6b2a2c..029c6ee7f3 100644 --- a/sdk/go/keepclient/keepclient.go +++ b/sdk/go/keepclient/keepclient.go @@ -4,24 +4,37 @@ package keepclient import ( "bytes" "crypto/md5" - "crypto/tls" "errors" "fmt" - "git.curoverse.com/arvados.git/sdk/go/arvadosclient" - "git.curoverse.com/arvados.git/sdk/go/streamer" "io" "io/ioutil" - "log" + "net" "net/http" "regexp" "strconv" "strings" "sync" + "time" + + "git.curoverse.com/arvados.git/sdk/go/arvadosclient" + "git.curoverse.com/arvados.git/sdk/go/streamer" ) // A Keep "block" is 64MB. const BLOCKSIZE = 64 * 1024 * 1024 +var ( + DefaultRequestTimeout = 20 * time.Second + DefaultConnectTimeout = 2 * time.Second + DefaultTLSHandshakeTimeout = 4 * time.Second + DefaultKeepAlive = 180 * time.Second + + DefaultProxyRequestTimeout = 300 * time.Second + DefaultProxyConnectTimeout = 30 * time.Second + DefaultProxyTLSHandshakeTimeout = 10 * time.Second + DefaultProxyKeepAlive = 120 * time.Second +) + // Error interface with an error and boolean indicating whether the error is temporary type Error interface { error @@ -49,8 +62,11 @@ type ErrNotFound struct { multipleResponseError } -var InsufficientReplicasError = errors.New("Could not write sufficient replicas") -var OversizeBlockError = errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")") +type InsufficientReplicasError error + +type OversizeBlockError error + +var ErrOversizeBlock = OversizeBlockError(errors.New("Exceeded maximum block size (" + strconv.Itoa(BLOCKSIZE) + ")")) var MissingArvadosApiHost = errors.New("Missing required environment variable ARVADOS_API_HOST") var MissingArvadosApiToken = errors.New("Missing required environment variable ARVADOS_API_TOKEN") var InvalidLocatorError = errors.New("Invalid locator") @@ -64,32 +80,42 @@ var ErrIncompleteIndex = errors.New("Got incomplete index") const X_Keep_Desired_Replicas = "X-Keep-Desired-Replicas" const X_Keep_Replicas_Stored = "X-Keep-Replicas-Stored" +type HTTPClient interface { + Do(*http.Request) (*http.Response, error) +} + // Information about Arvados and Keep servers. type KeepClient struct { Arvados *arvadosclient.ArvadosClient Want_replicas int - localRoots *map[string]string - writableLocalRoots *map[string]string - gatewayRoots *map[string]string + localRoots map[string]string + writableLocalRoots map[string]string + gatewayRoots map[string]string lock sync.RWMutex - Client *http.Client + HTTPClient HTTPClient Retries int + BlockCache *BlockCache // set to 1 if all writable services are of disk type, otherwise 0 replicasPerService int // Any non-disk typed services found in the list of keepservers? foundNonDiskSvc bool + + // Disable automatic discovery of keep services + disableDiscovery bool } -// MakeKeepClient creates a new KeepClient by contacting the API server to discover Keep servers. +// MakeKeepClient creates a new KeepClient, calls +// DiscoverKeepServices(), and returns when the client is ready to +// use. func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) { kc := New(arv) - return kc, kc.DiscoverKeepServers() + return kc, kc.discoverServices() } -// New func creates a new KeepClient struct. -// This func does not discover keep servers. It is the caller's responsibility. +// New creates a new KeepClient. Service discovery will occur on the +// next read/write operation. func New(arv *arvadosclient.ArvadosClient) *KeepClient { defaultReplicationLevel := 2 value, err := arv.Discovery("defaultCollectionReplication") @@ -99,15 +125,11 @@ func New(arv *arvadosclient.ArvadosClient) *KeepClient { defaultReplicationLevel = int(v) } } - - kc := &KeepClient{ + return &KeepClient{ Arvados: arv, Want_replicas: defaultReplicationLevel, - Client: &http.Client{Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: arv.ApiInsecure}}}, - Retries: 2, + Retries: 2, } - return kc } // Put a block given the block hash, a reader, and the number of bytes @@ -116,14 +138,14 @@ func New(arv *arvadosclient.ArvadosClient) *KeepClient { // Returns the locator for the written block, the number of replicas // written, and an error. // -// Returns an InsufficientReplicas error if 0 <= replicas < +// Returns an InsufficientReplicasError if 0 <= replicas < // kc.Wants_replicas. func (kc *KeepClient) PutHR(hash string, r io.Reader, dataBytes int64) (string, int, error) { // Buffer for reads from 'r' var bufsize int if dataBytes > 0 { if dataBytes > BLOCKSIZE { - return "", 0, OversizeBlockError + return "", 0, ErrOversizeBlock } bufsize = int(dataBytes) } else { @@ -169,6 +191,10 @@ func (kc *KeepClient) PutR(r io.Reader) (locator string, replicas int, err error } func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, int64, string, error) { + if strings.HasPrefix(locator, "d41d8cd98f00b204e9800998ecf8427e+0") { + return ioutil.NopCloser(bytes.NewReader(nil)), 0, "", nil + } + var errs []string tries_remaining := 1 + kc.Retries @@ -193,7 +219,7 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i continue } req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) - resp, err := kc.Client.Do(req) + resp, err := kc.httpClient().Do(req) if err != nil { // Probably a network error, may be transient, // can try again. @@ -233,7 +259,7 @@ func (kc *KeepClient) getOrHead(method string, locator string) (io.ReadCloser, i } serversToTry = retryList } - log.Printf("DEBUG: %s %s failed: %v", method, locator, errs) + DebugPrintf("DEBUG: %s %s failed: %v", method, locator, errs) var err error if count404 == numServers { @@ -294,7 +320,7 @@ func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error } req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", kc.Arvados.ApiToken)) - resp, err := kc.Client.Do(req) + resp, err := kc.httpClient().Do(req) if err != nil { return nil, err } @@ -325,55 +351,47 @@ func (kc *KeepClient) GetIndex(keepServiceUUID, prefix string) (io.Reader, error // LocalRoots() returns the map of local (i.e., disk and proxy) Keep // services: uuid -> baseURI. func (kc *KeepClient) LocalRoots() map[string]string { + kc.discoverServices() kc.lock.RLock() defer kc.lock.RUnlock() - return *kc.localRoots + return kc.localRoots } // GatewayRoots() returns the map of Keep remote gateway services: // uuid -> baseURI. func (kc *KeepClient) GatewayRoots() map[string]string { + kc.discoverServices() kc.lock.RLock() defer kc.lock.RUnlock() - return *kc.gatewayRoots + return kc.gatewayRoots } // WritableLocalRoots() returns the map of writable local Keep services: // uuid -> baseURI. func (kc *KeepClient) WritableLocalRoots() map[string]string { + kc.discoverServices() kc.lock.RLock() defer kc.lock.RUnlock() - return *kc.writableLocalRoots + return kc.writableLocalRoots } -// SetServiceRoots updates the localRoots and gatewayRoots maps, -// without risk of disrupting operations that are already in progress. +// SetServiceRoots disables service discovery and updates the +// localRoots and gatewayRoots maps, without disrupting operations +// that are already in progress. // -// The KeepClient makes its own copy of the supplied maps, so the -// caller can reuse/modify them after SetServiceRoots returns, but -// they should not be modified by any other goroutine while -// SetServiceRoots is running. -func (kc *KeepClient) SetServiceRoots(newLocals, newWritableLocals map[string]string, newGateways map[string]string) { - locals := make(map[string]string) - for uuid, root := range newLocals { - locals[uuid] = root - } - - writables := make(map[string]string) - for uuid, root := range newWritableLocals { - writables[uuid] = root - } - - gateways := make(map[string]string) - for uuid, root := range newGateways { - gateways[uuid] = root - } +// The supplied maps must not be modified after calling +// SetServiceRoots. +func (kc *KeepClient) SetServiceRoots(locals, writables, gateways map[string]string) { + kc.disableDiscovery = true + kc.setServiceRoots(locals, writables, gateways) +} +func (kc *KeepClient) setServiceRoots(locals, writables, gateways map[string]string) { kc.lock.Lock() defer kc.lock.Unlock() - kc.localRoots = &locals - kc.writableLocalRoots = &writables - kc.gatewayRoots = &gateways + kc.localRoots = locals + kc.writableLocalRoots = writables + kc.gatewayRoots = gateways } // getSortedRoots returns a list of base URIs of Keep services, in the @@ -404,6 +422,88 @@ func (kc *KeepClient) getSortedRoots(locator string) []string { return found } +func (kc *KeepClient) cache() *BlockCache { + if kc.BlockCache != nil { + return kc.BlockCache + } else { + return DefaultBlockCache + } +} + +var ( + // There are four global http.Client objects for the four + // possible permutations of TLS behavior (verify/skip-verify) + // and timeout settings (proxy/non-proxy). + defaultClient = map[bool]map[bool]HTTPClient{ + // defaultClient[false] is used for verified TLS reqs + false: {}, + // defaultClient[true] is used for unverified + // (insecure) TLS reqs + true: {}, + } + defaultClientMtx sync.Mutex +) + +// httpClient returns the HTTPClient field if it's not nil, otherwise +// whichever of the four global http.Client objects is suitable for +// the current environment (i.e., TLS verification on/off, keep +// services are/aren't proxies). +func (kc *KeepClient) httpClient() HTTPClient { + if kc.HTTPClient != nil { + return kc.HTTPClient + } + defaultClientMtx.Lock() + defer defaultClientMtx.Unlock() + if c, ok := defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc]; ok { + return c + } + + var requestTimeout, connectTimeout, keepAlive, tlsTimeout time.Duration + if kc.foundNonDiskSvc { + // Use longer timeouts when connecting to a proxy, + // because this usually means the intervening network + // is slower. + requestTimeout = DefaultProxyRequestTimeout + connectTimeout = DefaultProxyConnectTimeout + tlsTimeout = DefaultProxyTLSHandshakeTimeout + keepAlive = DefaultProxyKeepAlive + } else { + requestTimeout = DefaultRequestTimeout + connectTimeout = DefaultConnectTimeout + tlsTimeout = DefaultTLSHandshakeTimeout + keepAlive = DefaultKeepAlive + } + + transport, ok := http.DefaultTransport.(*http.Transport) + if ok { + copy := *transport + transport = © + } else { + // Evidently the application has replaced + // http.DefaultTransport with a different type, so we + // need to build our own from scratch using the Go 1.8 + // defaults. + transport = &http.Transport{ + MaxIdleConns: 100, + IdleConnTimeout: 90 * time.Second, + ExpectContinueTimeout: time.Second, + } + } + transport.DialContext = (&net.Dialer{ + Timeout: connectTimeout, + KeepAlive: keepAlive, + DualStack: true, + }).DialContext + transport.TLSHandshakeTimeout = tlsTimeout + transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure) + c := &http.Client{ + Timeout: requestTimeout, + Transport: transport, + } + defaultClient[kc.Arvados.ApiInsecure][kc.foundNonDiskSvc] = c + return c +} + type Locator struct { Hash string Size int // -1 if data size is not known