+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+
/* Provides low-level Get/Put primitives for accessing Arvados Keep blocks. */
package keepclient
// 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
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
HTTPClient HTTPClient
Retries 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, calls
// use.
func MakeKeepClient(arv *arvadosclient.ArvadosClient) (*KeepClient, error) {
kc := New(arv)
- return kc, kc.DiscoverKeepServers()
+ return kc, kc.discoverServices()
}
-// New creates a new KeepClient. The caller must call
-// DiscoverKeepServers() before using the returned client to read or
-// write data.
+// 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")
} else if resp.StatusCode == 404 {
count404++
}
+ } else if resp.ContentLength < 0 {
+ // Missing Content-Length
+ resp.Body.Close()
+ return nil, 0, "", fmt.Errorf("Missing Content-Length of block")
} else {
// Success.
if method == "GET" {
// 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, 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
}
}
+func (kc *KeepClient) ClearBlockCache() {
+ kc.cache().Clear()
+}
+
var (
// There are four global http.Client objects for the four
// possible permutations of TLS behavior (verify/skip-verify)
return c
}
- var requestTimeout, connectTimeout, keepAliveInterval, tlsTimeout time.Duration
+ 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 = 300 * time.Second
- connectTimeout = 30 * time.Second
- tlsTimeout = 10 * time.Second
- keepAliveInterval = 120 * time.Second
+ requestTimeout = DefaultProxyRequestTimeout
+ connectTimeout = DefaultProxyConnectTimeout
+ tlsTimeout = DefaultProxyTLSHandshakeTimeout
+ keepAlive = DefaultProxyKeepAlive
} else {
- requestTimeout = 20 * time.Second
- connectTimeout = 2 * time.Second
- tlsTimeout = 4 * time.Second
- keepAliveInterval = 180 * time.Second
+ requestTimeout = DefaultRequestTimeout
+ connectTimeout = DefaultConnectTimeout
+ tlsTimeout = DefaultTLSHandshakeTimeout
+ keepAlive = DefaultKeepAlive
}
- transport := &http.Transport{
- Dial: (&net.Dialer{
- Timeout: connectTimeout,
- KeepAlive: keepAliveInterval,
- }).Dial,
- TLSClientConfig: arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure),
- TLSHandshakeTimeout: tlsTimeout,
- }
- go func() {
- for range time.NewTicker(10 * time.Minute).C {
- transport.CloseIdleConnections()
+
+ 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,
projects / arvados.git / blobdiff