[MetricBeat] [AWS] Fix aws metric tags with resourcegroupstaggingapi …

…paginator (elastic#26385) (elastic#26443) (elastic#26630) (cherry picked from commit 179108c) Co-authored-by: martin <kwinstonix@users.noreply.github.com>
jsoriano · Jun 30, 2021 · f23335c · f23335c
1 parent 5217a13
commit f23335c
Show file tree

Hide file tree

Showing 4 changed files with 53 additions and 20 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -385,6 +385,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix azure billing dashboard. {pull}25554[25554]
 - Major refactor of system/cpu and system/core metrics. {pull}25771[25771]
 - Fix GCP Project ID being ingested as `cloud.account.id` in `gcp.billing` module {issue}26357[26357] {pull}26412[26412]
+- Fix aws metric tags with resourcegroupstaggingapi paginator.  {issue}26385[26385] {pull}26443[26443]
 
 *Packetbeat*
 

diff --git a/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go b/x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go
@@ -1309,7 +1309,15 @@ func (m *MockCloudWatchClientWithoutDim) GetMetricDataRequest(input *cloudwatch.
 func (m *MockResourceGroupsTaggingClient) GetResourcesRequest(input *resourcegroupstaggingapi.GetResourcesInput) resourcegroupstaggingapi.GetResourcesRequest {
 	httpReq, _ := http.NewRequest("", "", nil)
 	return resourcegroupstaggingapi.GetResourcesRequest{
+		Input: input,
+		Copy:  m.GetResourcesRequest,
 		Request: &awssdk.Request{
+			Operation: &awssdk.Operation{
+				Name:       "GetResources",
+				HTTPMethod: "POST",
+				HTTPPath:   "/",
+				Paginator:  nil,
+			},
 			Data: &resourcegroupstaggingapi.GetResourcesOutput{
 				PaginationToken: awssdk.String(""),
 				ResourceTagMappingList: []resourcegroupstaggingapi.ResourceTagMapping{

diff --git a/x-pack/metricbeat/module/aws/utils.go b/x-pack/metricbeat/module/aws/utils.go
@@ -177,39 +177,33 @@ func GetResourcesTags(svc resourcegroupstaggingapiiface.ClientAPI, resourceTypeF
 		ResourceTypeFilters: resourceTypeFilters,
 	}
 
-	init := true
-	for init || *getResourcesInput.PaginationToken != "" {
-		init = false
-		getResourcesRequest := svc.GetResourcesRequest(getResourcesInput)
-		output, err := getResourcesRequest.Send(context.TODO())
-		if err != nil {
-			err = errors.Wrap(err, "error GetResources")
-			return nil, err
-		}
-
-		getResourcesInput.PaginationToken = output.PaginationToken
-		if resourceTypeFilters == nil || len(output.ResourceTagMappingList) == 0 {
-			return nil, nil
-		}
-
-		for _, resourceTag := range output.ResourceTagMappingList {
+	getResourcesRequest := svc.GetResourcesRequest(getResourcesInput)
+	paginator := resourcegroupstaggingapi.NewGetResourcesPaginator(getResourcesRequest)
+	for paginator.Next(context.TODO()) {
+		page := paginator.CurrentPage()
+		for _, resourceTag := range page.ResourceTagMappingList {
 			shortIdentifier, err := FindShortIdentifierFromARN(*resourceTag.ResourceARN)
 			if err == nil {
 				resourceTagMap[shortIdentifier] = resourceTag.Tags
 			} else {
-				err = errors.Wrap(err, "error occurs when proccessing shortIdentifier")
+				err = errors.Wrap(err, "error occurs when processing shortIdentifier")
 				return nil, err
 			}
 
 			wholeIdentifier, err := FindWholeIdentifierFromARN(*resourceTag.ResourceARN)
 			if err == nil {
 				resourceTagMap[wholeIdentifier] = resourceTag.Tags
 			} else {
-				err = errors.Wrap(err, "error occurs when proccessing longIdentifier")
+				err = errors.Wrap(err, "error occurs when processing longIdentifier")
 				return nil, err
 			}
 		}
 	}
+
+	if err := paginator.Err(); err != nil {
+		err = errors.Wrap(err, "error GetResources with Paginator")
+		return nil, err
+	}
 	return resourceTagMap, nil
 }
 

diff --git a/x-pack/metricbeat/module/aws/utils_test.go b/x-pack/metricbeat/module/aws/utils_test.go
@@ -113,10 +113,17 @@ func (m *MockCloudWatchClient) GetMetricDataRequest(input *cloudwatch.GetMetricD
 
 func (m *MockResourceGroupsTaggingClient) GetResourcesRequest(input *resourcegroupstaggingapi.GetResourcesInput) resourcegroupstaggingapi.GetResourcesRequest {
 	httpReq, _ := http.NewRequest("", "", nil)
-	return resourcegroupstaggingapi.GetResourcesRequest{
+	op := &awssdk.Operation{
+		Name:       "GetResources",
+		HTTPMethod: "POST",
+		HTTPPath:   "/",
+		Paginator:  nil,
+	}
+	firstPageResult := resourcegroupstaggingapi.GetResourcesRequest{
 		Request: &awssdk.Request{
+			Operation: op,
 			Data: &resourcegroupstaggingapi.GetResourcesOutput{
-				PaginationToken: awssdk.String(""),
+				PaginationToken: awssdk.String("PaginationToken"),
 				ResourceTagMappingList: []resourcegroupstaggingapi.ResourceTagMapping{
 					{
 						ResourceARN: awssdk.String("arn:aws:rds:eu-west-1:123456789012:db:mysql-db-1"),
@@ -148,6 +155,29 @@ func (m *MockResourceGroupsTaggingClient) GetResourcesRequest(input *resourcegro
 			},
 			HTTPRequest: httpReq,
 		},
+		Input: input,
+		Copy:  m.GetResourcesRequest,
+	}
+
+	// aws resourcegroupstaggingapi default pagination size is 50, if resource amount is a
+	// multiple of 50, then last request has an empty result.
+	lastPageWithEmptyResult := resourcegroupstaggingapi.GetResourcesRequest{
+		Request: &awssdk.Request{
+			Data: &resourcegroupstaggingapi.GetResourcesOutput{
+				PaginationToken:        awssdk.String(""),
+				ResourceTagMappingList: []resourcegroupstaggingapi.ResourceTagMapping{},
+			},
+			HTTPRequest: httpReq,
+			Operation:   op,
+		},
+		Input: input,
+		Copy:  m.GetResourcesRequest,
+	}
+
+	if input.PaginationToken == nil {
+		return firstPageResult
+	} else {
+		return lastPageWithEmptyResult
 	}
 }