mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Joe ElliottandGitHub
95de404b8b
The query frontend treats all jobs as the same size when it farms them out to the queriers. This can cause querier instability b/c some jobs actually require quite a bit more resources to execute. By assigning weights to jobs we can reduce the amount each querier is asked to do will hopefully: reduce querier OOMs/timeouts/retries reduce querier latency increase total throughput Other changes Removed the roundtripper httpgrpc bridge and pushed the concept of pipeline.Request all the way down into the cortex frontend code. This can be a nice perf improvement b/c translating http -> httpgrpc is costly and we are pushing it to the last moment. Currently for some queries we are translating thousands of jobs and then throwing them away. Removed redundant parseQuery and createFetchSpansRequest to consolidate on the Compile function in pkg/traceql Check for context error before going through retry logic in retryWare. This causes retry metrics to be more accurate in the event of many cancelled jobs.
161 lines
4.2 KiB
Go
161 lines
4.2 KiB
Go
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"testing"
|
|
|
|
"github.com/grafana/dskit/httpgrpc"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/stretchr/testify/require"
|
|
"go.uber.org/atomic"
|
|
)
|
|
|
|
func TestRetry(t *testing.T) {
|
|
var try atomic.Int32
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
handler RoundTripper
|
|
maxRetries int
|
|
expectedTries int32
|
|
expectedRes *http.Response
|
|
expectedErr error
|
|
}{
|
|
{
|
|
name: "retry errors until success",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
if try.Inc() == 5 {
|
|
return &http.Response{StatusCode: 200}, nil
|
|
}
|
|
return nil, errors.New("this request failed")
|
|
}),
|
|
maxRetries: 5,
|
|
expectedTries: 5,
|
|
expectedRes: &http.Response{StatusCode: 200},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "don't retry 400's",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
return &http.Response{StatusCode: 400}, nil
|
|
}),
|
|
maxRetries: 5,
|
|
expectedTries: 1,
|
|
expectedRes: &http.Response{StatusCode: 400},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "don't retry GRPC request with HTTP 400's",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
return nil, httpgrpc.ErrorFromHTTPResponse(&httpgrpc.HTTPResponse{Code: 400})
|
|
}),
|
|
maxRetries: 5,
|
|
expectedTries: 1,
|
|
expectedRes: nil,
|
|
expectedErr: httpgrpc.ErrorFromHTTPResponse(&httpgrpc.HTTPResponse{Code: 400}),
|
|
},
|
|
{
|
|
name: "retry 500s",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
return &http.Response{StatusCode: 503}, nil
|
|
}),
|
|
maxRetries: 5,
|
|
expectedTries: 5,
|
|
expectedRes: &http.Response{StatusCode: 503},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "return last error",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
if try.Inc() == 5 {
|
|
return nil, errors.New("request failed")
|
|
}
|
|
return nil, errors.New("not the last request")
|
|
}),
|
|
maxRetries: 5,
|
|
expectedTries: 5,
|
|
expectedRes: nil,
|
|
expectedErr: errors.New("request failed"),
|
|
},
|
|
{
|
|
name: "maxRetries=1",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
return &http.Response{StatusCode: 500}, nil
|
|
}),
|
|
maxRetries: 1,
|
|
expectedTries: 1,
|
|
expectedRes: &http.Response{StatusCode: 500},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "maxRetries=0",
|
|
handler: RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
return &http.Response{StatusCode: 500}, nil
|
|
}),
|
|
maxRetries: 0,
|
|
expectedTries: 1,
|
|
expectedRes: &http.Response{StatusCode: 500},
|
|
expectedErr: nil,
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
try.Store(0)
|
|
|
|
retryWare := NewRetryWare(tc.maxRetries, true, prometheus.NewRegistry())
|
|
handler := retryWare.Wrap(tc.handler)
|
|
|
|
req := httptest.NewRequest("GET", "http://example.com", nil)
|
|
|
|
res, err := handler.RoundTrip(NewHTTPRequest(req))
|
|
|
|
require.Equal(t, tc.expectedTries, try.Load())
|
|
require.Equal(t, tc.expectedErr, err)
|
|
require.Equal(t, tc.expectedRes, res)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRetry_CancelledRequest(t *testing.T) {
|
|
var try atomic.Int32
|
|
|
|
// request is cancelled before first call
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
req, err := http.NewRequestWithContext(ctx, "GET", "http://example.com", nil)
|
|
require.NoError(t, err)
|
|
|
|
_, err = NewRetryWare(5, false, prometheus.NewRegistry()).
|
|
Wrap(RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
return nil, ctx.Err()
|
|
})).RoundTrip(NewHTTPRequest(req))
|
|
|
|
require.Equal(t, int32(0), try.Load())
|
|
require.Equal(t, ctx.Err(), err)
|
|
|
|
// request is cancelled after first call
|
|
ctx, cancel = context.WithCancel(context.Background())
|
|
|
|
req, err = http.NewRequestWithContext(ctx, "GET", "http://example.com", nil)
|
|
require.NoError(t, err)
|
|
|
|
_, err = NewRetryWare(5, false, prometheus.NewRegistry()).
|
|
Wrap(RoundTripperFunc(func(_ Request) (*http.Response, error) {
|
|
try.Inc()
|
|
cancel()
|
|
return nil, errors.New("this request failed")
|
|
})).RoundTrip(NewHTTPRequest(req))
|
|
|
|
require.Equal(t, int32(1), try.Load())
|
|
require.Equal(t, ctx.Err(), err)
|
|
}
|