Fix race condition when fetching resources (CoverGenius#23)

* Fix race condition when fetching resources (#4)

Currently when fetching resources an empty array for the current list of resources is initialized and is saved in place of the existing list. If this list is accessed by one of the metrics go routines before it has been fully populated then that metric will not be populated for the missing resources.

This diff adds locking and changes the new array to only be saved once fully populated to prevent such data races from occurring.

* Fix bug causing all default metrics to only use average statistic

Author: aywrite

base/config.go

@@ -13,8 +13,8 @@ type configMetric struct {
 	Dimensions    []*cloudwatch.Dimension `yaml:"dimensions"`     // The resource dimensions to generate individual series for (via labels)
 	Statistics    []*string               `yaml:"statistics"`     // List of AWS statistics to use.
 	OutputName    string                  `yaml:"output_name"`    // Allows override of the generate metric name
-	RangeSeconds  int64                   `yaml:"range_seconds"`  // How far back to request data for in seconds.
 	PeriodSeconds int64                   `yaml:"period_seconds"` // Granularity of results from cloudwatch API.
+	RangeSeconds  int64                   `yaml:"range_seconds"`  // How far back to request data for in seconds.
 }
 
 type metric struct {
@@ -43,8 +43,7 @@ type Config struct {
 func (c *Config) ConstructMetrics(defaults map[string]map[string]*MetricDescription) map[string][]*MetricDescription {
 	mds := make(map[string][]*MetricDescription)
 	for namespace, metrics := range c.Metrics.Data {
-
-		if len(metrics) <= 0 {
+		if len(metrics) == 0 {
 			if namespaceDefaults, ok := defaults[namespace]; ok {
 				for key, defaultMetric := range namespaceDefaults {
 					metrics = append(metrics, &configMetric{
@@ -54,6 +53,7 @@ func (c *Config) ConstructMetrics(defaults map[string]map[string]*MetricDescript
 						PeriodSeconds: defaultMetric.PeriodSeconds,
 						RangeSeconds:  defaultMetric.RangeSeconds,
 						Dimensions:    defaultMetric.Dimensions,
+						Statistics:    defaultMetric.Statistic,
 					})
 				}
 			}

base/controller.go

@@ -25,7 +25,7 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-var alphaRegex, _ = regexp.Compile("[^a-zA-Z0-9]+")
+var alphaRegex = regexp.MustCompile("[^a-zA-Z0-9]+")
 
 // TagDescription represents an AWS tag key value pair
 type TagDescription struct {
@@ -120,7 +120,7 @@ func (rd *RegionDescription) BuildARN(s *string, r *string) (string, error) {
 	return a.String(), nil
 }
 
-func (rd *RegionDescription) buildFilters() error {
+func (rd *RegionDescription) saveFilters() {
 	filters := []*ec2.Filter{}
 	for _, tag := range rd.Tags {
 		f := &ec2.Filter{
@@ -130,16 +130,20 @@ func (rd *RegionDescription) buildFilters() error {
 		filters = append(filters, f)
 	}
 	rd.Filters = filters
-	return nil
 }
 
 func (rd *RegionDescription) saveAccountID() error {
 	session := iam.New(rd.Session)
 	input := iam.GetUserInput{}
 	user, err := session.GetUser(&input)
-	h.LogError(err)
+	if err != nil {
+		return err
+	}
+
 	a, err := arn.Parse(*user.User.Arn)
-	h.LogError(err)
+	if err != nil {
+		return err
+	}
 	rd.AccountID = &a.AccountID
 
 	return nil
@@ -153,13 +157,16 @@ func (rd *RegionDescription) Init(s *session.Session, td []*TagDescription, metr
 	rd.Tags = td
 
 	err := rd.saveAccountID()
-	h.LogErrorExit(err)
+	if err != nil {
+		return fmt.Errorf("error saving account id: %s", err)
+	}
 
-	err = rd.buildFilters()
-	h.LogErrorExit(err)
+	rd.saveFilters()
 
 	err = rd.CreateNamespaceDescriptions(metrics)
-	h.LogErrorExit(err)
+	if err != nil {
+		return fmt.Errorf("error creating namespaces: %s", err)
+	}
 
 	return nil
 }
@@ -196,9 +203,9 @@ func (nd *NamespaceDescription) GatherMetrics(cw *cloudwatch.CloudWatch) {
 	for _, md := range nd.Metrics {
 		go func(md *MetricDescription) {
 			nd.Mutex.RLock()
-			result, err := md.getData(cw, nd.Resources, nd)
+			result, err := md.getData(cw, nd.Resources)
 			nd.Mutex.RUnlock()
-			h.LogError(err)
+			h.LogIfError(err)
 			md.saveData(result, *nd.Parent.Region)
 		}(md)
 	}
@@ -295,7 +302,7 @@ func (md *MetricDescription) saveData(c *cloudwatch.GetMetricDataOutput, region
 
 		labels, err := awsLabelsFromString(*data.Label)
 		if err != nil {
-			h.LogError(err)
+			h.LogIfError(err)
 			continue
 		}
 
@@ -320,7 +327,7 @@ func (md *MetricDescription) saveData(c *cloudwatch.GetMetricDataOutput, region
 			err = fmt.Errorf("unknown statistic type: %s", labels.statistic)
 		}
 		if err != nil {
-			h.LogError(err)
+			h.LogIfError(err)
 			continue
 		}
 
@@ -448,12 +455,12 @@ func (rd *RegionDescription) TagsFound(tl interface{}) bool {
 	return false
 }
 
-func (md *MetricDescription) getData(cw *cloudwatch.CloudWatch, rds []*ResourceDescription, nd *NamespaceDescription) (*cloudwatch.GetMetricDataOutput, error) {
+func (md *MetricDescription) getData(cw *cloudwatch.CloudWatch, rds []*ResourceDescription) (*cloudwatch.GetMetricDataOutput, error) {
 	query, err := md.BuildQuery(rds)
-	if len(query) < 1 {
+	if len(query) == 0 {
 		return &cloudwatch.GetMetricDataOutput{}, nil
 	}
-	h.LogError(err)
+	h.LogIfError(err)
 
 	end := time.Now().Round(5 * time.Minute)
 	start := end.Add(-time.Duration(md.RangeSeconds) * time.Second)
@@ -464,7 +471,7 @@ func (md *MetricDescription) getData(cw *cloudwatch.CloudWatch, rds []*ResourceD
 		MetricDataQueries: query,
 	}
 	result, err := cw.GetMetricData(&input)
-	h.LogError(err)
+	h.LogIfError(err)
 
 	return result, err
 }

base/prometheus.go

@@ -132,5 +132,4 @@ func NewBatchCounterVec(opts prometheus.Opts, labels []string) *BatchCounterVec
 			labels,
 		),
 	}
-
 }

build/cloudwatch-prometheus-exporter.spec

@@ -5,7 +5,7 @@
 
 
 Name: cloudwatch-prometheus-exporter
-Version: 0.1.0
+Version: 0.1.1
 Release: 0%{?dist}
 Summary: Cloudwatch Prometheus Exporter
 License: BSD
@@ -38,6 +38,8 @@ rm -rf %{buildroot}
 
 
 %changelog
+* Thu Apr 16 2020 Andrew Wright <andrew.w@covergenius.com>
+- Fix race condition when fetching resource list
 * Wed Apr 15 2020 Andrew Wright <andrew.w@covergenius.com>
 - Allow runtime configuration of metrics
 * Wed Apr 08 2020 Andrew Wright <andrew.w@covergenius.com>

ec2/controller.go

@@ -11,16 +11,17 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 )
 
-func CreateResourceDescription(nd *b.NamespaceDescription, instance *ec2.Instance) error {
+func createResourceDescription(nd *b.NamespaceDescription, instance *ec2.Instance) (*b.ResourceDescription, error) {
 	rd := b.ResourceDescription{}
 	dd := []*b.DimensionDescription{
 		{
 			Name:  aws.String("InstanceId"),
 			Value: instance.InstanceId,
 		},
 	}
-	err := rd.BuildDimensions(dd)
-	h.LogError(err)
+	if err := rd.BuildDimensions(dd); err != nil {
+		return nil, err
+	}
 
 	tags := make(map[string]*string)
 	for _, t := range instance.Tags {
@@ -34,29 +35,32 @@ func CreateResourceDescription(nd *b.NamespaceDescription, instance *ec2.Instanc
 	}
 	rd.Type = aws.String("ec2")
 	rd.Parent = nd
-	nd.Resources = append(nd.Resources, &rd)
 
-	return nil
+	return &rd, nil
 }
 
 // CreateResourceList fetches a list of all EC2 instances in the parent region
-func CreateResourceList(nd *b.NamespaceDescription, wg *sync.WaitGroup) error {
+func CreateResourceList(nd *b.NamespaceDescription, wg *sync.WaitGroup) {
 	defer wg.Done()
 	log.Debug("Creating EC2 resource list ...")
 
-	nd.Resources = []*b.ResourceDescription{}
 	session := ec2.New(nd.Parent.Session)
 	input := ec2.DescribeInstancesInput{
 		Filters: nd.Parent.Filters,
 	}
 	result, err := session.DescribeInstances(&input)
-	h.LogError(err)
+	h.LogIfError(err)
 
+	resources := []*b.ResourceDescription{}
 	for _, reservation := range result.Reservations {
 		for _, instance := range reservation.Instances {
-			err := CreateResourceDescription(nd, instance)
-			h.LogError(err)
+			if r, err := createResourceDescription(nd, instance); err == nil {
+				resources = append(resources, r)
+			}
+			h.LogIfError(err)
 		}
 	}
-	return nil
+	nd.Mutex.Lock()
+	nd.Resources = resources
+	nd.Mutex.Unlock()
 }

elasticache/controller.go

@@ -12,7 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/elasticache"
 )
 
-func CreateResourceDescription(nd *b.NamespaceDescription, cc *elasticache.CacheCluster) (*b.ResourceDescription, error) {
+func createResourceDescription(nd *b.NamespaceDescription, cc *elasticache.CacheCluster) (*b.ResourceDescription, error) {
 	rd := b.ResourceDescription{}
 	dd := []*b.DimensionDescription{
 		{
@@ -21,7 +21,7 @@ func CreateResourceDescription(nd *b.NamespaceDescription, cc *elasticache.Cache
 		},
 	}
 	err := rd.BuildDimensions(dd)
-	h.LogError(err)
+	h.LogIfError(err)
 	rd.ID = cc.CacheClusterId
 	rd.Name = cc.CacheClusterId
 	rd.Type = aws.String("elasticache")
@@ -31,48 +31,49 @@ func CreateResourceDescription(nd *b.NamespaceDescription, cc *elasticache.Cache
 }
 
 // CreateResourceList fetches a list of all Elasticache clusters in the parent region
-func CreateResourceList(nd *b.NamespaceDescription, wg *sync.WaitGroup) error {
+func CreateResourceList(nd *b.NamespaceDescription, wg *sync.WaitGroup) {
 	defer wg.Done()
 	log.Debug("Creating Elasticache resource list ...")
 
-	resources := []*b.ResourceDescription{}
 	session := elasticache.New(nd.Parent.Session)
 	input := elasticache.DescribeCacheClustersInput{}
 	result, err := session.DescribeCacheClusters(&input)
-	h.LogError(err)
+	h.LogIfError(err)
 	service := "elasticache"
 
 	var w sync.WaitGroup
 	w.Add(len(result.CacheClusters))
-	ch := make(chan (*b.ResourceDescription), len(result.CacheClusters))
+	ch := make(chan *b.ResourceDescription, len(result.CacheClusters))
 	for _, cc := range result.CacheClusters {
 		go func(cc *elasticache.CacheCluster, wg *sync.WaitGroup) {
 			defer wg.Done()
+
 			resource := strings.Join([]string{"cluster", *cc.CacheClusterId}, ":")
 			arn, err := nd.Parent.BuildARN(&service, &resource)
-			h.LogError(err)
+			h.LogIfError(err)
+
 			input := elasticache.ListTagsForResourceInput{
 				ResourceName: aws.String(arn),
 			}
 			tags, err := session.ListTagsForResource(&input)
-			h.LogError(err)
+			h.LogIfError(err)
 
 			if nd.Parent.TagsFound(tags) {
-				if r, err := CreateResourceDescription(nd, cc); err == nil {
+				if r, err := createResourceDescription(nd, cc); err == nil {
 					ch <- r
 				}
-				h.LogError(err)
+				h.LogIfError(err)
 			}
 		}(cc, &w)
 	}
 	w.Wait()
 	close(ch)
+
+	resources := []*b.ResourceDescription{}
 	for r := range ch {
 		resources = append(resources, r)
 	}
-
 	nd.Mutex.Lock()
 	nd.Resources = resources
 	nd.Mutex.Unlock()
-	return nil
 }

elb/controller.go

@@ -11,7 +11,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/elb"
 )
 
-func CreateResourceDescription(nd *b.NamespaceDescription, td *elb.TagDescription) error {
+func createResourceDescription(nd *b.NamespaceDescription, td *elb.TagDescription) (*b.ResourceDescription, error) {
 	rd := b.ResourceDescription{}
 	dd := []*b.DimensionDescription{
 		{
@@ -20,47 +20,50 @@ func CreateResourceDescription(nd *b.NamespaceDescription, td *elb.TagDescriptio
 		},
 	}
 	err := rd.BuildDimensions(dd)
-	h.LogError(err)
+	h.LogIfError(err)
 	rd.ID = td.LoadBalancerName
 	rd.Name = td.LoadBalancerName
 	rd.Type = aws.String("lb-classic")
 	rd.Parent = nd
-	nd.Resources = append(nd.Resources, &rd)
 
-	return nil
+	return &rd, nil
 }
 
 // CreateResourceList fetches a list of all Classic LB resources in the region
-func CreateResourceList(nd *b.NamespaceDescription, wg *sync.WaitGroup) error {
+func CreateResourceList(nd *b.NamespaceDescription, wg *sync.WaitGroup) {
 	defer wg.Done()
 	log.Debug("Creating Classic LB resource list ...")
-	nd.Resources = []*b.ResourceDescription{}
 	session := elb.New(nd.Parent.Session)
 	input := elb.DescribeLoadBalancersInput{}
 	result, err := session.DescribeLoadBalancers(&input)
-	h.LogError(err)
+	h.LogIfError(err)
 
 	resourceList := []*string{}
 	for _, lb := range result.LoadBalancerDescriptions {
 		resourceList = append(resourceList, lb.LoadBalancerName)
 	}
 	if len(resourceList) <= 0 {
-		return nil
+		return
 	}
 
 	dti := elb.DescribeTagsInput{
 		LoadBalancerNames: resourceList,
 	}
 	tags, err := session.DescribeTags(&dti)
-	h.LogError(err)
+	h.LogIfError(err)
 
+	resources := []*b.ResourceDescription{}
 	for _, td := range tags.TagDescriptions {
 		if nd.Parent.TagsFound(td) {
-			err := CreateResourceDescription(nd, td)
-			h.LogError(err)
+			if r, err := createResourceDescription(nd, td); err == nil {
+				resources = append(resources, r)
+			}
+			h.LogIfError(err)
 		} else {
 			continue
 		}
 	}
-	return nil
+	nd.Mutex.Lock()
+	nd.Resources = resources
+	nd.Mutex.Unlock()
 }