Merge pull request grafana#6221 from grafana/alerting_influxdb

Alerting support for influxdb

Author: bergquist

docker/blocks/influxdb/fig

@@ -1,5 +1,7 @@
 influxdb:
-  image: tutum/influxdb:0.12
+  #image: influxdb/influxdb:1.0-alpine
+  image: influxdb:latest
+  container_name: influxdb
   ports:
     - "2004:2004"
     - "8083:8083"

pkg/cmd/grafana-server/main.go

@@ -20,6 +20,7 @@ import (
 	_ "github.com/grafana/grafana/pkg/services/alerting/conditions"
 	_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
 	_ "github.com/grafana/grafana/pkg/tsdb/graphite"
+	_ "github.com/grafana/grafana/pkg/tsdb/influxdb"
 	_ "github.com/grafana/grafana/pkg/tsdb/prometheus"
 	_ "github.com/grafana/grafana/pkg/tsdb/testdata"
 )

pkg/services/alerting/conditions/evaluator.go

@@ -122,7 +122,7 @@ func NewAlertEvaluator(model *simplejson.Json) (AlertEvaluator, error) {
 		return &NoDataEvaluator{}, nil
 	}
 
-	return nil, alerting.ValidationError{Reason: "Evaludator invalid evaluator type"}
+	return nil, alerting.ValidationError{Reason: "Evaludator invalid evaluator type: " + typ}
 }
 
 func inSlice(a string, list []string) bool {

pkg/services/alerting/conditions/reducer_test.go

@@ -3,6 +3,8 @@ package conditions
 import (
 	"testing"
 
+	"gopkg.in/guregu/null.v3"
+
 	"github.com/grafana/grafana/pkg/tsdb"
 	. "github.com/smartystreets/goconvey/convey"
 )
@@ -43,7 +45,7 @@ func testReducer(typ string, datapoints ...float64) float64 {
 	}
 
 	for idx := range datapoints {
-		series.Points = append(series.Points, tsdb.NewTimePoint(datapoints[idx], 1234134))
+		series.Points = append(series.Points, tsdb.NewTimePoint(null.FloatFrom(datapoints[idx]), 1234134))
 	}
 
 	return reducer.Reduce(series).Float64

pkg/tsdb/influxdb/influxdb.go

@@ -0,0 +1,132 @@
+package influxdb
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	"golang.org/x/net/context/ctxhttp"
+
+	"github.com/grafana/grafana/pkg/log"
+	"github.com/grafana/grafana/pkg/tsdb"
+)
+
+type InfluxDBExecutor struct {
+	*tsdb.DataSourceInfo
+	QueryParser    *InfluxdbQueryParser
+	QueryBuilder   *QueryBuilder
+	ResponseParser *ResponseParser
+}
+
+func NewInfluxDBExecutor(dsInfo *tsdb.DataSourceInfo) tsdb.Executor {
+	return &InfluxDBExecutor{
+		DataSourceInfo: dsInfo,
+		QueryParser:    &InfluxdbQueryParser{},
+		QueryBuilder:   &QueryBuilder{},
+		ResponseParser: &ResponseParser{},
+	}
+}
+
+var (
+	glog       log.Logger
+	HttpClient *http.Client
+)
+
+func init() {
+	glog = log.New("tsdb.influxdb")
+	tsdb.RegisterExecutor("influxdb", NewInfluxDBExecutor)
+
+	tr := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+
+	HttpClient = &http.Client{
+		Timeout:   time.Duration(15 * time.Second),
+		Transport: tr,
+	}
+}
+
+func (e *InfluxDBExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, context *tsdb.QueryContext) *tsdb.BatchResult {
+	result := &tsdb.BatchResult{}
+
+	query, err := e.getQuery(queries, context)
+	if err != nil {
+		return result.WithError(err)
+	}
+
+	glog.Debug("Influxdb query", "raw query", query)
+
+	req, err := e.createRequest(query)
+	if err != nil {
+		return result.WithError(err)
+	}
+
+	resp, err := ctxhttp.Do(ctx, HttpClient, req)
+	if err != nil {
+		return result.WithError(err)
+	}
+
+	if resp.StatusCode/100 != 2 {
+		return result.WithError(fmt.Errorf("Influxdb returned statuscode invalid status code: %v", resp.Status))
+	}
+
+	var response Response
+	dec := json.NewDecoder(resp.Body)
+	dec.UseNumber()
+	err = dec.Decode(&response)
+	if err != nil {
+		return result.WithError(err)
+	}
+
+	result.QueryResults = make(map[string]*tsdb.QueryResult)
+	result.QueryResults["A"] = e.ResponseParser.Parse(&response)
+
+	return result
+}
+
+func (e *InfluxDBExecutor) getQuery(queries tsdb.QuerySlice, context *tsdb.QueryContext) (string, error) {
+	for _, v := range queries {
+		query, err := e.QueryParser.Parse(v.Model)
+		if err != nil {
+			return "", err
+		}
+
+		rawQuery, err := e.QueryBuilder.Build(query, context)
+		if err != nil {
+			return "", err
+		}
+
+		return rawQuery, nil
+	}
+
+	return "", fmt.Errorf("query request contains no queries")
+}
+
+func (e *InfluxDBExecutor) createRequest(query string) (*http.Request, error) {
+	u, _ := url.Parse(e.Url)
+	u.Path = path.Join(u.Path, "query")
+
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	params := req.URL.Query()
+	params.Set("q", query)
+	params.Set("db", e.Database)
+	params.Set("epoch", "s")
+	req.URL.RawQuery = params.Encode()
+
+	req.Header.Set("User-Agent", "Grafana")
+	if e.BasicAuth {
+		req.SetBasicAuth(e.BasicAuthUser, e.BasicAuthPassword)
+	}
+
+	glog.Debug("Influxdb request", "url", req.URL.String())
+	return req, nil
+}

