[autodiscovery] Wait for tag completeness before creating AD services (#48732)

### What does this PR do?

This PR is part of the effort to address the "missing tags" issue in the Agent. It introduces a new configuration option (`ad_tag_completeness_max_wait`) that makes autodiscovery wait until all tags are complete before processing an entity. The goal is to avoid scheduling checks with incomplete tags. When tags are not complete before the specified max wait, the check is scheduled as it is today with whatever tags are available at that moment.

This PR also adds a new telemetry metric (`autodiscovery.tag_completeness_delay`) that captures the delay in processing discovered entities.

The new option only applies to Kubernetes environments, because it's the only environment where tags can be detected as incomplete by workloadmeta. The other environment where this could happen is ECS EC2, but that has not been implemented yet.

The new option is disabled by default for now and not exposed. The reason is that, with the default config, tags can be delayed for several seconds. Mainly because the kubemetadata workloadmeta collector is pull-based and runs every 60s. It was recently converted to stream-based, but the option to use it is not enabled by default yet (needs more testing).


### Describe how you validated your changes

New unit tests plus tests in a local kind cluster:
- Deployed with the default config and verified that checks are scheduled immediately. The telemetry doesn't show any delays.
- Deployed with `ad_tag_completeness_max_wait=120`. Checks are a bit delayed as expected and the telemetry confirms it.
- Deployed with `ad_tag_completeness_max_wait=120` and kubemetadata streaming enabled (`kubernetes_metadata_streaming=true`). Checks are scheduled quickly and the telemetry confirms it.



Co-authored-by: david.ortiz <david.ortiz@datadoghq.com>

Author: davidor

comp/core/autodiscovery/listeners/kubelet.go

@@ -55,7 +55,8 @@ func NewKubeletListener(options ServiceListernerDeps) (ServiceListener, error) {
 		return nil, errors.New("workloadmeta store is not initialized")
 	}
 	var err error
-	l.workloadmetaListener, err = newWorkloadmetaListener(name, wmetaFilter, l.processPod, wmetaInstance, options.Telemetry)
+	maxWait := time.Duration(pkgconfigsetup.Datadog().GetInt("ad_tag_completeness_max_wait")) * time.Second
+	l.workloadmetaListener, err = newWorkloadmetaListenerWithTagWait(name, wmetaFilter, l.processPod, wmetaInstance, options.Telemetry, l.areTagsComplete, maxWait)
 	if err != nil {
 		return nil, err
 	}
@@ -215,3 +216,35 @@ func (l *KubeletListener) createContainerService(
 	podSvcID := buildSvcID(pod.GetID())
 	l.AddService(svcID, svc, podSvcID)
 }
+
+func (l *KubeletListener) areTagsComplete(entity workloadmeta.Entity) bool {
+	pod, ok := entity.(*workloadmeta.KubernetesPod)
+	if !ok {
+		log.Errorf("expected KubernetesPod entity, got %T", entity)
+		return true
+	}
+
+	podTaggerID := common.BuildTaggerEntityID(pod.GetID())
+	_, podComplete, err := l.tagger.TagWithCompleteness(podTaggerID, types.ChecksConfigCardinality)
+	if err != nil {
+		log.Debugf("error checking tag completeness for pod %s: %s", pod.ID, err)
+		return false
+	}
+	if !podComplete {
+		return false
+	}
+
+	for _, podContainer := range pod.GetAllContainers() {
+		containerTaggerID := types.NewEntityID(types.ContainerID, podContainer.ID)
+		_, containerComplete, err := l.tagger.TagWithCompleteness(containerTaggerID, types.ChecksConfigCardinality)
+		if err != nil {
+			log.Debugf("error checking tag completeness for container %s: %s", podContainer.ID, err)
+			return false
+		}
+		if !containerComplete {
+			return false
+		}
+	}
+
+	return true
+}

comp/core/autodiscovery/listeners/kubelet_test.go

@@ -12,8 +12,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def"
 	taggerfxmock "github.com/DataDog/datadog-agent/comp/core/tagger/fx-mock"
+	"github.com/DataDog/datadog-agent/comp/core/tagger/types"
 	workloadfilter "github.com/DataDog/datadog-agent/comp/core/workloadfilter/def"
 	workloadfilterfxmock "github.com/DataDog/datadog-agent/comp/core/workloadfilter/fx-mock"
 	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
@@ -663,6 +666,131 @@ func TestProcessPodWithEphemeralContainer(t *testing.T) {
 	wlm.assertServices(expectedServices)
 }
 
+func TestAreTagsComplete(t *testing.T) {
+	podEntityID := types.NewEntityID(types.KubernetesPodUID, podID)
+	containerEntityID := types.NewEntityID(types.ContainerID, containerID)
+
+	tests := []struct {
+		name     string
+		pod      *workloadmeta.KubernetesPod
+		tagInfos []*types.TagInfo
+		expected bool
+	}{
+		{
+			name: "pod and container complete",
+			pod: &workloadmeta.KubernetesPod{
+				EntityID: workloadmeta.EntityID{
+					Kind: workloadmeta.KindKubernetesPod,
+					ID:   podID,
+				},
+				Containers: []workloadmeta.OrchestratorContainer{
+					{
+						ID:   containerID,
+						Name: containerName,
+					},
+				},
+			},
+			tagInfos: []*types.TagInfo{
+				{
+					Source:      "source",
+					EntityID:    podEntityID,
+					LowCardTags: []string{"kube_namespace:default"},
+					IsComplete:  true,
+				},
+				{
+					Source:      "source",
+					EntityID:    containerEntityID,
+					LowCardTags: []string{"container_name:agent"},
+					IsComplete:  true,
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "pod incomplete",
+			pod: &workloadmeta.KubernetesPod{
+				EntityID: workloadmeta.EntityID{
+					Kind: workloadmeta.KindKubernetesPod,
+					ID:   podID,
+				},
+			},
+			tagInfos: []*types.TagInfo{
+				{
+					Source:      "source",
+					EntityID:    podEntityID,
+					LowCardTags: []string{"kube_namespace:default"},
+					IsComplete:  false,
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "container incomplete",
+			pod: &workloadmeta.KubernetesPod{
+				EntityID: workloadmeta.EntityID{
+					Kind: workloadmeta.KindKubernetesPod,
+					ID:   podID,
+				},
+				Containers: []workloadmeta.OrchestratorContainer{
+					{
+						ID:   containerID,
+						Name: containerName,
+					},
+				},
+			},
+			tagInfos: []*types.TagInfo{
+				{
+					Source:      "source",
+					EntityID:    podEntityID,
+					LowCardTags: []string{"kube_namespace:default"},
+					IsComplete:  true,
+				},
+				{
+					Source:      "source",
+					EntityID:    containerEntityID,
+					LowCardTags: []string{"container_name:agent"},
+					IsComplete:  false,
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "container not in tagger",
+			pod: &workloadmeta.KubernetesPod{
+				EntityID: workloadmeta.EntityID{
+					Kind: workloadmeta.KindKubernetesPod,
+					ID:   podID,
+				},
+				Containers: []workloadmeta.OrchestratorContainer{
+					{
+						ID:   "unknown-container",
+						Name: "unknown",
+					},
+				},
+			},
+			tagInfos: []*types.TagInfo{
+				{
+					Source:      "source",
+					EntityID:    podEntityID,
+					LowCardTags: []string{"kube_namespace:default"},
+					IsComplete:  true,
+				},
+			},
+			expected: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			taggerMock := taggerfxmock.SetupFakeTagger(t)
+			taggerMock.GetTagStore().ProcessTagInfo(test.tagInfos)
+			listener, _ := newKubeletListener(t, taggerMock)
+
+			assert.Equal(t, test.expected, listener.areTagsComplete(test.pod))
+		})
+	}
+}
+
 func newKubeletListener(t *testing.T, tagger tagger.Component) (*KubeletListener, *testWorkloadmetaListener) {
 	wlm := newTestWorkloadmetaListener(t)
 	filterStore := workloadfilterfxmock.SetupMockFilter(t)

comp/core/autodiscovery/listeners/workloadmeta.go

@@ -10,13 +10,16 @@ package listeners
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/telemetry"
 	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
 	"github.com/DataDog/datadog-agent/pkg/status/health"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
+const tagCompletenessRetryInterval = 1 * time.Second
+
 // workloadmetaListener is a generic subscriber to workloadmeta events that
 // generates AD services.
 type workloadmetaListener interface {
@@ -33,6 +36,11 @@ type workloadmetaListener interface {
 	AddService(svcID string, svc Service, parentSvcID string)
 }
 
+type pendingEntityInfo struct {
+	entity    workloadmeta.Entity
+	firstSeen time.Time
+}
+
 // workloadmetaListenerImpl implements workloadmetaListener.
 type workloadmetaListenerImpl struct {
 	name string
@@ -50,6 +58,10 @@ type workloadmetaListenerImpl struct {
 	delService chan<- Service
 
 	telemetryStore *telemetry.Store
+
+	isReadyFn       func(workloadmeta.Entity) bool // when nil, consider ready
+	pendingEntities map[string]pendingEntityInfo   // svcID → pending info
+	maxWaitDuration time.Duration
 }
 
 var _ workloadmetaListener = &workloadmetaListenerImpl{}
@@ -82,6 +94,36 @@ func newWorkloadmetaListener(
 	}, nil
 }
 
+// newWorkloadmetaListenerWithTagWait is like newWorkloadmetaListener but defers
+// processing until isReadyFn returns true. Disabled when maxWait is 0.
+func newWorkloadmetaListenerWithTagWait(
+	name string,
+	workloadFilters *workloadmeta.Filter,
+	processFn func(workloadmeta.Entity),
+	wmeta workloadmeta.Component,
+	telemetryStore *telemetry.Store,
+	isReadyFn func(workloadmeta.Entity) bool,
+	maxWait time.Duration,
+) (workloadmetaListener, error) {
+	base, err := newWorkloadmetaListener(name, workloadFilters, processFn, wmeta, telemetryStore)
+	if err != nil {
+		return nil, err
+	}
+
+	if maxWait > 0 {
+		listener, ok := base.(*workloadmetaListenerImpl)
+		if !ok {
+			return nil, fmt.Errorf("unexpected listener type %T", base)
+		}
+
+		listener.isReadyFn = isReadyFn
+		listener.pendingEntities = make(map[string]pendingEntityInfo)
+		listener.maxWaitDuration = maxWait
+	}
+
+	return base, nil
+}
+
 func (l *workloadmetaListenerImpl) Store() workloadmeta.Component {
 	return l.store
 }
@@ -126,7 +168,17 @@ func (l *workloadmetaListenerImpl) Listen(newSvc chan<- Service, delSvc chan<- S
 	log.Infof("%s initialized successfully", l.name)
 
 	go func() {
+		var retryTicker *time.Ticker
+		var retryChan <-chan time.Time
+		if l.isReadyFn != nil {
+			retryTicker = time.NewTicker(tagCompletenessRetryInterval)
+			retryChan = retryTicker.C
+		}
+
 		defer func() {
+			if retryTicker != nil {
+				retryTicker.Stop()
+			}
 			err := health.Deregister()
 			if err != nil {
 				log.Warnf("error de-registering health check: %s", err)
@@ -142,6 +194,9 @@ func (l *workloadmetaListenerImpl) Listen(newSvc chan<- Service, delSvc chan<- S
 
 				l.processEvents(evBundle)
 
+			case <-retryChan:
+				l.retryPendingEntities()
+
 			case <-health.C:
 
 			case <-l.stop:
@@ -181,6 +236,10 @@ func (l *workloadmetaListenerImpl) processEvents(evBundle workloadmeta.EventBund
 func (l *workloadmetaListenerImpl) processSetEntity(entity workloadmeta.Entity) {
 	svcID := buildSvcID(entity.GetID())
 
+	if l.isReadyFn != nil && l.waitIfNotReady(svcID, entity) {
+		return
+	}
+
 	// keep track of children of this entity from previous iterations ...
 	unseen := make(map[string]struct{})
 	for childSvcID := range l.children[svcID] {
@@ -204,10 +263,75 @@ func (l *workloadmetaListenerImpl) processSetEntity(entity workloadmeta.Entity)
 	}
 }
 
+// waitIfNotReady returns true if the entity is not ready and should be retried
+// later.
+func (l *workloadmetaListenerImpl) waitIfNotReady(svcID string, entity workloadmeta.Entity) bool {
+	if l.isReadyFn(entity) {
+		_, wasPending := l.pendingEntities[svcID]
+		l.resolvePending(svcID)
+		if !wasPending && l.telemetryStore != nil {
+			l.telemetryStore.TagCompletenessDelay.Observe(0, l.name)
+		}
+		return false
+	}
+
+	now := time.Now()
+
+	pending, exists := l.pendingEntities[svcID]
+	if !exists {
+		pending = pendingEntityInfo{
+			entity:    entity,
+			firstSeen: now,
+		}
+	} else {
+		pending.entity = entity
+	}
+
+	if now.Sub(pending.firstSeen) < l.maxWaitDuration {
+		l.pendingEntities[svcID] = pending
+		log.Debugf("%s not adding entity %s: tags not complete yet", l.name, svcID)
+		return true
+	}
+
+	// Timeout exceeded
+	log.Warnf("%s adding entity %s with potentially incomplete tags", l.name, svcID)
+	l.resolvePending(svcID)
+
+	return false
+}
+
+func (l *workloadmetaListenerImpl) resolvePending(svcID string) {
+	pending, wasPending := l.pendingEntities[svcID]
+	if !wasPending {
+		return
+	}
+
+	delay := time.Since(pending.firstSeen).Seconds()
+	if l.telemetryStore != nil {
+		l.telemetryStore.TagCompletenessDelay.Observe(delay, l.name)
+	}
+	delete(l.pendingEntities, svcID)
+}
+
+func (l *workloadmetaListenerImpl) retryPendingEntities() {
+	pending := make([]workloadmeta.Entity, 0, len(l.pendingEntities))
+	for _, pendingEntity := range l.pendingEntities {
+		pending = append(pending, pendingEntity.entity)
+	}
+
+	for _, entity := range pending {
+		l.processSetEntity(entity)
+	}
+}
+
 func (l *workloadmetaListenerImpl) processUnsetEntity(entity workloadmeta.Entity) {
 	entityID := entity.GetID()
 	parentSvcID := buildSvcID(entityID)
 
+	if l.pendingEntities != nil {
+		delete(l.pendingEntities, parentSvcID)
+	}
+
 	l.removeService(parentSvcID)
 
 	childrenSvcIDs := l.children[parentSvcID]