Merge branch 'master' of git://github.com/grafana/grafana into WPH95-feature/add_es_alerting
This commit is contained in:
Marcus Efraimsson
@@ -22,6 +22,7 @@ import (
|
||||
_ "github.com/grafana/grafana/pkg/services/alerting/conditions"
|
||||
_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/cloudwatch"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/elasticsearch"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/graphite"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/influxdb"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/mysql"
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/grafana/grafana/pkg/log"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
"golang.org/x/net/context/ctxhttp"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ElasticsearchExecutor struct {
|
||||
QueryParser *ElasticSearchQueryParser
|
||||
Transport *http.Transport
|
||||
}
|
||||
|
||||
var (
|
||||
glog log.Logger
|
||||
intervalCalculator tsdb.IntervalCalculator
|
||||
)
|
||||
|
||||
func NewElasticsearchExecutor(dsInfo *models.DataSource) (tsdb.TsdbQueryEndpoint, error) {
|
||||
transport, err := dsInfo.GetHttpTransport()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &ElasticsearchExecutor{
|
||||
Transport: transport,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func init() {
|
||||
glog = log.New("tsdb.elasticsearch")
|
||||
tsdb.RegisterTsdbQueryEndpoint("elasticsearch", NewElasticsearchExecutor)
|
||||
intervalCalculator = tsdb.NewIntervalCalculator(&tsdb.IntervalOptions{MinInterval: time.Millisecond * 1})
|
||||
}
|
||||
|
||||
func (e *ElasticsearchExecutor) Query(ctx context.Context, dsInfo *models.DataSource, tsdbQuery *tsdb.TsdbQuery) (*tsdb.Response, error) {
|
||||
result := &tsdb.Response{}
|
||||
result.Results = make(map[string]*tsdb.QueryResult)
|
||||
|
||||
queries, err := e.getQuery(dsInfo, tsdbQuery)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
buff := bytes.Buffer{}
|
||||
for _, q := range queries {
|
||||
s, err := q.Build(tsdbQuery, dsInfo)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
buff.WriteString(s)
|
||||
}
|
||||
payload := buff.String()
|
||||
|
||||
if setting.Env == setting.DEV {
|
||||
glog.Debug("Elasticsearch playload", "raw playload", payload)
|
||||
}
|
||||
glog.Info("Elasticsearch playload", "raw playload", payload)
|
||||
|
||||
req, err := e.createRequest(dsInfo, payload)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
httpClient, err := dsInfo.GetHttpClient()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := ctxhttp.Do(ctx, httpClient, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if resp.StatusCode/100 != 2 {
|
||||
return nil, fmt.Errorf("elasticsearch returned statuscode invalid status code: %v", resp.Status)
|
||||
}
|
||||
|
||||
var responses Responses
|
||||
dec := json.NewDecoder(resp.Body)
|
||||
defer resp.Body.Close()
|
||||
dec.UseNumber()
|
||||
err = dec.Decode(&responses)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, res := range responses.Responses {
|
||||
if res.Err != nil {
|
||||
return nil, errors.New(res.getErrMsg())
|
||||
}
|
||||
}
|
||||
responseParser := ElasticsearchResponseParser{responses.Responses, queries}
|
||||
queryRes := responseParser.getTimeSeries()
|
||||
result.Results["A"] = queryRes
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (e *ElasticsearchExecutor) getQuery(dsInfo *models.DataSource, context *tsdb.TsdbQuery) ([]*Query, error) {
|
||||
queries := make([]*Query, 0)
|
||||
if len(context.Queries) == 0 {
|
||||
return nil, fmt.Errorf("query request contains no queries")
|
||||
}
|
||||
for _, v := range context.Queries {
|
||||
|
||||
query, err := e.QueryParser.Parse(v.Model, dsInfo)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
queries = append(queries, query)
|
||||
|
||||
}
|
||||
return queries, nil
|
||||
|
||||
}
|
||||
|
||||
func (e *ElasticsearchExecutor) createRequest(dsInfo *models.DataSource, query string) (*http.Request, error) {
|
||||
u, _ := url.Parse(dsInfo.Url)
|
||||
u.Path = path.Join(u.Path, "_msearch")
|
||||
req, err := http.NewRequest(http.MethodPost, u.String(), strings.NewReader(query))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("User-Agent", "Grafana")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
if dsInfo.BasicAuth {
|
||||
req.SetBasicAuth(dsInfo.BasicAuthUser, dsInfo.BasicAuthPassword)
|
||||
}
|
||||
|
||||
if !dsInfo.BasicAuth && dsInfo.User != "" {
|
||||
req.SetBasicAuth(dsInfo.User, dsInfo.Password)
|
||||
}
|
||||
|
||||
glog.Debug("Elasticsearch request", "url", req.URL.String())
|
||||
glog.Debug("Elasticsearch request", "body", query)
|
||||
return req, nil
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"time"
|
||||
)
|
||||
|
||||
var avgWithMovingAvg = Query{
|
||||
TimeField: "timestamp",
|
||||
RawQuery: "(test:query) AND (name:sample)",
|
||||
Interval: time.Millisecond,
|
||||
BucketAggs: []*BucketAgg{{
|
||||
Field: "timestamp",
|
||||
ID: "2",
|
||||
Type: "date_histogram",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"interval": "auto",
|
||||
"min_doc_count": 0,
|
||||
"trimEdges": 0,
|
||||
}),
|
||||
}},
|
||||
Metrics: []*Metric{{
|
||||
Field: "value",
|
||||
ID: "1",
|
||||
Type: "avg",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"script": map[string]string{
|
||||
"inline": "_value * 2",
|
||||
},
|
||||
}),
|
||||
}, {
|
||||
Field: "1",
|
||||
ID: "3",
|
||||
Type: "moving_avg",
|
||||
PipelineAggregate: "1",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"minimize": false,
|
||||
"model": "simple",
|
||||
"window": 5,
|
||||
}),
|
||||
}},
|
||||
}
|
||||
|
||||
var wildcardsAndQuotes = Query{
|
||||
TimeField: "timestamp",
|
||||
RawQuery: "scope:$location.leagueconnect.api AND name:*CreateRegistration AND name:\"*.201-responses.rate\"",
|
||||
Interval: time.Millisecond,
|
||||
BucketAggs: []*BucketAgg{{
|
||||
Field: "timestamp",
|
||||
ID: "2",
|
||||
Type: "date_histogram",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{}),
|
||||
}},
|
||||
Metrics: []*Metric{{
|
||||
Field: "value",
|
||||
ID: "1",
|
||||
Type: "sum",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{}),
|
||||
}},
|
||||
}
|
||||
var termAggs = Query{
|
||||
TimeField: "timestamp",
|
||||
RawQuery: "(scope:*.hmp.metricsd) AND (name_raw:builtin.general.*_instance_count)",
|
||||
Interval: time.Millisecond,
|
||||
BucketAggs: []*BucketAgg{{
|
||||
Field: "name_raw",
|
||||
ID: "4",
|
||||
Type: "terms",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"order": "desc",
|
||||
"orderBy": "_term",
|
||||
"size": "10",
|
||||
}),
|
||||
}, {
|
||||
Field: "timestamp",
|
||||
ID: "2",
|
||||
Type: "date_histogram",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"interval": "auto",
|
||||
"min_doc_count": 0,
|
||||
"trimEdges": 0,
|
||||
}),
|
||||
}},
|
||||
Metrics: []*Metric{{
|
||||
Field: "value",
|
||||
ID: "1",
|
||||
Type: "sum",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{}),
|
||||
}},
|
||||
}
|
||||
|
||||
var filtersAggs = Query{
|
||||
TimeField: "time",
|
||||
RawQuery: "*",
|
||||
Interval: time.Millisecond,
|
||||
BucketAggs: []*BucketAgg{{
|
||||
ID: "3",
|
||||
Type: "filters",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"filters": []interface{}{
|
||||
map[string]interface{}{"label": "hello", "query": "host:\"67.65.185.232\""},
|
||||
},
|
||||
}),
|
||||
}, {
|
||||
Field: "timestamp",
|
||||
ID: "2",
|
||||
Type: "date_histogram",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{
|
||||
"interval": "auto",
|
||||
"min_doc_count": 0,
|
||||
"trimEdges": 0,
|
||||
}),
|
||||
}},
|
||||
Metrics: []*Metric{{
|
||||
Field: "bytesSent",
|
||||
ID: "1",
|
||||
Type: "count",
|
||||
PipelineAggregate: "select metric",
|
||||
Settings: simplejson.NewFromAny(map[string]interface{}{}),
|
||||
}},
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
"github.com/leibowitz/moment"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ElasticSearchQueryParser struct {
|
||||
}
|
||||
|
||||
func (qp *ElasticSearchQueryParser) Parse(model *simplejson.Json, dsInfo *models.DataSource) (*Query, error) {
|
||||
//payload := bytes.Buffer{}
|
||||
//queryHeader := qp.getQueryHeader()
|
||||
timeField, err := model.Get("timeField").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rawQuery := model.Get("query").MustString()
|
||||
bucketAggs, err := qp.parseBucketAggs(model)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
metrics, err := qp.parseMetrics(model)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
alias := model.Get("alias").MustString("")
|
||||
parsedInterval, err := tsdb.GetIntervalFrom(dsInfo, model, time.Millisecond)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &Query{timeField,
|
||||
rawQuery,
|
||||
bucketAggs,
|
||||
metrics,
|
||||
alias,
|
||||
parsedInterval}, nil
|
||||
}
|
||||
|
||||
func (qp *ElasticSearchQueryParser) parseBucketAggs(model *simplejson.Json) ([]*BucketAgg, error) {
|
||||
var err error
|
||||
var result []*BucketAgg
|
||||
for _, t := range model.Get("bucketAggs").MustArray() {
|
||||
aggJson := simplejson.NewFromAny(t)
|
||||
agg := &BucketAgg{}
|
||||
|
||||
agg.Type, err = aggJson.Get("type").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
agg.ID, err = aggJson.Get("id").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
agg.Field = aggJson.Get("field").MustString()
|
||||
agg.Settings = simplejson.NewFromAny(aggJson.Get("settings").MustMap())
|
||||
|
||||
result = append(result, agg)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (qp *ElasticSearchQueryParser) parseMetrics(model *simplejson.Json) ([]*Metric, error) {
|
||||
var err error
|
||||
var result []*Metric
|
||||
for _, t := range model.Get("metrics").MustArray() {
|
||||
metricJson := simplejson.NewFromAny(t)
|
||||
metric := &Metric{}
|
||||
|
||||
metric.Field = metricJson.Get("field").MustString()
|
||||
metric.Hide = metricJson.Get("hide").MustBool(false)
|
||||
metric.ID, err = metricJson.Get("id").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
metric.PipelineAggregate = metricJson.Get("pipelineAgg").MustString()
|
||||
metric.Settings = simplejson.NewFromAny(metricJson.Get("settings").MustMap())
|
||||
|
||||
metric.Type, err = metricJson.Get("type").String()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result = append(result, metric)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
func getRequestHeader(timeRange *tsdb.TimeRange, dsInfo *models.DataSource) *QueryHeader {
|
||||
var header QueryHeader
|
||||
esVersion := dsInfo.JsonData.Get("esVersion").MustInt()
|
||||
|
||||
searchType := "query_then_fetch"
|
||||
if esVersion < 5 {
|
||||
searchType = "count"
|
||||
}
|
||||
header.SearchType = searchType
|
||||
header.IgnoreUnavailable = true
|
||||
header.Index = getIndexList(dsInfo.Database, dsInfo.JsonData.Get("interval").MustString(), timeRange)
|
||||
|
||||
if esVersion >= 56 {
|
||||
header.MaxConcurrentShardRequests = dsInfo.JsonData.Get("maxConcurrentShardRequests").MustInt()
|
||||
}
|
||||
return &header
|
||||
}
|
||||
|
||||
func getIndexList(pattern string, interval string, timeRange *tsdb.TimeRange) string {
|
||||
if interval == "" {
|
||||
return pattern
|
||||
}
|
||||
|
||||
var indexes []string
|
||||
indexParts := strings.Split(strings.TrimLeft(pattern, "["), "]")
|
||||
indexBase := indexParts[0]
|
||||
if len(indexParts) <= 1 {
|
||||
return pattern
|
||||
}
|
||||
|
||||
indexDateFormat := indexParts[1]
|
||||
|
||||
start := moment.NewMoment(timeRange.MustGetFrom())
|
||||
end := moment.NewMoment(timeRange.MustGetTo())
|
||||
|
||||
indexes = append(indexes, fmt.Sprintf("%s%s", indexBase, start.Format(indexDateFormat)))
|
||||
for start.IsBefore(*end) {
|
||||
switch interval {
|
||||
case "Hourly":
|
||||
start = start.AddHours(1)
|
||||
|
||||
case "Daily":
|
||||
start = start.AddDay()
|
||||
|
||||
case "Weekly":
|
||||
start = start.AddWeeks(1)
|
||||
|
||||
case "Monthly":
|
||||
start = start.AddMonths(1)
|
||||
|
||||
case "Yearly":
|
||||
start = start.AddYears(1)
|
||||
}
|
||||
indexes = append(indexes, fmt.Sprintf("%s%s", indexBase, start.Format(indexDateFormat)))
|
||||
}
|
||||
return strings.Join(indexes, ",")
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
. "github.com/smartystreets/goconvey/convey"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func makeTime(hour int) string {
|
||||
//unixtime 1500000000 == 2017-07-14T02:40:00+00:00
|
||||
return strconv.Itoa((1500000000 + hour*60*60) * 1000)
|
||||
}
|
||||
|
||||
func getIndexListByTime(pattern string, interval string, hour int) string {
|
||||
timeRange := &tsdb.TimeRange{
|
||||
From: makeTime(0),
|
||||
To: makeTime(hour),
|
||||
}
|
||||
return getIndexList(pattern, interval, timeRange)
|
||||
}
|
||||
|
||||
func TestElasticsearchGetIndexList(t *testing.T) {
|
||||
Convey("Test Elasticsearch getIndex ", t, func() {
|
||||
|
||||
Convey("Parse Interval Formats", func() {
|
||||
So(getIndexListByTime("[logstash-]YYYY.MM.DD", "Daily", 48),
|
||||
ShouldEqual, "logstash-2017.07.14,logstash-2017.07.15,logstash-2017.07.16")
|
||||
|
||||
So(len(strings.Split(getIndexListByTime("[logstash-]YYYY.MM.DD.HH", "Hourly", 3), ",")),
|
||||
ShouldEqual, 4)
|
||||
|
||||
So(getIndexListByTime("[logstash-]YYYY.W", "Weekly", 100),
|
||||
ShouldEqual, "logstash-2017.28,logstash-2017.29")
|
||||
|
||||
So(getIndexListByTime("[logstash-]YYYY.MM", "Monthly", 700),
|
||||
ShouldEqual, "logstash-2017.07,logstash-2017.08")
|
||||
|
||||
So(getIndexListByTime("[logstash-]YYYY", "Yearly", 10000),
|
||||
ShouldEqual, "logstash-2017,logstash-2018,logstash-2019")
|
||||
})
|
||||
|
||||
Convey("No Interval", func() {
|
||||
index := getIndexListByTime("logstash-test", "", 1)
|
||||
So(index, ShouldEqual, "logstash-test")
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
)
|
||||
|
||||
type BucketAgg struct {
|
||||
Field string `json:"field"`
|
||||
ID string `json:"id"`
|
||||
Settings *simplejson.Json `json:"settings"`
|
||||
Type string `jsons:"type"`
|
||||
}
|
||||
|
||||
type Metric struct {
|
||||
Field string `json:"field"`
|
||||
Hide bool `json:"hide"`
|
||||
ID string `json:"id"`
|
||||
PipelineAggregate string `json:"pipelineAgg"`
|
||||
Settings *simplejson.Json `json:"settings"`
|
||||
Type string `json:"type"`
|
||||
}
|
||||
|
||||
type QueryHeader struct {
|
||||
SearchType string `json:"search_type"`
|
||||
IgnoreUnavailable bool `json:"ignore_unavailable"`
|
||||
Index interface{} `json:"index"`
|
||||
MaxConcurrentShardRequests int `json:"max_concurrent_shard_requests,omitempty"`
|
||||
}
|
||||
|
||||
func (q *QueryHeader) String() string {
|
||||
r, _ := json.Marshal(q)
|
||||
return string(r)
|
||||
}
|
||||
|
||||
type Request struct {
|
||||
Query map[string]interface{} `json:"query"`
|
||||
Aggs Aggs `json:"aggs"`
|
||||
Size int `json:"size"`
|
||||
}
|
||||
|
||||
type Aggs map[string]interface{}
|
||||
|
||||
type HistogramAgg struct {
|
||||
Interval string `json:"interval,omitempty"`
|
||||
Field string `json:"field"`
|
||||
MinDocCount int `json:"min_doc_count"`
|
||||
Missing string `json:"missing,omitempty"`
|
||||
}
|
||||
|
||||
type DateHistogramAgg struct {
|
||||
HistogramAgg
|
||||
ExtendedBounds ExtendedBounds `json:"extended_bounds"`
|
||||
Format string `json:"format"`
|
||||
}
|
||||
|
||||
type FiltersAgg struct {
|
||||
Filters map[string]interface{} `json:"filters"`
|
||||
}
|
||||
|
||||
type TermsAgg struct {
|
||||
Field string `json:"field"`
|
||||
Size int `json:"size"`
|
||||
Order map[string]interface{} `json:"order"`
|
||||
Missing string `json:"missing,omitempty"`
|
||||
}
|
||||
|
||||
type TermsAggWrap struct {
|
||||
Terms TermsAgg `json:"terms"`
|
||||
Aggs Aggs `json:"aggs"`
|
||||
}
|
||||
|
||||
type ExtendedBounds struct {
|
||||
Min string `json:"min"`
|
||||
Max string `json:"max"`
|
||||
}
|
||||
|
||||
type RangeFilter struct {
|
||||
Range map[string]RangeFilterSetting `json:"range"`
|
||||
}
|
||||
type RangeFilterSetting struct {
|
||||
Gte string `json:"gte"`
|
||||
Lte string `json:"lte"`
|
||||
Format string `json:"format"`
|
||||
}
|
||||
|
||||
func newRangeFilter(field string, rangeFilterSetting RangeFilterSetting) *RangeFilter {
|
||||
return &RangeFilter{
|
||||
map[string]RangeFilterSetting{field: rangeFilterSetting}}
|
||||
}
|
||||
|
||||
type QueryStringFilter struct {
|
||||
QueryString QueryStringFilterSetting `json:"query_string"`
|
||||
}
|
||||
type QueryStringFilterSetting struct {
|
||||
AnalyzeWildcard bool `json:"analyze_wildcard"`
|
||||
Query string `json:"query"`
|
||||
}
|
||||
|
||||
func newQueryStringFilter(analyzeWildcard bool, query string) *QueryStringFilter {
|
||||
return &QueryStringFilter{QueryStringFilterSetting{AnalyzeWildcard: analyzeWildcard, Query: query}}
|
||||
}
|
||||
|
||||
type BoolQuery struct {
|
||||
Filter []interface{} `json:"filter"`
|
||||
}
|
||||
|
||||
type Responses struct {
|
||||
Responses []Response `json:"responses"`
|
||||
}
|
||||
|
||||
type Response struct {
|
||||
Status int `json:"status"`
|
||||
Err map[string]interface{} `json:"error"`
|
||||
Aggregations map[string]interface{} `json:"aggregations"`
|
||||
}
|
||||
|
||||
func (r *Response) getErrMsg() string {
|
||||
var msg bytes.Buffer
|
||||
errJson := simplejson.NewFromAny(r.Err)
|
||||
errType, err := errJson.Get("type").String()
|
||||
if err == nil {
|
||||
msg.WriteString(fmt.Sprintf("type:%s", errType))
|
||||
}
|
||||
|
||||
reason, err := errJson.Get("type").String()
|
||||
if err == nil {
|
||||
msg.WriteString(fmt.Sprintf("reason:%s", reason))
|
||||
}
|
||||
return msg.String()
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
var rangeFilterSetting = RangeFilterSetting{Gte: "$timeFrom",
|
||||
Lte: "$timeTo",
|
||||
Format: "epoch_millis"}
|
||||
|
||||
type Query struct {
|
||||
TimeField string `json:"timeField"`
|
||||
RawQuery string `json:"query"`
|
||||
BucketAggs []*BucketAgg `json:"bucketAggs"`
|
||||
Metrics []*Metric `json:"metrics"`
|
||||
Alias string `json:"Alias"`
|
||||
Interval time.Duration
|
||||
}
|
||||
|
||||
func (q *Query) Build(queryContext *tsdb.TsdbQuery, dsInfo *models.DataSource) (string, error) {
|
||||
var req Request
|
||||
payload := bytes.Buffer{}
|
||||
|
||||
req.Size = 0
|
||||
q.renderReqQuery(&req)
|
||||
|
||||
// handle document query
|
||||
if q.isRawDocumentQuery() {
|
||||
return "", errors.New("alert not support Raw_Document")
|
||||
}
|
||||
|
||||
err := q.parseAggs(&req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
reqBytes, err := json.Marshal(req)
|
||||
reqHeader := getRequestHeader(queryContext.TimeRange, dsInfo)
|
||||
payload.WriteString(reqHeader.String() + "\n")
|
||||
payload.WriteString(string(reqBytes) + "\n")
|
||||
return q.renderTemplate(payload.String(), queryContext)
|
||||
}
|
||||
|
||||
func (q *Query) isRawDocumentQuery() bool {
|
||||
if len(q.BucketAggs) == 0 {
|
||||
if len(q.Metrics) > 0 {
|
||||
metric := simplejson.NewFromAny(q.Metrics[0])
|
||||
if metric.Get("type").MustString("") == "raw_document" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (q *Query) renderReqQuery(req *Request) {
|
||||
req.Query = make(map[string]interface{})
|
||||
boolQuery := BoolQuery{}
|
||||
boolQuery.Filter = append(boolQuery.Filter, newRangeFilter(q.TimeField, rangeFilterSetting))
|
||||
boolQuery.Filter = append(boolQuery.Filter, newQueryStringFilter(true, q.RawQuery))
|
||||
req.Query["bool"] = boolQuery
|
||||
}
|
||||
|
||||
func (q *Query) parseAggs(req *Request) error {
|
||||
aggs := make(Aggs)
|
||||
nestedAggs := aggs
|
||||
for _, agg := range q.BucketAggs {
|
||||
esAggs := make(Aggs)
|
||||
switch agg.Type {
|
||||
case "date_histogram":
|
||||
esAggs["date_histogram"] = q.getDateHistogramAgg(agg)
|
||||
case "histogram":
|
||||
esAggs["histogram"] = q.getHistogramAgg(agg)
|
||||
case "filters":
|
||||
esAggs["filters"] = q.getFilters(agg)
|
||||
case "terms":
|
||||
terms := q.getTerms(agg)
|
||||
esAggs["terms"] = terms.Terms
|
||||
esAggs["aggs"] = terms.Aggs
|
||||
case "geohash_grid":
|
||||
return errors.New("alert not support Geo_Hash_Grid")
|
||||
}
|
||||
|
||||
if _, ok := nestedAggs["aggs"]; !ok {
|
||||
nestedAggs["aggs"] = make(Aggs)
|
||||
}
|
||||
|
||||
if aggs, ok := (nestedAggs["aggs"]).(Aggs); ok {
|
||||
aggs[agg.ID] = esAggs
|
||||
}
|
||||
nestedAggs = esAggs
|
||||
|
||||
}
|
||||
nestedAggs["aggs"] = make(Aggs)
|
||||
|
||||
for _, metric := range q.Metrics {
|
||||
subAgg := make(Aggs)
|
||||
|
||||
if metric.Type == "count" {
|
||||
continue
|
||||
}
|
||||
settings := metric.Settings.MustMap(make(map[string]interface{}))
|
||||
|
||||
if isPipelineAgg(metric.Type) {
|
||||
if _, err := strconv.Atoi(metric.PipelineAggregate); err == nil {
|
||||
settings["buckets_path"] = metric.PipelineAggregate
|
||||
} else {
|
||||
continue
|
||||
}
|
||||
|
||||
} else {
|
||||
settings["field"] = metric.Field
|
||||
}
|
||||
|
||||
subAgg[metric.Type] = settings
|
||||
nestedAggs["aggs"].(Aggs)[metric.ID] = subAgg
|
||||
}
|
||||
req.Aggs = aggs["aggs"].(Aggs)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (q *Query) getDateHistogramAgg(target *BucketAgg) *DateHistogramAgg {
|
||||
agg := &DateHistogramAgg{}
|
||||
interval, err := target.Settings.Get("interval").String()
|
||||
if err == nil {
|
||||
agg.Interval = interval
|
||||
}
|
||||
agg.Field = q.TimeField
|
||||
agg.MinDocCount = target.Settings.Get("min_doc_count").MustInt(0)
|
||||
agg.ExtendedBounds = ExtendedBounds{"$timeFrom", "$timeTo"}
|
||||
agg.Format = "epoch_millis"
|
||||
|
||||
if agg.Interval == "auto" {
|
||||
agg.Interval = "$__interval"
|
||||
}
|
||||
|
||||
missing, err := target.Settings.Get("missing").String()
|
||||
if err == nil {
|
||||
agg.Missing = missing
|
||||
}
|
||||
return agg
|
||||
}
|
||||
|
||||
func (q *Query) getHistogramAgg(target *BucketAgg) *HistogramAgg {
|
||||
agg := &HistogramAgg{}
|
||||
interval, err := target.Settings.Get("interval").String()
|
||||
if err == nil {
|
||||
agg.Interval = interval
|
||||
}
|
||||
|
||||
if target.Field != "" {
|
||||
agg.Field = target.Field
|
||||
}
|
||||
agg.MinDocCount = target.Settings.Get("min_doc_count").MustInt(0)
|
||||
missing, err := target.Settings.Get("missing").String()
|
||||
if err == nil {
|
||||
agg.Missing = missing
|
||||
}
|
||||
return agg
|
||||
}
|
||||
|
||||
func (q *Query) getFilters(target *BucketAgg) *FiltersAgg {
|
||||
agg := &FiltersAgg{}
|
||||
agg.Filters = map[string]interface{}{}
|
||||
for _, filter := range target.Settings.Get("filters").MustArray() {
|
||||
filterJson := simplejson.NewFromAny(filter)
|
||||
query := filterJson.Get("query").MustString("")
|
||||
label := filterJson.Get("label").MustString("")
|
||||
if label == "" {
|
||||
label = query
|
||||
}
|
||||
|
||||
agg.Filters[label] = newQueryStringFilter(true, query)
|
||||
}
|
||||
return agg
|
||||
}
|
||||
|
||||
func (q *Query) getTerms(target *BucketAgg) *TermsAggWrap {
|
||||
agg := &TermsAggWrap{Aggs: make(Aggs)}
|
||||
agg.Terms.Field = target.Field
|
||||
if len(target.Settings.MustMap()) == 0 {
|
||||
return agg
|
||||
}
|
||||
sizeStr := target.Settings.Get("size").MustString("")
|
||||
size, err := strconv.Atoi(sizeStr)
|
||||
if err != nil {
|
||||
size = 500
|
||||
}
|
||||
agg.Terms.Size = size
|
||||
orderBy, err := target.Settings.Get("orderBy").String()
|
||||
if err == nil {
|
||||
agg.Terms.Order = make(map[string]interface{})
|
||||
agg.Terms.Order[orderBy] = target.Settings.Get("order").MustString("")
|
||||
if _, err := strconv.Atoi(orderBy); err != nil {
|
||||
for _, metricI := range q.Metrics {
|
||||
metric := simplejson.NewFromAny(metricI)
|
||||
metricId := metric.Get("id").MustString()
|
||||
if metricId == orderBy {
|
||||
subAggs := make(Aggs)
|
||||
metricField := metric.Get("field").MustString()
|
||||
metricType := metric.Get("type").MustString()
|
||||
subAggs[metricType] = map[string]string{"field": metricField}
|
||||
agg.Aggs = make(Aggs)
|
||||
agg.Aggs[metricId] = subAggs
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
missing, err := target.Settings.Get("missing").String()
|
||||
if err == nil {
|
||||
agg.Terms.Missing = missing
|
||||
}
|
||||
|
||||
return agg
|
||||
}
|
||||
|
||||
func (q *Query) renderTemplate(payload string, queryContext *tsdb.TsdbQuery) (string, error) {
|
||||
timeRange := queryContext.TimeRange
|
||||
interval := intervalCalculator.Calculate(timeRange, q.Interval)
|
||||
payload = strings.Replace(payload, "$timeFrom", fmt.Sprintf("%d", timeRange.GetFromAsMsEpoch()), -1)
|
||||
payload = strings.Replace(payload, "$timeTo", fmt.Sprintf("%d", timeRange.GetToAsMsEpoch()), -1)
|
||||
payload = strings.Replace(payload, "$interval", interval.Text, -1)
|
||||
payload = strings.Replace(payload, "$__interval_ms", strconv.FormatInt(interval.Value.Nanoseconds()/int64(time.Millisecond), 10), -1)
|
||||
payload = strings.Replace(payload, "$__interval", interval.Text, -1)
|
||||
return payload, nil
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package elasticsearch
|
||||
|
||||
var metricAggType = map[string]string{
|
||||
"count": "Count",
|
||||
"avg": "Average",
|
||||
"sum": "Sum",
|
||||
"max": "Max",
|
||||
"min": "Min",
|
||||
"extended_stats": "Extended Stats",
|
||||
"percentiles": "Percentiles",
|
||||
"cardinality": "Unique Count",
|
||||
"moving_avg": "Moving Average",
|
||||
"derivative": "Derivative",
|
||||
"raw_document": "Raw Document",
|
||||
}
|
||||
|
||||
var extendedStats = map[string]string{
|
||||
"avg": "Avg",
|
||||
"min": "Min",
|
||||
"max": "Max",
|
||||
"sum": "Sum",
|
||||
"count": "Count",
|
||||
"std_deviation": "Std Dev",
|
||||
"std_deviation_bounds_upper": "Std Dev Upper",
|
||||
"std_deviation_bounds_lower": "Std Dev Lower",
|
||||
}
|
||||
|
||||
var pipelineOptions = map[string]string{
|
||||
"moving_avg": "moving_avg",
|
||||
"derivative": "derivative",
|
||||
}
|
||||
|
||||
func isPipelineAgg(metricType string) bool {
|
||||
if _, ok := pipelineOptions[metricType]; ok {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func describeMetric(metricType, field string) string {
|
||||
text := metricAggType[metricType]
|
||||
return text + " " + field
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
. "github.com/smartystreets/goconvey/convey"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func testElasticSearchResponse(query Query, expectedElasticSearchRequestJSON string) {
|
||||
var queryExpectedJSONInterface, queryJSONInterface interface{}
|
||||
jsonDate, _ := simplejson.NewJson([]byte(`{"esVersion":2}`))
|
||||
dsInfo := &models.DataSource{
|
||||
Database: "grafana-test",
|
||||
JsonData: jsonDate,
|
||||
}
|
||||
|
||||
testTimeRange := tsdb.NewTimeRange("5m", "now")
|
||||
|
||||
s, err := query.Build(&tsdb.TsdbQuery{TimeRange: testTimeRange}, dsInfo)
|
||||
So(err, ShouldBeNil)
|
||||
queryJSON := strings.Split(s, "\n")[1]
|
||||
err = json.Unmarshal([]byte(queryJSON), &queryJSONInterface)
|
||||
So(err, ShouldBeNil)
|
||||
|
||||
expectedElasticSearchRequestJSON = strings.Replace(
|
||||
expectedElasticSearchRequestJSON,
|
||||
"<FROM_TIMESTAMP>",
|
||||
strconv.FormatInt(testTimeRange.GetFromAsMsEpoch(), 10),
|
||||
-1,
|
||||
)
|
||||
|
||||
expectedElasticSearchRequestJSON = strings.Replace(
|
||||
expectedElasticSearchRequestJSON,
|
||||
"<TO_TIMESTAMP>",
|
||||
strconv.FormatInt(testTimeRange.GetToAsMsEpoch(), 10),
|
||||
-1,
|
||||
)
|
||||
|
||||
err = json.Unmarshal([]byte(expectedElasticSearchRequestJSON), &queryExpectedJSONInterface)
|
||||
So(err, ShouldBeNil)
|
||||
|
||||
result := reflect.DeepEqual(queryExpectedJSONInterface, queryJSONInterface)
|
||||
if !result {
|
||||
fmt.Printf("ERROR: %s \n != \n %s", expectedElasticSearchRequestJSON, queryJSON)
|
||||
}
|
||||
So(result, ShouldBeTrue)
|
||||
}
|
||||
func TestElasticSearchQueryBuilder(t *testing.T) {
|
||||
Convey("Elasticsearch QueryBuilder query testing", t, func() {
|
||||
Convey("Build test average metric with moving average", func() {
|
||||
var expectedElasticsearchQueryJSON = `
|
||||
{
|
||||
"size": 0,
|
||||
"query": {
|
||||
"bool": {
|
||||
"filter": [
|
||||
{
|
||||
"range": {
|
||||
"timestamp": {
|
||||
"gte": "<FROM_TIMESTAMP>",
|
||||
"lte": "<TO_TIMESTAMP>",
|
||||
"format": "epoch_millis"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"query_string": {
|
||||
"analyze_wildcard": true,
|
||||
"query": "(test:query) AND (name:sample)"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"aggs": {
|
||||
"2": {
|
||||
"date_histogram": {
|
||||
"interval": "200ms",
|
||||
"field": "timestamp",
|
||||
"min_doc_count": 0,
|
||||
"extended_bounds": {
|
||||
"min": "<FROM_TIMESTAMP>",
|
||||
"max": "<TO_TIMESTAMP>"
|
||||
},
|
||||
"format": "epoch_millis"
|
||||
},
|
||||
"aggs": {
|
||||
"1": {
|
||||
"avg": {
|
||||
"field": "value",
|
||||
"script": {
|
||||
"inline": "_value * 2"
|
||||
}
|
||||
}
|
||||
},
|
||||
"3": {
|
||||
"moving_avg": {
|
||||
"buckets_path": "1",
|
||||
"window": 5,
|
||||
"model": "simple",
|
||||
"minimize": false
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}`
|
||||
|
||||
testElasticSearchResponse(avgWithMovingAvg, expectedElasticsearchQueryJSON)
|
||||
})
|
||||
Convey("Test Wildcards and Quotes", func() {
|
||||
expectedElasticsearchQueryJSON := `
|
||||
{
|
||||
"size": 0,
|
||||
"query": {
|
||||
"bool": {
|
||||
"filter": [
|
||||
{
|
||||
"range": {
|
||||
"timestamp": {
|
||||
"gte": "<FROM_TIMESTAMP>",
|
||||
"lte": "<TO_TIMESTAMP>",
|
||||
"format": "epoch_millis"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"query_string": {
|
||||
"analyze_wildcard": true,
|
||||
"query": "scope:$location.leagueconnect.api AND name:*CreateRegistration AND name:\"*.201-responses.rate\""
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"aggs": {
|
||||
"2": {
|
||||
"aggs": {
|
||||
"1": {
|
||||
"sum": {
|
||||
"field": "value"
|
||||
}
|
||||
}
|
||||
},
|
||||
"date_histogram": {
|
||||
"extended_bounds": {
|
||||
"max": "<TO_TIMESTAMP>",
|
||||
"min": "<FROM_TIMESTAMP>"
|
||||
},
|
||||
"field": "timestamp",
|
||||
"format": "epoch_millis",
|
||||
"min_doc_count": 0
|
||||
}
|
||||
}
|
||||
}
|
||||
}`
|
||||
|
||||
testElasticSearchResponse(wildcardsAndQuotes, expectedElasticsearchQueryJSON)
|
||||
})
|
||||
Convey("Test Term Aggregates", func() {
|
||||
expectedElasticsearchQueryJSON := `
|
||||
{
|
||||
"size": 0,
|
||||
"query": {
|
||||
"bool": {
|
||||
"filter": [
|
||||
{
|
||||
"range": {
|
||||
"timestamp": {
|
||||
"gte": "<FROM_TIMESTAMP>",
|
||||
"lte": "<TO_TIMESTAMP>",
|
||||
"format": "epoch_millis"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"query_string": {
|
||||
"analyze_wildcard": true,
|
||||
"query": "(scope:*.hmp.metricsd) AND (name_raw:builtin.general.*_instance_count)"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"aggs": {"4":{"aggs":{"2":{"aggs":{"1":{"sum":{"field":"value"}}},"date_histogram":{"extended_bounds":{"max":"<TO_TIMESTAMP>","min":"<FROM_TIMESTAMP>"},"field":"timestamp","format":"epoch_millis","interval":"200ms","min_doc_count":0}}},"terms":{"field":"name_raw","order":{"_term":"desc"},"size":10}}}
|
||||
}`
|
||||
|
||||
testElasticSearchResponse(termAggs, expectedElasticsearchQueryJSON)
|
||||
})
|
||||
Convey("Test Filters Aggregates", func() {
|
||||
expectedElasticsearchQueryJSON := `{
|
||||
"size": 0,
|
||||
"query": {
|
||||
"bool": {
|
||||
"filter": [
|
||||
{
|
||||
"range": {
|
||||
"time": {
|
||||
"gte": "<FROM_TIMESTAMP>",
|
||||
"lte": "<TO_TIMESTAMP>",
|
||||
"format": "epoch_millis"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"query_string": {
|
||||
"analyze_wildcard": true,
|
||||
"query": "*"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"aggs": {
|
||||
"3": {
|
||||
"filters": {
|
||||
"filters": {
|
||||
"hello": {
|
||||
"query_string": {
|
||||
"query": "host:\"67.65.185.232\"",
|
||||
"analyze_wildcard": true
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"aggs": {
|
||||
"2": {
|
||||
"date_histogram": {
|
||||
"interval": "200ms",
|
||||
"field": "time",
|
||||
"min_doc_count": 0,
|
||||
"extended_bounds": {
|
||||
"min": "<FROM_TIMESTAMP>",
|
||||
"max": "<TO_TIMESTAMP>"
|
||||
},
|
||||
"format": "epoch_millis"
|
||||
},
|
||||
"aggs": {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
`
|
||||
|
||||
testElasticSearchResponse(filtersAggs, expectedElasticsearchQueryJSON)
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,263 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/grafana/grafana/pkg/components/null"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type ElasticsearchResponseParser struct {
|
||||
Responses []Response
|
||||
Targets []*Query
|
||||
}
|
||||
|
||||
func (rp *ElasticsearchResponseParser) getTimeSeries() *tsdb.QueryResult {
|
||||
queryRes := tsdb.NewQueryResult()
|
||||
for i, res := range rp.Responses {
|
||||
target := rp.Targets[i]
|
||||
props := make(map[string]string)
|
||||
series := make([]*tsdb.TimeSeries, 0)
|
||||
rp.processBuckets(res.Aggregations, target, &series, props, 0)
|
||||
rp.nameSeries(&series, target)
|
||||
queryRes.Series = append(queryRes.Series, series...)
|
||||
}
|
||||
return queryRes
|
||||
}
|
||||
|
||||
func (rp *ElasticsearchResponseParser) processBuckets(aggs map[string]interface{}, target *Query, series *[]*tsdb.TimeSeries, props map[string]string, depth int) error {
|
||||
|
||||
var err error
|
||||
maxDepth := len(target.BucketAggs) - 1
|
||||
for aggId, v := range aggs {
|
||||
aggDef, _ := findAgg(target, aggId)
|
||||
esAgg := simplejson.NewFromAny(v)
|
||||
if aggDef == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if depth == maxDepth {
|
||||
if aggDef.Type == "date_histogram" {
|
||||
err = rp.processMetrics(esAgg, target, series, props)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
return fmt.Errorf("not support type:%s", aggDef.Type)
|
||||
}
|
||||
} else {
|
||||
for i, b := range esAgg.Get("buckets").MustArray() {
|
||||
bucket := simplejson.NewFromAny(b)
|
||||
newProps := props
|
||||
if key, err := bucket.Get("key").String(); err == nil {
|
||||
newProps[aggDef.Field] = key
|
||||
} else {
|
||||
props["filter"] = strconv.Itoa(i)
|
||||
}
|
||||
|
||||
if key, err := bucket.Get("key_as_string").String(); err == nil {
|
||||
props[aggDef.Field] = key
|
||||
}
|
||||
rp.processBuckets(bucket.MustMap(), target, series, newProps, depth+1)
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
func (rp *ElasticsearchResponseParser) processMetrics(esAgg *simplejson.Json, target *Query, series *[]*tsdb.TimeSeries, props map[string]string) error {
|
||||
for _, metric := range target.Metrics {
|
||||
if metric.Hide {
|
||||
continue
|
||||
}
|
||||
|
||||
switch metric.Type {
|
||||
case "count":
|
||||
newSeries := tsdb.TimeSeries{}
|
||||
for _, v := range esAgg.Get("buckets").MustArray() {
|
||||
bucket := simplejson.NewFromAny(v)
|
||||
value := castToNullFloat(bucket.Get("doc_count"))
|
||||
key := castToNullFloat(bucket.Get("key"))
|
||||
newSeries.Points = append(newSeries.Points, tsdb.TimePoint{value, key})
|
||||
}
|
||||
newSeries.Tags = props
|
||||
newSeries.Tags["metric"] = "count"
|
||||
*series = append(*series, &newSeries)
|
||||
|
||||
case "percentiles":
|
||||
buckets := esAgg.Get("buckets").MustArray()
|
||||
if len(buckets) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
firstBucket := simplejson.NewFromAny(buckets[0])
|
||||
percentiles := firstBucket.GetPath(metric.ID, "values").MustMap()
|
||||
|
||||
for percentileName := range percentiles {
|
||||
newSeries := tsdb.TimeSeries{}
|
||||
newSeries.Tags = props
|
||||
newSeries.Tags["metric"] = "p" + percentileName
|
||||
newSeries.Tags["field"] = metric.Field
|
||||
for _, v := range buckets {
|
||||
bucket := simplejson.NewFromAny(v)
|
||||
value := castToNullFloat(bucket.GetPath(metric.ID, "values", percentileName))
|
||||
key := castToNullFloat(bucket.Get("key"))
|
||||
newSeries.Points = append(newSeries.Points, tsdb.TimePoint{value, key})
|
||||
}
|
||||
*series = append(*series, &newSeries)
|
||||
}
|
||||
default:
|
||||
newSeries := tsdb.TimeSeries{}
|
||||
newSeries.Tags = map[string]string{}
|
||||
for k, v := range props {
|
||||
newSeries.Tags[k] = v
|
||||
}
|
||||
|
||||
newSeries.Tags["metric"] = metric.Type
|
||||
newSeries.Tags["field"] = metric.Field
|
||||
for _, v := range esAgg.Get("buckets").MustArray() {
|
||||
bucket := simplejson.NewFromAny(v)
|
||||
key := castToNullFloat(bucket.Get("key"))
|
||||
valueObj, err := bucket.Get(metric.ID).Map()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var value null.Float
|
||||
if _, ok := valueObj["normalized_value"]; ok {
|
||||
value = castToNullFloat(bucket.GetPath(metric.ID, "normalized_value"))
|
||||
} else {
|
||||
value = castToNullFloat(bucket.GetPath(metric.ID, "value"))
|
||||
}
|
||||
newSeries.Points = append(newSeries.Points, tsdb.TimePoint{value, key})
|
||||
}
|
||||
*series = append(*series, &newSeries)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rp *ElasticsearchResponseParser) nameSeries(seriesList *[]*tsdb.TimeSeries, target *Query) {
|
||||
set := make(map[string]string)
|
||||
for _, v := range *seriesList {
|
||||
if metricType, exists := v.Tags["metric"]; exists {
|
||||
if _, ok := set[metricType]; !ok {
|
||||
set[metricType] = ""
|
||||
}
|
||||
}
|
||||
}
|
||||
metricTypeCount := len(set)
|
||||
for _, series := range *seriesList {
|
||||
series.Name = rp.getSeriesName(series, target, metricTypeCount)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (rp *ElasticsearchResponseParser) getSeriesName(series *tsdb.TimeSeries, target *Query, metricTypeCount int) string {
|
||||
metricType := series.Tags["metric"]
|
||||
metricName := rp.getMetricName(metricType)
|
||||
delete(series.Tags, "metric")
|
||||
|
||||
field := ""
|
||||
if v, ok := series.Tags["field"]; ok {
|
||||
field = v
|
||||
delete(series.Tags, "field")
|
||||
}
|
||||
|
||||
if target.Alias != "" {
|
||||
var re = regexp.MustCompile(`{{([\s\S]+?)}}`)
|
||||
for _, match := range re.FindAllString(target.Alias, -1) {
|
||||
group := match[2 : len(match)-2]
|
||||
|
||||
if strings.HasPrefix(group, "term ") {
|
||||
if term, ok := series.Tags["term "]; ok {
|
||||
strings.Replace(target.Alias, match, term, 1)
|
||||
}
|
||||
}
|
||||
if v, ok := series.Tags[group]; ok {
|
||||
strings.Replace(target.Alias, match, v, 1)
|
||||
}
|
||||
|
||||
switch group {
|
||||
case "metric":
|
||||
strings.Replace(target.Alias, match, metricName, 1)
|
||||
case "field":
|
||||
strings.Replace(target.Alias, match, field, 1)
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
// todo, if field and pipelineAgg
|
||||
if field != "" && isPipelineAgg(metricType) {
|
||||
found := false
|
||||
for _, metric := range target.Metrics {
|
||||
if metric.ID == field {
|
||||
metricName += " " + describeMetric(metric.Type, field)
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
metricName = "Unset"
|
||||
}
|
||||
|
||||
} else if field != "" {
|
||||
metricName += " " + field
|
||||
}
|
||||
|
||||
if len(series.Tags) == 0 {
|
||||
return metricName
|
||||
}
|
||||
|
||||
name := ""
|
||||
for _, v := range series.Tags {
|
||||
name += v + " "
|
||||
}
|
||||
|
||||
if metricTypeCount == 1 {
|
||||
return strings.TrimSpace(name)
|
||||
}
|
||||
|
||||
return strings.TrimSpace(name) + " " + metricName
|
||||
|
||||
}
|
||||
|
||||
func (rp *ElasticsearchResponseParser) getMetricName(metric string) string {
|
||||
if text, ok := metricAggType[metric]; ok {
|
||||
return text
|
||||
}
|
||||
|
||||
if text, ok := extendedStats[metric]; ok {
|
||||
return text
|
||||
}
|
||||
|
||||
return metric
|
||||
}
|
||||
|
||||
func castToNullFloat(j *simplejson.Json) null.Float {
|
||||
f, err := j.Float64()
|
||||
if err == nil {
|
||||
return null.FloatFrom(f)
|
||||
}
|
||||
|
||||
s, err := j.String()
|
||||
if err == nil {
|
||||
v, _ := strconv.ParseFloat(s, 64)
|
||||
return null.FloatFromPtr(&v)
|
||||
}
|
||||
|
||||
return null.NewFloat(0, false)
|
||||
}
|
||||
|
||||
func findAgg(target *Query, aggId string) (*BucketAgg, error) {
|
||||
for _, v := range target.BucketAggs {
|
||||
if aggId == v.ID {
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
return nil, errors.New("can't found aggDef, aggID:" + aggId)
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
package elasticsearch
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
. "github.com/smartystreets/goconvey/convey"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func testElasticsearchResponse(body string, target Query) *tsdb.QueryResult {
|
||||
var responses Responses
|
||||
err := json.Unmarshal([]byte(body), &responses)
|
||||
So(err, ShouldBeNil)
|
||||
|
||||
responseParser := ElasticsearchResponseParser{responses.Responses, []*Query{&target}}
|
||||
return responseParser.getTimeSeries()
|
||||
}
|
||||
|
||||
func TestElasticSearchResponseParser(t *testing.T) {
|
||||
Convey("Elasticsearch Response query testing", t, func() {
|
||||
Convey("Build test average metric with moving average", func() {
|
||||
responses := `{
|
||||
"responses": [
|
||||
{
|
||||
"took": 1,
|
||||
"timed_out": false,
|
||||
"_shards": {
|
||||
"total": 5,
|
||||
"successful": 5,
|
||||
"skipped": 0,
|
||||
"failed": 0
|
||||
},
|
||||
"hits": {
|
||||
"total": 4500,
|
||||
"max_score": 0,
|
||||
"hits": []
|
||||
},
|
||||
"aggregations": {
|
||||
"2": {
|
||||
"buckets": [
|
||||
{
|
||||
"1": {
|
||||
"value": null
|
||||
},
|
||||
"key_as_string": "1522205880000",
|
||||
"key": 1522205880000,
|
||||
"doc_count": 0
|
||||
},
|
||||
{
|
||||
"1": {
|
||||
"value": 10
|
||||
},
|
||||
"key_as_string": "1522205940000",
|
||||
"key": 1522205940000,
|
||||
"doc_count": 300
|
||||
},
|
||||
{
|
||||
"1": {
|
||||
"value": 10
|
||||
},
|
||||
"3": {
|
||||
"value": 20
|
||||
},
|
||||
"key_as_string": "1522206000000",
|
||||
"key": 1522206000000,
|
||||
"doc_count": 300
|
||||
},
|
||||
{
|
||||
"1": {
|
||||
"value": 10
|
||||
},
|
||||
"3": {
|
||||
"value": 20
|
||||
},
|
||||
"key_as_string": "1522206060000",
|
||||
"key": 1522206060000,
|
||||
"doc_count": 300
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"status": 200
|
||||
}
|
||||
]
|
||||
}
|
||||
`
|
||||
res := testElasticsearchResponse(responses, avgWithMovingAvg)
|
||||
So(len(res.Series), ShouldEqual, 2)
|
||||
So(res.Series[0].Name, ShouldEqual, "Average value")
|
||||
So(len(res.Series[0].Points), ShouldEqual, 4)
|
||||
for i, p := range res.Series[0].Points {
|
||||
if i == 0 {
|
||||
So(p[0].Valid, ShouldBeFalse)
|
||||
} else {
|
||||
So(p[0].Float64, ShouldEqual, 10)
|
||||
}
|
||||
So(p[1].Float64, ShouldEqual, 1522205880000+60000*i)
|
||||
}
|
||||
|
||||
So(res.Series[1].Name, ShouldEqual, "Moving Average Average 1")
|
||||
So(len(res.Series[1].Points), ShouldEqual, 2)
|
||||
|
||||
for _, p := range res.Series[1].Points {
|
||||
So(p[0].Float64, ShouldEqual, 20)
|
||||
}
|
||||
|
||||
})
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user