mirror of https://github.com/grafana/loki
Adds the ability to hedge storage requests. (#4826)
* Adds the ability to hedge request for all backends Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Remove race from tests Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Remove the race Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Remove the race Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Testing Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * More testing Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Setup credentials to avoid auth Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * gomod Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * improve tests Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * gomod Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * changelog Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Group the configuration Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>pull/4851/head
Cyril Tovena
4 years ago
committed by
GitHub
parent
c6d3228263
commit
b27894fbf4
@ -0,0 +1,85 @@ |
||||
package azure |
||||
|
||||
import ( |
||||
"bytes" |
||||
"context" |
||||
"errors" |
||||
"net/http" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/stretchr/testify/require" |
||||
"go.uber.org/atomic" |
||||
|
||||
"github.com/grafana/loki/pkg/storage/chunk/hedging" |
||||
) |
||||
|
||||
type RoundTripperFunc func(*http.Request) (*http.Response, error) |
||||
|
||||
func (fn RoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { |
||||
return fn(req) |
||||
} |
||||
|
||||
func Test_Hedging(t *testing.T) { |
||||
for _, tc := range []struct { |
||||
name string |
||||
expectedCalls int32 |
||||
hedgeAt time.Duration |
||||
upTo int |
||||
do func(c *BlobStorage) |
||||
}{ |
||||
{ |
||||
"delete/put/list are not hedged", |
||||
3, |
||||
20 * time.Nanosecond, |
||||
10, |
||||
func(c *BlobStorage) { |
||||
_ = c.DeleteObject(context.Background(), "foo") |
||||
_, _, _ = c.List(context.Background(), "foo", "/") |
||||
_ = c.PutObject(context.Background(), "foo", bytes.NewReader([]byte("bar"))) |
||||
}, |
||||
}, |
||||
{ |
||||
"gets are hedged", |
||||
3, |
||||
20 * time.Nanosecond, |
||||
3, |
||||
func(c *BlobStorage) { |
||||
_, _ = c.GetObject(context.Background(), "foo") |
||||
}, |
||||
}, |
||||
{ |
||||
"gets are not hedged when not configured", |
||||
1, |
||||
0, |
||||
0, |
||||
func(c *BlobStorage) { |
||||
_, _ = c.GetObject(context.Background(), "foo") |
||||
}, |
||||
}, |
||||
} { |
||||
tc := tc |
||||
t.Run(tc.name, func(t *testing.T) { |
||||
count := atomic.NewInt32(0) |
||||
// hijack the client to count the number of calls
|
||||
defaultClient = &http.Client{ |
||||
Transport: RoundTripperFunc(func(req *http.Request) (*http.Response, error) { |
||||
count.Inc() |
||||
time.Sleep(200 * time.Millisecond) |
||||
return nil, errors.New("fo") |
||||
}), |
||||
} |
||||
c, err := NewBlobStorage(&BlobStorageConfig{ |
||||
ContainerName: "foo", |
||||
Environment: azureGlobal, |
||||
MaxRetries: 1, |
||||
}, hedging.Config{ |
||||
At: tc.hedgeAt, |
||||
UpTo: tc.upTo, |
||||
}) |
||||
require.NoError(t, err) |
||||
tc.do(c) |
||||
require.Equal(t, tc.expectedCalls, count.Load()) |
||||
}) |
||||
} |
||||
} |
||||
@ -0,0 +1,90 @@ |
||||
package gcp |
||||
|
||||
import ( |
||||
"bytes" |
||||
"context" |
||||
"net/http" |
||||
"net/http/httptest" |
||||
"testing" |
||||
"time" |
||||
|
||||
"cloud.google.com/go/storage" |
||||
"github.com/stretchr/testify/require" |
||||
"go.uber.org/atomic" |
||||
"google.golang.org/api/option" |
||||
|
||||
"github.com/grafana/loki/pkg/storage/chunk/hedging" |
||||
) |
||||
|
||||
func Test_Hedging(t *testing.T) { |
||||
for _, tc := range []struct { |
||||
name string |
||||
expectedCalls int32 |
||||
hedgeAt time.Duration |
||||
upTo int |
||||
do func(c *GCSObjectClient) |
||||
}{ |
||||
{ |
||||
"delete/put/list are not hedged", |
||||
3, |
||||
20 * time.Nanosecond, |
||||
10, |
||||
func(c *GCSObjectClient) { |
||||
_ = c.DeleteObject(context.Background(), "foo") |
||||
_, _, _ = c.List(context.Background(), "foo", "/") |
||||
_ = c.PutObject(context.Background(), "foo", bytes.NewReader([]byte("bar"))) |
||||
}, |
||||
}, |
||||
{ |
||||
"gets are hedged", |
||||
3, |
||||
20 * time.Nanosecond, |
||||
3, |
||||
func(c *GCSObjectClient) { |
||||
_, _ = c.GetObject(context.Background(), "foo") |
||||
}, |
||||
}, |
||||
{ |
||||
"gets are not hedged when not configured", |
||||
1, |
||||
0, |
||||
0, |
||||
func(c *GCSObjectClient) { |
||||
_, _ = c.GetObject(context.Background(), "foo") |
||||
}, |
||||
}, |
||||
} { |
||||
tc := tc |
||||
t.Run(tc.name, func(t *testing.T) { |
||||
count := atomic.NewInt32(0) |
||||
server := fakeServer(t, 200*time.Millisecond, count) |
||||
ctx := context.Background() |
||||
c, err := newGCSObjectClient(ctx, GCSConfig{ |
||||
BucketName: "test-bucket", |
||||
Insecure: true, |
||||
}, hedging.Config{ |
||||
At: tc.hedgeAt, |
||||
UpTo: tc.upTo, |
||||
}, func(ctx context.Context, opts ...option.ClientOption) (*storage.Client, error) { |
||||
opts = append(opts, option.WithEndpoint(server.URL)) |
||||
opts = append(opts, option.WithoutAuthentication()) |
||||
return storage.NewClient(ctx, opts...) |
||||
}) |
||||
require.NoError(t, err) |
||||
tc.do(c) |
||||
require.Equal(t, tc.expectedCalls, count.Load()) |
||||
}) |
||||
} |
||||
} |
||||
|
||||
func fakeServer(t *testing.T, returnIn time.Duration, counter *atomic.Int32) *httptest.Server { |
||||
server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
||||
counter.Inc() |
||||
time.Sleep(returnIn) |
||||
_, _ = w.Write([]byte(`{}`)) |
||||
})) |
||||
server.StartTLS() |
||||
t.Cleanup(server.Close) |
||||
|
||||
return server |
||||
} |
||||
@ -0,0 +1,42 @@ |
||||
package hedging |
||||
|
||||
import ( |
||||
"flag" |
||||
"net/http" |
||||
"time" |
||||
|
||||
"github.com/cristalhq/hedgedhttp" |
||||
) |
||||
|
||||
// Config is the configuration for hedging requests.
|
||||
type Config struct { |
||||
// At is the duration after which a second request will be issued.
|
||||
At time.Duration `yaml:"at"` |
||||
// UpTo is the maximum number of requests that will be issued.
|
||||
UpTo int `yaml:"up_to"` |
||||
} |
||||
|
||||
// RegisterFlags registers flags.
|
||||
func (cfg *Config) RegisterFlags(f *flag.FlagSet) { |
||||
cfg.RegisterFlagsWithPrefix("", f) |
||||
} |
||||
|
||||
// RegisterFlagsWithPrefix registers flags with prefix.
|
||||
func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) { |
||||
f.IntVar(&cfg.UpTo, prefix+"hedge-requests-up-to", 2, "The maximun of hedge requests allowed.") |
||||
f.DurationVar(&cfg.At, prefix+"hedge-requests-at", 0, "If set to a non-zero value a second request will be issued at the provided duration. Default is 0 (disabled)") |
||||
} |
||||
|
||||
func (cfg *Config) Client(client *http.Client) *http.Client { |
||||
if cfg.At == 0 { |
||||
return client |
||||
} |
||||
return hedgedhttp.NewClient(cfg.At, cfg.UpTo, client) |
||||
} |
||||
|
||||
func (cfg *Config) RoundTripper(next http.RoundTripper) http.RoundTripper { |
||||
if cfg.At == 0 { |
||||
return next |
||||
} |
||||
return hedgedhttp.NewRoundTripper(cfg.At, cfg.UpTo, next) |
||||
} |
||||
@ -0,0 +1,110 @@ |
||||
package openstack |
||||
|
||||
import ( |
||||
"bytes" |
||||
"context" |
||||
"net/http" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/cortexproject/cortex/pkg/storage/bucket/swift" |
||||
"github.com/stretchr/testify/require" |
||||
"go.uber.org/atomic" |
||||
|
||||
"github.com/grafana/loki/pkg/storage/chunk/hedging" |
||||
) |
||||
|
||||
type RoundTripperFunc func(*http.Request) (*http.Response, error) |
||||
|
||||
func (fn RoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { |
||||
return fn(req) |
||||
} |
||||
|
||||
func Test_Hedging(t *testing.T) { |
||||
for _, tc := range []struct { |
||||
name string |
||||
expectedCalls int32 |
||||
hedgeAt time.Duration |
||||
upTo int |
||||
do func(c *SwiftObjectClient) |
||||
}{ |
||||
{ |
||||
"delete/put/list are not hedged", |
||||
3, |
||||
20 * time.Nanosecond, |
||||
10, |
||||
func(c *SwiftObjectClient) { |
||||
_ = c.DeleteObject(context.Background(), "foo") |
||||
_, _, _ = c.List(context.Background(), "foo", "/") |
||||
_ = c.PutObject(context.Background(), "foo", bytes.NewReader([]byte("bar"))) |
||||
}, |
||||
}, |
||||
{ |
||||
"gets are hedged", |
||||
3, |
||||
20 * time.Nanosecond, |
||||
3, |
||||
func(c *SwiftObjectClient) { |
||||
_, _ = c.GetObject(context.Background(), "foo") |
||||
}, |
||||
}, |
||||
{ |
||||
"gets are not hedged when not configured", |
||||
1, |
||||
0, |
||||
0, |
||||
func(c *SwiftObjectClient) { |
||||
_, _ = c.GetObject(context.Background(), "foo") |
||||
}, |
||||
}, |
||||
} { |
||||
tc := tc |
||||
t.Run(tc.name, func(t *testing.T) { |
||||
count := atomic.NewInt32(0) |
||||
// hijack the transport to count the number of calls
|
||||
defaultTransport = RoundTripperFunc(func(req *http.Request) (*http.Response, error) { |
||||
// fake auth
|
||||
if req.Header.Get("X-Auth-Key") == "passwd" { |
||||
return &http.Response{ |
||||
StatusCode: http.StatusOK, |
||||
Body: http.NoBody, |
||||
Header: http.Header{ |
||||
"X-Storage-Url": []string{"http://swift.example.com/v1/AUTH_test"}, |
||||
"X-Auth-Token": []string{"token"}, |
||||
}, |
||||
}, nil |
||||
} |
||||
// fake container creation
|
||||
if req.Method == "PUT" && req.URL.Path == "/v1/AUTH_test/foo" { |
||||
return &http.Response{ |
||||
StatusCode: http.StatusCreated, |
||||
Body: http.NoBody, |
||||
}, nil |
||||
} |
||||
count.Inc() |
||||
time.Sleep(200 * time.Millisecond) |
||||
return &http.Response{ |
||||
StatusCode: http.StatusOK, |
||||
Body: http.NoBody, |
||||
}, nil |
||||
}) |
||||
|
||||
c, err := NewSwiftObjectClient(SwiftConfig{ |
||||
Config: swift.Config{ |
||||
MaxRetries: 1, |
||||
ContainerName: "foo", |
||||
AuthVersion: 1, |
||||
Password: "passwd", |
||||
ConnectTimeout: 10 * time.Second, |
||||
RequestTimeout: 10 * time.Second, |
||||
}, |
||||
}, hedging.Config{ |
||||
At: tc.hedgeAt, |
||||
UpTo: tc.upTo, |
||||
}) |
||||
require.NoError(t, err) |
||||
tc.do(c) |
||||
require.Equal(t, tc.expectedCalls, count.Load()) |
||||
}) |
||||
} |
||||
} |
||||
@ -0,0 +1,21 @@ |
||||
MIT License |
||||
|
||||
Copyright (c) 2021 cristaltech |
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy |
||||
of this software and associated documentation files (the "Software"), to deal |
||||
in the Software without restriction, including without limitation the rights |
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell |
||||
copies of the Software, and to permit persons to whom the Software is |
||||
furnished to do so, subject to the following conditions: |
||||
|
||||
The above copyright notice and this permission notice shall be included in all |
||||
copies or substantial portions of the Software. |
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR |
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, |
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE |
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER |
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE |
||||
SOFTWARE. |
||||
@ -0,0 +1,53 @@ |
||||
# hedgedhttp |
||||
|
||||
[![build-img]][build-url] |
||||
[![pkg-img]][pkg-url] |
||||
[![reportcard-img]][reportcard-url] |
||||
[![coverage-img]][coverage-url] |
||||
|
||||
Hedged HTTP client which helps to reduce tail latency at scale. |
||||
|
||||
## Rationale |
||||
|
||||
See paper [Tail at Scale](https://cacm.acm.org/magazines/2013/2/160173-the-tail-at-scale/fulltext) by Jeffrey Dean, Luiz André Barroso. In short: the client first sends one request, but then sends an additional request after a timeout if the previous hasn't returned an answer in the expected time. The client cancels remaining requests once the first result is received. |
||||
|
||||
## Acknowledge |
||||
|
||||
Thanks to [Bohdan Storozhuk](https://github.com/storozhukbm) for the review and powerful hints. |
||||
|
||||
## Features |
||||
|
||||
* Simple API. |
||||
* Easy to integrate. |
||||
* Optimized for speed. |
||||
* Clean and tested code. |
||||
* Dependency-free. |
||||
|
||||
## Install |
||||
|
||||
Go version 1.16+ |
||||
|
||||
``` |
||||
go get github.com/cristalhq/hedgedhttp |
||||
``` |
||||
|
||||
## Example |
||||
|
||||
TODO |
||||
|
||||
## Documentation |
||||
|
||||
See [these docs][pkg-url]. |
||||
|
||||
## License |
||||
|
||||
[MIT License](LICENSE). |
||||
|
||||
[build-img]: https://github.com/cristalhq/hedgedhttp/workflows/build/badge.svg |
||||
[build-url]: https://github.com/cristalhq/hedgedhttp/actions |
||||
[pkg-img]: https://pkg.go.dev/badge/cristalhq/hedgedhttp |
||||
[pkg-url]: https://pkg.go.dev/github.com/cristalhq/hedgedhttp |
||||
[reportcard-img]: https://goreportcard.com/badge/cristalhq/hedgedhttp |
||||
[reportcard-url]: https://goreportcard.com/report/cristalhq/hedgedhttp |
||||
[coverage-img]: https://codecov.io/gh/cristalhq/hedgedhttp/branch/main/graph/badge.svg |
||||
[coverage-url]: https://codecov.io/gh/cristalhq/hedgedhttp |
||||
@ -0,0 +1,345 @@ |
||||
package hedgedhttp |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"net/http" |
||||
"strings" |
||||
"sync" |
||||
"sync/atomic" |
||||
"time" |
||||
) |
||||
|
||||
const infiniteTimeout = 30 * 24 * time.Hour // domain specific infinite
|
||||
|
||||
// NewClient returns a new http.Client which implements hedged requests pattern.
|
||||
// Given Client starts a new request after a timeout from previous request.
|
||||
// Starts no more than upto requests.
|
||||
func NewClient(timeout time.Duration, upto int, client *http.Client) *http.Client { |
||||
newClient, _ := NewClientAndStats(timeout, upto, client) |
||||
return newClient |
||||
} |
||||
|
||||
// NewClientAndStats returns a new http.Client which implements hedged requests pattern
|
||||
// And Stats object that can be queried to obtain client's metrics.
|
||||
// Given Client starts a new request after a timeout from previous request.
|
||||
// Starts no more than upto requests.
|
||||
func NewClientAndStats(timeout time.Duration, upto int, client *http.Client) (*http.Client, *Stats) { |
||||
if client == nil { |
||||
client = &http.Client{ |
||||
Timeout: 5 * time.Second, |
||||
} |
||||
} |
||||
|
||||
newTransport, metrics := NewRoundTripperAndStats(timeout, upto, client.Transport) |
||||
|
||||
client.Transport = newTransport |
||||
|
||||
return client, metrics |
||||
} |
||||
|
||||
// NewRoundTripper returns a new http.RoundTripper which implements hedged requests pattern.
|
||||
// Given RoundTripper starts a new request after a timeout from previous request.
|
||||
// Starts no more than upto requests.
|
||||
func NewRoundTripper(timeout time.Duration, upto int, rt http.RoundTripper) http.RoundTripper { |
||||
newRT, _ := NewRoundTripperAndStats(timeout, upto, rt) |
||||
return newRT |
||||
} |
||||
|
||||
// NewRoundTripperAndStats returns a new http.RoundTripper which implements hedged requests pattern
|
||||
// And Stats object that can be queried to obtain client's metrics.
|
||||
// Given RoundTripper starts a new request after a timeout from previous request.
|
||||
// Starts no more than upto requests.
|
||||
func NewRoundTripperAndStats(timeout time.Duration, upto int, rt http.RoundTripper) (http.RoundTripper, *Stats) { |
||||
switch { |
||||
case timeout < 0: |
||||
panic("hedgedhttp: timeout cannot be negative") |
||||
case upto < 1: |
||||
panic("hedgedhttp: upto must be greater than 0") |
||||
} |
||||
|
||||
if rt == nil { |
||||
rt = http.DefaultTransport |
||||
} |
||||
|
||||
if timeout == 0 { |
||||
timeout = time.Nanosecond // smallest possible timeout if not set
|
||||
} |
||||
|
||||
hedged := &hedgedTransport{ |
||||
rt: rt, |
||||
timeout: timeout, |
||||
upto: upto, |
||||
metrics: &Stats{}, |
||||
} |
||||
return hedged, hedged.metrics |
||||
} |
||||
|
||||
type hedgedTransport struct { |
||||
rt http.RoundTripper |
||||
timeout time.Duration |
||||
upto int |
||||
metrics *Stats |
||||
} |
||||
|
||||
func (ht *hedgedTransport) RoundTrip(req *http.Request) (*http.Response, error) { |
||||
mainCtx := req.Context() |
||||
|
||||
timeout := ht.timeout |
||||
errOverall := &MultiError{} |
||||
resultCh := make(chan indexedResp, ht.upto) |
||||
errorCh := make(chan error, ht.upto) |
||||
|
||||
ht.metrics.requestedRoundTripsInc() |
||||
|
||||
resultIdx := -1 |
||||
cancels := make([]func(), ht.upto) |
||||
|
||||
defer runInPool(func() { |
||||
for i, cancel := range cancels { |
||||
if i != resultIdx && cancel != nil { |
||||
ht.metrics.canceledSubRequestsInc() |
||||
cancel() |
||||
} |
||||
} |
||||
}) |
||||
|
||||
for sent := 0; len(errOverall.Errors) < ht.upto; sent++ { |
||||
if sent < ht.upto { |
||||
idx := sent |
||||
subReq, cancel := reqWithCtx(req, mainCtx) |
||||
cancels[idx] = cancel |
||||
|
||||
runInPool(func() { |
||||
ht.metrics.actualRoundTripsInc() |
||||
resp, err := ht.rt.RoundTrip(subReq) |
||||
if err != nil { |
||||
ht.metrics.failedRoundTripsInc() |
||||
errorCh <- err |
||||
} else { |
||||
resultCh <- indexedResp{idx, resp} |
||||
} |
||||
}) |
||||
} |
||||
|
||||
// all request sent - effectively disabling timeout between requests
|
||||
if sent == ht.upto { |
||||
timeout = infiniteTimeout |
||||
} |
||||
resp, err := waitResult(mainCtx, resultCh, errorCh, timeout) |
||||
|
||||
switch { |
||||
case resp.Resp != nil: |
||||
resultIdx = resp.Index |
||||
return resp.Resp, nil |
||||
case mainCtx.Err() != nil: |
||||
ht.metrics.canceledByUserRoundTripsInc() |
||||
return nil, mainCtx.Err() |
||||
case err != nil: |
||||
errOverall.Errors = append(errOverall.Errors, err) |
||||
} |
||||
} |
||||
|
||||
// all request have returned errors
|
||||
return nil, errOverall |
||||
} |
||||
|
||||
func waitResult(ctx context.Context, resultCh <-chan indexedResp, errorCh <-chan error, timeout time.Duration) (indexedResp, error) { |
||||
// try to read result first before blocking on all other channels
|
||||
select { |
||||
case res := <-resultCh: |
||||
return res, nil |
||||
default: |
||||
timer := getTimer(timeout) |
||||
defer returnTimer(timer) |
||||
|
||||
select { |
||||
case res := <-resultCh: |
||||
return res, nil |
||||
|
||||
case reqErr := <-errorCh: |
||||
return indexedResp{}, reqErr |
||||
|
||||
case <-ctx.Done(): |
||||
return indexedResp{}, ctx.Err() |
||||
|
||||
case <-timer.C: |
||||
return indexedResp{}, nil // it's not a request timeout, it's timeout BETWEEN consecutive requests
|
||||
} |
||||
} |
||||
} |
||||
|
||||
type indexedResp struct { |
||||
Index int |
||||
Resp *http.Response |
||||
} |
||||
|
||||
func reqWithCtx(r *http.Request, ctx context.Context) (*http.Request, func()) { |
||||
ctx, cancel := context.WithCancel(ctx) |
||||
req := r.WithContext(ctx) |
||||
return req, cancel |
||||
} |
||||
|
||||
// atomicCounter is a false sharing safe counter.
|
||||
type atomicCounter struct { |
||||
count uint64 |
||||
_ [7]uint64 |
||||
} |
||||
|
||||
type cacheLine [64]byte |
||||
|
||||
// Stats object that can be queried to obtain certain metrics and get better observability.
|
||||
type Stats struct { |
||||
_ cacheLine |
||||
requestedRoundTrips atomicCounter |
||||
actualRoundTrips atomicCounter |
||||
failedRoundTrips atomicCounter |
||||
canceledByUserRoundTrips atomicCounter |
||||
canceledSubRequests atomicCounter |
||||
_ cacheLine |
||||
} |
||||
|
||||
func (s *Stats) requestedRoundTripsInc() { atomic.AddUint64(&s.requestedRoundTrips.count, 1) } |
||||
func (s *Stats) actualRoundTripsInc() { atomic.AddUint64(&s.actualRoundTrips.count, 1) } |
||||
func (s *Stats) failedRoundTripsInc() { atomic.AddUint64(&s.failedRoundTrips.count, 1) } |
||||
func (s *Stats) canceledByUserRoundTripsInc() { atomic.AddUint64(&s.canceledByUserRoundTrips.count, 1) } |
||||
func (s *Stats) canceledSubRequestsInc() { atomic.AddUint64(&s.canceledSubRequests.count, 1) } |
||||
|
||||
// RequestedRoundTrips returns count of requests that were requested by client.
|
||||
func (s *Stats) RequestedRoundTrips() uint64 { |
||||
return atomic.LoadUint64(&s.requestedRoundTrips.count) |
||||
} |
||||
|
||||
// ActualRoundTrips returns count of requests that were actually sent.
|
||||
func (s *Stats) ActualRoundTrips() uint64 { |
||||
return atomic.LoadUint64(&s.actualRoundTrips.count) |
||||
} |
||||
|
||||
// FailedRoundTrips returns count of requests that failed.
|
||||
func (s *Stats) FailedRoundTrips() uint64 { |
||||
return atomic.LoadUint64(&s.failedRoundTrips.count) |
||||
} |
||||
|
||||
// CanceledByUserRoundTrips returns count of requests that were canceled by user, using request context.
|
||||
func (s *Stats) CanceledByUserRoundTrips() uint64 { |
||||
return atomic.LoadUint64(&s.canceledByUserRoundTrips.count) |
||||
} |
||||
|
||||
// CanceledSubRequests returns count of hedged sub-requests that were canceled by transport.
|
||||
func (s *Stats) CanceledSubRequests() uint64 { |
||||
return atomic.LoadUint64(&s.canceledSubRequests.count) |
||||
} |
||||
|
||||
// StatsSnapshot is a snapshot of Stats.
|
||||
type StatsSnapshot struct { |
||||
RequestedRoundTrips uint64 // count of requests that were requested by client
|
||||
ActualRoundTrips uint64 // count of requests that were actually sent
|
||||
FailedRoundTrips uint64 // count of requests that failed
|
||||
CanceledByUserRoundTrips uint64 // count of requests that were canceled by user, using request context
|
||||
CanceledSubRequests uint64 // count of hedged sub-requests that were canceled by transport
|
||||
} |
||||
|
||||
// Snapshot of the stats.
|
||||
func (s *Stats) Snapshot() StatsSnapshot { |
||||
return StatsSnapshot{ |
||||
RequestedRoundTrips: s.RequestedRoundTrips(), |
||||
ActualRoundTrips: s.ActualRoundTrips(), |
||||
FailedRoundTrips: s.FailedRoundTrips(), |
||||
CanceledByUserRoundTrips: s.CanceledByUserRoundTrips(), |
||||
CanceledSubRequests: s.CanceledSubRequests(), |
||||
} |
||||
} |
||||
|
||||
var taskQueue = make(chan func()) |
||||
|
||||
func runInPool(task func()) { |
||||
select { |
||||
case taskQueue <- task: |
||||
// submited, everything is ok
|
||||
|
||||
default: |
||||
go func() { |
||||
// do the given task
|
||||
task() |
||||
|
||||
const cleanupDuration = 10 * time.Second |
||||
cleanupTicker := time.NewTicker(cleanupDuration) |
||||
defer cleanupTicker.Stop() |
||||
|
||||
for { |
||||
select { |
||||
case t := <-taskQueue: |
||||
t() |
||||
cleanupTicker.Reset(cleanupDuration) |
||||
case <-cleanupTicker.C: |
||||
return |
||||
} |
||||
} |
||||
}() |
||||
} |
||||
} |
||||
|
||||
// MultiError is an error type to track multiple errors. This is used to
|
||||
// accumulate errors in cases and return them as a single "error".
|
||||
// Insiper by https://github.com/hashicorp/go-multierror
|
||||
type MultiError struct { |
||||
Errors []error |
||||
ErrorFormatFn ErrorFormatFunc |
||||
} |
||||
|
||||
func (e *MultiError) Error() string { |
||||
fn := e.ErrorFormatFn |
||||
if fn == nil { |
||||
fn = listFormatFunc |
||||
} |
||||
return fn(e.Errors) |
||||
} |
||||
|
||||
func (e *MultiError) String() string { |
||||
return fmt.Sprintf("*%#v", e.Errors) |
||||
} |
||||
|
||||
// ErrorOrNil returns an error if there are some.
|
||||
func (e *MultiError) ErrorOrNil() error { |
||||
switch { |
||||
case e == nil || len(e.Errors) == 0: |
||||
return nil |
||||
default: |
||||
return e |
||||
} |
||||
} |
||||
|
||||
// ErrorFormatFunc is called by MultiError to return the list of errors as a string.
|
||||
type ErrorFormatFunc func([]error) string |
||||
|
||||
func listFormatFunc(es []error) string { |
||||
if len(es) == 1 { |
||||
return fmt.Sprintf("1 error occurred:\n\t* %s\n\n", es[0]) |
||||
} |
||||
|
||||
points := make([]string, len(es)) |
||||
for i, err := range es { |
||||
points[i] = fmt.Sprintf("* %s", err) |
||||
} |
||||
|
||||
return fmt.Sprintf("%d errors occurred:\n\t%s\n\n", len(es), strings.Join(points, "\n\t")) |
||||
} |
||||
|
||||
var timerPool = sync.Pool{New: func() interface{} { |
||||
return time.NewTimer(time.Second) |
||||
}} |
||||
|
||||
func getTimer(duration time.Duration) *time.Timer { |
||||
timer := timerPool.Get().(*time.Timer) |
||||
timer.Reset(duration) |
||||
return timer |
||||
} |
||||
|
||||
func returnTimer(timer *time.Timer) { |
||||
timer.Stop() |
||||
select { |
||||
case _ = <-timer.C: |
||||
default: |
||||
} |
||||
timerPool.Put(timer) |
||||
} |
||||
Loading…
Reference in new issue