[Metricbeat] Add latency config option into aws module

CHANGELOG.next.asciidoc

@@ -671,6 +671,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add host inventory metrics to azure compute_vm metricset. {pull}20641[20641]
 - Add host inventory metrics to googlecloud compute metricset. {pull}20391[20391]
 - Request prometheus endpoints to be gzipped by default {pull}20766[20766]
+- Add latency config parameter into aws module. {pull}20875[20875]
 - Release all kubernetes `state` metricsets as GA {pull}20901[20901]
 - Add billing metricset into googlecloud module. {pull}20812[20812] {issue}20738[20738]
 

metricbeat/docs/modules/aws.asciidoc

@@ -19,16 +19,27 @@ module. Please see <<aws-api-requests,AWS API requests>> for more details.
 [float]
 == Module-specific configuration notes
 
+* *AWS Credentials*
+
 The `aws` module requires AWS credentials configuration in order to make AWS API calls.
 Users can either use `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` and/or
 `AWS_SESSION_TOKEN`, or use shared AWS credentials file.
 Please see <<aws-credentials-config,AWS credentials options>> for more details.
 
+* *regions*
+
 This module also accepts optional configuration `regions` to specify which
 AWS regions to query metrics from. If the `regions` parameter is not set in the
 config file, then by default, the `aws` module will query metrics from all available
 AWS regions.
 
+* *latency*
+
+Some AWS services send monitoring metrics to CloudWatch with a latency to
+process larger than Metricbeat collection period. This case, please specify a
+`latency` parameter so collection start time and end time will be shifted by the
+given latency amount.
+
 The aws module comes with a predefined dashboard. For example:
 
 image::./images/metricbeat-aws-overview.png[]

x-pack/metricbeat/module/aws/_meta/config.yml

@@ -39,4 +39,8 @@
   period: 24h
   metricsets:
     - s3_daily_storage
+- module: aws
+  period: 1m
+  latency: 5m
+  metricsets:
     - s3_request

x-pack/metricbeat/module/aws/_meta/docs.asciidoc

@@ -11,16 +11,27 @@ module. Please see <<aws-api-requests,AWS API requests>> for more details.
 [float]
 == Module-specific configuration notes
 
+* *AWS Credentials*
+
 The `aws` module requires AWS credentials configuration in order to make AWS API calls.
 Users can either use `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY` and/or
 `AWS_SESSION_TOKEN`, or use shared AWS credentials file.
 Please see <<aws-credentials-config,AWS credentials options>> for more details.
 
+* *regions*
+
 This module also accepts optional configuration `regions` to specify which
 AWS regions to query metrics from. If the `regions` parameter is not set in the
 config file, then by default, the `aws` module will query metrics from all available
 AWS regions.
 
+* *latency*
+
+Some AWS services send monitoring metrics to CloudWatch with a latency to
+process larger than Metricbeat collection period. This case, please specify a
+`latency` parameter so collection start time and end time will be shifted by the
+given latency amount.
+
 The aws module comes with a predefined dashboard. For example:
 
 image::./images/metricbeat-aws-overview.png[]

x-pack/metricbeat/module/aws/aws.go