pkg/tsdb/influxdb/model_parser.go

@@ -0,0 +1,153 @@
+package influxdb
+
+import (
+	"strconv"
+
+	"github.com/grafana/grafana/pkg/components/simplejson"
+)
+
+type InfluxdbQueryParser struct{}
+
+func (qp *InfluxdbQueryParser) Parse(model *simplejson.Json) (*Query, error) {
+	policy := model.Get("policy").MustString("default")
+
+	measurement, err := model.Get("measurement").String()
+	if err != nil {
+		return nil, err
+	}
+
+	resultFormat, err := model.Get("resultFormat").String()
+	if err != nil {
+		return nil, err
+	}
+
+	tags, err := qp.parseTags(model)
+	if err != nil {
+		return nil, err
+	}
+
+	groupBys, err := qp.parseGroupBy(model)
+	if err != nil {
+		return nil, err
+	}
+
+	selects, err := qp.parseSelects(model)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Query{
+		Measurement:  measurement,
+		Policy:       policy,
+		ResultFormat: resultFormat,
+		GroupBy:      groupBys,
+		Tags:         tags,
+		Selects:      selects,
+	}, nil
+}
+
+func (qp *InfluxdbQueryParser) parseSelects(model *simplejson.Json) ([]*Select, error) {
+	var result []*Select
+
+	for _, selectObj := range model.Get("select").MustArray() {
+		selectJson := simplejson.NewFromAny(selectObj)
+		var parts Select
+
+		for _, partObj := range selectJson.MustArray() {
+			part := simplejson.NewFromAny(partObj)
+			queryPart, err := qp.parseQueryPart(part)
+			if err != nil {
+				return nil, err
+			}
+
+			parts = append(parts, *queryPart)
+		}
+
+		result = append(result, &parts)
+	}
+
+	return result, nil
+}
+
+func (*InfluxdbQueryParser) parseTags(model *simplejson.Json) ([]*Tag, error) {
+	var result []*Tag
+	for _, t := range model.Get("tags").MustArray() {
+		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 (*InfluxdbQueryParser) 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 (qp *InfluxdbQueryParser) parseGroupBy(model *simplejson.Json) ([]*QueryPart, error) {
+	var result []*QueryPart
+
+	for _, groupObj := range model.Get("groupBy").MustArray() {
+		groupJson := simplejson.NewFromAny(groupObj)
+		queryPart, err := qp.parseQueryPart(groupJson)
+
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, queryPart)
+	}
+
+	return result, nil
+}