InfluxDB: Fix sending retention policy with the backend request (#72763)
* Add retention policy to the request * refactor * refactor influxql query flow * fix healthcheck * organize imports * handle queries separately * fix tests * update bench test
This commit is contained in:
ismail simsek
@@ -0,0 +1,178 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
)
|
||||
|
||||
type InfluxdbQueryParser struct{}
|
||||
|
||||
func QueryParse(query backend.DataQuery) (*Query, error) {
|
||||
model, err := simplejson.NewJson(query.JSON)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("couldn't unmarshal query")
|
||||
}
|
||||
|
||||
policy := model.Get("policy").MustString("default")
|
||||
rawQuery := model.Get("query").MustString("")
|
||||
useRawQuery := model.Get("rawQuery").MustBool(false)
|
||||
alias := model.Get("alias").MustString("")
|
||||
tz := model.Get("tz").MustString("")
|
||||
limit := model.Get("limit").MustString("")
|
||||
slimit := model.Get("slimit").MustString("")
|
||||
orderByTime := model.Get("orderByTime").MustString("")
|
||||
|
||||
measurement := model.Get("measurement").MustString("")
|
||||
|
||||
tags, err := parseTags(model)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
groupBys, err := parseGroupBy(model)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
selects, err := parseSelects(model)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
interval := query.Interval
|
||||
|
||||
// we make sure it is at least 1 millisecond
|
||||
minInterval := time.Millisecond
|
||||
|
||||
if interval < minInterval {
|
||||
interval = minInterval
|
||||
}
|
||||
|
||||
return &Query{
|
||||
Measurement: measurement,
|
||||
Policy: policy,
|
||||
GroupBy: groupBys,
|
||||
Tags: tags,
|
||||
Selects: selects,
|
||||
RawQuery: rawQuery,
|
||||
Interval: interval,
|
||||
Alias: alias,
|
||||
UseRawQuery: useRawQuery,
|
||||
Tz: tz,
|
||||
Limit: limit,
|
||||
Slimit: slimit,
|
||||
OrderByTime: orderByTime,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func parseSelects(model *simplejson.Json) ([]*Select, error) {
|
||||
selectObjs := model.Get("select").MustArray()
|
||||
result := make([]*Select, 0, len(selectObjs))
|
||||
|
||||
for _, selectObj := range selectObjs {
|
||||
selectJson := simplejson.NewFromAny(selectObj)
|
||||
var parts Select
|
||||
|
||||
for _, partObj := range selectJson.MustArray() {
|
||||
part := simplejson.NewFromAny(partObj)
|
||||
queryPart, err := parseQueryPart(part)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
parts = append(parts, *queryPart)
|
||||
}
|
||||
|
||||
result = append(result, &parts)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func parseTags(model *simplejson.Json) ([]*Tag, error) {
|
||||
tags := model.Get("tags").MustArray()
|
||||
result := make([]*Tag, 0, len(tags))
|
||||
for _, t := range tags {
|
||||
tagJson := simplejson.NewFromAny(t)
|
||||
tag := &Tag{}
|
||||
var err error
|
||||
|
||||
tag.Key, err = tagJson.Get("key").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tag.Value, err = tagJson.Get("value").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
operator, err := tagJson.Get("operator").String()
|
||||
if err == nil {
|
||||
tag.Operator = operator
|
||||
}
|
||||
|
||||
condition, err := tagJson.Get("condition").String()
|
||||
if err == nil {
|
||||
tag.Condition = condition
|
||||
}
|
||||
|
||||
result = append(result, tag)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func parseQueryPart(model *simplejson.Json) (*QueryPart, error) {
|
||||
typ, err := model.Get("type").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var params []string
|
||||
for _, paramObj := range model.Get("params").MustArray() {
|
||||
param := simplejson.NewFromAny(paramObj)
|
||||
|
||||
stringParam, err := param.String()
|
||||
if err == nil {
|
||||
params = append(params, stringParam)
|
||||
continue
|
||||
}
|
||||
|
||||
intParam, err := param.Int()
|
||||
if err == nil {
|
||||
params = append(params, strconv.Itoa(intParam))
|
||||
continue
|
||||
}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
qp, err := NewQueryPart(typ, params)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return qp, nil
|
||||
}
|
||||
|
||||
func parseGroupBy(model *simplejson.Json) ([]*QueryPart, error) {
|
||||
groupBy := model.Get("groupBy").MustArray()
|
||||
result := make([]*QueryPart, 0, len(groupBy))
|
||||
for _, groupObj := range groupBy {
|
||||
groupJson := simplejson.NewFromAny(groupObj)
|
||||
queryPart, err := parseQueryPart(groupJson)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result = append(result, queryPart)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
@@ -0,0 +1,201 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestInfluxdbQueryParser_Parse(t *testing.T) {
|
||||
t.Run("can parse influxdb json model", func(t *testing.T) {
|
||||
json := `
|
||||
{
|
||||
"groupBy": [
|
||||
{
|
||||
"params": [
|
||||
"$interval"
|
||||
],
|
||||
"type": "time"
|
||||
},
|
||||
{
|
||||
"params": [
|
||||
"datacenter"
|
||||
],
|
||||
"type": "tag"
|
||||
},
|
||||
{
|
||||
"params": [
|
||||
"none"
|
||||
],
|
||||
"type": "fill"
|
||||
}
|
||||
],
|
||||
"measurement": "logins.count",
|
||||
"tz": "Europe/Paris",
|
||||
"limit": "1",
|
||||
"slimit": "1",
|
||||
"orderByTime": "ASC",
|
||||
"policy": "default",
|
||||
"refId": "B",
|
||||
"resultFormat": "time_series",
|
||||
"select": [
|
||||
[
|
||||
{
|
||||
"type": "field",
|
||||
"params": [
|
||||
"value"
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "count",
|
||||
"params": []
|
||||
}
|
||||
],
|
||||
[
|
||||
{
|
||||
"type": "field",
|
||||
"params": [
|
||||
"value"
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "bottom",
|
||||
"params": [
|
||||
3
|
||||
]
|
||||
}
|
||||
],
|
||||
[
|
||||
{
|
||||
"type": "field",
|
||||
"params": [
|
||||
"value"
|
||||
]
|
||||
},
|
||||
{
|
||||
"type": "mean",
|
||||
"params": []
|
||||
},
|
||||
{
|
||||
"type": "math",
|
||||
"params": [
|
||||
" / 100"
|
||||
]
|
||||
}
|
||||
]
|
||||
],
|
||||
"alias": "series alias",
|
||||
"tags": [
|
||||
{
|
||||
"key": "datacenter",
|
||||
"operator": "=",
|
||||
"value": "America"
|
||||
},
|
||||
{
|
||||
"condition": "OR",
|
||||
"key": "hostname",
|
||||
"operator": "=",
|
||||
"value": "server1"
|
||||
}
|
||||
]
|
||||
}
|
||||
`
|
||||
|
||||
query := backend.DataQuery{
|
||||
JSON: []byte(json),
|
||||
Interval: time.Second * 20,
|
||||
}
|
||||
|
||||
res, err := QueryParse(query)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, res.GroupBy, 3)
|
||||
require.Len(t, res.Selects, 3)
|
||||
require.Len(t, res.Tags, 2)
|
||||
require.Equal(t, "Europe/Paris", res.Tz)
|
||||
require.Equal(t, "1", res.Limit)
|
||||
require.Equal(t, "1", res.Slimit)
|
||||
require.Equal(t, "ASC", res.OrderByTime)
|
||||
require.Equal(t, time.Second*20, res.Interval)
|
||||
require.Equal(t, "series alias", res.Alias)
|
||||
})
|
||||
|
||||
t.Run("can parse raw query json model", func(t *testing.T) {
|
||||
json := `
|
||||
{
|
||||
"groupBy": [
|
||||
{
|
||||
"params": [
|
||||
"$interval"
|
||||
],
|
||||
"type": "time"
|
||||
},
|
||||
{
|
||||
"params": [
|
||||
"null"
|
||||
],
|
||||
"type": "fill"
|
||||
}
|
||||
],
|
||||
"interval": ">10s",
|
||||
"policy": "default",
|
||||
"query": "RawDummyQuery",
|
||||
"rawQuery": true,
|
||||
"refId": "A",
|
||||
"resultFormat": "time_series",
|
||||
"select": [
|
||||
[
|
||||
{
|
||||
"params": [
|
||||
"value"
|
||||
],
|
||||
"type": "field"
|
||||
},
|
||||
{
|
||||
"params": [
|
||||
|
||||
],
|
||||
"type": "mean"
|
||||
}
|
||||
]
|
||||
],
|
||||
"tags": [
|
||||
|
||||
]
|
||||
}
|
||||
`
|
||||
|
||||
query := backend.DataQuery{
|
||||
JSON: []byte(json),
|
||||
Interval: time.Second * 10,
|
||||
}
|
||||
|
||||
res, err := QueryParse(query)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "RawDummyQuery", res.RawQuery)
|
||||
require.Len(t, res.GroupBy, 2)
|
||||
require.Len(t, res.Selects, 1)
|
||||
require.Empty(t, res.Tags)
|
||||
require.Equal(t, time.Second*10, res.Interval)
|
||||
})
|
||||
|
||||
t.Run("will enforce a minInterval of 1 millisecond", func(t *testing.T) {
|
||||
json := `
|
||||
{
|
||||
"query": "RawDummyQuery",
|
||||
"rawQuery": true,
|
||||
"resultFormat": "time_series"
|
||||
}
|
||||
`
|
||||
|
||||
query := backend.DataQuery{
|
||||
JSON: []byte(json),
|
||||
Interval: time.Millisecond * 0,
|
||||
}
|
||||
|
||||
res, err := QueryParse(query)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, time.Millisecond*1, res.Interval)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package models
|
||||
|
||||
import "time"
|
||||
|
||||
type Query struct {
|
||||
Measurement string
|
||||
Policy string
|
||||
Tags []*Tag
|
||||
GroupBy []*QueryPart
|
||||
Selects []*Select
|
||||
RawQuery string
|
||||
UseRawQuery bool
|
||||
Alias string
|
||||
Interval time.Duration
|
||||
Tz string
|
||||
Limit string
|
||||
Slimit string
|
||||
OrderByTime string
|
||||
RefID string
|
||||
}
|
||||
|
||||
type Tag struct {
|
||||
Key string
|
||||
Operator string
|
||||
Value string
|
||||
Condition string
|
||||
}
|
||||
|
||||
type Select []QueryPart
|
||||
|
||||
type Response struct {
|
||||
Results []Result
|
||||
Error string
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
Series []Row
|
||||
Messages []*Message
|
||||
Error string
|
||||
}
|
||||
|
||||
type Message struct {
|
||||
Level string `json:"level,omitempty"`
|
||||
Text string `json:"text,omitempty"`
|
||||
}
|
||||
|
||||
type Row struct {
|
||||
Name string `json:"name,omitempty"`
|
||||
Tags map[string]string `json:"tags,omitempty"`
|
||||
Columns []string `json:"columns,omitempty"`
|
||||
Values [][]interface{} `json:"values,omitempty"`
|
||||
}
|
||||
@@ -0,0 +1,205 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
|
||||
"github.com/grafana/grafana/pkg/tsdb/intervalv2"
|
||||
)
|
||||
|
||||
var (
|
||||
regexpOperatorPattern = regexp.MustCompile(`^\/.*\/$`)
|
||||
regexpMeasurementPattern = regexp.MustCompile(`^\/.*\/$`)
|
||||
)
|
||||
|
||||
func (query *Query) Build(queryContext *backend.QueryDataRequest) (string, error) {
|
||||
var res string
|
||||
if query.UseRawQuery && query.RawQuery != "" {
|
||||
res = query.RawQuery
|
||||
} else {
|
||||
res = query.renderSelectors(queryContext)
|
||||
res += query.renderMeasurement()
|
||||
res += query.renderWhereClause()
|
||||
res += query.renderTimeFilter(queryContext)
|
||||
res += query.renderGroupBy(queryContext)
|
||||
res += query.renderOrderByTime()
|
||||
res += query.renderLimit()
|
||||
res += query.renderSlimit()
|
||||
res += query.renderTz()
|
||||
}
|
||||
|
||||
intervalText := intervalv2.FormatDuration(query.Interval)
|
||||
intervalMs := int64(query.Interval / time.Millisecond)
|
||||
|
||||
res = strings.ReplaceAll(res, "$timeFilter", query.renderTimeFilter(queryContext))
|
||||
res = strings.ReplaceAll(res, "$interval", intervalText)
|
||||
res = strings.ReplaceAll(res, "$__interval_ms", strconv.FormatInt(intervalMs, 10))
|
||||
res = strings.ReplaceAll(res, "$__interval", intervalText)
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (query *Query) renderTags() []string {
|
||||
res := make([]string, 0, len(query.Tags))
|
||||
for i, tag := range query.Tags {
|
||||
str := ""
|
||||
|
||||
if i > 0 {
|
||||
if tag.Condition == "" {
|
||||
str += "AND"
|
||||
} else {
|
||||
str += tag.Condition
|
||||
}
|
||||
str += " "
|
||||
}
|
||||
|
||||
// If the operator is missing we fall back to sensible defaults
|
||||
if tag.Operator == "" {
|
||||
if regexpOperatorPattern.MatchString(tag.Value) {
|
||||
tag.Operator = "=~"
|
||||
} else {
|
||||
tag.Operator = "="
|
||||
}
|
||||
}
|
||||
|
||||
// quote value unless regex or number
|
||||
var textValue string
|
||||
switch tag.Operator {
|
||||
case "=~", "!~":
|
||||
textValue = tag.Value
|
||||
case "<", ">":
|
||||
textValue = tag.Value
|
||||
default:
|
||||
textValue = fmt.Sprintf("'%s'", strings.ReplaceAll(tag.Value, `\`, `\\`))
|
||||
}
|
||||
|
||||
escapedKey := fmt.Sprintf(`"%s"`, tag.Key)
|
||||
|
||||
if strings.HasSuffix(tag.Key, "::tag") {
|
||||
escapedKey = fmt.Sprintf(`"%s"::tag`, strings.TrimSuffix(tag.Key, "::tag"))
|
||||
}
|
||||
|
||||
if strings.HasSuffix(tag.Key, "::field") {
|
||||
escapedKey = fmt.Sprintf(`"%s"::field`, strings.TrimSuffix(tag.Key, "::field"))
|
||||
}
|
||||
|
||||
res = append(res, fmt.Sprintf(`%s%s %s %s`, str, escapedKey, tag.Operator, textValue))
|
||||
}
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
func (query *Query) renderTimeFilter(queryContext *backend.QueryDataRequest) string {
|
||||
from, to := epochMStoInfluxTime(&queryContext.Queries[0].TimeRange)
|
||||
return fmt.Sprintf("time >= %s and time <= %s", from, to)
|
||||
}
|
||||
|
||||
func (query *Query) renderSelectors(queryContext *backend.QueryDataRequest) string {
|
||||
res := "SELECT "
|
||||
|
||||
selectors := make([]string, 0, len(query.Selects))
|
||||
for _, sel := range query.Selects {
|
||||
stk := ""
|
||||
for _, s := range *sel {
|
||||
stk = s.Render(query, queryContext, stk)
|
||||
}
|
||||
selectors = append(selectors, stk)
|
||||
}
|
||||
|
||||
return res + strings.Join(selectors, ", ")
|
||||
}
|
||||
|
||||
func (query *Query) renderMeasurement() string {
|
||||
var policy string
|
||||
if query.Policy == "" || query.Policy == "default" {
|
||||
policy = ""
|
||||
} else {
|
||||
policy = `"` + query.Policy + `".`
|
||||
}
|
||||
|
||||
measurement := query.Measurement
|
||||
|
||||
if !regexpMeasurementPattern.MatchString(measurement) {
|
||||
measurement = fmt.Sprintf(`"%s"`, measurement)
|
||||
}
|
||||
|
||||
return fmt.Sprintf(` FROM %s%s`, policy, measurement)
|
||||
}
|
||||
|
||||
func (query *Query) renderWhereClause() string {
|
||||
res := " WHERE "
|
||||
conditions := query.renderTags()
|
||||
if len(conditions) > 0 {
|
||||
if len(conditions) > 1 {
|
||||
res += "(" + strings.Join(conditions, " ") + ")"
|
||||
} else {
|
||||
res += conditions[0]
|
||||
}
|
||||
res += " AND "
|
||||
}
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
func (query *Query) renderGroupBy(queryContext *backend.QueryDataRequest) string {
|
||||
groupBy := ""
|
||||
for i, group := range query.GroupBy {
|
||||
if i == 0 {
|
||||
groupBy += " GROUP BY"
|
||||
}
|
||||
|
||||
if i > 0 && group.Type != "fill" {
|
||||
groupBy += ", " // fill is so very special. fill is a creep, fill is a weirdo
|
||||
} else {
|
||||
groupBy += " "
|
||||
}
|
||||
|
||||
groupBy += group.Render(query, queryContext, "")
|
||||
}
|
||||
|
||||
return groupBy
|
||||
}
|
||||
|
||||
func (query *Query) renderOrderByTime() string {
|
||||
orderByTime := query.OrderByTime
|
||||
if orderByTime == "" {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf(" ORDER BY time %s", orderByTime)
|
||||
}
|
||||
|
||||
func (query *Query) renderTz() string {
|
||||
tz := query.Tz
|
||||
if tz == "" {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf(" tz('%s')", tz)
|
||||
}
|
||||
|
||||
func (query *Query) renderLimit() string {
|
||||
limit := query.Limit
|
||||
if limit == "" {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf(" limit %s", limit)
|
||||
}
|
||||
|
||||
func (query *Query) renderSlimit() string {
|
||||
slimit := query.Slimit
|
||||
if slimit == "" {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf(" slimit %s", slimit)
|
||||
}
|
||||
|
||||
func epochMStoInfluxTime(tr *backend.TimeRange) (string, string) {
|
||||
from := tr.From.UnixNano() / int64(time.Millisecond)
|
||||
to := tr.To.UnixNano() / int64(time.Millisecond)
|
||||
|
||||
return fmt.Sprintf("%dms", from), fmt.Sprintf("%dms", to)
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
)
|
||||
|
||||
var renders map[string]QueryDefinition
|
||||
|
||||
type DefinitionParameters struct {
|
||||
Name string
|
||||
Type string
|
||||
}
|
||||
|
||||
type QueryDefinition struct {
|
||||
Renderer func(query *Query, queryContext *backend.QueryDataRequest, part *QueryPart, innerExpr string) string
|
||||
Params []DefinitionParameters
|
||||
}
|
||||
|
||||
func init() {
|
||||
renders = make(map[string]QueryDefinition)
|
||||
|
||||
renders["field"] = QueryDefinition{Renderer: fieldRenderer}
|
||||
|
||||
renders["spread"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["count"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["distinct"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["integral"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["mean"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["median"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["sum"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["mode"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["cumulative_sum"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["non_negative_difference"] = QueryDefinition{Renderer: functionRenderer}
|
||||
|
||||
renders["holt_winters"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "number", Type: "number"}, {Name: "season", Type: "number"}},
|
||||
}
|
||||
renders["holt_winters_with_fit"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "number", Type: "number"}, {Name: "season", Type: "number"}},
|
||||
}
|
||||
|
||||
renders["derivative"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "duration", Type: "interval"}},
|
||||
}
|
||||
|
||||
renders["non_negative_derivative"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "duration", Type: "interval"}},
|
||||
}
|
||||
renders["difference"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["moving_average"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "window", Type: "number"}},
|
||||
}
|
||||
renders["stddev"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["time"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "interval", Type: "time"}, {Name: "offset", Type: "time"}},
|
||||
}
|
||||
renders["fill"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "fill", Type: "string"}},
|
||||
}
|
||||
renders["elapsed"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "duration", Type: "interval"}},
|
||||
}
|
||||
renders["bottom"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "count", Type: "int"}},
|
||||
}
|
||||
|
||||
renders["first"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["last"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["max"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["min"] = QueryDefinition{Renderer: functionRenderer}
|
||||
renders["percentile"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "nth", Type: "int"}},
|
||||
}
|
||||
renders["top"] = QueryDefinition{
|
||||
Renderer: functionRenderer,
|
||||
Params: []DefinitionParameters{{Name: "count", Type: "int"}},
|
||||
}
|
||||
renders["tag"] = QueryDefinition{
|
||||
Renderer: fieldRenderer,
|
||||
Params: []DefinitionParameters{{Name: "tag", Type: "string"}},
|
||||
}
|
||||
|
||||
renders["math"] = QueryDefinition{Renderer: suffixRenderer}
|
||||
renders["alias"] = QueryDefinition{Renderer: aliasRenderer}
|
||||
}
|
||||
|
||||
func fieldRenderer(query *Query, queryContext *backend.QueryDataRequest, part *QueryPart, innerExpr string) string {
|
||||
param := part.Params[0]
|
||||
|
||||
if param == "*" {
|
||||
return "*"
|
||||
}
|
||||
|
||||
escapedParam := fmt.Sprintf(`"%s"`, param)
|
||||
|
||||
if strings.HasSuffix(param, "::tag") {
|
||||
escapedParam = fmt.Sprintf(`"%s"::tag`, strings.TrimSuffix(param, "::tag"))
|
||||
}
|
||||
|
||||
if strings.HasSuffix(param, "::field") {
|
||||
escapedParam = fmt.Sprintf(`"%s"::field`, strings.TrimSuffix(param, "::field"))
|
||||
}
|
||||
|
||||
return escapedParam
|
||||
}
|
||||
|
||||
func functionRenderer(query *Query, queryContext *backend.QueryDataRequest, part *QueryPart, innerExpr string) string {
|
||||
for i, param := range part.Params {
|
||||
if part.Type == "time" && param == "auto" {
|
||||
part.Params[i] = "$__interval"
|
||||
}
|
||||
}
|
||||
|
||||
if innerExpr != "" {
|
||||
part.Params = append([]string{innerExpr}, part.Params...)
|
||||
}
|
||||
|
||||
params := strings.Join(part.Params, ", ")
|
||||
|
||||
return fmt.Sprintf("%s(%s)", part.Type, params)
|
||||
}
|
||||
|
||||
func suffixRenderer(query *Query, queryContext *backend.QueryDataRequest, part *QueryPart, innerExpr string) string {
|
||||
return fmt.Sprintf("%s %s", innerExpr, part.Params[0])
|
||||
}
|
||||
|
||||
func aliasRenderer(query *Query, queryContext *backend.QueryDataRequest, part *QueryPart, innerExpr string) string {
|
||||
return fmt.Sprintf(`%s AS "%s"`, innerExpr, part.Params[0])
|
||||
}
|
||||
|
||||
func NewQueryPart(typ string, params []string) (*QueryPart, error) {
|
||||
def, exist := renders[typ]
|
||||
if !exist {
|
||||
return nil, fmt.Errorf("missing query definition for %q", typ)
|
||||
}
|
||||
|
||||
return &QueryPart{
|
||||
Type: typ,
|
||||
Params: params,
|
||||
Def: def,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type QueryPart struct {
|
||||
Def QueryDefinition
|
||||
Type string
|
||||
Params []string
|
||||
}
|
||||
|
||||
func (qp *QueryPart) Render(query *Query, queryContext *backend.QueryDataRequest, expr string) string {
|
||||
return qp.Def.Renderer(query, queryContext, qp, expr)
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
)
|
||||
|
||||
func TestInfluxdbQueryPart(t *testing.T) {
|
||||
tcs := []struct {
|
||||
mode string
|
||||
input string
|
||||
params []string
|
||||
expected string
|
||||
}{
|
||||
{mode: "field", params: []string{"value"}, input: "value", expected: `"value"`},
|
||||
{mode: "derivative", params: []string{"10s"}, input: "mean(value)", expected: `derivative(mean(value), 10s)`},
|
||||
{mode: "bottom", params: []string{"3"}, input: "value", expected: `bottom(value, 3)`},
|
||||
{mode: "time", params: []string{"$interval"}, input: "", expected: `time($interval)`},
|
||||
{mode: "time", params: []string{"auto"}, input: "", expected: `time($__interval)`},
|
||||
{mode: "spread", params: []string{}, input: "value", expected: `spread(value)`},
|
||||
{mode: "math", params: []string{"/ 100"}, input: "mean(value)", expected: `mean(value) / 100`},
|
||||
{mode: "alias", params: []string{"test"}, input: "mean(value)", expected: `mean(value) AS "test"`},
|
||||
{mode: "count", params: []string{}, input: "distinct(value)", expected: `count(distinct(value))`},
|
||||
{mode: "mode", params: []string{}, input: "value", expected: `mode(value)`},
|
||||
{mode: "cumulative_sum", params: []string{}, input: "mean(value)", expected: `cumulative_sum(mean(value))`},
|
||||
{mode: "non_negative_difference", params: []string{}, input: "max(value)", expected: `non_negative_difference(max(value))`},
|
||||
}
|
||||
|
||||
queryContext := &backend.QueryDataRequest{}
|
||||
query := &Query{}
|
||||
|
||||
for _, tc := range tcs {
|
||||
part, err := NewQueryPart(tc.mode, tc.params)
|
||||
if err != nil {
|
||||
t.Errorf("Expected NewQueryPart to not return an error. error: %v", err)
|
||||
}
|
||||
|
||||
res := part.Render(query, queryContext, tc.input)
|
||||
if res != tc.expected {
|
||||
t.Errorf("expected %v to render into %s", tc, tc.expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,255 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestInfluxdbQueryBuilder(t *testing.T) {
|
||||
t.Run("Influxdb query builder", func(t *testing.T) {
|
||||
qp1, _ := NewQueryPart("field", []string{"value"})
|
||||
qp2, _ := NewQueryPart("mean", []string{})
|
||||
|
||||
mathPartDivideBy100, _ := NewQueryPart("math", []string{"/ 100"})
|
||||
mathPartDivideByIntervalMs, _ := NewQueryPart("math", []string{"/ $__interval_ms"})
|
||||
|
||||
groupBy1, _ := NewQueryPart("time", []string{"$__interval"})
|
||||
groupBy2, _ := NewQueryPart("tag", []string{"datacenter"})
|
||||
groupBy3, _ := NewQueryPart("fill", []string{"null"})
|
||||
|
||||
groupByOldInterval, _ := NewQueryPart("time", []string{"$interval"})
|
||||
|
||||
tag1 := &Tag{Key: "hostname", Value: "server1", Operator: "="}
|
||||
tag2 := &Tag{Key: "hostname", Value: "server2", Operator: "=", Condition: "OR"}
|
||||
|
||||
timeRange := backend.TimeRange{
|
||||
From: time.Date(2020, 8, 1, 0, 0, 0, 0, time.UTC),
|
||||
To: time.Date(2020, 8, 1, 0, 5, 0, 0, time.UTC),
|
||||
}
|
||||
queryContext := &backend.QueryDataRequest{
|
||||
Queries: []backend.DataQuery{
|
||||
{
|
||||
TimeRange: timeRange,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
t.Run("can build simple query", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2}},
|
||||
Measurement: "cpu",
|
||||
Policy: "policy",
|
||||
GroupBy: []*QueryPart{groupBy1, groupBy3},
|
||||
Interval: time.Second * 10,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery, `SELECT mean("value") FROM "policy"."cpu" WHERE time >= 1596240000000ms and time <= 1596240300000ms GROUP BY time(10s) fill(null)`)
|
||||
})
|
||||
|
||||
t.Run("can build query with tz", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2}},
|
||||
Measurement: "cpu",
|
||||
GroupBy: []*QueryPart{groupBy1},
|
||||
Tz: "Europe/Paris",
|
||||
Interval: time.Second * 5,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery,
|
||||
`SELECT mean("value") FROM "cpu" WHERE time >= 1596240000000ms and time <= 1596240300000ms GROUP BY time(5s) tz('Europe/Paris')`)
|
||||
})
|
||||
|
||||
t.Run("can build query with tz, limit, slimit, orderByTime and puts them in the correct order", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2}},
|
||||
Measurement: "cpu",
|
||||
GroupBy: []*QueryPart{groupBy1},
|
||||
Tz: "Europe/Paris",
|
||||
Limit: "1",
|
||||
Slimit: "1",
|
||||
OrderByTime: "ASC",
|
||||
Interval: time.Second * 5,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery,
|
||||
`SELECT mean("value") FROM "cpu" WHERE time >= 1596240000000ms and time <= 1596240300000ms GROUP BY time(5s) ORDER BY time ASC limit 1 slimit 1 tz('Europe/Paris')`)
|
||||
})
|
||||
|
||||
t.Run("can build query with group bys", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2}},
|
||||
Measurement: "cpu",
|
||||
GroupBy: []*QueryPart{groupBy1, groupBy2, groupBy3},
|
||||
Tags: []*Tag{tag1, tag2},
|
||||
Interval: time.Second * 5,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery, `SELECT mean("value") FROM "cpu" WHERE ("hostname" = 'server1' OR "hostname" = 'server2') AND time >= 1596240000000ms and time <= 1596240300000ms GROUP BY time(5s), "datacenter" fill(null)`)
|
||||
})
|
||||
|
||||
t.Run("can build query with math part", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2, *mathPartDivideBy100}},
|
||||
Measurement: "cpu",
|
||||
Interval: time.Second * 5,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery,
|
||||
`SELECT mean("value") / 100 FROM "cpu" WHERE time >= 1596240000000ms and time <= 1596240300000ms`)
|
||||
})
|
||||
|
||||
t.Run("can build query with math part using $__interval_ms variable", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2, *mathPartDivideByIntervalMs}},
|
||||
Measurement: "cpu",
|
||||
Interval: time.Second * 5,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery,
|
||||
`SELECT mean("value") / 5000 FROM "cpu" WHERE time >= 1596240000000ms and time <= 1596240300000ms`)
|
||||
})
|
||||
|
||||
t.Run("can build query with old $interval variable", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2}},
|
||||
Measurement: "cpu",
|
||||
Policy: "",
|
||||
GroupBy: []*QueryPart{groupByOldInterval},
|
||||
Interval: time.Millisecond * 200,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery,
|
||||
`SELECT mean("value") FROM "cpu" WHERE time >= 1596240000000ms and time <= 1596240300000ms GROUP BY time(200ms)`)
|
||||
})
|
||||
|
||||
t.Run("can render time range", func(t *testing.T) {
|
||||
query := Query{}
|
||||
t.Run("render from: 2h to now-1h", func(t *testing.T) {
|
||||
query := Query{}
|
||||
timeRange = backend.TimeRange{
|
||||
From: time.Date(2020, 8, 1, 0, 0, 0, 0, time.UTC),
|
||||
To: time.Date(2020, 8, 1, 1, 0, 0, 0, time.UTC),
|
||||
}
|
||||
queryContext = &backend.QueryDataRequest{
|
||||
Queries: []backend.DataQuery{
|
||||
{
|
||||
TimeRange: timeRange,
|
||||
},
|
||||
},
|
||||
}
|
||||
require.Equal(t, query.renderTimeFilter(queryContext),
|
||||
"time >= 1596240000000ms and time <= 1596243600000ms")
|
||||
})
|
||||
|
||||
t.Run("render from: 10m", func(t *testing.T) {
|
||||
timeRange = backend.TimeRange{
|
||||
From: time.Date(2020, 8, 1, 0, 0, 0, 0, time.UTC),
|
||||
To: time.Date(2020, 8, 1, 0, 10, 0, 0, time.UTC),
|
||||
}
|
||||
queryContext = &backend.QueryDataRequest{
|
||||
Queries: []backend.DataQuery{
|
||||
{
|
||||
TimeRange: timeRange,
|
||||
},
|
||||
},
|
||||
}
|
||||
require.Equal(t, query.renderTimeFilter(queryContext),
|
||||
"time >= 1596240000000ms and time <= 1596240600000ms")
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("can build query from raw query", func(t *testing.T) {
|
||||
query := &Query{
|
||||
Selects: []*Select{{*qp1, *qp2}},
|
||||
Measurement: "cpu",
|
||||
Policy: "policy",
|
||||
GroupBy: []*QueryPart{groupBy1, groupBy3},
|
||||
Interval: time.Second * 10,
|
||||
RawQuery: "Raw query",
|
||||
UseRawQuery: true,
|
||||
}
|
||||
|
||||
rawQuery, err := query.Build(queryContext)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, rawQuery, `Raw query`)
|
||||
})
|
||||
|
||||
t.Run("can render normal tags without operator", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "", Value: `value`, Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" = 'value'`)
|
||||
})
|
||||
|
||||
t.Run("can render regex tags without operator", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "", Value: `/value/`, Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" =~ /value/`)
|
||||
})
|
||||
|
||||
t.Run("can render regex tags", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "=~", Value: `/value/`, Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" =~ /value/`)
|
||||
})
|
||||
|
||||
t.Run("can render number tags", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "=", Value: "10001", Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" = '10001'`)
|
||||
})
|
||||
|
||||
t.Run("can render numbers less then condition tags", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "<", Value: "10001", Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" < 10001`)
|
||||
})
|
||||
|
||||
t.Run("can render number greater then condition tags", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: ">", Value: "10001", Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" > 10001`)
|
||||
})
|
||||
|
||||
t.Run("can render string tags", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "=", Value: "value", Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" = 'value'`)
|
||||
})
|
||||
|
||||
t.Run("can escape backslashes when rendering string tags", func(t *testing.T) {
|
||||
query := &Query{Tags: []*Tag{{Operator: "=", Value: `C:\test\`, Key: "key"}}}
|
||||
|
||||
require.Equal(t, strings.Join(query.renderTags(), ""), `"key" = 'C:\\test\\'`)
|
||||
})
|
||||
|
||||
t.Run("can render regular measurement", func(t *testing.T) {
|
||||
query := &Query{Measurement: `apa`, Policy: "policy"}
|
||||
|
||||
require.Equal(t, query.renderMeasurement(), ` FROM "policy"."apa"`)
|
||||
})
|
||||
|
||||
t.Run("can render regexp measurement", func(t *testing.T) {
|
||||
query := &Query{Measurement: `/apa/`, Policy: "policy"}
|
||||
|
||||
require.Equal(t, query.renderMeasurement(), ` FROM "policy"./apa/`)
|
||||
})
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user