@@ -26,6 +26,7 @@ import (
 type Config struct {
 	Period     time.Duration       `config:"period" validate:"nonzero,required"`
 	Regions    []string            `config:"regions"`
+	Latency    time.Duration       `config:"latency"`
 	AWSConfig  awscommon.ConfigAWS `config:",inline"`
 	TagsFilter []Tag               `config:"tags_filter"`
 }
@@ -36,6 +37,7 @@ type MetricSet struct {
 	RegionsList []string
 	Endpoint    string
 	Period      time.Duration
+	Latency     time.Duration
 	AwsConfig   *awssdk.Config
 	AccountName string
 	AccountID   string
@@ -86,6 +88,7 @@ func NewMetricSet(base mb.BaseMetricSet) (*MetricSet, error) {
 	metricSet := MetricSet{
 		BaseMetricSet: base,
 		Period:        config.Period,
+		Latency:       config.Latency,
 		AwsConfig:     &awsConfig,
 		TagsFilter:    config.TagsFilter,
 	}

x-pack/metricbeat/module/aws/cloudwatch/cloudwatch.go

@@ -141,7 +141,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	// Get startTime and endTime
-	startTime, endTime := aws.GetStartTimeEndTime(m.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.Period, m.Latency)
+	m.Logger().Debugf("startTime = %s, endTime = %s", startTime, endTime)
 
 	// Check statistic method in config
 	err := m.checkStatistics()

x-pack/metricbeat/module/aws/cloudwatch/cloudwatch_test.go

@@ -1357,7 +1357,7 @@ func TestCreateEventsWithIdentifier(t *testing.T) {
 			Value: "test-ec2",
 		},
 	}
-	startTime, endTime := aws.GetStartTimeEndTime(m.MetricSet.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.MetricSet.Period, m.MetricSet.Latency)
 
 	events, err := m.createEvents(mockCloudwatchSvc, mockTaggingSvc, listMetricWithStatsTotal, resourceTypeTagFilters, regionName, startTime, endTime)
 	assert.NoError(t, err)
@@ -1399,7 +1399,7 @@ func TestCreateEventsWithoutIdentifier(t *testing.T) {
 	}
 
 	resourceTypeTagFilters := map[string][]aws.Tag{}
-	startTime, endTime := aws.GetStartTimeEndTime(m.MetricSet.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.MetricSet.Period, m.MetricSet.Latency)
 
 	events, err := m.createEvents(mockCloudwatchSvc, mockTaggingSvc, listMetricWithStatsTotal, resourceTypeTagFilters, regionName, startTime, endTime)
 	assert.NoError(t, err)
@@ -1447,7 +1447,7 @@ func TestCreateEventsWithTagsFilter(t *testing.T) {
 			Value: "foo",
 		},
 	}
-	startTime, endTime := aws.GetStartTimeEndTime(m.MetricSet.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.MetricSet.Period, m.MetricSet.Latency)
 
 	events, err := m.createEvents(mockCloudwatchSvc, mockTaggingSvc, listMetricWithStatsTotal, resourceTypeTagFilters, regionName, startTime, endTime)
 	assert.NoError(t, err)

x-pack/metricbeat/module/aws/ec2/ec2.go

@@ -88,7 +88,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	// Get startTime and endTime
-	startTime, endTime := aws.GetStartTimeEndTime(m.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.Period, m.Latency)
+	m.Logger().Debugf("startTime = %s, endTime = %s", startTime, endTime)
 
 	for _, regionName := range m.MetricSet.RegionsList {
 		awsConfig := m.MetricSet.AwsConfig.Copy()

x-pack/metricbeat/module/aws/rds/rds.go

@@ -79,7 +79,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	// Get startTime and endTime
-	startTime, endTime := aws.GetStartTimeEndTime(m.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.Period, m.Latency)
+	m.Logger().Debugf("startTime = %s, endTime = %s", startTime, endTime)
 
 	for _, regionName := range m.MetricSet.RegionsList {
 		awsConfig := m.MetricSet.AwsConfig.Copy()

x-pack/metricbeat/module/aws/s3_daily_storage/s3_daily_storage.go

@@ -69,7 +69,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	namespace := "AWS/S3"
 	// Get startTime and endTime
-	startTime, endTime := aws.GetStartTimeEndTime(m.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.Period, m.Latency)
+	m.Logger().Debugf("startTime = %s, endTime = %s", startTime, endTime)
 
 	// GetMetricData for AWS S3 from Cloudwatch
 	for _, regionName := range m.MetricSet.RegionsList {

x-pack/metricbeat/module/aws/s3_request/s3_request.go

@@ -69,7 +69,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	namespace := "AWS/S3"
 	// Get startTime and endTime
-	startTime, endTime := aws.GetStartTimeEndTime(m.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.Period, m.Latency)
+	m.Logger().Debugf("startTime = %s, endTime = %s", startTime, endTime)
 
 	// GetMetricData for AWS S3 from Cloudwatch
 	for _, regionName := range m.MetricSet.RegionsList {

x-pack/metricbeat/module/aws/sqs/sqs.go

@@ -67,7 +67,8 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
 	namespace := "AWS/SQS"
 	// Get startTime and endTime
-	startTime, endTime := aws.GetStartTimeEndTime(m.Period)
+	startTime, endTime := aws.GetStartTimeEndTime(m.Period, m.Latency)
+	m.Logger().Debugf("startTime = %s, endTime = %s", startTime, endTime)
 
 	for _, regionName := range m.MetricSet.RegionsList {
 		awsConfig := m.MetricSet.AwsConfig.Copy()

x-pack/metricbeat/module/aws/utils.go

@@ -22,8 +22,13 @@ import (
 )
 
 // GetStartTimeEndTime function uses durationString to create startTime and endTime for queries.
-func GetStartTimeEndTime(period time.Duration) (time.Time, time.Time) {
+func GetStartTimeEndTime(period time.Duration, latency time.Duration) (time.Time, time.Time) {
 	endTime := time.Now()
+	if latency != 0 {
+		// add latency if config is not 0
+		endTime = endTime.Add(latency * -1)
+	}
+
 	// Set startTime double the period earlier than the endtime in order to
 	// make sure GetMetricDataRequest gets the latest data point for each metric.
 	return endTime.Add(period * -2), endTime

x-pack/metricbeat/module/aws/utils_test.go

@@ -176,7 +176,7 @@ func TestGetListMetricsOutputWithWildcard(t *testing.T) {
 }
 
 func TestGetMetricDataPerRegion(t *testing.T) {
-	startTime, endTime := GetStartTimeEndTime(10 * time.Minute)
+	startTime, endTime := GetStartTimeEndTime(10*time.Minute, 0)
 
 	mockSvc := &MockCloudWatchClient{}
 	var metricDataQueries []cloudwatch.MetricDataQuery
@@ -205,7 +205,7 @@ func TestGetMetricDataPerRegion(t *testing.T) {
 }
 
 func TestGetMetricDataResults(t *testing.T) {
-	startTime, endTime := GetStartTimeEndTime(10 * time.Minute)
+	startTime, endTime := GetStartTimeEndTime(10*time.Minute, 0)
 
 	mockSvc := &MockCloudWatchClient{}
 	metricInfo := cloudwatch.Metric{

x-pack/metricbeat/modules.d/aws.yml.disabled

@@ -42,4 +42,8 @@
   period: 24h
   metricsets:
     - s3_daily_storage
+- module: aws
+  period: 1m
+  latency: 5m
+  metricsets:
     - s3_request