PMM-13912 Custom delay for Cloud Watch.

basic/collector.go

@@ -31,19 +31,21 @@ type Metric struct {
 }
 
 type Collector struct {
-	config   *config.Config
-	sessions *sessions.Sessions
-	metrics  []Metric
-	l        log.Logger
+	config          *config.Config
+	sessions        *sessions.Sessions
+	metrics         []Metric
+	l               log.Logger
+	cloudwatchDelay time.Duration
 }
 
 // New creates a new instance of a Collector.
-func New(config *config.Config, sessions *sessions.Sessions, logger log.Logger) *Collector {
+func New(config *config.Config, sessions *sessions.Sessions, logger log.Logger, delay time.Duration) *Collector {
 	return &Collector{
-		config:   config,
-		sessions: sessions,
-		metrics:  Metrics,
-		l:        log.With(logger, "component", "basic"),
+		config:          config,
+		sessions:        sessions,
+		metrics:         Metrics,
+		l:               log.With(logger, "component", "basic"),
+		cloudwatchDelay: delay,
 	}
 }
 
@@ -73,7 +75,11 @@ func (e *Collector) collect(ch chan<- prometheus.Metric) {
 		go func() {
 			defer wg.Done()
 
-			s := NewScraper(&instance, e, ch)
+			if e.cloudwatchDelay != defaultDelay {
+				level.Warn(e.l).Log("msg", fmt.Sprintf("Using custom CloudWatch delay %s for %s. Setting a very small delay may result in missing or incomplete metrics, as CloudWatch may not have published the latest datapoints yet.", e.cloudwatchDelay, instance))
+			}
+
+			s := NewScraper(&instance, e, ch, e.cloudwatchDelay)
 			if s == nil {
 				level.Error(e.l).Log("msg", fmt.Sprintf("No scraper for %s, skipping.", instance))
 				return

basic/collector_test.go

@@ -6,11 +6,11 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/percona/exporter_shared/helpers"
 	"github.com/prometheus/common/promlog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/percona/exporter_shared/helpers"
 	"github.com/percona/rds_exporter/client"
 	"github.com/percona/rds_exporter/config"
 	"github.com/percona/rds_exporter/sessions"
@@ -24,7 +24,7 @@ func TestCollector(t *testing.T) {
 	sess, err := sessions.New(cfg.Instances, client.HTTP(), logger, false)
 	require.NoError(t, err)
 
-	c := New(cfg, sess, logger)
+	c := New(cfg, sess, logger, defaultDelay)
 
 	actualMetrics := helpers.ReadMetrics(helpers.CollectMetrics(c))
 	sort.Slice(actualMetrics, func(i, j int) bool { return actualMetrics[i].Less(actualMetrics[j]) })
@@ -68,7 +68,7 @@ func TestCollectorDisableBasicMetrics(t *testing.T) {
 	sess, err := sessions.New(cfg.Instances, client.HTTP(), logger, false)
 	require.NoError(t, err)
 
-	c := New(cfg, sess, logger)
+	c := New(cfg, sess, logger, defaultDelay)
 
 	actualMetrics := helpers.ReadMetrics(helpers.CollectMetrics(c))
 	actualLines := helpers.Format(helpers.WriteMetrics(actualMetrics))

basic/scraper.go

@@ -16,22 +16,24 @@ import (
 
 var (
 	Period = 60 * time.Second
-	Delay  = 600 * time.Second
 	Range  = 600 * time.Second
+
+	defaultDelay = 600 * time.Second
 )
 
 type Scraper struct {
 	// params
 	instance  *config.Instance
 	collector *Collector
 	ch        chan<- prometheus.Metric
+	delay     time.Duration
 
 	// internal
 	svc         *cloudwatch.Client
 	constLabels prometheus.Labels
 }
 
-func NewScraper(instance *config.Instance, collector *Collector, ch chan<- prometheus.Metric) *Scraper {
+func NewScraper(instance *config.Instance, collector *Collector, ch chan<- prometheus.Metric, delay time.Duration) *Scraper {
 	cfg, _ := collector.sessions.GetConfig(instance.Region, instance.Instance)
 	if cfg == nil {
 		return nil
@@ -55,6 +57,7 @@ func NewScraper(instance *config.Instance, collector *Collector, ch chan<- prome
 		instance:  instance,
 		collector: collector,
 		ch:        ch,
+		delay:     delay,
 
 		// internal
 		svc:         svc,
@@ -93,7 +96,7 @@ func (s *Scraper) Scrape() {
 
 func (s *Scraper) scrapeMetric(metric Metric) error {
 	now := time.Now()
-	end := now.Add(-Delay)
+	end := now.Add(-s.delay)
 
 	params := &cloudwatch.GetMetricStatisticsInput{
 		EndTime:    aws.Time(end),

main.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -28,6 +29,7 @@ var (
 	enhancedMetricsPathF = kingpin.Flag("web.enhanced-telemetry-path", "Path under which to expose exporter's enhanced metrics.").Default("/enhanced").String()
 	configFileF          = kingpin.Flag("config.file", "Path to configuration file.").Default("config.yml").String()
 	logTraceF            = kingpin.Flag("log.trace", "Enable verbose tracing of AWS requests (will log credentials).").Default("false").Bool()
+	cloudwatchDelayF     = kingpin.Flag("cloudwatch-delay", "Delay (in seconds) for CloudWatch metrics scraping. Default is 600.").Default("600").Int()
 	logger               = log.NewNopLogger()
 )
 
@@ -54,9 +56,11 @@ func main() {
 		os.Exit(1)
 	}
 
+	cloudwatchDelay := time.Duration(*cloudwatchDelayF) * time.Second
+
 	// basic metrics + client metrics + exporter own metrics (ProcessCollector and GoCollector)
 	{
-		prometheus.MustRegister(basic.New(cfg, sess, logger))
+		prometheus.MustRegister(basic.New(cfg, sess, logger, cloudwatchDelay))
 		prometheus.MustRegister(client)
 		http.Handle(*basicMetricsPathF, promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{
 			//ErrorLog:      log.NewErrorLogger(), TODO TS