[libbeat] AWS ELB Monitoring

vendor/vendor.json

@@ -234,10 +234,18 @@
 		{
 			"checksumSHA1": "427ei4bQ6ap8N/Hrw7NpxOqjd+Y=",
 			"path": "github.com/aws/aws-sdk-go-v2/aws",
-			"revision": "da6472f8447d3fef65272d1e396e892df15310d5",
-			"revisionTime": "2019-05-29T23:50:13Z",
-			"version": "v2.0.0-preview.5",
-			"versionExact": "v2.0.0-preview.5"
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
+		},
+		{
+			"checksumSHA1": "hKCG2tJrxNuV9sHh8TWDQ643i8Y=",
+			"path": "github.com/aws/aws-sdk-go-v2/aws/arn",
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
 			"checksumSHA1": "DVN6AdXnfTwEOLJwCOhFqJAEF1E=",
@@ -418,18 +426,18 @@
 		{
 			"checksumSHA1": "PKzWnwKTiLmXB2b3PAplL0zLyCM=",
 			"path": "github.com/aws/aws-sdk-go-v2/service/cloudformation",
-			"revision": "da6472f8447d3fef65272d1e396e892df15310d5",
-			"revisionTime": "2019-05-29T23:50:13Z",
-			"version": "v2.0.0-preview.5",
-			"versionExact": "v2.0.0-preview.5"
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
 			"checksumSHA1": "PfI5CDu8P1Gr+iZnGUWlL27uqU8=",
 			"path": "github.com/aws/aws-sdk-go-v2/service/cloudformation/cloudformationiface",
-			"revision": "da6472f8447d3fef65272d1e396e892df15310d5",
-			"revisionTime": "2019-05-29T23:50:13Z",
-			"version": "v2.0.0-preview.5",
-			"versionExact": "v2.0.0-preview.5"
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
 			"checksumSHA1": "swltJAy/rsn+Q/uXLk1ziBU5kFg=",
@@ -464,32 +472,44 @@
 			"versionExact": "v0.9.0"
 		},
 		{
-			"checksumSHA1": "qUIf77TBORA1MZoNFNdiR5kFRjI=",
-			"path": "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2",
-			"revision": "da6472f8447d3fef65272d1e396e892df15310d5",
-			"revisionTime": "2019-05-29T23:50:13Z"
+			"checksumSHA1": "eMHyTG27+3dZ56gst6ofgI4axsQ=",
+			"path": "github.com/aws/aws-sdk-go-v2/service/rds",
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
+		},
+		{
+			"checksumSHA1": "FOeCFbLZpbW9dehoSTok/SIMBGQ=",
+			"path": "github.com/aws/aws-sdk-go-v2/service/rds/rdsiface",
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
-			"checksumSHA1": "kCyrtNPVHzoajcxIYke+pHqWkR0=",
-			"path": "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/elasticloadbalancingv2iface",
-			"revision": "da6472f8447d3fef65272d1e396e892df15310d5",
-			"revisionTime": "2019-05-29T23:50:13Z"
+			"checksumSHA1": "X8PEjygPDSFiWO03ckxEiZRvKhY=",
+			"path": "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi",
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
-			"checksumSHA1": "QX85Yp1ihkxh8ye/FktHZUVsT6A=",
-			"path": "github.com/aws/aws-sdk-go-v2/service/lambda",
-			"revision": "d52522b5f4b95591ff6528d7c54923951aadf099",
-			"revisionTime": "2018-09-27T22:51:20Z",
-			"version": "v2.0.0-preview.5",
-			"versionExact": "v2.0.0-preview.5"
+			"checksumSHA1": "KH7LAWZzbM7BJZYYghpbFedl84Y=",
+			"path": "github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi/resourcegroupstaggingapiiface",
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
 			"checksumSHA1": "H0fRQXfKJ5ZjJC7d0NgWgEcW2iI=",
 			"path": "github.com/aws/aws-sdk-go-v2/service/s3",
-			"revision": "da6472f8447d3fef65272d1e396e892df15310d5",
-			"revisionTime": "2019-05-29T23:50:13Z",
-			"version": "v2.0.0-preview.5",
-			"versionExact": "v2.0.0-preview.5"
+			"revision": "098e15df3044cf1b04a222c1c33c3e6135ac89f3",
+			"revisionTime": "2019-05-28T21:51:27Z",
+			"version": "v0.9.0",
+			"versionExact": "v0.9.0"
 		},
 		{
 			"checksumSHA1": "VOdJIdzSIvHiu23lwsxFOTeBrWk=",

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

@@ -7,6 +7,8 @@ package elb
 import (
 	"time"
 
+	"github.com/elastic/beats/x-pack/libbeat/common/aws"
+
 	"github.com/elastic/beats/libbeat/autodiscover/template"
 	"github.com/elastic/beats/libbeat/common"
 )
@@ -28,10 +30,8 @@ type Config struct {
 
 	// AWS Specific autodiscover fields
 
-	Regions         []string `config:"regions" validate:"required"`
-	AccessKeyID     string   `config:"access_key_id"`
-	SecretAccessKey string   `config:"secret_access_key"`
-	SessionToken    string   `config:"session_token"`
+	Regions []string `config:"regions" validate:"required"`
+	aws.ConfigAWS
 }
 
 func defaultConfig() *Config {

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

@@ -6,20 +6,21 @@ package elb
 
 import (
 	"context"
-	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/elasticloadbalancingv2iface"
 	"sync"
 
+	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/elasticloadbalancingv2iface"
+
 	"github.com/elastic/beats/libbeat/logp"
 
 	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2"
 	"go.uber.org/multierr"
-
-	"github.com/elastic/beats/libbeat/common/atomic"
 )
 
+const logSelector = "autodiscover-elb-fetch"
+
 // fetcher is an interface that can fetch a list of lbListener (load balancer + listener) objects without pagination being necessary.
 type fetcher interface {
-	fetch() ([]*lbListener, error)
+	fetch(ctx context.Context) ([]*lbListener, error)
 }
 
 // apiMultiFetcher fetches results from multiple clients concatenating their results together
@@ -28,14 +29,14 @@ type apiMultiFetcher struct {
 	fetchers []fetcher
 }
 
-func (amf *apiMultiFetcher) fetch() ([]*lbListener, error) {
+func (amf *apiMultiFetcher) fetch(ctx context.Context) ([]*lbListener, error) {
 	fetchResults := make(chan []*lbListener)
 	fetchErr := make(chan error)
 
 	// Simultaneously fetch all from each region
 	for _, f := range amf.fetchers {
 		go func(f fetcher) {
-			fres, ferr := f.fetch()
+			fres, ferr := f.fetch(ctx)
 			if ferr != nil {
 				fetchErr <- ferr
 			} else {
@@ -64,7 +65,7 @@ type apiFetcher struct {
 	client elasticloadbalancingv2iface.ClientAPI
 }
 
-func newAPIFetcher(clients []elasticloadbalancingv2iface.ClientAPI) fetcher {
+func newAPIFetcher(ctx context.Context, clients []elasticloadbalancingv2iface.ClientAPI) fetcher {
 	fetchers := make([]fetcher, len(clients))
 	for idx, client := range clients {
 		fetchers[idx] = &apiFetcher{client}
@@ -77,7 +78,7 @@ func newAPIFetcher(clients []elasticloadbalancingv2iface.ClientAPI) fetcher {
 // per listener API request. Each page of results has O(n)+1 perf since we need that
 // additional fetch per lb. We let the goroutine scheduler sort things out, and use
 // a sync.Pool to limit the number of in-flight requests.
-func (f *apiFetcher) fetch() ([]*lbListener, error) {
+func (f *apiFetcher) fetch(ctx context.Context) ([]*lbListener, error) {
 	var pageSize int64 = 50
 
 	req := f.client.DescribeLoadBalancersRequest(&elasticloadbalancingv2.DescribeLoadBalancersInput{PageSize: &pageSize})
@@ -88,15 +89,14 @@ func (f *apiFetcher) fetch() ([]*lbListener, error) {
 	for i := 0; i < 10; i++ {
 		taskPool.Put(nil)
 	}
+
+	ctx, cancel := context.WithCancel(ctx)
 	ir := &fetchRequest{
-		elasticloadbalancingv2.NewDescribeLoadBalancersPaginator(req),
-		f.client,
-		atomic.MakeBool(true),
-		[]*lbListener{},
-		[]error{},
-		sync.Mutex{},
-		taskPool,
-		sync.WaitGroup{},
+		paginator: elasticloadbalancingv2.NewDescribeLoadBalancersPaginator(req),
+		client:    f.client,
+		taskPool:  taskPool,
+		context:   ctx,
+		cancel:    cancel,
 	}
 
 	return ir.fetch()
@@ -107,12 +107,13 @@ func (f *apiFetcher) fetch() ([]*lbListener, error) {
 type fetchRequest struct {
 	paginator    elasticloadbalancingv2.DescribeLoadBalancersPaginator
 	client       elasticloadbalancingv2iface.ClientAPI
-	running      atomic.Bool
 	lbListeners  []*lbListener
 	errs         []error
 	resultsLock  sync.Mutex
 	taskPool     sync.Pool
 	pendingTasks sync.WaitGroup
+	context      context.Context
+	cancel       func()
 }
 
 func (p *fetchRequest) fetch() ([]*lbListener, error) {
@@ -136,13 +137,21 @@ func (p *fetchRequest) fetch() ([]*lbListener, error) {
 
 func (p *fetchRequest) fetchAllPages() {
 	// Keep fetching pages unless we're stopped OR there are no pages left
-	for p.running.Load() && p.fetchNextPage() {
-		logp.Debug("autodiscover-elb", "API page fetched")
+Outer:
+	for {
+		select {
+		case <-p.context.Done():
+			logp.Debug(logSelector, "done fetching ELB pages, context cancelled")
+			break Outer
+		default:
+			p.fetchNextPage()
+			logp.Debug(logSelector, "API page fetched")
+		}
 	}
 }
 
 func (p *fetchRequest) fetchNextPage() (more bool) {
-	success := p.paginator.Next(context.TODO())
+	success := p.paginator.Next(p.context)
 
 	if success {
 		for _, lb := range p.paginator.CurrentPage().LoadBalancers {
@@ -175,14 +184,23 @@ func (p *fetchRequest) dispatch(fn func()) {
 func (p *fetchRequest) fetchListeners(lb elasticloadbalancingv2.LoadBalancer) {
 	listenReq := p.client.DescribeListenersRequest(&elasticloadbalancingv2.DescribeListenersInput{LoadBalancerArn: lb.LoadBalancerArn})
 	listen := elasticloadbalancingv2.NewDescribeListenersPaginator(listenReq)
-	for listen.Next(context.TODO()) && p.running.Load() {
-		for _, listener := range listen.CurrentPage().Listeners {
-			p.recordGoodResult(&lb, &listener)
-		}
-	}
+
 	if listen.Err() != nil {
 		p.recordErrResult(listen.Err())
 	}
+
+Outer:
+	for listen.Next(p.context) {
+		select {
+		case <-p.context.Done():
+			break Outer
+		default:
+			for _, listener := range listen.CurrentPage().Listeners {
+				p.recordGoodResult(&lb, &listener)
+			}
+		}
+
+	}
 }
 
 func (p *fetchRequest) recordGoodResult(lb *elasticloadbalancingv2.LoadBalancer, lbl *elasticloadbalancingv2.Listener) {
@@ -198,6 +216,5 @@ func (p *fetchRequest) recordErrResult(err error) {
 
 	p.errs = append(p.errs, err)
 
-	// Try to stop execution early
-	p.running.Store(false)
+	p.cancel()
 }

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

@@ -5,6 +5,7 @@
 package elb
 
 import (
+	"context"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/elasticloadbalancingv2iface"
@@ -13,6 +14,6 @@ import (
 
 func Test_newAPIFetcher(t *testing.T) {
 	client := newMockELBClient(0)
-	fetcher := newAPIFetcher([]elasticloadbalancingv2iface.ClientAPI{client})
+	fetcher := newAPIFetcher(context.TODO(), []elasticloadbalancingv2iface.ClientAPI{client})
 	require.NotNil(t, fetcher)
 }

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

@@ -18,23 +18,55 @@ type lbListener struct {
 
 // toMap converts this lbListener into the form consumed as metadata in the autodiscovery process.
 func (l *lbListener) toMap() common.MapStr {
+	// We fully spell out listener_arn to avoid confusion with the ARN for the whole ELB
+	m := common.MapStr{
+		"listener_arn":       l.listener.ListenerArn,
+		"load_balancer_arn":  safeStrp(l.lb.LoadBalancerArn),
+		"host":               safeStrp(l.lb.DNSName),
+		"protocol":           l.listener.Protocol,
+		"type":               string(l.lb.Type),
+		"scheme":             l.lb.Scheme,
+		"availability_zones": l.azStrings(),
+		"created":            l.lb.CreatedTime,
+		"state":              l.stateMap(),
+		"ip_address_type":    string(l.lb.IpAddressType),
+		"security_groups":    l.lb.SecurityGroups,
+		"vpc_id":             safeStrp(l.lb.VpcId),
+		"ssl_policy":         l.listener.SslPolicy,
+	}
+
+	if l.listener.Port != nil {
+		m["port"] = *l.listener.Port
+	}
+
+	return m
+}
+
+// safeStrp makes handling AWS *string types easier.
+// The AWS lib never returns plain strings, always using pointers, probably for memory efficiency reasons.
+// This is a bit odd, because strings are just pointers into byte arrays, however this is the choice they've made.
+// This will return the plain version of the given string or an empty string if the pointer is null
+func safeStrp(strp *string) string {
+	if strp == nil {
+		return ""
+	}
+
+	return *strp
+}
+
+func (l *lbListener) toCloudMap() common.MapStr {
 	m := common.MapStr{}
 
-	// We fully spell out listener_arn to avoid confusion with the ARN for the whole ELB
-	m["listener_arn"] = l.listener.ListenerArn
-	m["load_balancer_arn"] = *l.lb.LoadBalancerArn
-	m["host"] = *l.lb.DNSName
-	m["port"] = *l.listener.Port
-	m["protocol"] = l.listener.Protocol
-	m["type"] = string(l.lb.Type)
-	m["scheme"] = l.lb.Scheme
-	m["availability_zones"] = l.azStrings()
-	m["created"] = l.lb.CreatedTime
-	m["state"] = l.stateMap()
-	m["ip_address_type"] = string(l.lb.IpAddressType)
-	m["security_groups"] = l.lb.SecurityGroups
-	m["vpc_id"] = *l.lb.VpcId
-	m["ssl_policy"] = l.listener.SslPolicy
+	var azs []string
+	for _, az := range l.lb.AvailabilityZones {
+		azs = append(azs, *az.ZoneName)
+	}
+	m["availability_zone"] = azs
+	m["provider"] = "aws"
+
+	// The region is just an AZ with the last character removed
+	firstAz := azs[0]
+	m["region"] = firstAz[:len(firstAz)-2]
 
 	return m
 }

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

@@ -5,6 +5,7 @@
 package elb
 
 import (
+	"context"
 	"sync"
 	"time"
 
@@ -22,7 +23,7 @@ func newMockFetcher(lbListeners []*lbListener, err error) *mockFetcher {
 	return &mockFetcher{lblListeners: lbListeners, err: err}
 }
 
-func (f *mockFetcher) fetch() ([]*lbListener, error) {
+func (f *mockFetcher) fetch(ctx context.Context) ([]*lbListener, error) {
 	f.lock.Lock()
 	defer f.lock.Unlock()