[libbeat] AWS ELB Monitoring

heartbeat/docs/autodiscover-aws-elb-config.asciidoc

@@ -22,3 +22,5 @@ heartbeat.autodiscover:
 -------------------------------------------------------------------------------------
 
 This configuration launches a `tcp` monitor for all ELBs that have a declared port (TODO: come up with a better condition).
+
+This autodiscover provider takes our standard AWS credentials options. Please see <<aws-credentials-config>> for more details.

libbeat/docs/shared-autodiscover.asciidoc

@@ -265,6 +265,8 @@ ifdef::autodiscoverAWSELB[]
 [float]
 ===== Amazon ELBs
 
+*Note: This provider is experimental*
+
 The Amazon ELB autodiscover provider discovers https://aws.amazon.com/elasticloadbalancing/[ELBs] and their listeners. This is useful when you don't want to connect
 directly to a service, but rather to the ELB fronting a pool of services.
 

x-pack/libbeat/autodiscover/providers/aws/elb/config.go

@@ -30,8 +30,8 @@ type Config struct {
 
 	// AWS Specific autodiscover fields
 
-	Regions []string `config:"regions" validate:"required"`
-	aws.ConfigAWS
+	Regions   []string      `config:"regions" validate:"required"`
+	AWSConfig aws.ConfigAWS `config:",inline"`
 }
 
 func defaultConfig() *Config {

x-pack/libbeat/autodiscover/providers/aws/elb/fetch.go

@@ -8,12 +8,11 @@ import (
 	"context"
 	"sync"
 
+	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
 	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/elasticloadbalancingv2iface"
+	"go.uber.org/multierr"
 
 	"github.com/elastic/beats/libbeat/logp"
-
-	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
-	"go.uber.org/multierr"
 )
 
 const logSelector = "autodiscover-elb-fetch"
@@ -137,15 +136,17 @@ func (p *fetchRequest) fetch() ([]*lbListener, error) {
 
 func (p *fetchRequest) fetchAllPages() {
 	// Keep fetching pages unless we're stopped OR there are no pages left
-Outer:
 	for {
 		select {
 		case <-p.context.Done():
 			logp.Debug(logSelector, "done fetching ELB pages, context cancelled")
-			break Outer
+			return
 		default:
-			p.fetchNextPage()
-			logp.Debug(logSelector, "API page fetched")
+			if !p.fetchNextPage() {
+				logp.Debug(logSelector, "fetched all ELB pages")
+				return
+			}
+			logp.Debug(logSelector, "fetched ELB page")
 		}
 	}
 }
@@ -189,12 +190,15 @@ func (p *fetchRequest) fetchListeners(lb elasticloadbalancingv2.LoadBalancer) {
 		p.recordErrResult(listen.Err())
 	}
 
-Outer:
-	for listen.Next(p.context) {
+	for {
 		select {
 		case <-p.context.Done():
-			break Outer
+			return
 		default:
+			if !listen.Next(p.context) {
+				return
+			}
+
 			for _, listener := range listen.CurrentPage().Listeners {
 				p.recordGoodResult(&lb, &listener)
 			}

x-pack/libbeat/autodiscover/providers/aws/elb/provider.go

@@ -54,10 +54,10 @@ func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config) (autodis
 	var clients []elasticloadbalancingv2iface.ClientAPI
 	for _, region := range config.Regions {
 		awsCfg, err := awscommon.GetAWSCredentials(awscommon.ConfigAWS{
-			AccessKeyID:     config.AccessKeyID,
-			SecretAccessKey: config.SecretAccessKey,
-			SessionToken:    config.SessionToken,
-			ProfileName:     config.ProfileName,
+			AccessKeyID:     config.AWSConfig.AccessKeyID,
+			SecretAccessKey: config.AWSConfig.SecretAccessKey,
+			SessionToken:    config.AWSConfig.SessionToken,
+			ProfileName:     config.AWSConfig.ProfileName,
 		})
 		if err != nil {
 			logp.Err("error loading AWS config for aws_elb autodiscover provider: %s", err)

x-pack/libbeat/autodiscover/providers/aws/elb/provider_test.go

@@ -9,13 +9,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/elastic/beats/libbeat/common"
-
 	"github.com/gofrs/uuid"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/bus"
 )
 

x-pack/libbeat/autodiscover/providers/aws/elb/watch.go

@@ -19,22 +19,22 @@ type watcher struct {
 	onStop      func(uuid string)
 	done        chan struct{}
 	ticker      *time.Ticker
-	interval    time.Duration
+	period      time.Duration
 	lbListeners map[string]uint64
 }
 
 func newWatcher(
 	fetcher fetcher,
-	interval time.Duration,
+	period time.Duration,
 	onStart func(uuid string, lblMap *lbListener),
 	onStop func(uuid string)) *watcher {
 	return &watcher{
 		fetcher:     fetcher,
 		onStart:     onStart,
 		onStop:      onStop,
 		done:        make(chan struct{}),
-		ticker:      time.NewTicker(interval),
-		interval:    interval,
+		ticker:      time.NewTicker(period),
+		period:      period,
 		lbListeners: map[string]uint64{},
 	}
 }
@@ -57,7 +57,6 @@ func (w *watcher) forever() {
 			err := w.once()
 			if err != nil {
 				logp.Err("error while fetching AWS ELBs: %s", err)
-				return
 			}
 		}
 	}
@@ -66,11 +65,12 @@ func (w *watcher) forever() {
 // once executes the watch loop a single time.
 // This is mostly useful for testing.
 func (w *watcher) once() error {
-	ctx, _ := context.WithTimeout(context.Background(), w.interval)
+	ctx, _ := context.WithTimeout(context.Background(), w.period)
 	fetchedLbls, err := w.fetcher.fetch(ctx)
 	if err != nil {
 		return err
 	}
+	logp.Info("fetched %d load balancer listeners from AWS for autodiscovery", len(fetchedLbls))
 
 	oldGen := w.gen
 	w.gen++