Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-20 07:19:17 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

lib/promscrape/discoveryutils: switch to native http client from fasthttp (#3568)

Browse Source
This commit is contained in:
Zakhar Bessarab 2023-01-06 07:34:47 +04:00 committed by GitHub
parent 5bdd880142
commit bced9fb978
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 209 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
lib/promscrape/client.go
View File

@ -14,7 +14,6 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
"github.com/VictoriaMetrics/fasthttp"
"github.com/VictoriaMetrics/metrics"
@ -349,7 +348,32 @@ var (
)
func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response, deadline time.Time) error {
return discoveryutils.DoRequestWithPossibleRetry(hc, req, resp, deadline, scrapeRequests, scrapeRetries)
sleepTime := time.Second
scrapeRequests.Inc()
for {
// Use DoDeadline instead of Do even if hc.ReadTimeout is already set in order to guarantee the given deadline
// across multiple retries.
err := hc.DoDeadline(req, resp, deadline)
if err == nil {
statusCode := resp.StatusCode()
if statusCode != fasthttp.StatusTooManyRequests {
return nil
}
} else if err != fasthttp.ErrConnectionClosed && !strings.Contains(err.Error(), "broken pipe") {
return err
}
// Retry request after exponentially increased sleep.
maxSleepTime := time.Until(deadline)
if sleepTime > maxSleepTime {
return fmt.Errorf("the server closes all the connection attempts: %w", err)
}
sleepTime += sleepTime
if sleepTime > maxSleepTime {
sleepTime = maxSleepTime
}
time.Sleep(sleepTime)
scrapeRetries.Inc()
}
}
type streamReader struct {

13
lib/promscrape/discovery/azure/api.go
View File

@ -3,6 +3,8 @@ package azure
import (
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"os"
"strconv"
@ -13,7 +15,6 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
"github.com/VictoriaMetrics/fasthttp"
)
var configMap = discoveryutils.NewConfigMap()
@ -157,7 +158,7 @@ func readCloudEndpointsFromFile(filePath string) (*cloudEnvironmentEndpoints, er
func getRefreshTokenFunc(sdc *SDConfig, ac, proxyAC *promauth.Config, env *cloudEnvironmentEndpoints) (refreshTokenFunc, error) {
var tokenEndpoint, tokenAPIPath string
var modifyRequest func(request *fasthttp.Request)
var modifyRequest func(request *http.Request)
authenticationMethod := sdc.AuthenticationMethod
if authenticationMethod == "" {
authenticationMethod = "OAuth"
@ -182,9 +183,9 @@ func getRefreshTokenFunc(sdc *SDConfig, ac, proxyAC *promauth.Config, env *cloud
authParams := q.Encode()
tokenAPIPath = "/" + sdc.TenantID + "/oauth2/token"
tokenEndpoint = env.ActiveDirectoryEndpoint
modifyRequest = func(request *fasthttp.Request) {
request.SetBodyString(authParams)
request.Header.SetMethod("POST")
modifyRequest = func(request *http.Request) {
request.Body = io.NopCloser(strings.NewReader(authParams))
request.Method = http.MethodPost
}
case "managedidentity":
endpoint := "http://169.254.169.254/metadata/identity/oauth2/token"
@ -210,7 +211,7 @@ func getRefreshTokenFunc(sdc *SDConfig, ac, proxyAC *promauth.Config, env *cloud
endpointURL.RawQuery = q.Encode()
tokenAPIPath = endpointURL.RequestURI()
tokenEndpoint = endpointURL.Scheme + "://" + endpointURL.Host
modifyRequest = func(request *fasthttp.Request) {
modifyRequest = func(request *http.Request) {
if msiSecret != "" {
request.Header.Set("secret", msiSecret)
} else {

4
lib/promscrape/discovery/azure/machine.go
View File

@ -3,11 +3,11 @@ package azure
import (
"encoding/json"
"fmt"
"net/http"
"net/url"
"sync"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
"github.com/VictoriaMetrics/fasthttp"
)
// virtualMachine represents an Azure virtual machine (which can also be created by a VMSS)
@ -64,7 +64,7 @@ type listAPIResponse struct {
func visitAllAPIObjects(ac *apiConfig, apiURL string, cb func(data json.RawMessage) error) error {
nextLinkURI := apiURL
for {
resp, err := ac.c.GetAPIResponseWithReqParams(nextLinkURI, func(request *fasthttp.Request) {
resp, err := ac.c.GetAPIResponseWithReqParams(nextLinkURI, func(request *http.Request) {
request.Header.Set("Authorization", "Bearer "+ac.mustGetAuthToken())
})
if err != nil {

5
lib/promscrape/discovery/azure/nic.go
View File

@ -3,8 +3,7 @@ package azure
import (
"encoding/json"
"fmt"
"github.com/VictoriaMetrics/fasthttp"
"net/http"
)
// networkInterface a network interface in a resource group.
@ -66,7 +65,7 @@ func getNIC(ac *apiConfig, id string, isScaleSetVM bool) (*networkInterface, err
apiQueryParams = "api-version=2021-03-01&$expand=ipConfigurations/publicIPAddress"
}
apiURL := id + "?" + apiQueryParams
resp, err := ac.c.GetAPIResponseWithReqParams(apiURL, func(request *fasthttp.Request) {
resp, err := ac.c.GetAPIResponseWithReqParams(apiURL, func(request *http.Request) {
request.Header.Set("Authorization", "Bearer "+ac.mustGetAuthToken())
})
if err != nil {

8
lib/promscrape/discovery/consul/api.go
View File

@ -3,6 +3,7 @@ package consul
import (
"flag"
"fmt"
"net/http"
"os"
"strconv"
"strings"
@ -11,7 +12,6 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
"github.com/VictoriaMetrics/fasthttp"
)
var waitTime = flag.Duration("promscrape.consul.waitTime", 0, "Wait time used by Consul service discovery. Default value is used if not set")
@ -157,13 +157,13 @@ func maxWaitTime() time.Duration {
func getBlockingAPIResponse(client *discoveryutils.Client, path string, index int64) ([]byte, int64, error) {
path += "&index=" + strconv.FormatInt(index, 10)
path += "&wait=" + fmt.Sprintf("%ds", int(maxWaitTime().Seconds()))
getMeta := func(resp *fasthttp.Response) {
ind := resp.Header.Peek("X-Consul-Index")
getMeta := func(resp *http.Response) {
ind := resp.Header.Get("X-Consul-Index")
if len(ind) == 0 {
logger.Errorf("cannot find X-Consul-Index header in response from %q", path)
return
}
newIndex, err := strconv.ParseInt(string(ind), 10, 64)
newIndex, err := strconv.ParseInt(ind, 10, 64)
if err != nil {
logger.Errorf("cannot parse X-Consul-Index header value in response from %q: %s", path, err)
return

9
lib/promscrape/discovery/consul/watch.go
View File

@ -1,7 +1,9 @@
package consul
import (
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"net/url"
@ -79,9 +81,7 @@ func newConsulWatcher(client *discoveryutils.Client, sdc *SDConfig, datacenter,
func (cw *consulWatcher) mustStop() {
close(cw.stopCh)
// Do not wait for the watcher to stop, since it may take
// up to discoveryutils.BlockingClientReadTimeout to complete.
// TODO: add ability to cancel blocking requests.
cw.client.Stop()
}
func (cw *consulWatcher) updateServices(serviceNames []string) {
@ -225,6 +225,9 @@ func (sw *serviceWatcher) watchForServiceNodesUpdates(cw *consulWatcher, initWG
f := func() {
data, newIndex, err := getBlockingAPIResponse(cw.client, path, index)
if err != nil {
if errors.Is(err, context.Canceled) {
return
}
logger.Errorf("cannot obtain Consul serviceNodes for serviceName=%q from %q: %s", sw.serviceName, clientAddr, err)
return
}

4
lib/promscrape/discovery/http/api.go
View File

@ -3,12 +3,12 @@ package http
import (
"encoding/json"
"fmt"
"net/http"
"net/url"
"strconv"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
"github.com/VictoriaMetrics/fasthttp"
"github.com/VictoriaMetrics/metrics"
)
@ -66,7 +66,7 @@ func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
}
func getHTTPTargets(cfg *apiConfig) ([]httpGroupTarget, error) {
data, err := cfg.client.GetAPIResponseWithReqParams(cfg.path, func(request *fasthttp.Request) {
data, err := cfg.client.GetAPIResponseWithReqParams(cfg.path, func(request *http.Request) {
request.Header.Set("X-Prometheus-Refresh-Interval-Seconds", strconv.FormatFloat(SDCheckInterval.Seconds(), 'f', 0, 64))
request.Header.Set("Accept", "application/json")
})

282
lib/promscrape/discoveryutils/client.go
View File

@ -1,11 +1,15 @@
package discoveryutils
import (
"compress/gzip"
"context"
"crypto/tls"
"flag"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"sync"
"time"
@ -13,7 +17,6 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
"github.com/VictoriaMetrics/fasthttp"
"github.com/VictoriaMetrics/metrics"
)
@ -27,6 +30,26 @@ var defaultClient = &http.Client{
Timeout: 30 * time.Second,
}
var (
concurrencyLimitCh chan struct{}
concurrencyLimitChOnce sync.Once
)
const (
// BlockingClientReadTimeout is the maximum duration for waiting the response from GetBlockingAPI*
BlockingClientReadTimeout = 10 * time.Minute
// DefaultClientReadTimeout is the maximum duration for waiting the response from GetAPI*
DefaultClientReadTimeout = time.Minute
// DefaultClientWriteTimeout is the maximum duration for waiting the request to be sent to GetAPI* and GetBlockingAPI*
DefaultClientWriteTimeout = 10 * time.Second
)
func concurrencyLimitChInit() {
concurrencyLimitCh = make(chan struct{}, *maxConcurrency)
}
// GetHTTPClient returns default client for http API requests.
func GetHTTPClient() *http.Client {
return defaultClient
@ -34,18 +57,28 @@ func GetHTTPClient() *http.Client {
// Client is http client, which talks to the given apiServer.
type Client struct {
// hc is used for short requests.
hc *fasthttp.HostClient
// client is used for short requests.
client *HTTPClient
// blockingClient is used for long-polling requests.
blockingClient *fasthttp.HostClient
blockingClient *HTTPClient
apiServer string
hostPort string
setFasthttpHeaders func(req *fasthttp.Request)
setFasthttpProxyHeaders func(req *fasthttp.Request)
sendFullURL bool
dialAddr string
setHTTPHeaders func(req *http.Request)
setHTTPProxyHeaders func(req *http.Request)
clientCtx context.Context
clientCancel context.CancelFunc
}
// HTTPClient is a wrapper around http.Client with timeouts.
type HTTPClient struct {
client *http.Client
ReadTimeout time.Duration
WriteTimeout time.Duration
}
func addMissingPort(addr string, isTLS bool) string {
@ -60,45 +93,30 @@ func addMissingPort(addr string, isTLS bool) string {
// NewClient returns new Client for the given args.
func NewClient(apiServer string, ac *promauth.Config, proxyURL *proxy.URL, proxyAC *promauth.Config) (*Client, error) {
var u fasthttp.URI
u.Update(apiServer)
u, err := url.Parse(apiServer)
if err != nil {
return nil, fmt.Errorf("cannot parse provided url %q: %w", apiServer, err)
}
// special case for unix socket connection
var dialFunc fasthttp.DialFunc
if string(u.Scheme()) == "unix" {
dialAddr := string(u.Path())
var dialFunc func(addr string) (net.Conn, error)
if string(u.Scheme) == "unix" {
dialAddr := u.Path
apiServer = "http://"
dialFunc = func(_ string) (net.Conn, error) {
return net.Dial("unix", dialAddr)
}
}
hostPort := string(u.Host())
dialAddr := hostPort
isTLS := string(u.Scheme()) == "https"
dialAddr := u.Host
isTLS := string(u.Scheme) == "https"
var tlsCfg *tls.Config
if isTLS {
tlsCfg = ac.NewTLSConfig()
}
sendFullURL := !isTLS && proxyURL.IsHTTPOrHTTPS()
setFasthttpProxyHeaders := func(req *fasthttp.Request) {}
if sendFullURL {
// Send full urls in requests to a proxy host for non-TLS apiServer
// like net/http package from Go does.
// See https://en.wikipedia.org/wiki/Proxy_server#Web_proxy_servers
pu := proxyURL.GetURL()
dialAddr = pu.Host
isTLS = pu.Scheme == "https"
if isTLS {
tlsCfg = proxyAC.NewTLSConfig()
}
proxyURLOrig := proxyURL
setFasthttpProxyHeaders = func(req *fasthttp.Request) {
proxyURLOrig.SetFasthttpHeaders(proxyAC, req)
}
proxyURL = &proxy.URL{}
}
hostPort = addMissingPort(hostPort, isTLS)
setHTTPProxyHeaders := func(req *http.Request) {}
dialAddr = addMissingPort(dialAddr, isTLS)
if dialFunc == nil {
var err error
@ -106,64 +124,66 @@ func NewClient(apiServer string, ac *promauth.Config, proxyURL *proxy.URL, proxy
if err != nil {
return nil, err
}
if proxyAC != nil {
setHTTPProxyHeaders = func(req *http.Request) {
proxyURL.SetHeaders(proxyAC, req)
}
}
}
hc := &fasthttp.HostClient{
Addr: dialAddr,
Name: "vm_promscrape/discovery",
IsTLS: isTLS,
TLSConfig: tlsCfg,
ReadTimeout: time.Minute,
WriteTimeout: 10 * time.Second,
MaxResponseBodySize: 300 * 1024 * 1024,
MaxConns: 2 * *maxConcurrency,
Dial: dialFunc,
hcTransport := &http.Transport{
TLSClientConfig: tlsCfg,
MaxConnsPerHost: 2 * *maxConcurrency,
ResponseHeaderTimeout: *maxWaitTime,
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
return dialFunc(dialAddr)
},
}
blockingClient := &fasthttp.HostClient{
Addr: dialAddr,
Name: "vm_promscrape/discovery",
IsTLS: isTLS,
TLSConfig: tlsCfg,
ReadTimeout: BlockingClientReadTimeout,
WriteTimeout: 10 * time.Second,
MaxResponseBodySize: 300 * 1024 * 1024,
MaxConns: 64 * 1024,
Dial: dialFunc,
hc := &http.Client{
Timeout: DefaultClientReadTimeout,
Transport: hcTransport,
}
setFasthttpHeaders := func(req *fasthttp.Request) {}
blockingTransport := &http.Transport{
TLSClientConfig: tlsCfg,
MaxConnsPerHost: 64 * 1024,
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
return dialFunc(dialAddr)
},
}
blockingClient := &http.Client{
Timeout: BlockingClientReadTimeout,
Transport: blockingTransport,
}
setHTTPHeaders := func(req *http.Request) {}
if ac != nil {
setFasthttpHeaders = func(req *fasthttp.Request) { ac.SetFasthttpHeaders(req, true) }
setHTTPHeaders = func(req *http.Request) { ac.SetHeaders(req, true) }
}
ctx, cancel := context.WithCancel(context.Background())
return &Client{
hc: hc,
blockingClient: blockingClient,
apiServer: apiServer,
hostPort: hostPort,
setFasthttpHeaders: setFasthttpHeaders,
setFasthttpProxyHeaders: setFasthttpProxyHeaders,
sendFullURL: sendFullURL,
client: &HTTPClient{client: hc, ReadTimeout: DefaultClientReadTimeout, WriteTimeout: DefaultClientWriteTimeout},
blockingClient: &HTTPClient{client: blockingClient, ReadTimeout: BlockingClientReadTimeout, WriteTimeout: DefaultClientWriteTimeout},
apiServer: apiServer,
dialAddr: dialAddr,
setHTTPHeaders: setHTTPHeaders,
setHTTPProxyHeaders: setHTTPProxyHeaders,
clientCtx: ctx,
clientCancel: cancel,
}, nil
}
// BlockingClientReadTimeout is the maximum duration for waiting the response from GetBlockingAPI*
const BlockingClientReadTimeout = 10 * time.Minute
var (
concurrencyLimitCh chan struct{}
concurrencyLimitChOnce sync.Once
)
func concurrencyLimitChInit() {
concurrencyLimitCh = make(chan struct{}, *maxConcurrency)
}
// Addr returns the address the client connects to.
func (c *Client) Addr() string {
return c.hc.Addr
return c.dialAddr
}
// GetAPIResponseWithReqParams returns response for given absolute path with optional callback for request.
// modifyRequestParams should never reference data from request.
func (c *Client) GetAPIResponseWithReqParams(path string, modifyRequestParams func(request *fasthttp.Request)) ([]byte, error) {
func (c *Client) GetAPIResponseWithReqParams(path string, modifyRequestParams func(request *http.Request)) ([]byte, error) {
return c.getAPIResponse(path, modifyRequestParams)
}
@ -173,7 +193,7 @@ func (c *Client) GetAPIResponse(path string) ([]byte, error) {
}
// GetAPIResponse returns response for the given absolute path with optional callback for request.
func (c *Client) getAPIResponse(path string, modifyRequest func(request *fasthttp.Request)) ([]byte, error) {
func (c *Client) getAPIResponse(path string, modifyRequest func(request *http.Request)) ([]byte, error) {
// Limit the number of concurrent API requests.
concurrencyLimitChOnce.Do(concurrencyLimitChInit)
t := timerpool.Get(*maxWaitTime)
@ -186,56 +206,66 @@ func (c *Client) getAPIResponse(path string, modifyRequest func(request *fasthtt
c.apiServer, *maxWaitTime, *maxConcurrency)
}
defer func() { <-concurrencyLimitCh }()
return c.getAPIResponseWithParamsAndClient(c.hc, path, modifyRequest, nil)
return c.getAPIResponseWithParamsAndClient(c.client, path, modifyRequest, nil)
}
// GetBlockingAPIResponse returns response for given absolute path with blocking client and optional callback for api response,
// inspectResponse - should never reference data from response.
func (c *Client) GetBlockingAPIResponse(path string, inspectResponse func(resp *fasthttp.Response)) ([]byte, error) {
func (c *Client) GetBlockingAPIResponse(path string, inspectResponse func(resp *http.Response)) ([]byte, error) {
return c.getAPIResponseWithParamsAndClient(c.blockingClient, path, nil, inspectResponse)
}
// getAPIResponseWithParamsAndClient returns response for the given absolute path with optional callback for request and for response.
func (c *Client) getAPIResponseWithParamsAndClient(client *fasthttp.HostClient, path string, modifyRequest func(req *fasthttp.Request), inspectResponse func(resp *fasthttp.Response)) ([]byte, error) {
func (c *Client) getAPIResponseWithParamsAndClient(client *HTTPClient, path string, modifyRequest func(req *http.Request), inspectResponse func(resp *http.Response)) ([]byte, error) {
requestURL := c.apiServer + path
var u fasthttp.URI
u.Update(requestURL)
var req fasthttp.Request
if c.sendFullURL {
req.SetRequestURIBytes(u.FullURI())
} else {
req.SetRequestURIBytes(u.RequestURI())
u, err := url.Parse(requestURL)
if err != nil {
return nil, fmt.Errorf("cannot parse %q: %w", requestURL, err)
}
req.Header.SetHost(c.hostPort)
req.Header.Set("Accept-Encoding", "gzip")
c.setFasthttpHeaders(&req)
c.setFasthttpProxyHeaders(&req)
if modifyRequest != nil {
modifyRequest(&req)
u.Host = c.dialAddr
deadline := time.Now().Add(client.WriteTimeout)
ctx, cancel := context.WithDeadline(c.clientCtx, deadline)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "GET", u.String(), nil)
if err != nil {
return nil, fmt.Errorf("cannot create request for %q: %w", requestURL, err)
}
var resp fasthttp.Response
deadline := time.Now().Add(client.ReadTimeout)
if err := doRequestWithPossibleRetry(client, &req, &resp, deadline); err != nil {
req.Header.Set("Host", c.dialAddr)
req.Header.Set("Accept-Encoding", "gzip")
c.setHTTPHeaders(req)
c.setHTTPProxyHeaders(req)
if modifyRequest != nil {
modifyRequest(req)
}
resp, err := doRequestWithPossibleRetry(client, req)
if err != nil {
return nil, fmt.Errorf("cannot fetch %q: %w", requestURL, err)
}
var data []byte
if ce := resp.Header.Peek("Content-Encoding"); string(ce) == "gzip" {
dst, err := fasthttp.AppendGunzipBytes(nil, resp.Body())
reader := resp.Body
if resp.Header.Get("Content-Encoding") == "gzip" {
reader, err = gzip.NewReader(resp.Body)
if err != nil {
return nil, fmt.Errorf("cannot ungzip response from %q: %w", requestURL, err)
return nil, fmt.Errorf("cannot create gzip reader for %q: %w", requestURL, err)
}
data = dst
} else {
data = append(data[:0], resp.Body()...)
}
data, err := io.ReadAll(reader)
if err != nil {
return nil, fmt.Errorf("cannot ungzip response from %q: %w", requestURL, err)
}
_ = resp.Body.Close()
if inspectResponse != nil {
inspectResponse(&resp)
inspectResponse(resp)
}
statusCode := resp.StatusCode()
if statusCode != fasthttp.StatusOK {
statusCode := resp.StatusCode
if statusCode != http.StatusOK {
return nil, fmt.Errorf("unexpected status code returned from %q: %d; expecting %d; response body: %q",
requestURL, statusCode, fasthttp.StatusOK, data)
requestURL, statusCode, http.StatusOK, data)
}
return data, nil
}
@ -245,26 +275,34 @@ func (c *Client) APIServer() string {
return c.apiServer
}
// DoRequestWithPossibleRetry performs the given req at hc and stores the response at resp.
func DoRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response, deadline time.Time, requestCounter, retryCounter *metrics.Counter) error {
// Stop cancels all in-flight requests
func (c *Client) Stop() {
c.clientCancel()
}
// DoRequestWithPossibleRetry performs the given req at client and stores the response at resp.
func DoRequestWithPossibleRetry(hc *HTTPClient, req *http.Request, requestCounter, retryCounter *metrics.Counter) (*http.Response, error) {
sleepTime := time.Second
requestCounter.Inc()
deadline, ok := req.Context().Deadline()
if !ok {
deadline = time.Now().Add(hc.WriteTimeout)
}
for {
// Use DoDeadline instead of Do even if hc.ReadTimeout is already set in order to guarantee the given deadline
// across multiple retries.
err := hc.DoDeadline(req, resp, deadline)
resp, err := hc.client.Do(req)
if err == nil {
statusCode := resp.StatusCode()
if statusCode != fasthttp.StatusTooManyRequests {
return nil
statusCode := resp.StatusCode
if statusCode != http.StatusTooManyRequests {
return resp, nil
}
} else if err != fasthttp.ErrConnectionClosed && !strings.Contains(err.Error(), "broken pipe") {
return err
} else if err != net.ErrClosed && !strings.Contains(err.Error(), "broken pipe") {
return nil, err
}
// Retry request after exponentially increased sleep.
maxSleepTime := time.Until(deadline)
if sleepTime > maxSleepTime {
return fmt.Errorf("the server closes all the connection attempts: %w", err)
return nil, fmt.Errorf("the server closes all the connection attempts: %w", err)
}
sleepTime += sleepTime
if sleepTime > maxSleepTime {
@ -275,8 +313,8 @@ func DoRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request,
}
}
func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response, deadline time.Time) error {
return DoRequestWithPossibleRetry(hc, req, resp, deadline, discoveryRequests, discoveryRetries)
func doRequestWithPossibleRetry(hc *HTTPClient, req *http.Request) (*http.Response, error) {
return DoRequestWithPossibleRetry(hc, req, discoveryRequests, discoveryRetries)
}
var (