forked from grafana/grafana
/
log_sync_query.go
124 lines (102 loc) · 3.5 KB
/
log_sync_query.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
package cloudwatch
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
"github.com/aws/aws-sdk-go/service/cloudwatchlogs/cloudwatchlogsiface"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/tsdb/cloudwatch/kinds/dataquery"
"github.com/grafana/grafana/pkg/tsdb/cloudwatch/models"
)
const initialAlertPollPeriod = time.Second
var executeSyncLogQuery = func(ctx context.Context, e *cloudWatchExecutor, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
resp := backend.NewQueryDataResponse()
instance, err := e.getInstance(ctx, req.PluginContext)
if err != nil {
return nil, err
}
for _, q := range req.Queries {
var logsQuery models.LogsQuery
err := json.Unmarshal(q.JSON, &logsQuery)
if err != nil {
continue
}
logsQuery.Subtype = "StartQuery"
if logsQuery.Expression != nil {
logsQuery.QueryString = *logsQuery.Expression
}
region := logsQuery.Region
if logsQuery.Region == "" || region == defaultRegion {
logsQuery.Region = instance.Settings.Region
}
logsClient, err := e.getCWLogsClient(ctx, req.PluginContext, region)
if err != nil {
return nil, err
}
getQueryResultsOutput, err := e.syncQuery(ctx, logsClient, q, logsQuery, instance.Settings.LogsTimeout.Duration)
if err != nil {
return nil, err
}
dataframe, err := logsResultsToDataframes(getQueryResultsOutput)
if err != nil {
return nil, err
}
var frames []*data.Frame
if len(logsQuery.StatsGroups) > 0 && len(dataframe.Fields) > 0 {
frames, err = groupResults(dataframe, logsQuery.StatsGroups, true)
if err != nil {
return nil, err
}
} else {
frames = data.Frames{dataframe}
}
refId := "A"
if q.RefID != "" {
refId = q.RefID
}
respD := resp.Responses[refId]
respD.Frames = frames
resp.Responses[refId] = respD
}
return resp, nil
}
func (e *cloudWatchExecutor) syncQuery(ctx context.Context, logsClient cloudwatchlogsiface.CloudWatchLogsAPI,
queryContext backend.DataQuery, logsQuery models.LogsQuery, logsTimeout time.Duration) (*cloudwatchlogs.GetQueryResultsOutput, error) {
startQueryOutput, err := e.executeStartQuery(ctx, logsClient, logsQuery, queryContext.TimeRange)
if err != nil {
return nil, err
}
requestParams := models.LogsQuery{
CloudWatchLogsQuery: dataquery.CloudWatchLogsQuery{
Region: logsQuery.Region,
},
QueryId: *startQueryOutput.QueryId,
}
/*
Unlike many other data sources, with Cloudwatch Logs query requests don't receive the results as the response
to the query, but rather an ID is first returned. Following this, a client is expected to send requests along
with the ID until the status of the query is complete, receiving (possibly partial) results each time. For
queries made via dashboards and Explore, the logic of making these repeated queries is handled on the
frontend, but because alerts and expressions are executed on the backend the logic needs to be reimplemented here.
*/
ticker := time.NewTicker(initialAlertPollPeriod)
defer ticker.Stop()
attemptCount := 1
for range ticker.C {
res, err := e.executeGetQueryResults(ctx, logsClient, requestParams)
if err != nil {
return nil, fmt.Errorf("CloudWatch Error: %w", err)
}
if isTerminated(*res.Status) {
return res, err
}
if time.Duration(attemptCount)*time.Second >= logsTimeout {
return res, fmt.Errorf("time to fetch query results exceeded logs timeout")
}
attemptCount++
}
return nil, nil
}