public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cloudwatch: Migrate queries that use multiple stats to one query per stat (#36925)

Browse Source 
* migrate queries that use multiple stats - squash commits

* fix typo
This commit is contained in:
Erik Sundell
2021-09-08 16:06:43 +02:00
committed by GitHub
parent ae9343f8ae
commit 5e38b02f94
34 changed files with 2303 additions and 1492 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/cloudwatch/response_parser.go
+173 -121
View File
@@ -8,86 +8,119 @@ import (
"time"
"github.com/aws/aws-sdk-go/service/cloudwatch"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/components/simplejson"
)
func (e *cloudWatchExecutor) parseResponse(metricDataOutputs []*cloudwatch.GetMetricDataOutput,
queries map[string]*cloudWatchQuery) ([]*cloudwatchResponse, error) {
// Map from result ID -> label -> result
mdrs := make(map[string]map[string]*cloudwatch.MetricDataResult)
labels := map[string][]string{}
for _, mdo := range metricDataOutputs {
requestExceededMaxLimit := false
for _, message := range mdo.Messages {
if *message.Code == "MaxMetricsExceeded" {
requestExceededMaxLimit = true
}
}
for _, r := range mdo.MetricDataResults {
id := *r.Id
label := *r.Label
if _, exists := mdrs[id]; !exists {
mdrs[id] = make(map[string]*cloudwatch.MetricDataResult)
mdrs[id][label] = r
labels[id] = append(labels[id], label)
} else if _, exists := mdrs[id][label]; !exists {
mdrs[id][label] = r
labels[id] = append(labels[id], label)
} else {
mdr := mdrs[id][label]
mdr.Timestamps = append(mdr.Timestamps, r.Timestamps...)
mdr.Values = append(mdr.Values, r.Values...)
if *r.StatusCode == "Complete" {
mdr.StatusCode = r.StatusCode
}
}
queries[id].RequestExceededMaxLimit = requestExceededMaxLimit
}
func (e *cloudWatchExecutor) parseResponse(startTime time.Time, endTime time.Time, metricDataOutputs []*cloudwatch.GetMetricDataOutput,
queries []*cloudWatchQuery) ([]*responseWrapper, error) {
aggregatedResponse := aggregateResponse(metricDataOutputs)
queriesById := map[string]*cloudWatchQuery{}
for _, query := range queries {
queriesById[query.Id] = query
}
cloudWatchResponses := make([]*cloudwatchResponse, 0, len(mdrs))
for id, lr := range mdrs {
query := queries[id]
frames, partialData, err := parseMetricResults(lr, labels[id], query)
results := []*responseWrapper{}
for id, response := range aggregatedResponse {
queryRow := queriesById[id]
dataRes := backend.DataResponse{}
if response.HasArithmeticError {
dataRes.Error = fmt.Errorf("ArithmeticError in query %q: %s", queryRow.RefId, response.ArithmeticErrorMessage)
}
var err error
dataRes.Frames, err = buildDataFrames(startTime, endTime, response, queryRow)
if err != nil {
return nil, err
}
response := &cloudwatchResponse{
DataFrames: frames,
Period: query.Period,
Expression: query.UsedExpression,
RefId: query.RefId,
Id: query.Id,
RequestExceededMaxLimit: query.RequestExceededMaxLimit,
PartialData: partialData,
}
cloudWatchResponses = append(cloudWatchResponses, response)
results = append(results, &responseWrapper{
DataResponse: &dataRes,
RefId: queryRow.RefId,
})
}
return cloudWatchResponses, nil
return results, nil
}
func parseMetricResults(results map[string]*cloudwatch.MetricDataResult, labels []string,
query *cloudWatchQuery) (data.Frames, bool, error) {
partialData := false
frames := data.Frames{}
for _, label := range labels {
result := results[label]
if *result.StatusCode != "Complete" {
partialData = true
}
for _, message := range result.Messages {
if *message.Code == "ArithmeticError" {
return nil, false, fmt.Errorf("ArithmeticError in query %q: %s", query.RefId, *message.Value)
func aggregateResponse(getMetricDataOutputs []*cloudwatch.GetMetricDataOutput) map[string]queryRowResponse {
responseByID := make(map[string]queryRowResponse)
for _, gmdo := range getMetricDataOutputs {
requestExceededMaxLimit := false
for _, message := range gmdo.Messages {
if *message.Code == "MaxMetricsExceeded" {
requestExceededMaxLimit = true
}
}
for _, r := range gmdo.MetricDataResults {
id := *r.Id
label := *r.Label
response := newQueryRowResponse(id)
if _, exists := responseByID[id]; exists {
response = responseByID[id]
}
for _, message := range r.Messages {
if *message.Code == "ArithmeticError" {
response.addArithmeticError(message.Value)
}
}
if _, exists := response.Metrics[label]; !exists {
response.addMetricDataResult(r)
} else {
response.appendTimeSeries(r)
}
response.RequestExceededMaxLimit = response.RequestExceededMaxLimit || requestExceededMaxLimit
responseByID[id] = response
}
}
return responseByID
}
func getLabels(cloudwatchLabel string, query *cloudWatchQuery) data.Labels {
dims := make([]string, 0, len(query.Dimensions))
for k := range query.Dimensions {
dims = append(dims, k)
}
sort.Strings(dims)
labels := data.Labels{}
for _, dim := range dims {
values := query.Dimensions[dim]
if len(values) == 1 && values[0] != "*" {
labels[dim] = values[0]
} else {
for _, value := range values {
if value == cloudwatchLabel || value == "*" {
labels[dim] = cloudwatchLabel
} else if strings.Contains(cloudwatchLabel, value) {
labels[dim] = value
}
}
}
}
return labels
}
func buildDataFrames(startTime time.Time, endTime time.Time, aggregatedResponse queryRowResponse,
query *cloudWatchQuery) (data.Frames, error) {
frames := data.Frames{}
for _, label := range aggregatedResponse.Labels {
metric := aggregatedResponse.Metrics[label]
deepLink, err := query.buildDeepLink(startTime, endTime)
if err != nil {
return nil, err
}
// In case a multi-valued dimension is used and the cloudwatch query yields no values, create one empty time
// series for each dimension value. Use that dimension value to expand the alias field
if len(result.Values) == 0 && query.isMultiValuedDimensionExpression() {
if len(metric.Values) == 0 && query.isMultiValuedDimensionExpression() {
series := 0
multiValuedDimension := ""
for key, values := range query.Dimensions {
@@ -98,18 +131,18 @@ func parseMetricResults(results map[string]*cloudwatch.MetricDataResult, labels
}
for _, value := range query.Dimensions[multiValuedDimension] {
tags := map[string]string{multiValuedDimension: value}
labels := map[string]string{multiValuedDimension: value}
for key, values := range query.Dimensions {
if key != multiValuedDimension && len(values) > 0 {
tags[key] = values[0]
labels[key] = values[0]
}
}
timeField := data.NewField(data.TimeSeriesTimeFieldName, nil, []*time.Time{})
valueField := data.NewField(data.TimeSeriesValueFieldName, tags, []*float64{})
valueField := data.NewField(data.TimeSeriesValueFieldName, labels, []*float64{})
frameName := formatAlias(query, query.Stats, tags, label)
valueField.SetConfig(&data.FieldConfig{DisplayNameFromDS: frameName})
frameName := formatAlias(query, query.Statistic, labels, label)
valueField.SetConfig(&data.FieldConfig{DisplayNameFromDS: frameName, Links: createDataLinks(deepLink)})
emptyFrame := data.Frame{
Name: frameName,
@@ -118,66 +151,63 @@ func parseMetricResults(results map[string]*cloudwatch.MetricDataResult, labels
valueField,
},
RefID: query.RefId,
Meta: createMeta(query),
}
frames = append(frames, &emptyFrame)
}
} else {
dims := make([]string, 0, len(query.Dimensions))
for k := range query.Dimensions {
dims = append(dims, k)
}
sort.Strings(dims)
tags := data.Labels{}
for _, dim := range dims {
values := query.Dimensions[dim]
if len(values) == 1 && values[0] != "*" {
tags[dim] = values[0]
} else {
for _, value := range values {
if value == label || value == "*" {
tags[dim] = label
} else if strings.Contains(label, value) {
tags[dim] = value
}
}
}
}
timestamps := []*time.Time{}
points := []*float64{}
for j, t := range result.Timestamps {
if j > 0 {
expectedTimestamp := result.Timestamps[j-1].Add(time.Duration(query.Period) * time.Second)
if expectedTimestamp.Before(*t) {
timestamps = append(timestamps, &expectedTimestamp)
points = append(points, nil)
}
}
val := result.Values[j]
timestamps = append(timestamps, t)
points = append(points, val)
}
timeField := data.NewField(data.TimeSeriesTimeFieldName, nil, timestamps)
valueField := data.NewField(data.TimeSeriesValueFieldName, tags, points)
frameName := formatAlias(query, query.Stats, tags, label)
valueField.SetConfig(&data.FieldConfig{DisplayNameFromDS: frameName})
frame := data.Frame{
Name: frameName,
Fields: []*data.Field{
timeField,
valueField,
},
RefID: query.RefId,
}
frames = append(frames, &frame)
continue
}
labels := getLabels(label, query)
timestamps := []*time.Time{}
points := []*float64{}
for j, t := range metric.Timestamps {
if j > 0 {
expectedTimestamp := metric.Timestamps[j-1].Add(time.Duration(query.Period) * time.Second)
if expectedTimestamp.Before(*t) {
timestamps = append(timestamps, &expectedTimestamp)
points = append(points, nil)
}
}
val := metric.Values[j]
timestamps = append(timestamps, t)
points = append(points, val)
}
timeField := data.NewField(data.TimeSeriesTimeFieldName, nil, timestamps)
valueField := data.NewField(data.TimeSeriesValueFieldName, labels, points)
frameName := formatAlias(query, query.Statistic, labels, label)
valueField.SetConfig(&data.FieldConfig{DisplayNameFromDS: frameName, Links: createDataLinks(deepLink)})
frame := data.Frame{
Name: frameName,
Fields: []*data.Field{
timeField,
valueField,
},
RefID: query.RefId,
Meta: createMeta(query),
}
if aggregatedResponse.RequestExceededMaxLimit {
frame.AppendNotices(data.Notice{
Severity: data.NoticeSeverityWarning,
Text: "cloudwatch GetMetricData error: Maximum number of allowed metrics exceeded. Your search may have been limited",
})
}
if aggregatedResponse.StatusCode != "Complete" {
frame.AppendNotices(data.Notice{
Severity: data.NoticeSeverityWarning,
Text: "cloudwatch GetMetricData error: Too many datapoints requested - your search has been limited. Please try to reduce the time range",
})
}
frames = append(frames, &frame)
}
return frames, partialData, nil
return frames, nil
}
func formatAlias(query *cloudWatchQuery, stat string, dimensions map[string]string, label string) string {
@@ -231,3 +261,25 @@ func formatAlias(query *cloudWatchQuery, stat string, dimensions map[string]stri
return string(result)
}
func createDataLinks(link string) []data.DataLink {
dataLinks := []data.DataLink{}
if link != "" {
dataLinks = append(dataLinks, data.DataLink{
Title: "View in CloudWatch console",
TargetBlank: true,
URL: link,
})
}
return dataLinks
}
func createMeta(query *cloudWatchQuery) *data.FrameMeta {
return &data.FrameMeta{
ExecutedQueryString: query.UsedExpression,
Custom: simplejson.NewFromAny(map[string]interface{}{
"period": query.Period,
"id": query.Id,
}),
}
}