-
Notifications
You must be signed in to change notification settings - Fork 543
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Streaming PromQL engine: optionally fall back to Prometheus' engine i…
…f query is not supported (#7898) * Initial commit. * Log and add a trace event when a query falls back to Prometheus' engine. * Add CLI flag to enable falling back to Prometheus' engine, and enable fallback by default. * Add changelog and docs * Address PR feedback * Rename metrics These metrics could be emitted by query-frontends too, so "querier" in the name could be misleading.
- Loading branch information
1 parent
37e2aa1
commit bce1bbc
Showing
10 changed files
with
302 additions
and
12 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
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,88 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
|
|
||
| package streamingpromql | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/go-kit/log/level" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promauto" | ||
| "github.com/prometheus/prometheus/promql" | ||
| "github.com/prometheus/prometheus/storage" | ||
|
|
||
| "github.com/grafana/mimir/pkg/util/spanlogger" | ||
| ) | ||
|
|
||
| type EngineWithFallback struct { | ||
| preferred promql.QueryEngine | ||
| fallback promql.QueryEngine | ||
|
|
||
| supportedQueries prometheus.Counter | ||
| unsupportedQueries *prometheus.CounterVec | ||
|
|
||
| logger log.Logger | ||
| } | ||
|
|
||
| func NewEngineWithFallback(preferred, fallback promql.QueryEngine, reg prometheus.Registerer, logger log.Logger) promql.QueryEngine { | ||
| return &EngineWithFallback{ | ||
| preferred: preferred, | ||
| fallback: fallback, | ||
|
|
||
| supportedQueries: promauto.With(reg).NewCounter(prometheus.CounterOpts{ | ||
| Name: "cortex_streaming_promql_engine_supported_queries_total", | ||
| Help: "Total number of queries that were supported by the streaming engine.", | ||
| }), | ||
| unsupportedQueries: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ | ||
| Name: "cortex_streaming_promql_engine_unsupported_queries_total", | ||
| Help: "Total number of queries that were not supported by the streaming engine and so fell back to Prometheus' engine.", | ||
| }, []string{"reason"}), | ||
|
|
||
| logger: logger, | ||
| } | ||
| } | ||
|
|
||
| func (e EngineWithFallback) NewInstantQuery(ctx context.Context, q storage.Queryable, opts promql.QueryOpts, qs string, ts time.Time) (promql.Query, error) { | ||
| query, err := e.preferred.NewInstantQuery(ctx, q, opts, qs, ts) | ||
|
|
||
| if err == nil { | ||
| e.supportedQueries.Inc() | ||
| return query, nil | ||
| } | ||
|
|
||
| notSupportedErr := NotSupportedError{} | ||
| if !errors.As(err, ¬SupportedErr) { | ||
| // Don't bother trying the fallback engine if we failed for a reason other than the expression not being supported. | ||
| return nil, err | ||
| } | ||
|
|
||
| logger := spanlogger.FromContext(ctx, e.logger) | ||
| level.Info(logger).Log("msg", "falling back to Prometheus' PromQL engine", "reason", notSupportedErr.reason, "expr", qs) | ||
| e.unsupportedQueries.WithLabelValues(notSupportedErr.reason).Inc() | ||
|
|
||
| return e.fallback.NewInstantQuery(ctx, q, opts, qs, ts) | ||
| } | ||
|
|
||
| func (e EngineWithFallback) NewRangeQuery(ctx context.Context, q storage.Queryable, opts promql.QueryOpts, qs string, start, end time.Time, interval time.Duration) (promql.Query, error) { | ||
| query, err := e.preferred.NewRangeQuery(ctx, q, opts, qs, start, end, interval) | ||
|
|
||
| if err == nil { | ||
| e.supportedQueries.Inc() | ||
| return query, nil | ||
| } | ||
|
|
||
| notSupportedErr := NotSupportedError{} | ||
| if !errors.As(err, ¬SupportedErr) { | ||
| // Don't bother trying the fallback engine if we failed for a reason other than the expression not being supported. | ||
| return nil, err | ||
| } | ||
|
|
||
| logger := spanlogger.FromContext(ctx, e.logger) | ||
| level.Info(logger).Log("msg", "falling back to Prometheus' PromQL engine", "reason", notSupportedErr.reason, "expr", qs) | ||
| e.unsupportedQueries.WithLabelValues(notSupportedErr.reason).Inc() | ||
|
|
||
| return e.fallback.NewRangeQuery(ctx, q, opts, qs, start, end, interval) | ||
| } |
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,166 @@ | ||
| // SPDX-License-Identifier: AGPL-3.0-only | ||
|
|
||
| package streamingpromql | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "strings" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| promtest "github.com/prometheus/client_golang/prometheus/testutil" | ||
| "github.com/prometheus/prometheus/promql" | ||
| "github.com/prometheus/prometheus/promql/parser" | ||
| "github.com/prometheus/prometheus/storage" | ||
| "github.com/prometheus/prometheus/util/stats" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestEngineWithFallback(t *testing.T) { | ||
| ctx := context.Background() | ||
| logger := log.NewNopLogger() | ||
|
|
||
| generators := map[string]func(engine promql.QueryEngine, expr string) (promql.Query, error){ | ||
| "instant query": func(engine promql.QueryEngine, expr string) (promql.Query, error) { | ||
| return engine.NewInstantQuery(ctx, nil, nil, expr, time.Now()) | ||
| }, | ||
| "range query": func(engine promql.QueryEngine, expr string) (promql.Query, error) { | ||
| return engine.NewRangeQuery(ctx, nil, nil, expr, time.Now(), time.Now().Add(-time.Minute), time.Second) | ||
| }, | ||
| } | ||
|
|
||
| for name, createQuery := range generators { | ||
| t.Run(name, func(t *testing.T) { | ||
| t.Run("should not fall back for supported expressions", func(t *testing.T) { | ||
| reg := prometheus.NewPedanticRegistry() | ||
| preferredEngine := newFakeEngineThatSupportsLimitedQueries() | ||
| fallbackEngine := newFakeEngineThatSupportsAllQueries() | ||
| engineWithFallback := NewEngineWithFallback(preferredEngine, fallbackEngine, reg, logger) | ||
|
|
||
| query, err := createQuery(engineWithFallback, "a_supported_expression") | ||
| require.NoError(t, err) | ||
| require.Equal(t, preferredEngine.query, query, "should return query from preferred engine") | ||
| require.False(t, fallbackEngine.wasCalled, "should not call fallback engine if expression is supported by preferred engine") | ||
|
|
||
| require.NoError(t, promtest.GatherAndCompare(reg, strings.NewReader(` | ||
| # HELP cortex_streaming_promql_engine_supported_queries_total Total number of queries that were supported by the streaming engine. | ||
| # TYPE cortex_streaming_promql_engine_supported_queries_total counter | ||
| cortex_streaming_promql_engine_supported_queries_total 1 | ||
| `), "cortex_streaming_promql_engine_supported_queries_total", "cortex_streaming_promql_engine_unsupported_queries_total")) | ||
| }) | ||
|
|
||
| t.Run("should fall back for unsupported expressions", func(t *testing.T) { | ||
| reg := prometheus.NewPedanticRegistry() | ||
| preferredEngine := newFakeEngineThatSupportsLimitedQueries() | ||
| fallbackEngine := newFakeEngineThatSupportsAllQueries() | ||
| engineWithFallback := NewEngineWithFallback(preferredEngine, fallbackEngine, reg, logger) | ||
|
|
||
| query, err := createQuery(engineWithFallback, "a_non_supported_expression") | ||
| require.NoError(t, err) | ||
| require.Equal(t, fallbackEngine.query, query, "should return query from fallback engine if expression is not supported by preferred engine") | ||
|
|
||
| require.NoError(t, promtest.GatherAndCompare(reg, strings.NewReader(` | ||
| # HELP cortex_streaming_promql_engine_supported_queries_total Total number of queries that were supported by the streaming engine. | ||
| # TYPE cortex_streaming_promql_engine_supported_queries_total counter | ||
| cortex_streaming_promql_engine_supported_queries_total 0 | ||
| # HELP cortex_streaming_promql_engine_unsupported_queries_total Total number of queries that were not supported by the streaming engine and so fell back to Prometheus' engine. | ||
| # TYPE cortex_streaming_promql_engine_unsupported_queries_total counter | ||
| cortex_streaming_promql_engine_unsupported_queries_total{reason="this expression is not supported"} 1 | ||
| `), "cortex_streaming_promql_engine_supported_queries_total", "cortex_streaming_promql_engine_unsupported_queries_total")) | ||
| }) | ||
|
|
||
| t.Run("should not fall back if creating query fails for another reason", func(t *testing.T) { | ||
| reg := prometheus.NewPedanticRegistry() | ||
| preferredEngine := newFakeEngineThatSupportsLimitedQueries() | ||
| fallbackEngine := newFakeEngineThatSupportsAllQueries() | ||
| engineWithFallback := NewEngineWithFallback(preferredEngine, fallbackEngine, reg, logger) | ||
|
|
||
| _, err := createQuery(engineWithFallback, "an_invalid_expression") | ||
| require.EqualError(t, err, "the query is invalid") | ||
| require.False(t, fallbackEngine.wasCalled, "should not call fallback engine if creating query fails for another reason") | ||
| }) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| type fakeEngineThatSupportsAllQueries struct { | ||
| query *fakeQuery | ||
| wasCalled bool | ||
| } | ||
|
|
||
| func newFakeEngineThatSupportsAllQueries() *fakeEngineThatSupportsAllQueries { | ||
| return &fakeEngineThatSupportsAllQueries{ | ||
| query: &fakeQuery{"query from fallback engine"}, | ||
| } | ||
| } | ||
|
|
||
| func (f *fakeEngineThatSupportsAllQueries) NewInstantQuery(context.Context, storage.Queryable, promql.QueryOpts, string, time.Time) (promql.Query, error) { | ||
| f.wasCalled = true | ||
| return f.query, nil | ||
| } | ||
|
|
||
| func (f *fakeEngineThatSupportsAllQueries) NewRangeQuery(context.Context, storage.Queryable, promql.QueryOpts, string, time.Time, time.Time, time.Duration) (promql.Query, error) { | ||
| f.wasCalled = true | ||
| return f.query, nil | ||
| } | ||
|
|
||
| type fakeEngineThatSupportsLimitedQueries struct { | ||
| query *fakeQuery | ||
| } | ||
|
|
||
| func newFakeEngineThatSupportsLimitedQueries() *fakeEngineThatSupportsLimitedQueries { | ||
| return &fakeEngineThatSupportsLimitedQueries{ | ||
| query: &fakeQuery{"query from preferred engine"}, | ||
| } | ||
| } | ||
|
|
||
| func (f *fakeEngineThatSupportsLimitedQueries) NewInstantQuery(_ context.Context, _ storage.Queryable, _ promql.QueryOpts, qs string, _ time.Time) (promql.Query, error) { | ||
| if qs == "a_supported_expression" { | ||
| return f.query, nil | ||
| } else if qs == "an_invalid_expression" { | ||
| return nil, errors.New("the query is invalid") | ||
| } | ||
|
|
||
| return nil, NewNotSupportedError("this expression is not supported") | ||
| } | ||
|
|
||
| func (f *fakeEngineThatSupportsLimitedQueries) NewRangeQuery(_ context.Context, _ storage.Queryable, _ promql.QueryOpts, qs string, _, _ time.Time, _ time.Duration) (promql.Query, error) { | ||
| if qs == "a_supported_expression" { | ||
| return f.query, nil | ||
| } else if qs == "an_invalid_expression" { | ||
| return nil, errors.New("the query is invalid") | ||
| } | ||
|
|
||
| return nil, NewNotSupportedError("this expression is not supported") | ||
| } | ||
|
|
||
| type fakeQuery struct { | ||
| name string | ||
| } | ||
|
|
||
| func (f fakeQuery) Exec(context.Context) *promql.Result { | ||
| panic("fakeQuery: Exec() not supported") | ||
| } | ||
|
|
||
| func (f fakeQuery) Close() { | ||
| panic("fakeQuery: Close() not supported") | ||
| } | ||
|
|
||
| func (f fakeQuery) Statement() parser.Statement { | ||
| panic("fakeQuery: Statement() not supported") | ||
| } | ||
|
|
||
| func (f fakeQuery) Stats() *stats.Statistics { | ||
| panic("fakeQuery: Stats() not supported") | ||
| } | ||
|
|
||
| func (f fakeQuery) Cancel() { | ||
| panic("fakeQuery: Cancel() not supported") | ||
| } | ||
|
|
||
| func (f fakeQuery) String() string { | ||
| panic("fakeQuery: String() not supported") | ||
| } |