feat(prometheus): initial support for prometheus
This commit is contained in:
@@ -2,6 +2,8 @@ package conditions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/bus"
|
||||
"github.com/grafana/grafana/pkg/components/simplejson"
|
||||
@@ -32,7 +34,8 @@ type AlertQuery struct {
|
||||
}
|
||||
|
||||
func (c *QueryCondition) Eval(context *alerting.EvalContext) {
|
||||
seriesList, err := c.executeQuery(context)
|
||||
timerange := tsdb.NewTimerange(c.Query.From, c.Query.To)
|
||||
seriesList, err := c.executeQuery(context, timerange)
|
||||
if err != nil {
|
||||
context.Error = err
|
||||
return
|
||||
@@ -66,7 +69,7 @@ func (c *QueryCondition) Eval(context *alerting.EvalContext) {
|
||||
context.Firing = len(context.EvalMatches) > 0
|
||||
}
|
||||
|
||||
func (c *QueryCondition) executeQuery(context *alerting.EvalContext) (tsdb.TimeSeriesSlice, error) {
|
||||
func (c *QueryCondition) executeQuery(context *alerting.EvalContext, timerange tsdb.TimeRange) (tsdb.TimeSeriesSlice, error) {
|
||||
getDsInfo := &m.GetDataSourceByIdQuery{
|
||||
Id: c.Query.DatasourceId,
|
||||
OrgId: context.Rule.OrgId,
|
||||
@@ -76,7 +79,7 @@ func (c *QueryCondition) executeQuery(context *alerting.EvalContext) (tsdb.TimeS
|
||||
return nil, fmt.Errorf("Could not find datasource")
|
||||
}
|
||||
|
||||
req := c.getRequestForAlertRule(getDsInfo.Result)
|
||||
req := c.getRequestForAlertRule(getDsInfo.Result, timerange)
|
||||
result := make(tsdb.TimeSeriesSlice, 0)
|
||||
|
||||
resp, err := c.HandleRequest(req)
|
||||
@@ -102,12 +105,9 @@ func (c *QueryCondition) executeQuery(context *alerting.EvalContext) (tsdb.TimeS
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (c *QueryCondition) getRequestForAlertRule(datasource *m.DataSource) *tsdb.Request {
|
||||
func (c *QueryCondition) getRequestForAlertRule(datasource *m.DataSource, timerange tsdb.TimeRange) *tsdb.Request {
|
||||
req := &tsdb.Request{
|
||||
TimeRange: tsdb.TimeRange{
|
||||
From: c.Query.From,
|
||||
To: c.Query.To,
|
||||
},
|
||||
TimeRange: timerange,
|
||||
Queries: []*tsdb.Query{
|
||||
{
|
||||
RefId: "A",
|
||||
@@ -141,6 +141,15 @@ func NewQueryCondition(model *simplejson.Json, index int) (*QueryCondition, erro
|
||||
condition.Query.Model = queryJson.Get("model")
|
||||
condition.Query.From = queryJson.Get("params").MustArray()[1].(string)
|
||||
condition.Query.To = queryJson.Get("params").MustArray()[2].(string)
|
||||
|
||||
if err := validateFromValue(condition.Query.From); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := validateToValue(condition.Query.To); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
condition.Query.DatasourceId = queryJson.Get("datasourceId").MustInt64()
|
||||
|
||||
reducerJson := model.Get("reducer")
|
||||
@@ -155,3 +164,25 @@ func NewQueryCondition(model *simplejson.Json, index int) (*QueryCondition, erro
|
||||
condition.Evaluator = evaluator
|
||||
return &condition, nil
|
||||
}
|
||||
|
||||
func validateFromValue(from string) error {
|
||||
fromRaw := strings.Replace(from, "now-", "", 1)
|
||||
|
||||
_, err := time.ParseDuration("-" + fromRaw)
|
||||
return err
|
||||
}
|
||||
|
||||
func validateToValue(to string) error {
|
||||
if to == "now" {
|
||||
return nil
|
||||
} else if strings.HasPrefix(to, "now-") {
|
||||
withoutNow := strings.Replace(to, "now-", "", 1)
|
||||
|
||||
_, err := time.ParseDuration("-" + withoutNow)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
return fmt.Errorf("cannot parse to value %s", to)
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/graphite"
|
||||
_ "github.com/grafana/grafana/pkg/tsdb/prometheus"
|
||||
)
|
||||
|
||||
var engine *alerting.Engine
|
||||
|
||||
+1
-1
@@ -26,7 +26,7 @@ func (bg *Batch) process(context *QueryContext) {
|
||||
if executor == nil {
|
||||
bg.Done = true
|
||||
result := &BatchResult{
|
||||
Error: errors.New("Could not find executor for data source type " + bg.Queries[0].DataSource.PluginId),
|
||||
Error: errors.New("Could not find executor for data source type: " + bg.Queries[0].DataSource.PluginId),
|
||||
QueryResults: make(map[string]*QueryResult),
|
||||
}
|
||||
for _, query := range bg.Queries {
|
||||
|
||||
+9
-3
@@ -1,10 +1,16 @@
|
||||
package tsdb
|
||||
|
||||
type TimeRange struct {
|
||||
From string
|
||||
To string
|
||||
type Query struct {
|
||||
RefId string
|
||||
Query string
|
||||
Depends []string
|
||||
DataSource *DataSourceInfo
|
||||
Results []*TimeSeries
|
||||
Exclude bool
|
||||
}
|
||||
|
||||
type QuerySlice []*Query
|
||||
|
||||
type Request struct {
|
||||
TimeRange TimeRange
|
||||
MaxDataPoints int
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
package prometheus
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/log"
|
||||
"github.com/grafana/grafana/pkg/tsdb"
|
||||
"github.com/prometheus/client_golang/api/prometheus"
|
||||
pmodel "github.com/prometheus/common/model"
|
||||
)
|
||||
|
||||
type PrometheusExecutor struct {
|
||||
*tsdb.DataSourceInfo
|
||||
}
|
||||
|
||||
func NewPrometheusExecutor(dsInfo *tsdb.DataSourceInfo) tsdb.Executor {
|
||||
return &PrometheusExecutor{dsInfo}
|
||||
}
|
||||
|
||||
var (
|
||||
plog log.Logger
|
||||
HttpClient http.Client
|
||||
)
|
||||
|
||||
func init() {
|
||||
plog = log.New("tsdb.prometheus")
|
||||
tsdb.RegisterExecutor("prometheus", NewPrometheusExecutor)
|
||||
}
|
||||
|
||||
func (e *PrometheusExecutor) getClient() (prometheus.QueryAPI, error) {
|
||||
cfg := prometheus.Config{
|
||||
Address: e.DataSourceInfo.Url,
|
||||
}
|
||||
|
||||
client, err := prometheus.New(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return prometheus.NewQueryAPI(client), nil
|
||||
}
|
||||
|
||||
func (e *PrometheusExecutor) Execute(queries tsdb.QuerySlice, queryContext *tsdb.QueryContext) *tsdb.BatchResult {
|
||||
result := &tsdb.BatchResult{}
|
||||
|
||||
client, err := e.getClient()
|
||||
if err != nil {
|
||||
result.Error = err
|
||||
return result
|
||||
}
|
||||
|
||||
from, _ := queryContext.TimeRange.FromTime()
|
||||
to, _ := queryContext.TimeRange.ToTime()
|
||||
timeRange := prometheus.Range{
|
||||
Start: from,
|
||||
End: to,
|
||||
Step: time.Second,
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
value, err := client.QueryRange(ctx, "counters_logins", timeRange)
|
||||
|
||||
if err != nil {
|
||||
result.Error = err
|
||||
return result
|
||||
}
|
||||
|
||||
result.QueryResults = parseResponse(value)
|
||||
return result
|
||||
}
|
||||
|
||||
func parseResponse(value pmodel.Value) map[string]*tsdb.QueryResult {
|
||||
queryResults := make(map[string]*tsdb.QueryResult)
|
||||
queryRes := &tsdb.QueryResult{}
|
||||
|
||||
data := value.(pmodel.Matrix)
|
||||
|
||||
for _, v := range data {
|
||||
var points [][2]*float64
|
||||
for _, k := range v.Values {
|
||||
dummie := float64(k.Timestamp)
|
||||
d2 := float64(k.Value)
|
||||
points = append(points, [2]*float64{&d2, &dummie})
|
||||
}
|
||||
|
||||
queryRes.Series = append(queryRes.Series, &tsdb.TimeSeries{
|
||||
Name: v.Metric.String(),
|
||||
Points: points,
|
||||
})
|
||||
}
|
||||
|
||||
queryResults["A"] = queryRes
|
||||
return queryResults
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
package prometheus
|
||||
@@ -1,12 +0,0 @@
|
||||
package tsdb
|
||||
|
||||
type Query struct {
|
||||
RefId string
|
||||
Query string
|
||||
Depends []string
|
||||
DataSource *DataSourceInfo
|
||||
Results []*TimeSeries
|
||||
Exclude bool
|
||||
}
|
||||
|
||||
type QuerySlice []*Query
|
||||
@@ -0,0 +1,77 @@
|
||||
package tsdb
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewTimerange(from, to string) TimeRange {
|
||||
return TimeRange{
|
||||
From: from,
|
||||
To: to,
|
||||
Now: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
type TimeRange struct {
|
||||
From string
|
||||
To string
|
||||
Now time.Time
|
||||
}
|
||||
|
||||
func (tr TimeRange) FromUnix() (int64, error) {
|
||||
fromRaw := strings.Replace(tr.From, "now-", "", 1)
|
||||
|
||||
diff, err := time.ParseDuration("-" + fromRaw)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return tr.Now.Add(diff).Unix(), nil
|
||||
}
|
||||
|
||||
func (tr TimeRange) FromTime() (time.Time, error) {
|
||||
fromRaw := strings.Replace(tr.From, "now-", "", 1)
|
||||
|
||||
diff, err := time.ParseDuration("-" + fromRaw)
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
|
||||
return tr.Now.Add(diff), nil
|
||||
}
|
||||
|
||||
func (tr TimeRange) ToUnix() (int64, error) {
|
||||
if tr.To == "now" {
|
||||
return tr.Now.Unix(), nil
|
||||
} else if strings.HasPrefix(tr.To, "now-") {
|
||||
withoutNow := strings.Replace(tr.To, "now-", "", 1)
|
||||
|
||||
diff, err := time.ParseDuration("-" + withoutNow)
|
||||
if err != nil {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
return tr.Now.Add(diff).Unix(), nil
|
||||
}
|
||||
|
||||
return 0, fmt.Errorf("cannot parse to value %s", tr.To)
|
||||
}
|
||||
|
||||
func (tr TimeRange) ToTime() (time.Time, error) {
|
||||
if tr.To == "now" {
|
||||
return tr.Now, nil
|
||||
} else if strings.HasPrefix(tr.To, "now-") {
|
||||
withoutNow := strings.Replace(tr.To, "now-", "", 1)
|
||||
|
||||
diff, err := time.ParseDuration("-" + withoutNow)
|
||||
if err != nil {
|
||||
return time.Time{}, nil
|
||||
}
|
||||
|
||||
return tr.Now.Add(diff), nil
|
||||
}
|
||||
|
||||
return time.Time{}, fmt.Errorf("cannot parse to value %s", tr.To)
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package tsdb
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
. "github.com/smartystreets/goconvey/convey"
|
||||
)
|
||||
|
||||
func TestTimeRange(t *testing.T) {
|
||||
Convey("Time range", t, func() {
|
||||
|
||||
now := time.Now()
|
||||
|
||||
Convey("Can parse 5m, now", func() {
|
||||
tr := TimeRange{
|
||||
From: "5m",
|
||||
To: "now",
|
||||
Now: now,
|
||||
}
|
||||
|
||||
Convey("5m ago ", func() {
|
||||
fiveMinAgo, _ := time.ParseDuration("-5m")
|
||||
expected := now.Add(fiveMinAgo)
|
||||
|
||||
res, err := tr.FromUnix()
|
||||
So(err, ShouldBeNil)
|
||||
So(res, ShouldAlmostEqual, expected.Unix())
|
||||
})
|
||||
|
||||
Convey("now ", func() {
|
||||
res, err := tr.ToUnix()
|
||||
So(err, ShouldBeNil)
|
||||
So(res, ShouldAlmostEqual, now.Unix())
|
||||
})
|
||||
})
|
||||
|
||||
Convey("Can parse 5h, now-10m", func() {
|
||||
tr := TimeRange{
|
||||
From: "5h",
|
||||
To: "now-10m",
|
||||
Now: now,
|
||||
}
|
||||
|
||||
Convey("5h ago ", func() {
|
||||
fiveMinAgo, _ := time.ParseDuration("-5h")
|
||||
expected := now.Add(fiveMinAgo)
|
||||
|
||||
res, err := tr.FromUnix()
|
||||
So(err, ShouldBeNil)
|
||||
So(res, ShouldAlmostEqual, expected.Unix())
|
||||
})
|
||||
|
||||
Convey("now-10m ", func() {
|
||||
fiveMinAgo, _ := time.ParseDuration("-10m")
|
||||
expected := now.Add(fiveMinAgo)
|
||||
res, err := tr.ToUnix()
|
||||
So(err, ShouldBeNil)
|
||||
So(res, ShouldAlmostEqual, expected.Unix())
|
||||
})
|
||||
})
|
||||
|
||||
Convey("Cannot parse asdf", func() {
|
||||
var err error
|
||||
tr := TimeRange{
|
||||
From: "asdf",
|
||||
To: "asdf",
|
||||
Now: now,
|
||||
}
|
||||
|
||||
_, err = tr.FromUnix()
|
||||
So(err, ShouldNotBeNil)
|
||||
|
||||
_, err = tr.ToUnix()
|
||||
So(err, ShouldNotBeNil)
|
||||
})
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user