feat(influxdb): add support for serie alias replacement

ref #6510
This commit is contained in:
bergquist
2016-11-09 10:22:11 +01:00
parent 24a3a100ed
commit c897d39d5e
4 changed files with 190 additions and 53 deletions
+15 -12
View File
@@ -11,6 +11,7 @@ import (
"golang.org/x/net/context/ctxhttp" "golang.org/x/net/context/ctxhttp"
"github.com/grafana/grafana/pkg/log" "github.com/grafana/grafana/pkg/log"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/tsdb" "github.com/grafana/grafana/pkg/tsdb"
) )
@@ -50,9 +51,16 @@ func (e *InfluxDBExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice,
return result.WithError(err) return result.WithError(err)
} }
glog.Debug("Influxdb query", "raw query", query) rawQuery, err := e.QueryBuilder.Build(query, context)
if err != nil {
return result.WithError(err)
}
req, err := e.createRequest(query) if setting.Env == setting.DEV {
glog.Debug("Influxdb query", "raw query", query)
}
req, err := e.createRequest(rawQuery)
if err != nil { if err != nil {
return result.WithError(err) return result.WithError(err)
} }
@@ -77,28 +85,23 @@ func (e *InfluxDBExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice,
} }
result.QueryResults = make(map[string]*tsdb.QueryResult) result.QueryResults = make(map[string]*tsdb.QueryResult)
result.QueryResults["A"] = e.ResponseParser.Parse(&response) result.QueryResults["A"] = e.ResponseParser.Parse(&response, query)
return result return result
} }
func (e *InfluxDBExecutor) getQuery(queries tsdb.QuerySlice, context *tsdb.QueryContext) (string, error) { func (e *InfluxDBExecutor) getQuery(queries tsdb.QuerySlice, context *tsdb.QueryContext) (*Query, error) {
for _, v := range queries { for _, v := range queries {
query, err := e.QueryParser.Parse(v.Model, e.DataSourceInfo) query, err := e.QueryParser.Parse(v.Model, e.DataSourceInfo)
if err != nil { if err != nil {
return "", err return nil, err
} }
rawQuery, err := e.QueryBuilder.Build(query, context) return query, nil
if err != nil {
return "", err
}
return rawQuery, nil
} }
return "", fmt.Errorf("query request contains no queries") return nil, fmt.Errorf("query request contains no queries")
} }
func (e *InfluxDBExecutor) createRequest(query string) (*http.Request, error) { func (e *InfluxDBExecutor) createRequest(query string) (*http.Request, error) {
+48 -5
View File
@@ -3,6 +3,7 @@ package influxdb
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"regexp"
"strings" "strings"
"github.com/grafana/grafana/pkg/tsdb" "github.com/grafana/grafana/pkg/tsdb"
@@ -11,17 +12,25 @@ import (
type ResponseParser struct{} type ResponseParser struct{}
func (rp *ResponseParser) Parse(response *Response) *tsdb.QueryResult { var (
legendFormat *regexp.Regexp
)
func init() {
legendFormat = regexp.MustCompile(`\[\[(\w+?)*\]\]*|\$\s*(\w+?)*`)
}
func (rp *ResponseParser) Parse(response *Response, query *Query) *tsdb.QueryResult {
queryRes := tsdb.NewQueryResult() queryRes := tsdb.NewQueryResult()
for _, result := range response.Results { for _, result := range response.Results {
queryRes.Series = append(queryRes.Series, rp.transformRows(result.Series, queryRes)...) queryRes.Series = append(queryRes.Series, rp.transformRows(result.Series, queryRes, query)...)
} }
return queryRes return queryRes
} }
func (rp *ResponseParser) transformRows(rows []Row, queryResult *tsdb.QueryResult) tsdb.TimeSeriesSlice { func (rp *ResponseParser) transformRows(rows []Row, queryResult *tsdb.QueryResult, query *Query) tsdb.TimeSeriesSlice {
var result tsdb.TimeSeriesSlice var result tsdb.TimeSeriesSlice
for _, row := range rows { for _, row := range rows {
@@ -38,7 +47,7 @@ func (rp *ResponseParser) transformRows(rows []Row, queryResult *tsdb.QueryResul
} }
} }
result = append(result, &tsdb.TimeSeries{ result = append(result, &tsdb.TimeSeries{
Name: rp.formatSerieName(row, column), Name: rp.formatSerieName(row, column, query),
Points: points, Points: points,
}) })
} }
@@ -47,7 +56,41 @@ func (rp *ResponseParser) transformRows(rows []Row, queryResult *tsdb.QueryResul
return result return result
} }
func (rp *ResponseParser) formatSerieName(row Row, column string) string { func (rp *ResponseParser) formatSerieName(row Row, column string, query *Query) string {
if query.Alias == "" {
return rp.buildSerieNameFromQuery(row, column)
}
result := legendFormat.ReplaceAllFunc([]byte(query.Alias), func(in []byte) []byte {
aliasFormat := string(in)
aliasFormat = strings.Replace(aliasFormat, "[[", "", 1)
aliasFormat = strings.Replace(aliasFormat, "]]", "", 1)
aliasFormat = strings.Replace(aliasFormat, "$", "", 1)
if aliasFormat == "m" || aliasFormat == "measurement" {
return []byte(query.Measurement)
}
if aliasFormat == "col" {
return []byte(column)
}
if !strings.HasPrefix(aliasFormat, "tag_") {
return in
}
tagKey := strings.Replace(aliasFormat, "tag_", "", 1)
tagValue, exist := row.Tags[tagKey]
if exist {
return []byte(tagValue)
}
return in
})
return string(result)
}
func (rp *ResponseParser) buildSerieNameFromQuery(row Row, column string) string {
var tags []string var tags []string
for k, v := range row.Tags { for k, v := range row.Tags {
+123 -32
View File
@@ -4,56 +4,147 @@ import (
"encoding/json" "encoding/json"
"testing" "testing"
"github.com/grafana/grafana/pkg/setting"
. "github.com/smartystreets/goconvey/convey" . "github.com/smartystreets/goconvey/convey"
) )
func TestInfluxdbResponseParser(t *testing.T) { func TestInfluxdbResponseParser(t *testing.T) {
Convey("Influxdb response parser", t, func() { Convey("Influxdb response parser", t, func() {
Convey("Response parser", func() {
parser := &ResponseParser{}
parser := &ResponseParser{} setting.NewConfigContext(&setting.CommandLineArgs{
HomePath: "../../../",
})
response := &Response{ response := &Response{
Results: []Result{ Results: []Result{
Result{ Result{
Series: []Row{ Series: []Row{
{ {
Name: "cpu", Name: "cpu",
Columns: []string{"time", "mean", "sum"}, Columns: []string{"time", "mean", "sum"},
Tags: map[string]string{"datacenter": "America"}, Tags: map[string]string{"datacenter": "America"},
Values: [][]interface{}{ Values: [][]interface{}{
{json.Number("111"), json.Number("222"), json.Number("333")}, {json.Number("111"), json.Number("222"), json.Number("333")},
{json.Number("111"), json.Number("222"), json.Number("333")}, {json.Number("111"), json.Number("222"), json.Number("333")},
{json.Number("111"), json.Number("null"), json.Number("333")}, {json.Number("111"), json.Number("null"), json.Number("333")},
},
}, },
}, },
}, },
}, },
}, }
}
result := parser.Parse(response) query := &Query{}
Convey("can parse all series", func() { result := parser.Parse(response, query)
So(len(result.Series), ShouldEqual, 2)
Convey("can parse all series", func() {
So(len(result.Series), ShouldEqual, 2)
})
Convey("can parse all points", func() {
So(len(result.Series[0].Points), ShouldEqual, 3)
So(len(result.Series[1].Points), ShouldEqual, 3)
})
Convey("can parse multi row result", func() {
So(result.Series[0].Points[1][0].Float64, ShouldEqual, float64(222))
So(result.Series[1].Points[1][0].Float64, ShouldEqual, float64(333))
})
Convey("can parse null points", func() {
So(result.Series[0].Points[2][0].Valid, ShouldBeFalse)
})
Convey("can format serie names", func() {
So(result.Series[0].Name, ShouldEqual, "cpu.mean { datacenter: America }")
So(result.Series[1].Name, ShouldEqual, "cpu.sum { datacenter: America }")
})
}) })
Convey("can parse all points", func() { Convey("Response parser with alias", func() {
So(len(result.Series[0].Points), ShouldEqual, 3) parser := &ResponseParser{}
So(len(result.Series[1].Points), ShouldEqual, 3)
})
Convey("can parse multi row result", func() { response := &Response{
So(result.Series[0].Points[1][0].Float64, ShouldEqual, float64(222)) Results: []Result{
So(result.Series[1].Points[1][0].Float64, ShouldEqual, float64(333)) Result{
}) Series: []Row{
{
Name: "cpu",
Columns: []string{"time", "mean", "sum"},
Tags: map[string]string{"datacenter": "America"},
Values: [][]interface{}{
{json.Number("111"), json.Number("222"), json.Number("333")},
},
},
},
},
},
}
Convey("can parse null points", func() { Convey("$ alias", func() {
So(result.Series[0].Points[2][0].Valid, ShouldBeFalse) Convey("simple alias", func() {
}) query := &Query{Alias: "serie alias"}
result := parser.Parse(response, query)
Convey("can format serie names", func() { So(result.Series[0].Name, ShouldEqual, "serie alias")
So(result.Series[0].Name, ShouldEqual, "cpu.mean { datacenter: America }") })
So(result.Series[1].Name, ShouldEqual, "cpu.sum { datacenter: America }")
Convey("measurement alias", func() {
query := &Query{Alias: "alias $m $measurement", Measurement: "10m"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "alias 10m 10m")
})
Convey("column alias", func() {
query := &Query{Alias: "alias $col", Measurement: "10m"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "alias mean")
So(result.Series[1].Name, ShouldEqual, "alias sum")
})
Convey("tag alias", func() {
query := &Query{Alias: "alias $tag_datacenter"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "alias America")
})
})
Convey("[[]] alias", func() {
Convey("simple alias", func() {
query := &Query{Alias: "serie alias"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "serie alias")
})
Convey("measurement alias", func() {
query := &Query{Alias: "alias [[m]] [[measurement]]", Measurement: "10m"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "alias 10m 10m")
})
Convey("column alias", func() {
query := &Query{Alias: "alias [[col]]", Measurement: "10m"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "alias mean")
So(result.Series[1].Name, ShouldEqual, "alias sum")
})
Convey("tag alias", func() {
query := &Query{Alias: "alias [[tag_datacenter]]"}
result := parser.Parse(response, query)
So(result.Series[0].Name, ShouldEqual, "alias America")
})
})
}) })
}) })
} }
+4 -4
View File
@@ -24,12 +24,14 @@ func NewPrometheusExecutor(dsInfo *tsdb.DataSourceInfo) tsdb.Executor {
} }
var ( var (
plog log.Logger plog log.Logger
legendFormat *regexp.Regexp
) )
func init() { func init() {
plog = log.New("tsdb.prometheus") plog = log.New("tsdb.prometheus")
tsdb.RegisterExecutor("prometheus", NewPrometheusExecutor) tsdb.RegisterExecutor("prometheus", NewPrometheusExecutor)
legendFormat = regexp.MustCompile(`\{\{\s*(.+?)\s*\}\}`)
} }
func (e *PrometheusExecutor) getClient() (prometheus.QueryAPI, error) { func (e *PrometheusExecutor) getClient() (prometheus.QueryAPI, error) {
@@ -79,13 +81,11 @@ func (e *PrometheusExecutor) Execute(ctx context.Context, queries tsdb.QuerySlic
} }
func formatLegend(metric pmodel.Metric, query *PrometheusQuery) string { func formatLegend(metric pmodel.Metric, query *PrometheusQuery) string {
reg, _ := regexp.Compile(`\{\{\s*(.+?)\s*\}\}`)
if query.LegendFormat == "" { if query.LegendFormat == "" {
return metric.String() return metric.String()
} }
result := reg.ReplaceAllFunc([]byte(query.LegendFormat), func(in []byte) []byte { result := legendFormat.ReplaceAllFunc([]byte(query.LegendFormat), func(in []byte) []byte {
labelName := strings.Replace(string(in), "{{", "", 1) labelName := strings.Replace(string(in), "{{", "", 1)
labelName = strings.Replace(labelName, "}}", "", 1) labelName = strings.Replace(labelName, "}}", "", 1)
labelName = strings.TrimSpace(labelName) labelName = strings.TrimSpace(labelName)