Skip to content
This repository was archived by the owner on Mar 8, 2023. It is now read-only.

Parallel fetch opto #17

Open
tul wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Parallel fetch opto #17

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
599ae0e
extract scrapeMetric into separate function (no functional changes)
tul Mar 29, 2019
cc0d849
issue all cloudwatch requests in parallel
tul Mar 29, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
243 changes: 128 additions & 115 deletions aws.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@ import (
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/cloudwatch"
"github.com/prometheus/client_golang/prometheus"
"time"
"regexp"
"strings"
"sync"
"time"
)

func getLatestDatapoint(datapoints []*cloudwatch.Datapoint) *cloudwatch.Datapoint {
Expand All @@ -26,163 +27,176 @@ func getLatestDatapoint(datapoints []*cloudwatch.Datapoint) *cloudwatch.Datapoin
// scrape makes the required calls to AWS CloudWatch by using the parameters in the cwCollector
// Once converted into Prometheus format, the metrics are pushed on the ch channel.
func scrape(collector *cwCollector, ch chan<- prometheus.Metric) {
session := session.Must(session.NewSession(&aws.Config{
awsSession := session.Must(session.NewSession(&aws.Config{
Region: aws.String(collector.Region),
}))

svc := cloudwatch.New(session)
svc := cloudwatch.New(awsSession)
wg := sync.WaitGroup{}
for m := range collector.Template.Metrics {
metric := &collector.Template.Metrics[m]
wg.Add(1)
go func(metric *cwMetric) {
scrapeMetric(collector, &wg, ch, metric, svc)
wg.Done()
}(&collector.Template.Metrics[m])
}
wg.Wait()
}

now := time.Now()
end := now.Add(time.Duration(-metric.ConfMetric.DelaySeconds) * time.Second)
func scrapeMetric(collector *cwCollector, wg *sync.WaitGroup, ch chan<- prometheus.Metric, metric *cwMetric, svc *cloudwatch.CloudWatch) {
now := time.Now()
end := now.Add(time.Duration(-metric.ConfMetric.DelaySeconds) * time.Second)

params := &cloudwatch.GetMetricStatisticsInput{
EndTime: aws.Time(end),
StartTime: aws.Time(end.Add(time.Duration(-metric.ConfMetric.RangeSeconds) * time.Second)),
params := &cloudwatch.GetMetricStatisticsInput{
EndTime: aws.Time(end),
StartTime: aws.Time(end.Add(time.Duration(-metric.ConfMetric.RangeSeconds) * time.Second)),

Period: aws.Int64(int64(metric.ConfMetric.PeriodSeconds)),
MetricName: aws.String(metric.ConfMetric.Name),
Namespace: aws.String(metric.ConfMetric.Namespace),
Dimensions: []*cloudwatch.Dimension{},
Statistics: []*string{},
Unit: nil,
}

dimensions:=[]*cloudwatch.Dimension{}
Period: aws.Int64(int64(metric.ConfMetric.PeriodSeconds)),
MetricName: aws.String(metric.ConfMetric.Name),
Namespace: aws.String(metric.ConfMetric.Namespace),
Dimensions: []*cloudwatch.Dimension{},
Statistics: []*string{},
Unit: nil,
}

//This map will hold dimensions name which has been already collected
valueCollected := map[string]bool{}
dimensions := []*cloudwatch.Dimension{}

//This map will hold dimensions name which has been already collected
valueCollected := map[string]bool{}

if len(metric.ConfMetric.DimensionsSelectRegex) == 0 {
metric.ConfMetric.DimensionsSelectRegex = map[string]string{}
}
if len(metric.ConfMetric.DimensionsSelectRegex) == 0 {
metric.ConfMetric.DimensionsSelectRegex = map[string]string{}
}

//Check for dimensions who does not have either select or dimensions select_regex and make them select everything using regex
for _,dimension := range metric.ConfMetric.Dimensions {
_, found := metric.ConfMetric.DimensionsSelect[dimension]
_, found2 := metric.ConfMetric.DimensionsSelectRegex[dimension]
if !found && !found2 {
metric.ConfMetric.DimensionsSelectRegex[dimension]=".*"
}
//Check for dimensions who does not have either select or dimensions select_regex and make them select everything using regex
for _, dimension := range metric.ConfMetric.Dimensions {
_, found := metric.ConfMetric.DimensionsSelect[dimension]
_, found2 := metric.ConfMetric.DimensionsSelectRegex[dimension]
if !found && !found2 {
metric.ConfMetric.DimensionsSelectRegex[dimension] = ".*"
}
}

for _, stat := range metric.ConfMetric.Statistics {
params.Statistics = append(params.Statistics, aws.String(stat))
}

labels := make([]string, 0, len(metric.LabelNames))

for _, stat := range metric.ConfMetric.Statistics {
params.Statistics = append(params.Statistics, aws.String(stat))
}
// Loop through the dimensions selects to build the filters and the labels array
for dim := range metric.ConfMetric.DimensionsSelect {
for val := range metric.ConfMetric.DimensionsSelect[dim] {
dimValue := metric.ConfMetric.DimensionsSelect[dim][val]

labels := make([]string, 0, len(metric.LabelNames))
// Replace $_target token by the actual URL target
if dimValue == "$_target" {
dimValue = collector.Target
}

// Loop through the dimensions selects to build the filters and the labels array
for dim := range metric.ConfMetric.DimensionsSelect {
for val := range metric.ConfMetric.DimensionsSelect[dim] {
dimValue := metric.ConfMetric.DimensionsSelect[dim][val]
dimensions = append(dimensions, &cloudwatch.Dimension{
Name: aws.String(dim),
Value: aws.String(dimValue),
})

// Replace $_target token by the actual URL target
if dimValue == "$_target" {
dimValue = collector.Target
}
labels = append(labels, dimValue)
}
}

dimensions = append(dimensions, &cloudwatch.Dimension{
Name: aws.String(dim),
Value: aws.String(dimValue),
})
if len(dimensions) > 0 || len(metric.ConfMetric.Dimensions) == 0 {
labels = append(labels, collector.Template.Task.Name)
paramsCopy := *params
paramsCopy.Dimensions = dimensions
wg.Add(1)
go func() {
scrapeSingleDataPoint(collector, ch, &paramsCopy, metric, labels, svc)
wg.Done()
}()
}

labels = append(labels, dimValue)
}
}
//If no regex is specified, continue
if len(metric.ConfMetric.DimensionsSelectRegex) == 0 {
return
}

if len(dimensions) > 0 || len(metric.ConfMetric.Dimensions) ==0 {
labels = append(labels, collector.Template.Task.Name)
params.Dimensions=dimensions
scrapeSingleDataPoint(collector,ch,params,metric,labels,svc)
}
// Get all the metric to select the ones who'll match the regex
result, err := svc.ListMetrics(&cloudwatch.ListMetricsInput{
MetricName: aws.String(metric.ConfMetric.Name),
Namespace: aws.String(metric.ConfMetric.Namespace),
})
nextToken := result.NextToken
metrics := result.Metrics
totalRequests.Inc()

//If no regex is specified, continue
if (len(metric.ConfMetric.DimensionsSelectRegex)==0){
continue
}
if err != nil {
fmt.Println(err)
return
}


// Get all the metric to select the ones who'll match the regex
for nextToken != nil {
result, err := svc.ListMetrics(&cloudwatch.ListMetricsInput{
MetricName: aws.String(metric.ConfMetric.Name),
Namespace: aws.String(metric.ConfMetric.Namespace),
NextToken: nextToken,
})
nextToken:=result.NextToken
metrics:=result.Metrics
totalRequests.Inc()

if err != nil {
fmt.Println(err)
continue
}
nextToken = result.NextToken
metrics = append(metrics, result.Metrics...)
}

//For each metric returned by aws
for _, met := range result.Metrics {
labels := make([]string, 0, len(metric.LabelNames))
dimensions = []*cloudwatch.Dimension{}

//Try to match each dimensions to the regex
for _, dim := range met.Dimensions {
dimRegex := metric.ConfMetric.DimensionsSelectRegex[*dim.Name]
if dimRegex == "" {
dimRegex = "\\b" + strings.Join(metric.ConfMetric.DimensionsSelect[*dim.Name], "\\b|\\b") + "\\b"
}

match, _ := regexp.MatchString(dimRegex, *dim.Value)
if match {
dimensions = append(dimensions, &cloudwatch.Dimension{
Name: aws.String(*dim.Name),
Value: aws.String(*dim.Value),
})
labels = append(labels, *dim.Value)

for nextToken!=nil {
result, err := svc.ListMetrics(&cloudwatch.ListMetricsInput{
MetricName: aws.String(metric.ConfMetric.Name),
Namespace: aws.String(metric.ConfMetric.Namespace),
NextToken: nextToken,
})
if err != nil {
fmt.Println(err)
continue
}
nextToken=result.NextToken
metrics=append(metrics,result.Metrics...)
}

//For each metric returned by aws
for _,met := range result.Metrics {
labels := make([]string, 0, len(metric.LabelNames))
dimensions=[]*cloudwatch.Dimension{}

//Try to match each dimensions to the regex
for _,dim := range met.Dimensions {
dimRegex:=metric.ConfMetric.DimensionsSelectRegex[*dim.Name]
if(dimRegex==""){
dimRegex="\\b"+strings.Join(metric.ConfMetric.DimensionsSelect[*dim.Name],"\\b|\\b")+"\\b"
}

match,_:=regexp.MatchString(dimRegex,*dim.Value)
if match {
dimensions=append(dimensions, &cloudwatch.Dimension{
Name: aws.String(*dim.Name),
Value: aws.String(*dim.Value),
})
labels = append(labels, *dim.Value)


}

//Cheking if all dimensions matched
if len(labels) == len(metric.ConfMetric.Dimensions) {

//Checking if this couple of dimensions has already been scraped
if _, ok := valueCollected[strings.Join(labels, ";")]; ok {
continue
}

//Cheking if all dimensions matched
if len(labels) == len(metric.ConfMetric.Dimensions) {
//If no, then scrape them
valueCollected[strings.Join(labels, ";")] = true

//Checking if this couple of dimensions has already been scraped
if _, ok := valueCollected[strings.Join(labels,";")]; ok {
continue
}
paramsCopy := *params
paramsCopy.Dimensions = dimensions

//If no, then scrape them
valueCollected[strings.Join(labels,";")]=true

params.Dimensions = dimensions
labels = append(labels, collector.Template.Task.Name)
wg.Add(1)
go func() {
scrapeSingleDataPoint(collector, ch, &paramsCopy, metric, labels, svc)
wg.Done()
}()

labels = append(labels, collector.Template.Task.Name)
scrapeSingleDataPoint(collector,ch,params,metric,labels,svc)

}
}
}
}

//Send a single dataPoint to the Prometheus lib
func scrapeSingleDataPoint(collector *cwCollector, ch chan<- prometheus.Metric,params *cloudwatch.GetMetricStatisticsInput,metric *cwMetric,labels []string,svc *cloudwatch.CloudWatch) error {
func scrapeSingleDataPoint(collector *cwCollector, ch chan<- prometheus.Metric, params *cloudwatch.GetMetricStatisticsInput, metric *cwMetric, labels []string, svc *cloudwatch.CloudWatch) error {

resp, err := svc.GetMetricStatistics(params)
totalRequests.Inc()

Expand All @@ -199,7 +213,6 @@ func scrapeSingleDataPoint(collector *cwCollector, ch chan<- prometheus.Metric,p
// Pick the latest datapoint
dp := getLatestDatapoint(resp.Datapoints)


if dp.Sum != nil {
ch <- prometheus.MustNewConstMetric(metric.Desc, metric.ValType, float64(*dp.Sum), labels...)
}
Expand Down