fix(add_cloud_metadata): propagate context and wait for init to complete on shutdown

libbeat/processors/add_cloud_metadata/add_cloud_metadata.go

@@ -18,6 +18,7 @@
 package add_cloud_metadata
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -44,11 +45,13 @@ func init() {
 }
 
 type addCloudMetadata struct {
-	initOnce sync.Once
-	initData *initData
-	initDone chan struct{}
-	metadata mapstr.M
-	logger   *logp.Logger
+	baseCtx       context.Context
+	baseCtxCancel context.CancelFunc
+	initOnce      sync.Once
+	initData      *initData
+	initDone      chan struct{}
+	metadata      mapstr.M
+	logger        *logp.Logger
 }
 
 type initData struct {
@@ -75,15 +78,18 @@ func New(c *cfg.C, log *logp.Logger) (beat.Processor, error) {
 	if err != nil {
 		return nil, err
 	}
+	ctx, cancel := context.WithCancel(context.Background())
 	p := &addCloudMetadata{
 		initData: &initData{
 			fetchers:  fetchers,
 			timeout:   config.Timeout,
 			tlsConfig: tlsConfig,
 			overwrite: config.Overwrite,
 		},
-		initDone: make(chan struct{}),
-		logger:   log.Named("add_cloud_metadata"),
+		initDone:      make(chan struct{}),
+		logger:        log.Named("add_cloud_metadata"),
+		baseCtx:       ctx,
+		baseCtxCancel: cancel,
 	}
 
 	go p.init()
@@ -98,7 +104,7 @@ func (r result) String() string {
 func (p *addCloudMetadata) init() {
 	p.initOnce.Do(func() { // fetch metadata only once
 		defer close(p.initDone) // signal that init() completed
-		result := p.fetchMetadata()
+		result := p.fetchMetadata(p.baseCtx)
 		if result == nil {
 			p.logger.Info("add_cloud_metadata: hosting provider type not detected.")
 			return
@@ -138,6 +144,12 @@ func (p *addCloudMetadata) String() string {
 	return "add_cloud_metadata=" + metadataStr
 }
 
+func (p *addCloudMetadata) Close() error {
+	p.baseCtxCancel()
+	p.initOnce.Do(func() {})
+	return nil
+}
+
 func (p *addCloudMetadata) addMeta(event *beat.Event, meta mapstr.M) error {
 	for key, metaVal := range meta {
 		// If key exists in event already and overwrite flag is set to false, this processor will not overwrite the

libbeat/processors/add_cloud_metadata/providers.go

@@ -153,7 +153,7 @@ func setupFetchers(providers map[string]provider, c *conf.C, logger *logp.Logger
 // hosting providers supported by this processor. It will wait for the results to
 // be returned or for a timeout to occur then returns the first result that
 // completed in time.
-func (p *addCloudMetadata) fetchMetadata() *result {
+func (p *addCloudMetadata) fetchMetadata(ctx context.Context) *result {
 	p.logger.Debugf("add_cloud_metadata: starting to fetch metadata, timeout=%v", p.initData.timeout)
 	start := time.Now()
 	defer func() {
@@ -174,7 +174,7 @@ func (p *addCloudMetadata) fetchMetadata() *result {
 	}
 
 	// Create context to enable explicit cancellation of the http requests.
-	ctx, cancel := context.WithTimeout(context.TODO(), p.initData.timeout)
+	ctx, cancel := context.WithTimeout(ctx, p.initData.timeout)
 	defer cancel()
 
 	results := make(chan result)