From 772e8cbf6005002a8fa0ee3b622976d440700fa9 Mon Sep 17 00:00:00 2001 From: Ivana Huckova <30407135+ivanahuckova@users.noreply.github.com> Date: Thu, 5 Jan 2023 12:26:27 +0100 Subject: [PATCH] Elasticsearch: Use interval provided by data request in backend (#60480) * Elasticsearch: Remove interval caulation and use interval provises by grafana * Remove redundant code * Adjust snapshot tests * Update test * Fix lint --- pkg/tsdb/elasticsearch/client/client.go | 4 +- pkg/tsdb/elasticsearch/client/client_test.go | 3 +- pkg/tsdb/elasticsearch/client/models.go | 5 +- .../elasticsearch/client/search_request.go | 9 +- .../client/search_request_test.go | 9 +- pkg/tsdb/elasticsearch/elasticsearch.go | 10 +- pkg/tsdb/elasticsearch/models.go | 4 +- pkg/tsdb/elasticsearch/parse_query.go | 4 +- pkg/tsdb/elasticsearch/parse_query_test.go | 2 +- pkg/tsdb/elasticsearch/querydata_test.go | 6 +- .../testdata_request/logs.queries.json | 3 +- .../metric_multi.queries.json | 6 +- pkg/tsdb/elasticsearch/time_series_query.go | 21 +-- .../elasticsearch/time_series_query_test.go | 132 +++++++++--------- 14 files changed, 104 insertions(+), 114 deletions(-) diff --git a/pkg/tsdb/elasticsearch/client/client.go b/pkg/tsdb/elasticsearch/client/client.go index d95ac7a0503..6382b443871 100644 --- a/pkg/tsdb/elasticsearch/client/client.go +++ b/pkg/tsdb/elasticsearch/client/client.go @@ -89,7 +89,7 @@ func (c *baseClientImpl) GetMinInterval(queryInterval string) (time.Duration, er type multiRequest struct { header map[string]interface{} body interface{} - interval intervalv2.Interval + interval time.Duration } func (c *baseClientImpl) executeBatchRequest(uriPath, uriQuery string, requests []*multiRequest) (*http.Response, error) { @@ -119,7 +119,7 @@ func (c *baseClientImpl) encodeBatchRequests(requests []*multiRequest) ([]byte, body := string(reqBody) body = strings.ReplaceAll(body, "$__interval_ms", strconv.FormatInt(r.interval.Milliseconds(), 10)) - body = strings.ReplaceAll(body, "$__interval", r.interval.Text) + body = strings.ReplaceAll(body, "$__interval", r.interval.String()) payload.WriteString(body + "\n") } diff --git a/pkg/tsdb/elasticsearch/client/client_test.go b/pkg/tsdb/elasticsearch/client/client_test.go index 94a366bdbf1..0e0bb1bed42 100644 --- a/pkg/tsdb/elasticsearch/client/client_test.go +++ b/pkg/tsdb/elasticsearch/client/client_test.go @@ -12,7 +12,6 @@ import ( "github.com/Masterminds/semver" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana/pkg/components/simplejson" - "github.com/grafana/grafana/pkg/tsdb/intervalv2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -114,7 +113,7 @@ func createMultisearchForTest(t *testing.T, c Client) (*MultiSearchRequest, erro t.Helper() msb := c.MultiSearch() - s := msb.Search(intervalv2.Interval{Value: 15 * time.Second, Text: "15s"}) + s := msb.Search(15 * time.Second) s.Agg().DateHistogram("2", "@timestamp", func(a *DateHistogramAgg, ab AggBuilder) { a.FixedInterval = "$__interval" diff --git a/pkg/tsdb/elasticsearch/client/models.go b/pkg/tsdb/elasticsearch/client/models.go index 9c32439f1aa..8ac01ea8620 100644 --- a/pkg/tsdb/elasticsearch/client/models.go +++ b/pkg/tsdb/elasticsearch/client/models.go @@ -2,14 +2,13 @@ package es import ( "encoding/json" - - "github.com/grafana/grafana/pkg/tsdb/intervalv2" + "time" ) // SearchRequest represents a search request type SearchRequest struct { Index string - Interval intervalv2.Interval + Interval time.Duration Size int Sort map[string]interface{} Query *Query diff --git a/pkg/tsdb/elasticsearch/client/search_request.go b/pkg/tsdb/elasticsearch/client/search_request.go index 162e24d22d1..f4a4ce97755 100644 --- a/pkg/tsdb/elasticsearch/client/search_request.go +++ b/pkg/tsdb/elasticsearch/client/search_request.go @@ -2,8 +2,7 @@ package es import ( "strings" - - "github.com/grafana/grafana/pkg/tsdb/intervalv2" + "time" ) const ( @@ -14,7 +13,7 @@ const ( // SearchRequestBuilder represents a builder which can build a search request type SearchRequestBuilder struct { - interval intervalv2.Interval + interval time.Duration index string size int // Currently sort is map, but based in examples it should be an array https://www.elastic.co/guide/en/elasticsearch/reference/current/sort-search-results.html @@ -25,7 +24,7 @@ type SearchRequestBuilder struct { } // NewSearchRequestBuilder create a new search request builder -func NewSearchRequestBuilder(interval intervalv2.Interval) *SearchRequestBuilder { +func NewSearchRequestBuilder(interval time.Duration) *SearchRequestBuilder { builder := &SearchRequestBuilder{ interval: interval, sort: make(map[string]interface{}), @@ -137,7 +136,7 @@ func NewMultiSearchRequestBuilder() *MultiSearchRequestBuilder { } // Search initiates and returns a new search request builder -func (m *MultiSearchRequestBuilder) Search(interval intervalv2.Interval) *SearchRequestBuilder { +func (m *MultiSearchRequestBuilder) Search(interval time.Duration) *SearchRequestBuilder { b := NewSearchRequestBuilder(interval) m.requestBuilders = append(m.requestBuilders, b) return b diff --git a/pkg/tsdb/elasticsearch/client/search_request_test.go b/pkg/tsdb/elasticsearch/client/search_request_test.go index 502b05d96f2..618249d1619 100644 --- a/pkg/tsdb/elasticsearch/client/search_request_test.go +++ b/pkg/tsdb/elasticsearch/client/search_request_test.go @@ -6,7 +6,6 @@ import ( "time" "github.com/grafana/grafana/pkg/components/simplejson" - "github.com/grafana/grafana/pkg/tsdb/intervalv2" "github.com/stretchr/testify/require" ) @@ -15,7 +14,7 @@ func TestSearchRequest(t *testing.T) { timeField := "@timestamp" setup := func() *SearchRequestBuilder { - return NewSearchRequestBuilder(intervalv2.Interval{Value: 15 * time.Second, Text: "15s"}) + return NewSearchRequestBuilder(15 * time.Second) } t.Run("When building search request", func(t *testing.T) { @@ -401,7 +400,7 @@ func TestSearchRequest(t *testing.T) { func TestMultiSearchRequest(t *testing.T) { t.Run("When adding one search request", func(t *testing.T) { b := NewMultiSearchRequestBuilder() - b.Search(intervalv2.Interval{Value: 15 * time.Second, Text: "15s"}) + b.Search(15 * time.Second) t.Run("When building search request should contain one search request", func(t *testing.T) { mr, err := b.Build() @@ -412,8 +411,8 @@ func TestMultiSearchRequest(t *testing.T) { t.Run("When adding two search requests", func(t *testing.T) { b := NewMultiSearchRequestBuilder() - b.Search(intervalv2.Interval{Value: 15 * time.Second, Text: "15s"}) - b.Search(intervalv2.Interval{Value: 15 * time.Second, Text: "15s"}) + b.Search(15 * time.Second) + b.Search(15 * time.Second) t.Run("When building search request should contain two search requests", func(t *testing.T) { mr, err := b.Build() diff --git a/pkg/tsdb/elasticsearch/elasticsearch.go b/pkg/tsdb/elasticsearch/elasticsearch.go index 5e3b07defab..7d84751a60d 100644 --- a/pkg/tsdb/elasticsearch/elasticsearch.go +++ b/pkg/tsdb/elasticsearch/elasticsearch.go @@ -15,14 +15,12 @@ import ( "github.com/grafana/grafana/pkg/infra/httpclient" "github.com/grafana/grafana/pkg/infra/log" es "github.com/grafana/grafana/pkg/tsdb/elasticsearch/client" - "github.com/grafana/grafana/pkg/tsdb/intervalv2" ) var eslog = log.New("tsdb.elasticsearch") type Service struct { httpClientProvider httpclient.Provider - intervalCalculator intervalv2.Calculator im instancemgmt.InstanceManager } @@ -32,7 +30,6 @@ func ProvideService(httpClientProvider httpclient.Provider) *Service { return &Service{ im: datasource.NewInstanceManager(newInstanceSettings(httpClientProvider)), httpClientProvider: httpClientProvider, - intervalCalculator: intervalv2.NewCalculator(), } } @@ -42,11 +39,11 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest) return &backend.QueryDataResponse{}, err } - return queryData(ctx, req.Queries, dsInfo, s.intervalCalculator) + return queryData(ctx, req.Queries, dsInfo) } // separate function to allow testing the whole transformation and query flow -func queryData(ctx context.Context, queries []backend.DataQuery, dsInfo *es.DatasourceInfo, intervalCalculator intervalv2.Calculator) (*backend.QueryDataResponse, error) { +func queryData(ctx context.Context, queries []backend.DataQuery, dsInfo *es.DatasourceInfo) (*backend.QueryDataResponse, error) { // Support for version after their end-of-life (currently <7.10.0) was removed lastSupportedVersion, _ := semver.NewVersion("7.10.0") if dsInfo.ESVersion.LessThan(lastSupportedVersion) { @@ -61,8 +58,7 @@ func queryData(ctx context.Context, queries []backend.DataQuery, dsInfo *es.Data if err != nil { return &backend.QueryDataResponse{}, err } - - query := newTimeSeriesQuery(client, queries, intervalCalculator) + query := newTimeSeriesQuery(client, queries) return query.execute() } diff --git a/pkg/tsdb/elasticsearch/models.go b/pkg/tsdb/elasticsearch/models.go index 2740e1def1d..cdf6c299d91 100644 --- a/pkg/tsdb/elasticsearch/models.go +++ b/pkg/tsdb/elasticsearch/models.go @@ -1,6 +1,8 @@ package elasticsearch import ( + "time" + "github.com/grafana/grafana/pkg/components/simplejson" ) @@ -11,7 +13,7 @@ type Query struct { BucketAggs []*BucketAgg `json:"bucketAggs"` Metrics []*MetricAgg `json:"metrics"` Alias string `json:"alias"` - Interval string + Interval time.Duration IntervalMs int64 RefID string MaxDataPoints int64 diff --git a/pkg/tsdb/elasticsearch/parse_query.go b/pkg/tsdb/elasticsearch/parse_query.go index 8ffae3a216f..7bfbb443e70 100644 --- a/pkg/tsdb/elasticsearch/parse_query.go +++ b/pkg/tsdb/elasticsearch/parse_query.go @@ -26,7 +26,8 @@ func parseQuery(tsdbQuery []backend.DataQuery) ([]*Query, error) { return nil, err } alias := model.Get("alias").MustString("") - interval := model.Get("interval").MustString("") + intervalMs := model.Get("intervalMs").MustInt64(0) + interval := q.Interval queries = append(queries, &Query{ TimeField: timeField, @@ -35,6 +36,7 @@ func parseQuery(tsdbQuery []backend.DataQuery) ([]*Query, error) { Metrics: metrics, Alias: alias, Interval: interval, + IntervalMs: intervalMs, RefID: q.RefID, MaxDataPoints: q.MaxDataPoints, }) diff --git a/pkg/tsdb/elasticsearch/parse_query_test.go b/pkg/tsdb/elasticsearch/parse_query_test.go index fc6cf047270..26b79569ca3 100644 --- a/pkg/tsdb/elasticsearch/parse_query_test.go +++ b/pkg/tsdb/elasticsearch/parse_query_test.go @@ -70,7 +70,7 @@ func TestParseQuery(t *testing.T) { require.Equal(t, q.TimeField, "@timestamp") require.Equal(t, q.RawQuery, "@metric:cpu") require.Equal(t, q.Alias, "{{@hostname}} {{metric}}") - require.Equal(t, q.Interval, "10m") + require.Equal(t, q.Interval.String(), "10s") require.Len(t, q.Metrics, 2) require.Equal(t, q.Metrics[0].Field, "@value") diff --git a/pkg/tsdb/elasticsearch/querydata_test.go b/pkg/tsdb/elasticsearch/querydata_test.go index 162fb0cb8f1..c3161d4b1b3 100644 --- a/pkg/tsdb/elasticsearch/querydata_test.go +++ b/pkg/tsdb/elasticsearch/querydata_test.go @@ -12,7 +12,6 @@ import ( "github.com/Masterminds/semver" "github.com/grafana/grafana-plugin-sdk-go/backend" es "github.com/grafana/grafana/pkg/tsdb/elasticsearch/client" - "github.com/grafana/grafana/pkg/tsdb/intervalv2" ) type queryDataTestRoundTripper struct { @@ -56,6 +55,7 @@ func newFlowTestDsInfo(body []byte, reuestCallback func(req *http.Request) error type queryDataTestQueryJSON struct { IntervalMs int64 + Interval time.Duration MaxDataPoints int64 RefID string } @@ -90,7 +90,7 @@ func newFlowTestQueries(allJsonBytes []byte) ([]backend.DataQuery, error) { query := backend.DataQuery{ RefID: jsonInfo.RefID, MaxDataPoints: jsonInfo.MaxDataPoints, - Interval: time.Duration(jsonInfo.IntervalMs) * time.Millisecond, + Interval: jsonInfo.Interval, TimeRange: timeRange, JSON: jsonBytes, } @@ -130,7 +130,7 @@ func queryDataTest(queriesBytes []byte, responseBytes []byte) (queryDataTestResu return nil }) - result, err := queryData(context.Background(), queries, dsInfo, intervalv2.NewCalculator()) + result, err := queryData(context.Background(), queries, dsInfo) if err != nil { return queryDataTestResult{}, err } diff --git a/pkg/tsdb/elasticsearch/testdata_request/logs.queries.json b/pkg/tsdb/elasticsearch/testdata_request/logs.queries.json index 781bf4309a8..773dde53fe8 100644 --- a/pkg/tsdb/elasticsearch/testdata_request/logs.queries.json +++ b/pkg/tsdb/elasticsearch/testdata_request/logs.queries.json @@ -17,7 +17,8 @@ "timeField": "testtime", "key": "Q-ee8fea91-a4c4-4ded-9827-b362476a4083-0", "datasourceId": 39, - "intervalMs": 2000, + "intervalMs": 1000, + "interval": 1000000000, "maxDataPoints": 1318 } ] \ No newline at end of file diff --git a/pkg/tsdb/elasticsearch/testdata_request/metric_multi.queries.json b/pkg/tsdb/elasticsearch/testdata_request/metric_multi.queries.json index 547950c1d63..f39757ece31 100644 --- a/pkg/tsdb/elasticsearch/testdata_request/metric_multi.queries.json +++ b/pkg/tsdb/elasticsearch/testdata_request/metric_multi.queries.json @@ -15,7 +15,8 @@ } ], "timeField": "testtime", - "intervalMs": 30000, + "intervalMs": 1000, + "interval": 1000000000, "maxDataPoints": 814 }, { @@ -35,7 +36,8 @@ } ], "timeField": "testtime", - "intervalMs": 30000, + "intervalMs": 1000, + "interval": 1000000000, "maxDataPoints": 814 } ] diff --git a/pkg/tsdb/elasticsearch/time_series_query.go b/pkg/tsdb/elasticsearch/time_series_query.go index 076d041fd04..f41db468917 100644 --- a/pkg/tsdb/elasticsearch/time_series_query.go +++ b/pkg/tsdb/elasticsearch/time_series_query.go @@ -9,7 +9,6 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana/pkg/components/simplejson" es "github.com/grafana/grafana/pkg/tsdb/elasticsearch/client" - "github.com/grafana/grafana/pkg/tsdb/intervalv2" ) const ( @@ -17,17 +16,14 @@ const ( ) type timeSeriesQuery struct { - client es.Client - dataQueries []backend.DataQuery - intervalCalculator intervalv2.Calculator + client es.Client + dataQueries []backend.DataQuery } -var newTimeSeriesQuery = func(client es.Client, dataQuery []backend.DataQuery, - intervalCalculator intervalv2.Calculator) *timeSeriesQuery { +var newTimeSeriesQuery = func(client es.Client, dataQuery []backend.DataQuery) *timeSeriesQuery { return &timeSeriesQuery{ - client: client, - dataQueries: dataQuery, - intervalCalculator: intervalCalculator, + client: client, + dataQueries: dataQuery, } } @@ -65,14 +61,9 @@ func (e *timeSeriesQuery) execute() (*backend.QueryDataResponse, error) { func (e *timeSeriesQuery) processQuery(q *Query, ms *es.MultiSearchRequestBuilder, from, to int64, result backend.QueryDataResponse) error { - minInterval, err := e.client.GetMinInterval(q.Interval) - if err != nil { - return err - } - interval := e.intervalCalculator.Calculate(e.dataQueries[0].TimeRange, minInterval, q.MaxDataPoints) defaultTimeField := e.client.GetTimeField() - b := ms.Search(interval) + b := ms.Search(q.Interval) b.Size(0) filters := b.Query().Bool().Filter() filters.AddDateRangeFilter(e.client.GetTimeField(), to, from, es.DateFormatEpochMS) diff --git a/pkg/tsdb/elasticsearch/time_series_query_test.go b/pkg/tsdb/elasticsearch/time_series_query_test.go index 287bcfe0882..fe9cd710c4b 100644 --- a/pkg/tsdb/elasticsearch/time_series_query_test.go +++ b/pkg/tsdb/elasticsearch/time_series_query_test.go @@ -7,7 +7,6 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend" es "github.com/grafana/grafana/pkg/tsdb/elasticsearch/client" - "github.com/grafana/grafana/pkg/tsdb/intervalv2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -25,7 +24,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "timeField": "@timestamp", "bucketAggs": [{ "type": "date_histogram", "field": "@timestamp", "id": "2" }], "metrics": [{"type": "count", "id": "0" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] rangeFilter := sr.Query.Bool.Filters[0].(*es.RangeFilter) @@ -45,7 +44,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "timeField": "@timestamp", "bucketAggs": [{ "type": "date_histogram", "field": "@timestamp", "id": "1" }], "metrics": [{"type": "avg", "id": "0", "settings": {"missing": "null", "script": "1" } }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -63,7 +62,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { { "type": "date_histogram", "field": "@timestamp", "id": "3" } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -84,7 +83,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { { "type": "date_histogram", "field": "@timestamp", "id": "2" } ], "metrics": [{"type": "avg", "field": "@value", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -113,7 +112,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { {"type": "count", "id": "1" }, {"type": "avg", "field": "@value", "id": "5" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -137,7 +136,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { {"type": "count", "id": "1" }, {"type": "avg", "field": "@value", "id": "5" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -168,7 +167,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "metrics": [ {"type": "count", "id": "1" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -193,7 +192,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { }, { "type": "date_histogram", "field": "@timestamp", "id": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -219,7 +218,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "metrics": [ {"type": "percentiles", "field": "@value", "id": "1", "settings": { "percents": ["95","99"] } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -248,7 +247,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "metrics": [ {"type": "extended_stats", "field": "@value", "id": "1", "meta": { "std_deviation": true } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -279,7 +278,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { {"type": "count", "id": "1" }, {"type": "avg", "field": "@value", "id": "5" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -305,7 +304,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "metrics": [ {"type": "count", "id": "1" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -332,7 +331,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -364,7 +363,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { { "type": "date_histogram", "field": "@timestamp", "id": "4" } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -395,7 +394,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { { "type": "date_histogram", "field": "@timestamp", "id": "4" } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -417,7 +416,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "timeField": "@timestamp", "bucketAggs": [], "metrics": [{ "id": "1", "type": "raw_document", "settings": {} }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -430,7 +429,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "timeField": "@timestamp", "bucketAggs": [], "metrics": [{ "id": "1", "type": "raw_document", "settings": {} }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -452,7 +451,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "timeField": "@timestamp", "bucketAggs": [], "metrics": [{ "id": "1", "type": "raw_data", "settings": {} }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -474,7 +473,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "timeField": "@timestamp", "bucketAggs": [], "metrics": [{ "id": "1", "type": "raw_document", "settings": { "size": 1337 } }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -494,7 +493,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -521,7 +520,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -544,7 +543,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -567,7 +566,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -589,7 +588,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -616,7 +615,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -642,7 +641,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { } ], "metrics": [{"type": "count", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -669,7 +668,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "field": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -705,7 +704,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -742,7 +741,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "field": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -774,7 +773,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -809,7 +808,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "type": "moving_avg" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -846,7 +845,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "Metric to apply moving average" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -872,7 +871,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "metrics": [ { "id": "2", "type": "top_metrics", "settings": { "order": "desc", "orderBy": "@timestamp", "metrics": ["@value"]} } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] firstLevel := sr.Aggs[0] @@ -902,7 +901,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -940,7 +939,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -976,7 +975,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "Metric to apply cumulative sum" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1007,7 +1006,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1037,7 +1036,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1066,7 +1065,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "field": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1097,7 +1096,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "settings": { "lag": "5" } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1128,7 +1127,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "settings": { "lag": "5" } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1157,7 +1156,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "pipelineAgg": "3" } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1191,7 +1190,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "settings": { "script": "params.var1 * params.var2" } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1228,7 +1227,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "settings": { "script": "params.var1 * params.var2" } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1263,7 +1262,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "settings": { "script": "params.var1 * 1000" } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1297,7 +1296,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { "settings": { "script": "params.var1 * 1000" } } ] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1318,7 +1317,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { _, err := executeTsdbQuery(c, `{ "timeField": "@timestamp", "query": "foo" - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] filter := sr.Query.Bool.Filters[1].(*es.QueryStringFilter) @@ -1331,7 +1330,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { _, err := executeTsdbQuery(c, `{ "timeField": "@timestamp", "query": "foo" - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] filter := sr.Query.Bool.Filters[1].(*es.QueryStringFilter) @@ -1344,7 +1343,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { _, err := executeTsdbQuery(c, `{ "timeField": "@timestamp", "metrics": [{ "type": "logs", "id": "1"}] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] require.Equal(t, sr.Size, defaultSize) @@ -1377,7 +1376,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { _, err := executeTsdbQuery(c, `{ "timeField": "@timestamp", "metrics": [{ "type": "logs", "id": "1", "settings": { "limit": 1000 }}] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] require.Equal(t, sr.Size, 1000) @@ -1388,7 +1387,7 @@ func TestExecuteTimeSeriesQuery(t *testing.T) { _, err := executeTsdbQuery(c, `{ "timeField": "@timestamp", "metrics": [{ "type": "logs", "id": "1" }] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] require.Equal(t, sr.CustomProps["highlight"], map[string]interface{}{ @@ -1431,7 +1430,7 @@ func TestSettingsCasting(t *testing.T) { } ], "bucketAggs": [{"type": "date_histogram", "field": "@timestamp", "id": "1"}] - }`, from, to, 15*time.Second) + }`, from, to) require.NoError(t, err) sr := c.multisearchRequests[0].Requests[0] movingAvgSettings := sr.Aggs[0].Aggregation.Aggs[1].Aggregation.Aggregation.(*es.PipelineAggregation).Settings @@ -1475,7 +1474,7 @@ func TestSettingsCasting(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1510,7 +1509,7 @@ func TestSettingsCasting(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] serialDiffSettings := sr.Aggs[0].Aggregation.Aggs[1].Aggregation.Aggregation.(*es.PipelineAggregation).Settings @@ -1537,7 +1536,7 @@ func TestSettingsCasting(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1573,7 +1572,7 @@ func TestSettingsCasting(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1608,7 +1607,7 @@ func TestSettingsCasting(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1635,7 +1634,7 @@ func TestSettingsCasting(t *testing.T) { "metrics": [ { "id": "1", "type": "average", "field": "@value" } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1672,7 +1671,7 @@ func TestSettingsCasting(t *testing.T) { } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1694,7 +1693,7 @@ func TestSettingsCasting(t *testing.T) { "bucketAggs": [ { "type": "date_histogram", "id": "2", "settings": { "min_doc_count": "1" } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1710,7 +1709,7 @@ func TestSettingsCasting(t *testing.T) { "bucketAggs": [ { "type": "date_histogram", "id": "2", "field": "@time", "settings": { "min_doc_count": "1" } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1726,7 +1725,7 @@ func TestSettingsCasting(t *testing.T) { "bucketAggs": [ { "type": "date_histogram", "id": "2", "field": "@time", "settings": { "min_doc_count": "1", "interval": "1d" } } ] - }`, from, to, 15*time.Second) + }`, from, to) assert.Nil(t, err) sr := c.multisearchRequests[0].Requests[0] @@ -1774,13 +1773,14 @@ func newDataQuery(body string) (backend.QueryDataRequest, error) { return backend.QueryDataRequest{ Queries: []backend.DataQuery{ { - JSON: json.RawMessage(body), + JSON: json.RawMessage(body), + Interval: 10 * time.Second, }, }, }, nil } -func executeTsdbQuery(c es.Client, body string, from, to time.Time, minInterval time.Duration) ( +func executeTsdbQuery(c es.Client, body string, from, to time.Time) ( *backend.QueryDataResponse, error) { timeRange := backend.TimeRange{ From: from, @@ -1794,6 +1794,6 @@ func executeTsdbQuery(c es.Client, body string, from, to time.Time, minInterval }, }, } - query := newTimeSeriesQuery(c, dataRequest.Queries, intervalv2.NewCalculator(intervalv2.CalculatorOptions{MinInterval: minInterval})) + query := newTimeSeriesQuery(c, dataRequest.Queries) return query.execute() }