-
Notifications
You must be signed in to change notification settings - Fork 544
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Optionally wait for the query-frontend to start up before rejecting r…
…equests (#6621) * Backoff and retry requests received while the query-frontend is starting up * Add changelog entry. * Address PR feedback: disable backoff by default, rename flag, make `isTerminalError` resilient to other states added in the future * Make formatting of newQueryTripperware consistent. * Check frontend is running earlier, to avoid unnecessary work and avoid other middleware running while frontend is in an inconsistent state * Refactor to use `service.AwaitRunning()` * Update changelog entry to reflect new behaviour and config option name * Address PR feedback: `ie.` -> `i.e.` * Address PR feedback: pass interface, not function * Simplify code. * Apply same logic to cardinality and series endpoints as well. * Address PR feedback: remove implicit circular dependency. * Update outdated comment. * Make method and trace span names consistent
- Loading branch information
1 parent
880d23f
commit a27952a
Showing
11 changed files
with
206 additions
and
23 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
|
|
||
| package querymiddleware | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/go-kit/log/level" | ||
| "github.com/grafana/dskit/services" | ||
| "github.com/opentracing/opentracing-go/ext" | ||
|
|
||
| apierror "github.com/grafana/mimir/pkg/api/error" | ||
| "github.com/grafana/mimir/pkg/util/spanlogger" | ||
| ) | ||
|
|
||
| func NewFrontendRunningRoundTripper(next http.RoundTripper, service services.Service, timeout time.Duration, log log.Logger) http.RoundTripper { | ||
| return &frontendRunningRoundTripper{ | ||
| next: next, | ||
| service: service, | ||
| timeout: timeout, | ||
| log: log, | ||
| } | ||
| } | ||
|
|
||
| type frontendRunningRoundTripper struct { | ||
| next http.RoundTripper | ||
| service services.Service | ||
| timeout time.Duration | ||
| log log.Logger | ||
| } | ||
|
|
||
| func (f *frontendRunningRoundTripper) RoundTrip(request *http.Request) (*http.Response, error) { | ||
| if err := awaitQueryFrontendServiceRunning(request.Context(), f.service, f.timeout, f.log); err != nil { | ||
| return nil, apierror.New(apierror.TypeUnavailable, err.Error()) | ||
| } | ||
|
|
||
| return f.next.RoundTrip(request) | ||
| } | ||
|
|
||
| // This method is not on frontendRunningRoundTripper to make it easier to test this logic. | ||
| func awaitQueryFrontendServiceRunning(ctx context.Context, service services.Service, timeout time.Duration, log log.Logger) error { | ||
| if state := service.State(); state == services.Running { | ||
| // Fast path: frontend is already running, nothing more to do. | ||
| return nil | ||
| } else if timeout == 0 { | ||
| // If waiting for the frontend to be ready is disabled by config, and it's not ready, abort now. | ||
| return fmt.Errorf("frontend not running: %v", state) | ||
| } | ||
|
|
||
| spanLog, ctx := spanlogger.NewWithLogger(ctx, log, "awaitQueryFrontendServiceRunning") | ||
| defer spanLog.Finish() | ||
|
|
||
| ctx, cancel := context.WithTimeout(ctx, timeout) | ||
| defer cancel() | ||
|
|
||
| if err := service.AwaitRunning(ctx); err != nil { | ||
| ext.Error.Set(spanLog.Span, true) | ||
|
|
||
| if ctx.Err() != nil { | ||
| level.Warn(spanLog).Log("msg", "frontend not running, timed out waiting for it to be running", "state", service.State(), "timeout", timeout) | ||
| return fmt.Errorf("frontend not running (is %v), timed out waiting for it to be running after %v", service.State(), timeout) | ||
| } | ||
|
|
||
| level.Warn(spanLog).Log("msg", "failed waiting for frontend to be running", "err", err) | ||
| return fmt.Errorf("frontend not running: %w", err) | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
|
|
||
| package querymiddleware | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/grafana/dskit/services" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestAwaitQueryFrontendServiceRunning_ServiceIsReady(t *testing.T) { | ||
| run := func(ctx context.Context) error { | ||
| <-ctx.Done() | ||
| return nil | ||
| } | ||
|
|
||
| service := services.NewBasicService(nil, run, nil) | ||
| require.NoError(t, services.StartAndAwaitRunning(context.Background(), service)) | ||
| defer func() { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), service)) }() | ||
|
|
||
| err := awaitQueryFrontendServiceRunning(context.Background(), service, time.Second, log.NewNopLogger()) | ||
| require.NoError(t, err) | ||
| } | ||
|
|
||
| func TestAwaitQueryFrontendServiceRunning_ServiceIsNotReadyWaitDisabled(t *testing.T) { | ||
| startChan := make(chan struct{}) | ||
| start := func(ctx context.Context) error { | ||
| <-startChan | ||
| return nil | ||
| } | ||
|
|
||
| service := services.NewBasicService(start, nil, nil) | ||
| require.NoError(t, service.StartAsync(context.Background())) | ||
| defer func() { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), service)) }() | ||
|
|
||
| err := awaitQueryFrontendServiceRunning(context.Background(), service, 0, log.NewNopLogger()) | ||
| require.EqualError(t, err, "frontend not running: Starting") | ||
|
|
||
| close(startChan) | ||
| } | ||
|
|
||
| func TestAwaitQueryFrontendServiceRunning_ServiceIsNotReadyInitially(t *testing.T) { | ||
| startChan := make(chan struct{}) | ||
| start := func(ctx context.Context) error { | ||
| <-startChan | ||
| return nil | ||
| } | ||
| run := func(ctx context.Context) error { | ||
| <-ctx.Done() | ||
| return nil | ||
| } | ||
|
|
||
| service := services.NewBasicService(start, run, nil) | ||
| require.NoError(t, service.StartAsync(context.Background())) | ||
| defer func() { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), service)) }() | ||
|
|
||
| go func() { | ||
| time.Sleep(500 * time.Millisecond) | ||
| close(startChan) | ||
| }() | ||
|
|
||
| err := awaitQueryFrontendServiceRunning(context.Background(), service, time.Second, log.NewNopLogger()) | ||
| require.NoError(t, err) | ||
| } | ||
|
|
||
| func TestAwaitQueryFrontendServiceRunning_ServiceIsNotReadyAfterTimeout(t *testing.T) { | ||
| serviceChan := make(chan struct{}) | ||
| start := func(ctx context.Context) error { | ||
| <-serviceChan | ||
| return nil | ||
| } | ||
|
|
||
| service := services.NewBasicService(start, nil, nil) | ||
| require.NoError(t, service.StartAsync(context.Background())) | ||
| defer func() { require.NoError(t, services.StopAndAwaitTerminated(context.Background(), service)) }() | ||
|
|
||
| err := awaitQueryFrontendServiceRunning(context.Background(), service, 100*time.Millisecond, log.NewNopLogger()) | ||
| require.EqualError(t, err, "frontend not running (is Starting), timed out waiting for it to be running after 100ms") | ||
|
|
||
| close(serviceChan) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters