X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/cac9d1111f86a7ff6da2176e3069dec4484154d4..afbea0d985d273232291f03f343baed727393108:/sdk/go/arvados/client.go diff --git a/sdk/go/arvados/client.go b/sdk/go/arvados/client.go index 13bb3bf80d..05176214ae 100644 --- a/sdk/go/arvados/client.go +++ b/sdk/go/arvados/client.go @@ -7,18 +7,23 @@ package arvados import ( "bytes" "context" + "crypto/rand" "crypto/tls" "encoding/json" "errors" "fmt" "io" + "io/fs" "io/ioutil" "log" + "math/big" + "net" "net/http" "net/url" "os" "regexp" "strings" + "sync/atomic" "time" "git.arvados.org/arvados.git/sdk/go/httpserver" @@ -72,6 +77,14 @@ type Client struct { // APIHost and AuthToken were loaded from ARVADOS_* env vars // (used to customize "no host/token" error messages) loadedFromEnv bool + + // Track/limit concurrent outgoing API calls. Note this + // differs from an outgoing connection limit (a feature + // provided by http.Transport) when concurrent calls are + // multiplexed on a single http2 connection. + requestLimiter requestLimiter + + last503 atomic.Value } // InsecureHTTPClient is the default http.Client used by a Client with @@ -93,7 +106,40 @@ func NewClientFromConfig(cluster *Cluster) (*Client, error) { if ctrlURL.Host == "" { return nil, fmt.Errorf("no host in config Services.Controller.ExternalURL: %v", ctrlURL) } + var hc *http.Client + if srvaddr := os.Getenv("ARVADOS_SERVER_ADDRESS"); srvaddr != "" { + // When this client is used to make a request to + // https://{ctrlhost}:port/ (any port), it dials the + // indicated port on ARVADOS_SERVER_ADDRESS instead. + // + // This is invoked by arvados-server boot to ensure + // that server->server traffic (e.g., + // keepproxy->controller) only hits local interfaces, + // even if the Controller.ExternalURL host is a load + // balancer / gateway and not a local interface + // address (e.g., when running on a cloud VM). + // + // This avoids unnecessary delay/cost of routing + // external traffic, and also allows controller to + // recognize other services as internal clients based + // on the connection source address. + divertedHost := (*url.URL)(&cluster.Services.Controller.ExternalURL).Hostname() + var dialer net.Dialer + hc = &http.Client{ + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: cluster.TLS.Insecure}, + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + host, port, err := net.SplitHostPort(addr) + if err == nil && network == "tcp" && host == divertedHost { + addr = net.JoinHostPort(srvaddr, port) + } + return dialer.DialContext(ctx, network, addr) + }, + }, + } + } return &Client{ + Client: hc, Scheme: ctrlURL.Scheme, APIHost: ctrlURL.Host, Insecure: cluster.TLS.Insecure, @@ -102,11 +148,60 @@ func NewClientFromConfig(cluster *Cluster) (*Client, error) { } // NewClientFromEnv creates a new Client that uses the default HTTP -// client with the API endpoint and credentials given by the -// ARVADOS_API_* environment variables. +// client, and loads API endpoint and credentials from ARVADOS_* +// environment variables (if set) and +// $HOME/.config/arvados/settings.conf (if readable). +// +// If a config exists in both locations, the environment variable is +// used. +// +// If there is an error (other than ENOENT) reading settings.conf, +// NewClientFromEnv logs the error to log.Default(), then proceeds as +// if settings.conf did not exist. +// +// Space characters are trimmed when reading the settings file, so +// these are equivalent: +// +// ARVADOS_API_HOST=localhost\n +// ARVADOS_API_HOST=localhost\r\n +// ARVADOS_API_HOST = localhost \n +// \tARVADOS_API_HOST = localhost\n func NewClientFromEnv() *Client { + vars := map[string]string{} + home := os.Getenv("HOME") + conffile := home + "/.config/arvados/settings.conf" + if home == "" { + // no $HOME => just use env vars + } else if settings, err := os.ReadFile(conffile); errors.Is(err, fs.ErrNotExist) { + // no config file => just use env vars + } else if err != nil { + // config file unreadable => log message, then use env vars + log.Printf("continuing without loading %s: %s", conffile, err) + } else { + for _, line := range bytes.Split(settings, []byte{'\n'}) { + kv := bytes.SplitN(line, []byte{'='}, 2) + k := string(bytes.TrimSpace(kv[0])) + if len(kv) != 2 || !strings.HasPrefix(k, "ARVADOS_") { + // Same behavior as python sdk: + // silently skip leading # (comments), + // blank lines, typos, and non-Arvados + // vars. + continue + } + vars[k] = string(bytes.TrimSpace(kv[1])) + } + } + for _, env := range os.Environ() { + if !strings.HasPrefix(env, "ARVADOS_") { + continue + } + kv := strings.SplitN(env, "=", 2) + if len(kv) == 2 { + vars[kv[0]] = kv[1] + } + } var svcs []string - for _, s := range strings.Split(os.Getenv("ARVADOS_KEEP_SERVICES"), " ") { + for _, s := range strings.Split(vars["ARVADOS_KEEP_SERVICES"], " ") { if s == "" { continue } else if u, err := url.Parse(s); err != nil { @@ -118,13 +213,13 @@ func NewClientFromEnv() *Client { } } var insecure bool - if s := strings.ToLower(os.Getenv("ARVADOS_API_HOST_INSECURE")); s == "1" || s == "yes" || s == "true" { + if s := strings.ToLower(vars["ARVADOS_API_HOST_INSECURE"]); s == "1" || s == "yes" || s == "true" { insecure = true } return &Client{ Scheme: "https", - APIHost: os.Getenv("ARVADOS_API_HOST"), - AuthToken: os.Getenv("ARVADOS_API_TOKEN"), + APIHost: vars["ARVADOS_API_HOST"], + AuthToken: vars["ARVADOS_API_TOKEN"], Insecure: insecure, KeepServiceURIs: svcs, Timeout: 5 * time.Minute, @@ -134,10 +229,12 @@ func NewClientFromEnv() *Client { var reqIDGen = httpserver.IDGenerator{Prefix: "req-"} -// Do adds Authorization and X-Request-Id headers and then calls +// Do adds Authorization and X-Request-Id headers, delays in order to +// comply with rate-limiting restrictions, and then calls // (*http.Client)Do(). func (c *Client) Do(req *http.Request) (*http.Response, error) { - if auth, _ := req.Context().Value(contextKeyAuthorization{}).(string); auth != "" { + ctx := req.Context() + if auth, _ := ctx.Value(contextKeyAuthorization{}).(string); auth != "" { req.Header.Add("Authorization", auth) } else if c.AuthToken != "" { req.Header.Add("Authorization", "OAuth2 "+c.AuthToken) @@ -145,7 +242,7 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) { if req.Header.Get("X-Request-Id") == "" { var reqid string - if ctxreqid, _ := req.Context().Value(contextKeyRequestID{}).(string); ctxreqid != "" { + if ctxreqid, _ := ctx.Value(contextKeyRequestID{}).(string); ctxreqid != "" { reqid = ctxreqid } else if c.defaultRequestID != "" { reqid = c.defaultRequestID @@ -160,23 +257,48 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) { } var cancel context.CancelFunc if c.Timeout > 0 { - ctx := req.Context() ctx, cancel = context.WithDeadline(ctx, time.Now().Add(c.Timeout)) req = req.WithContext(ctx) + } else { + cancel = context.CancelFunc(func() {}) } + + c.requestLimiter.Acquire(ctx) + if ctx.Err() != nil { + c.requestLimiter.Release() + return nil, ctx.Err() + } + + // Attach Release() to cancel func, see cancelOnClose below. + cancelOrig := cancel + cancel = func() { + c.requestLimiter.Release() + cancelOrig() + } + resp, err := c.httpClient().Do(req) - if err == nil && cancel != nil { + if c.requestLimiter.Report(resp, err) { + c.last503.Store(time.Now()) + } + if err == nil { // We need to call cancel() eventually, but we can't // use "defer cancel()" because the context has to // stay alive until the caller has finished reading // the response body. resp.Body = cancelOnClose{ReadCloser: resp.Body, cancel: cancel} - } else if cancel != nil { + } else { cancel() } return resp, err } +// Last503 returns the time of the most recent HTTP 503 (Service +// Unavailable) response. Zero time indicates never. +func (c *Client) Last503() time.Time { + t, _ := c.last503.Load().(time.Time) + return t +} + // cancelOnClose calls a provided CancelFunc when its wrapped // ReadCloser's Close() method is called. type cancelOnClose struct { @@ -217,6 +339,8 @@ func (c *Client) DoAndDecode(dst interface{}, req *http.Request) error { return err } switch { + case resp.StatusCode == http.StatusNoContent: + return nil case resp.StatusCode == http.StatusOK && dst == nil: return nil case resp.StatusCode == http.StatusOK: @@ -242,11 +366,11 @@ func (c *Client) DoAndDecode(dst interface{}, req *http.Request) error { // Convert an arbitrary struct to url.Values. For example, // -// Foo{Bar: []int{1,2,3}, Baz: "waz"} +// Foo{Bar: []int{1,2,3}, Baz: "waz"} // // becomes // -// url.Values{`bar`:`{"a":[1,2,3]}`,`Baz`:`waz`} +// url.Values{`bar`:`{"a":[1,2,3]}`,`Baz`:`waz`} // // params itself is returned if it is already an url.Values. func anythingToValues(params interface{}) (url.Values, error) { @@ -513,3 +637,17 @@ func (c *Client) PathForUUID(method, uuid string) (string, error) { } return path, nil } + +var maxUUIDInt = (&big.Int{}).Exp(big.NewInt(36), big.NewInt(15), nil) + +func RandomUUID(clusterID, infix string) string { + n, err := rand.Int(rand.Reader, maxUUIDInt) + if err != nil { + panic(err) + } + nstr := n.Text(36) + for len(nstr) < 15 { + nstr = "0" + nstr + } + return clusterID + "-" + infix + "-" + nstr +}