[ASCII-1281] Ensure host metadata information information is sent first (#23620)

* add new priority provider

* filter and get non optional providers in the constructor

* fix linters

* apply feedback

* add test for the priority provider logic

* ensure test consistency

Author: Gustavo Caso

comp/metadata/host/hostimpl/host.go

@@ -66,7 +66,7 @@ type provides struct {
 	fx.Out
 
 	Comp                 hostComp.Component
-	MetadataProvider     runnerimpl.Provider
+	MetadataProvider     runnerimpl.PriorityProvider
 	FlareProvider        flaretypes.Provider
 	StatusHeaderProvider status.HeaderInformationProvider
 }
@@ -102,7 +102,7 @@ func newHostProvider(deps dependencies) provides {
 	}
 	return provides{
 		Comp:             &h,
-		MetadataProvider: runnerimpl.NewProvider(h.collect),
+		MetadataProvider: runnerimpl.NewPriorityProvider(h.collect),
 		FlareProvider:    flaretypes.NewProvider(h.fillFlare),
 		StatusHeaderProvider: status.NewHeaderInformationProvider(StatusProvider{
 			Config: h.config,

comp/metadata/runner/runnerimpl/runner.go

@@ -27,13 +27,18 @@ func Module() fxutil.Module {
 // MetadataProvider is the provider for metadata
 type MetadataProvider func(context.Context) time.Duration
 
+// PriorityMetadataProvider is the provider for metadata that needs to be execute at start time of the agent.
+// Right now the agent needs the host metadata provider to execute first to ensure host tags are being reported correctly.
+// This is a temporary workaorund until we figure a more permanent solution in the backend
+// If you need to use this provide please contact the agent-shared-components team
+type PriorityMetadataProvider func(context.Context) time.Duration
+
 type runnerImpl struct {
 	log    log.Component
 	config config.Component
 
-	// providers are the metada providers to run. They're Optional because some of them can be disabled through the
-	// configuration
-	providers []optional.Option[MetadataProvider]
+	providers         []MetadataProvider
+	priorityProviders []PriorityMetadataProvider
 
 	wg       sync.WaitGroup
 	stopChan chan struct{}
@@ -45,7 +50,8 @@ type dependencies struct {
 	Log    log.Component
 	Config config.Component
 
-	Providers []optional.Option[MetadataProvider] `group:"metadata_provider"`
+	Providers         []optional.Option[MetadataProvider]         `group:"metadata_provider"`
+	PriorityProviders []optional.Option[PriorityMetadataProvider] `group:"metadata_priority_provider"`
 }
 
 // Provider represents the callback from a metada provider. This is returned by 'NewProvider' helper.
@@ -55,13 +61,27 @@ type Provider struct {
 	Callback optional.Option[MetadataProvider] `group:"metadata_provider"`
 }
 
+// PriorityProvider represents the callback from a priority metada provider. This is returned by 'NewPriorityProvider' helper.
+type PriorityProvider struct {
+	fx.Out
+
+	Callback optional.Option[PriorityMetadataProvider] `group:"metadata_priority_provider"`
+}
+
 // NewProvider registers a new metadata provider by adding a callback to the runner.
 func NewProvider(callback MetadataProvider) Provider {
 	return Provider{
 		Callback: optional.NewOption[MetadataProvider](callback),
 	}
 }
 
+// NewPriorityProvider registers a new metadata provider by adding a callback to the runner.
+func NewPriorityProvider(callback PriorityMetadataProvider) PriorityProvider {
+	return PriorityProvider{
+		Callback: optional.NewOption[PriorityMetadataProvider](callback),
+	}
+}
+
 // NewEmptyProvider returns a empty provider which is not going to register anything. This is useful for providers that
 // can be enabled/disabled through configuration.
 func NewEmptyProvider() Provider {
@@ -72,11 +92,30 @@ func NewEmptyProvider() Provider {
 
 // createRunner instantiates a runner object
 func createRunner(deps dependencies) *runnerImpl {
+	providers := []MetadataProvider{}
+	priorityProviders := []PriorityMetadataProvider{}
+
+	nonNilProviders := fxutil.GetAndFilterGroup(deps.Providers)
+	nonNilPriorityProviders := fxutil.GetAndFilterGroup(deps.PriorityProviders)
+
+	for _, optionaP := range nonNilProviders {
+		if p, isSet := optionaP.Get(); isSet {
+			providers = append(providers, p)
+		}
+	}
+
+	for _, optionaP := range nonNilPriorityProviders {
+		if p, isSet := optionaP.Get(); isSet {
+			priorityProviders = append(priorityProviders, p)
+		}
+	}
+
 	return &runnerImpl{
-		log:       deps.Log,
-		config:    deps.Config,
-		providers: fxutil.GetAndFilterGroup(deps.Providers),
-		stopChan:  make(chan struct{}),
+		log:               deps.Log,
+		config:            deps.Config,
+		providers:         providers,
+		priorityProviders: priorityProviders,
+		stopChan:          make(chan struct{}),
 	}
 }
 
@@ -100,7 +139,7 @@ func newRunner(lc fx.Lifecycle, deps dependencies) runner.Component {
 }
 
 // handleProvider runs a provider at regular interval until the runner is stopped
-func (r *runnerImpl) handleProvider(p MetadataProvider) {
+func (r *runnerImpl) handleProvider(p func(context.Context) time.Duration) {
 	r.log.Debugf("Starting runner for MetadataProvider %#v", p)
 	r.wg.Add(1)
 
@@ -137,12 +176,21 @@ func (r *runnerImpl) handleProvider(p MetadataProvider) {
 // start is called by FX when the application starts. Lifecycle hooks are blocking and called sequencially. We should
 // not block here.
 func (r *runnerImpl) start() error {
-	r.log.Debugf("Starting metadata runner with %d providers", len(r.providers))
-	for _, optionaP := range r.providers {
-		if p, isSet := optionaP.Get(); isSet {
-			go r.handleProvider(p)
+	r.log.Debugf("Starting metadata runner with %d priority providers and %d regular providers", len(r.priorityProviders), len(r.providers))
+
+	go func() {
+		for _, priorityProvider := range r.priorityProviders {
+			// Execute synchronously the priority provider
+			priorityProvider(context.Background())
+
+			go r.handleProvider(priorityProvider)
 		}
-	}
+
+		for _, provider := range r.providers {
+			go r.handleProvider(provider)
+		}
+	}()
+
 	return nil
 }
 

comp/metadata/runner/runnerimpl/runner_test.go

@@ -8,6 +8,7 @@ package runnerimpl
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -21,12 +22,15 @@ import (
 )
 
 func TestHandleProvider(t *testing.T) {
-	called := make(chan struct{})
+	wg := sync.WaitGroup{}
+
 	provider := func(context.Context) time.Duration {
-		called <- struct{}{}
+		wg.Done()
 		return 1 * time.Minute // Long timeout to block
 	}
 
+	wg.Add(1)
+
 	r := createRunner(
 		fxutil.Test[dependencies](
 			t,
@@ -36,18 +40,21 @@ func TestHandleProvider(t *testing.T) {
 		))
 
 	r.start()
-	// either called receive a value or the test will fail as a timeout
-	<-called
+	// either the provider call wg.Done() or the test will fail as a timeout
+	wg.Wait()
 	assert.NoError(t, r.stop())
 }
 
 func TestRunnerCreation(t *testing.T) {
-	called := make(chan struct{})
-	callback := func(context.Context) time.Duration {
-		called <- struct{}{}
+	wg := sync.WaitGroup{}
+
+	provider := func(context.Context) time.Duration {
+		wg.Done()
 		return 1 * time.Minute // Long timeout to block
 	}
 
+	wg.Add(1)
+
 	lc := fxtest.NewLifecycle(t)
 	fxutil.Test[runner.Component](
 		t,
@@ -56,14 +63,62 @@ func TestRunnerCreation(t *testing.T) {
 		config.MockModule(),
 		Module(),
 		// Supplying our provider by using the helper function
-		fx.Supply(NewProvider(callback)),
+		fx.Supply(NewProvider(provider)),
 	)
 
 	ctx := context.Background()
 	lc.Start(ctx)
 
-	// either called receive a value or the test will fail as a timeout
-	<-called
+	// either the provider call wg.Done() or the test will fail as a timeout
+	wg.Wait()
+
+	assert.NoError(t, lc.Stop(ctx))
+}
+
+func TestPriorityProviderOrder(t *testing.T) {
+	wg := sync.WaitGroup{}
+	m := sync.Mutex{}
+	eventSequence := []string{}
+
+	provider := func(context.Context) time.Duration {
+		m.Lock()
+		eventSequence = append(eventSequence, "provider")
+		m.Unlock()
+		wg.Done()
+		return 1 * time.Minute // Long timeout to block
+	}
+
+	priorityProvider := func(context.Context) time.Duration {
+		m.Lock()
+		eventSequence = append(eventSequence, "priorityProvider")
+		m.Unlock()
+		wg.Done()
+		return 1 * time.Minute // Long timeout to block
+	}
+
+	// We add 3 work unit because the priority provider are called twice at startup. One synchronousily and one asynchronosuly
+	wg.Add(3)
+
+	lc := fxtest.NewLifecycle(t)
+	fxutil.Test[runner.Component](
+		t,
+		fx.Supply(lc),
+		logimpl.MockModule(),
+		config.MockModule(),
+		Module(),
+		// Supplying our provider by using the helper function
+		fx.Supply(NewProvider(provider)),
+		fx.Supply(NewPriorityProvider(priorityProvider)),
+	)
+
+	ctx := context.Background()
+	lc.Start(ctx)
 
+	// either the provider call wg.Done() or the test will fail as a timeout
+	wg.Wait()
+	// it is expected to see three events. Priority providers are called twice at start up
+	assert.Equal(t, 3, len(eventSequence))
+	// ensure priority provider is the first provider to be executed
+	assert.Equal(t, "priorityProvider", eventSequence[0])
 	assert.NoError(t, lc.Stop(ctx))
 }