prometheus.go 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. package prometheus
  2. import (
  3. "context"
  4. "fmt"
  5. "regexp"
  6. "strings"
  7. "time"
  8. "github.com/opentracing/opentracing-go"
  9. "net/http"
  10. "github.com/grafana/grafana/pkg/components/null"
  11. "github.com/grafana/grafana/pkg/log"
  12. "github.com/grafana/grafana/pkg/models"
  13. "github.com/grafana/grafana/pkg/tsdb"
  14. api "github.com/prometheus/client_golang/api"
  15. apiv1 "github.com/prometheus/client_golang/api/prometheus/v1"
  16. "github.com/prometheus/common/model"
  17. //api "github.com/prometheus/client_golang/api"
  18. //apiv1 "github.com/prometheus/client_golang/api/prometheus/v1"
  19. )
  20. type PrometheusExecutor struct {
  21. *models.DataSource
  22. Transport *http.Transport
  23. }
  24. type basicAuthTransport struct {
  25. *http.Transport
  26. username string
  27. password string
  28. }
  29. func (bat basicAuthTransport) RoundTrip(req *http.Request) (*http.Response, error) {
  30. req.SetBasicAuth(bat.username, bat.password)
  31. return bat.Transport.RoundTrip(req)
  32. }
  33. func NewPrometheusExecutor(dsInfo *models.DataSource) (tsdb.TsdbQueryEndpoint, error) {
  34. transport, err := dsInfo.GetHttpTransport()
  35. if err != nil {
  36. return nil, err
  37. }
  38. return &PrometheusExecutor{
  39. DataSource: dsInfo,
  40. Transport: transport,
  41. }, nil
  42. }
  43. var (
  44. plog log.Logger
  45. legendFormat *regexp.Regexp
  46. )
  47. func init() {
  48. plog = log.New("tsdb.prometheus")
  49. tsdb.RegisterTsdbQueryEndpoint("prometheus", NewPrometheusExecutor)
  50. legendFormat = regexp.MustCompile(`\{\{\s*(.+?)\s*\}\}`)
  51. }
  52. func (e *PrometheusExecutor) getClient() (apiv1.API, error) {
  53. cfg := api.Config{
  54. Address: e.DataSource.Url,
  55. RoundTripper: e.Transport,
  56. }
  57. if e.BasicAuth {
  58. cfg.RoundTripper = basicAuthTransport{
  59. Transport: e.Transport,
  60. username: e.BasicAuthUser,
  61. password: e.BasicAuthPassword,
  62. }
  63. }
  64. client, err := api.NewClient(cfg)
  65. if err != nil {
  66. return nil, err
  67. }
  68. return apiv1.NewAPI(client), nil
  69. }
  70. func (e *PrometheusExecutor) Query(ctx context.Context, queryContext *tsdb.TsdbQuery) *tsdb.BatchResult {
  71. result := &tsdb.BatchResult{}
  72. client, err := e.getClient()
  73. if err != nil {
  74. return result.WithError(err)
  75. }
  76. query, err := parseQuery(queryContext.Queries, queryContext)
  77. if err != nil {
  78. return result.WithError(err)
  79. }
  80. timeRange := apiv1.Range{
  81. Start: query.Start,
  82. End: query.End,
  83. Step: query.Step,
  84. }
  85. span, ctx := opentracing.StartSpanFromContext(ctx, "alerting.prometheus")
  86. span.SetTag("expr", query.Expr)
  87. span.SetTag("start_unixnano", int64(query.Start.UnixNano()))
  88. span.SetTag("stop_unixnano", int64(query.End.UnixNano()))
  89. defer span.Finish()
  90. value, err := client.QueryRange(ctx, query.Expr, timeRange)
  91. if err != nil {
  92. return result.WithError(err)
  93. }
  94. queryResult, err := parseResponse(value, query)
  95. if err != nil {
  96. return result.WithError(err)
  97. }
  98. result.QueryResults = queryResult
  99. return result
  100. }
  101. func formatLegend(metric model.Metric, query *PrometheusQuery) string {
  102. if query.LegendFormat == "" {
  103. return metric.String()
  104. }
  105. result := legendFormat.ReplaceAllFunc([]byte(query.LegendFormat), func(in []byte) []byte {
  106. labelName := strings.Replace(string(in), "{{", "", 1)
  107. labelName = strings.Replace(labelName, "}}", "", 1)
  108. labelName = strings.TrimSpace(labelName)
  109. if val, exists := metric[model.LabelName(labelName)]; exists {
  110. return []byte(val)
  111. }
  112. return in
  113. })
  114. return string(result)
  115. }
  116. func parseQuery(queries []*tsdb.Query, queryContext *tsdb.TsdbQuery) (*PrometheusQuery, error) {
  117. queryModel := queries[0]
  118. expr, err := queryModel.Model.Get("expr").String()
  119. if err != nil {
  120. return nil, err
  121. }
  122. step, err := queryModel.Model.Get("step").Int64()
  123. if err != nil {
  124. return nil, err
  125. }
  126. format := queryModel.Model.Get("legendFormat").MustString("")
  127. start, err := queryContext.TimeRange.ParseFrom()
  128. if err != nil {
  129. return nil, err
  130. }
  131. end, err := queryContext.TimeRange.ParseTo()
  132. if err != nil {
  133. return nil, err
  134. }
  135. return &PrometheusQuery{
  136. Expr: expr,
  137. Step: time.Second * time.Duration(step),
  138. LegendFormat: format,
  139. Start: start,
  140. End: end,
  141. }, nil
  142. }
  143. func parseResponse(value model.Value, query *PrometheusQuery) (map[string]*tsdb.QueryResult, error) {
  144. queryResults := make(map[string]*tsdb.QueryResult)
  145. queryRes := tsdb.NewQueryResult()
  146. data, ok := value.(model.Matrix)
  147. if !ok {
  148. return queryResults, fmt.Errorf("Unsupported result format: %s", value.Type().String())
  149. }
  150. for _, v := range data {
  151. series := tsdb.TimeSeries{
  152. Name: formatLegend(v.Metric, query),
  153. Tags: map[string]string{},
  154. }
  155. for k, v := range v.Metric {
  156. series.Tags[string(k)] = string(v)
  157. }
  158. for _, k := range v.Values {
  159. series.Points = append(series.Points, tsdb.NewTimePoint(null.FloatFrom(float64(k.Value)), float64(k.Timestamp.Unix()*1000)))
  160. }
  161. queryRes.Series = append(queryRes.Series, &series)
  162. }
  163. queryResults["A"] = queryRes
  164. return queryResults, nil
  165. }