azuremonitor: builds a query and sends it to Azure on the backend

Lots of edge cases not covered and the response is not parsed. It only
handles one service and will have to be refactored to handle multiple

pkg/cmd/grafana-server/main.go

@@ -19,6 +19,7 @@ import (
 	_ "github.com/grafana/grafana/pkg/services/alerting/conditions"
 	_ "github.com/grafana/grafana/pkg/services/alerting/conditions"
 	_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
 	_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/setting"
+	_ "github.com/grafana/grafana/pkg/tsdb/azuremonitor"
 	_ "github.com/grafana/grafana/pkg/tsdb/cloudwatch"
 	_ "github.com/grafana/grafana/pkg/tsdb/cloudwatch"
 	_ "github.com/grafana/grafana/pkg/tsdb/elasticsearch"
 	_ "github.com/grafana/grafana/pkg/tsdb/elasticsearch"
 	_ "github.com/grafana/grafana/pkg/tsdb/graphite"
 	_ "github.com/grafana/grafana/pkg/tsdb/graphite"

pkg/models/datasource.go

@@ -23,7 +23,7 @@ const (
 	DS_ACCESS_DIRECT = "direct"
 	DS_ACCESS_DIRECT = "direct"
 	DS_ACCESS_PROXY  = "proxy"
 	DS_ACCESS_PROXY  = "proxy"
 	DS_STACKDRIVER   = "stackdriver"
 	DS_STACKDRIVER   = "stackdriver"
-	DS_AZURE_MONITOR = "azure-monitor"
+	DS_AZURE_MONITOR = "grafana-azure-monitor-datasource"
 )
 )
 
 
 var (
 var (

pkg/tsdb/azuremonitor/azuremonitor.go

+package azuremonitor
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	// "github.com/grafana/grafana/pkg/components/null"
+	"github.com/grafana/grafana/pkg/api/pluginproxy"
+	"github.com/grafana/grafana/pkg/components/simplejson"
+	"github.com/grafana/grafana/pkg/log"
+	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/grafana/grafana/pkg/tsdb"
+	"github.com/opentracing/opentracing-go"
+	"golang.org/x/net/context/ctxhttp"
+)
+
+var (
+	slog log.Logger
+)
+
+// AzureMonitorExecutor executes queries for the Azure Monitor datasource - all four services
+type AzureMonitorExecutor struct {
+	httpClient *http.Client
+	dsInfo     *models.DataSource
+}
+
+// NewAzureMonitorExecutor initializes a http client
+func NewAzureMonitorExecutor(dsInfo *models.DataSource) (tsdb.TsdbQueryEndpoint, error) {
+	httpClient, err := dsInfo.GetHttpClient()
+	if err != nil {
+		return nil, err
+	}
+
+	return &AzureMonitorExecutor{
+		httpClient: httpClient,
+		dsInfo:     dsInfo,
+	}, nil
+}
+
+func init() {
+	slog = log.New("tsdb.azuremonitor")
+	tsdb.RegisterTsdbQueryEndpoint("grafana-azure-monitor-datasource", NewAzureMonitorExecutor)
+}
+
+// Query takes in the frontend queries, parses them into the query format
+// expected by chosen Azure Monitor service (Azure Monitor, App Insights etc.)
+// executes the queries against the API and parses the response into
+// the right format
+func (e *AzureMonitorExecutor) Query(ctx context.Context, dsInfo *models.DataSource, tsdbQuery *tsdb.TsdbQuery) (*tsdb.Response, error) {
+	var result *tsdb.Response
+	var err error
+	queryType := tsdbQuery.Queries[0].Model.Get("queryType").MustString("")
+
+	switch queryType {
+	case "azureMonitorTimeSeriesQuery":
+	case "Azure Monitor":
+		fallthrough
+	default:
+		result, err = e.executeTimeSeriesQuery(ctx, tsdbQuery)
+	}
+
+	return result, err
+}
+
+func (e *AzureMonitorExecutor) executeTimeSeriesQuery(ctx context.Context, tsdbQuery *tsdb.TsdbQuery) (*tsdb.Response, error) {
+	result := &tsdb.Response{
+		Results: make(map[string]*tsdb.QueryResult),
+	}
+
+	queries, err := e.buildQueries(tsdbQuery)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, query := range queries {
+		queryRes, resp, err := e.executeQuery(ctx, query, tsdbQuery)
+		if err != nil {
+			return nil, err
+		}
+		err = e.parseResponse(queryRes, resp, query)
+		if err != nil {
+			queryRes.Error = err
+		}
+		result.Results[query.RefID] = queryRes
+	}
+
+	return result, nil
+}
+
+func (e *AzureMonitorExecutor) buildQueries(tsdbQuery *tsdb.TsdbQuery) ([]*AzureMonitorQuery, error) {
+	azureMonitorQueries := []*AzureMonitorQuery{}
+	startTime, err := tsdbQuery.TimeRange.ParseFrom()
+	if err != nil {
+		return nil, err
+	}
+
+	endTime, err := tsdbQuery.TimeRange.ParseTo()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, query := range tsdbQuery.Queries {
+		var target string
+
+		azureMonitorTarget := query.Model.Get("azureMonitor").MustMap()
+
+		resourceGroup := azureMonitorTarget["resourceGroup"].(string)
+		metricDefinition := azureMonitorTarget["metricDefinition"].(string)
+		resourceName := azureMonitorTarget["resourceName"].(string)
+		azureURL := fmt.Sprintf("resourceGroups/%s/providers/%s/%s/providers/microsoft.insights/metrics", resourceGroup, metricDefinition, resourceName)
+
+		alias := azureMonitorTarget["alias"].(string)
+
+		params := url.Values{}
+		params.Add("api-version", "2018-01-01")
+		params.Add("timespan", fmt.Sprintf("%v/%v", startTime.UTC().Format(time.RFC3339), endTime.UTC().Format(time.RFC3339)))
+		params.Add("interval", azureMonitorTarget["timeGrain"].(string))
+		params.Add("aggregation", azureMonitorTarget["aggregation"].(string))
+		params.Add("metricnames", azureMonitorTarget["metricName"].(string))
+		target = params.Encode()
+
+		if setting.Env == setting.DEV {
+			slog.Debug("Azuremonitor request", "params", params)
+		}
+
+		azureMonitorQueries = append(azureMonitorQueries, &AzureMonitorQuery{
+			URL:    azureURL,
+			Target: target,
+			Params: params,
+			RefID:  query.RefId,
+			Alias:  alias,
+		})
+	}
+
+	return azureMonitorQueries, nil
+}
+
+func (e *AzureMonitorExecutor) executeQuery(ctx context.Context, query *AzureMonitorQuery, tsdbQuery *tsdb.TsdbQuery) (*tsdb.QueryResult, AzureMonitorResponse, error) {
+	queryResult := &tsdb.QueryResult{Meta: simplejson.New(), RefId: query.RefID}
+
+	req, err := e.createRequest(ctx, e.dsInfo)
+	if err != nil {
+		queryResult.Error = err
+		return queryResult, AzureMonitorResponse{}, nil
+	}
+
+	req.URL.Path = path.Join(req.URL.Path, query.URL)
+	req.URL.RawQuery = query.Params.Encode()
+	queryResult.Meta.Set("rawQuery", req.URL.RawQuery)
+
+	span, ctx := opentracing.StartSpanFromContext(ctx, "azuremonitor query")
+	span.SetTag("target", query.Target)
+	span.SetTag("from", tsdbQuery.TimeRange.From)
+	span.SetTag("until", tsdbQuery.TimeRange.To)
+	span.SetTag("datasource_id", e.dsInfo.Id)
+	span.SetTag("org_id", e.dsInfo.OrgId)
+
+	defer span.Finish()
+
+	opentracing.GlobalTracer().Inject(
+		span.Context(),
+		opentracing.HTTPHeaders,
+		opentracing.HTTPHeadersCarrier(req.Header))
+
+	res, err := ctxhttp.Do(ctx, e.httpClient, req)
+	if err != nil {
+		queryResult.Error = err
+		return queryResult, AzureMonitorResponse{}, nil
+	}
+
+	data, err := e.unmarshalResponse(res)
+	if err != nil {
+		queryResult.Error = err
+		return queryResult, AzureMonitorResponse{}, nil
+	}
+
+	return queryResult, data, nil
+}
+
+func (e *AzureMonitorExecutor) createRequest(ctx context.Context, dsInfo *models.DataSource) (*http.Request, error) {
+	// find plugin
+	plugin, ok := plugins.DataSources[dsInfo.Type]
+	if !ok {
+		return nil, errors.New("Unable to find datasource plugin Azure Monitor")
+	}
+
+	var azureMonitorRoute *plugins.AppPluginRoute
+	for _, route := range plugin.Routes {
+		if route.Path == "azuremonitor" {
+			azureMonitorRoute = route
+			break
+		}
+	}
+
+	cloudName := dsInfo.JsonData.Get("cloudName").MustString("azuremonitor")
+	subscriptionID := dsInfo.JsonData.Get("subscriptionId").MustString()
+	proxyPass := fmt.Sprintf("%s/subscriptions/%s", cloudName, subscriptionID)
+
+	u, _ := url.Parse(dsInfo.Url)
+	u.Path = path.Join(u.Path, "render")
+
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		slog.Error("Failed to create request", "error", err)
+		return nil, fmt.Errorf("Failed to create request. error: %v", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("User-Agent", fmt.Sprintf("Grafana/%s", setting.BuildVersion))
+
+	pluginproxy.ApplyRoute(ctx, req, proxyPass, azureMonitorRoute, dsInfo)
+
+	return req, nil
+}
+
+func (e *AzureMonitorExecutor) unmarshalResponse(res *http.Response) (AzureMonitorResponse, error) {
+	body, err := ioutil.ReadAll(res.Body)
+	defer res.Body.Close()
+	if err != nil {
+		return AzureMonitorResponse{}, err
+	}
+
+	if res.StatusCode/100 != 2 {
+		slog.Error("Request failed", "status", res.Status, "body", string(body))
+		return AzureMonitorResponse{}, fmt.Errorf(string(body))
+	}
+
+	var data AzureMonitorResponse
+	err = json.Unmarshal(body, &data)
+	if err != nil {
+		slog.Error("Failed to unmarshal AzureMonitor response", "error", err, "status", res.Status, "body", string(body))
+		return AzureMonitorResponse{}, err
+	}
+
+	return data, nil
+}
+
+func (e *AzureMonitorExecutor) parseResponse(queryRes *tsdb.QueryResult, data AzureMonitorResponse, query *AzureMonitorQuery) error {
+	slog.Info("AzureMonitor", "Response", data)
+
+	return nil
+}

pkg/tsdb/azuremonitor/azuremonitor_test.go

+package azuremonitor
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/grafana/grafana/pkg/components/simplejson"
+	"github.com/grafana/grafana/pkg/tsdb"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestAzureMonitor(t *testing.T) {
+	Convey("AzureMonitor", t, func() {
+		executor := &AzureMonitorExecutor{}
+
+		Convey("Parse queries from frontend and build AzureMonitor API queries", func() {
+			fromStart := time.Date(2018, 3, 15, 13, 0, 0, 0, time.UTC).In(time.Local)
+			tsdbQuery := &tsdb.TsdbQuery{
+				TimeRange: &tsdb.TimeRange{
+					From: fmt.Sprintf("%v", fromStart.Unix()*1000),
+					To:   fmt.Sprintf("%v", fromStart.Add(34*time.Minute).Unix()*1000),
+				},
+				Queries: []*tsdb.Query{
+					{
+						Model: simplejson.NewFromAny(map[string]interface{}{
+							"azureMonitor": map[string]interface{}{
+								"timeGrain":        "PT1M",
+								"aggregation":      "Average",
+								"resourceGroup":    "grafanastaging",
+								"resourceName":     "grafana",
+								"metricDefinition": "Microsoft.Compute/virtualMachines",
+								"metricName":       "Percentage CPU",
+								"alias":            "testalias",
+								"queryType":        "Azure Monitor",
+							},
+						}),
+						RefId: "A",
+					},
+				},
+			}
+			Convey("and is a normal query", func() {
+				queries, err := executor.buildQueries(tsdbQuery)
+				So(err, ShouldBeNil)
+
+				So(len(queries), ShouldEqual, 1)
+				So(queries[0].RefID, ShouldEqual, "A")
+				So(queries[0].URL, ShouldEqual, "resourceGroups/grafanastaging/providers/Microsoft.Compute/virtualMachines/grafana/providers/microsoft.insights/metrics")
+				So(queries[0].Target, ShouldEqual, "aggregation=Average&api-version=2018-01-01&interval=PT1M&metricnames=Percentage+CPU&timespan=2018-03-15T13%3A00%3A00Z%2F2018-03-15T13%3A34%3A00Z")
+				So(len(queries[0].Params), ShouldEqual, 5)
+				So(queries[0].Params["timespan"][0], ShouldEqual, "2018-03-15T13:00:00Z/2018-03-15T13:34:00Z")
+				So(queries[0].Params["api-version"][0], ShouldEqual, "2018-01-01")
+				So(queries[0].Params["aggregation"][0], ShouldEqual, "Average")
+				So(queries[0].Params["metricnames"][0], ShouldEqual, "Percentage CPU")
+				So(queries[0].Params["interval"][0], ShouldEqual, "PT1M")
+				So(queries[0].Alias, ShouldEqual, "testalias")
+			})
+		})
+	})
+}

pkg/tsdb/azuremonitor/types.go

+package azuremonitor
+
+import (
+	"net/url"
+	"time"
+)
+
+// AzureMonitorQuery is the query for all the services as they have similar queries
+// with a url, a querystring and an alias field
+type AzureMonitorQuery struct {
+	URL    string
+	Target string
+	Params url.Values
+	RefID  string
+	Alias  string
+}
+
+// AzureMonitorResponse is the json response from the Azure Monitor API
+type AzureMonitorResponse struct {
+	Cost     int    `json:"cost"`
+	Timespan string `json:"timespan"`
+	Interval string `json:"interval"`
+	Value    []struct {
+		ID   string `json:"id"`
+		Type string `json:"type"`
+		Name struct {
+			Value          string `json:"value"`
+			LocalizedValue string `json:"localizedValue"`
+		} `json:"name"`
+		Unit       string `json:"unit"`
+		Timeseries []struct {
+			Metadatavalues []struct {
+				Name struct {
+					Value          string `json:"value"`
+					LocalizedValue string `json:"localizedValue"`
+				} `json:"name"`
+				Value string `json:"value"`
+			} `json:"metadatavalues"`
+			Data []struct {
+				TimeStamp time.Time `json:"timeStamp"`
+				Average   float64   `json:"average"`
+			} `json:"data"`
+		} `json:"timeseries"`
+	} `json:"value"`
+	Namespace      string `json:"namespace"`
+	Resourceregion string `json:"resourceregion"`
+}
+
+// ApplicationInsightsResponse is the json response from the Application Insights API
+type ApplicationInsightsResponse struct {
+	Tables []struct {
+		TableName string `json:"TableName"`
+		Columns   []struct {
+			ColumnName string `json:"ColumnName"`
+			DataType   string `json:"DataType"`
+			ColumnType string `json:"ColumnType"`
+		} `json:"Columns"`
+		Rows [][]interface{} `json:"Rows"`
+	} `json:"Tables"`
+}
+
+// AzureLogAnalyticsResponse is the json response object from the Azure Log Analytics API.
+type AzureLogAnalyticsResponse struct {
+	Tables []struct {
+		Name    string `json:"name"`
+		Columns []struct {
+			Name string `json:"name"`
+			Type string `json:"type"`
+		} `json:"columns"`
+		Rows [][]interface{} `json:"rows"`
+	} `json:"tables"`
+}