CAP-3858 fix missing AgentVersion on manifest payload (#53915)

<!--Please give us some feedback on your experience writing this PR ! https://app.datadoghq.com/forms/43db4c02-6837-400c-8083-692e141b1b88 !-->

### What does this PR do?
Fix missing `AgentVersion` on `CollectorManifest` payloads for all non-pod Kubernetes resources.                                                                                                                        
                                                                                                                                                                                                                          
For buffered manifest payloads (Deployments, Services, Nodes, and every other non-pod resource collected by the Cluster Agent orchestrator check), we have `AgentVersion = nil`.                          
                                                                                                                                                                                                                          
Root cause: `ManifestBuffer.flushManifest` built a synthetic processor context from a snapshot of a handful of config fields and that snapshot silently omitted `AgentVersion`. Pods weren't affected because they're collected by the process-agent on each node, not through this buffer.                                                                                                                                                   
                                                                                                                                                                                                               

### What changed 
                                                                                                                                                                                                   
  - **Fix the drop.** `ManifestBuffer` now holds a reference to the owning `OrchestratorCheck` and builds the flush context via a new `newFlushContext()` helper that reads shared per-check fields live from the check.                                                                                                                                                           
  - **Prevent the class of bug.** Removed the identity fields from `ManifestBufferConfig` so it only holds buffer tuning knobs. Any future per-check field will flow through `newFlushContext` automatically instead of needing to be manually copied into a snapshot.                                                                                                                                                                          
  - **Also fixed on the way through:**                                                                                                                                                                                    
  - Terminated-resource run config was dropping `AgentVersion` too — propagated it from the main run config.                                                                                                            
  - `newOrchestratorCheck` wasn't copying the `Meta` field from `version.Version` into `model.AgentVersion` — added it.                                                                                                 
                                                                                                                                                                                                                                                                
### Describe how you validated your changes

Added a regression assertion in `TestFlushManifest` that fails if `sentManifest.AgentVersion` is nil or doesn't match the check's version. All 647 package tests pass.           

### Additional Notes


Co-authored-by: kangyi.li <kangyi.li@datadoghq.com>

Author: kangyili

pkg/collector/corechecks/cluster/orchestrator/collector_bundle.go

@@ -117,6 +117,7 @@ func NewCollectorBundle(chk *OrchestratorCheck) *CollectorBundle {
 		ClusterID:             runCfg.ClusterID,
 		Config:                runCfg.Config,
 		MsgGroupRef:           runCfg.MsgGroupRef,
+		AgentVersion:          runCfg.AgentVersion,
 		TerminatedResources:   true,
 	}
 

pkg/collector/corechecks/cluster/orchestrator/manifest_buffer.go

@@ -20,7 +20,6 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/collector/corechecks/cluster/orchestrator/processors"
 	"github.com/DataDog/datadog-agent/pkg/collector/corechecks/cluster/orchestrator/processors/common"
 	"github.com/DataDog/datadog-agent/pkg/orchestrator"
-	"github.com/DataDog/datadog-agent/pkg/orchestrator/config"
 	pkgorchestratormodel "github.com/DataDog/datadog-agent/pkg/orchestrator/model"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
@@ -40,17 +39,14 @@ func init() {
 	bufferExpVars.Set("BufferFlushed", bufferFlushedTotal)
 }
 
-// ManifestBufferConfig contains information about buffering manifests.
+// ManifestBufferConfig holds buffer tuning knobs. Per-check identity
+// (cluster, agent version, config, ...) lives on ManifestBuffer.chk and is
+// read live at flush time via newFlushContext.
 type ManifestBufferConfig struct {
-	KubeClusterName             string
-	ClusterID                   string
-	MaxPerMessage               int
-	MaxWeightPerMessageBytes    int
 	MsgGroupRef                 *atomic.Int32
 	BufferedManifestEnabled     bool
 	MaxBufferedManifests        int
 	ManifestBufferFlushInterval time.Duration
-	ExtraTags                   []string
 }
 
 // ManifestBuffer is a buffer of manifest sent from all collectors
@@ -59,6 +55,7 @@ type ManifestBufferConfig struct {
 // and gets stopped after the check is done.
 type ManifestBuffer struct {
 	Cfg               *ManifestBufferConfig
+	chk               *OrchestratorCheck
 	ManifestChan      chan interface{}
 	bufferedManifests []interface{}
 	stopCh            chan struct{}
@@ -68,16 +65,12 @@ type ManifestBuffer struct {
 // NewManifestBuffer returns a new ManifestBuffer
 func NewManifestBuffer(chk *OrchestratorCheck) *ManifestBuffer {
 	manifestBuffer := &ManifestBuffer{
+		chk: chk,
 		Cfg: &ManifestBufferConfig{
-			ClusterID:                   chk.clusterID,
-			KubeClusterName:             chk.orchestratorConfig.KubeClusterName,
 			MsgGroupRef:                 chk.groupID,
-			MaxPerMessage:               chk.orchestratorConfig.MaxPerMessage,
-			MaxWeightPerMessageBytes:    chk.orchestratorConfig.MaxWeightPerMessageBytes,
 			BufferedManifestEnabled:     chk.orchestratorConfig.BufferedManifestEnabled,
 			MaxBufferedManifests:        chk.orchestratorConfig.MaxPerMessage,
 			ManifestBufferFlushInterval: chk.orchestratorConfig.ManifestBufferFlushInterval,
-			ExtraTags:                   chk.orchestratorConfig.ExtraTags,
 		},
 		ManifestChan: make(chan interface{}),
 		stopCh:       make(chan struct{}),
@@ -87,23 +80,26 @@ func NewManifestBuffer(chk *OrchestratorCheck) *ManifestBuffer {
 	return manifestBuffer
 }
 
-// flushManifest flushes manifests by chunking them first then sending them to the sender
-func (cb *ManifestBuffer) flushManifest(sender sender.Sender) {
-	manifests := cb.bufferedManifests
-	ctx := &processors.K8sProcessorContext{
+// newFlushContext is the single source of truth for shared per-check fields
+// on the buffered manifest flush path.
+func (cb *ManifestBuffer) newFlushContext() *processors.K8sProcessorContext {
+	return &processors.K8sProcessorContext{
 		BaseProcessorContext: processors.BaseProcessorContext{
-			MsgGroupID: cb.Cfg.MsgGroupRef.Inc(),
-			Cfg: &config.OrchestratorConfig{
-				KubeClusterName:          cb.Cfg.KubeClusterName,
-				MaxPerMessage:            cb.Cfg.MaxPerMessage,
-				MaxWeightPerMessageBytes: cb.Cfg.MaxWeightPerMessageBytes,
-				ExtraTags:                cb.Cfg.ExtraTags,
-			},
-			ClusterID: cb.Cfg.ClusterID,
+			Cfg:          cb.chk.orchestratorConfig,
+			MsgGroupID:   cb.Cfg.MsgGroupRef.Inc(),
+			ClusterID:    cb.chk.clusterID,
+			AgentVersion: cb.chk.agentVersion,
 		},
+		APIClient: cb.chk.apiClient,
 	}
+}
+
+// flushManifest flushes manifests by chunking them first then sending them to the sender
+func (cb *ManifestBuffer) flushManifest(sender sender.Sender) {
+	manifests := cb.bufferedManifests
+	ctx := cb.newFlushContext()
 	manifestMessages := processors.ChunkManifest(ctx, common.BaseHandlers{}.BuildManifestMessageBody, manifests)
-	sender.OrchestratorManifest(manifestMessages, cb.Cfg.ClusterID)
+	sender.OrchestratorManifest(manifestMessages, cb.chk.clusterID)
 	setManifestStats(manifests)
 	cb.bufferedManifests = cb.bufferedManifests[:0]
 }

pkg/collector/corechecks/cluster/orchestrator/manifest_buffer_test.go

@@ -123,11 +123,8 @@ func TestNewManifestBuffer(t *testing.T) {
 	assert.Equal(t, 0, len(mb.bufferedManifests))
 	assert.Equal(t, cap(mb.bufferedManifests), mb.Cfg.MaxBufferedManifests)
 
-	// Verify configuration was copied correctly
-	assert.Equal(t, orchCheck.clusterID, mb.Cfg.ClusterID)
-	assert.Equal(t, orchCheck.orchestratorConfig.KubeClusterName, mb.Cfg.KubeClusterName)
-	assert.Equal(t, orchCheck.orchestratorConfig.MaxPerMessage, mb.Cfg.MaxPerMessage)
-	assert.Equal(t, orchCheck.orchestratorConfig.MaxWeightPerMessageBytes, mb.Cfg.MaxWeightPerMessageBytes)
+	assert.Equal(t, orchCheck.orchestratorConfig.MaxPerMessage, mb.Cfg.MaxBufferedManifests)
+	assert.Same(t, orchCheck, mb.chk)
 }
 
 func TestFlushManifest(t *testing.T) {
@@ -171,10 +168,12 @@ func TestFlushManifest(t *testing.T) {
 
 	// Verify the manifest contains expected data
 	assert.Equal(t, "buffer-cluster", sentManifest.ClusterName)
-	assert.Equal(t, mb.Cfg.ClusterID, sentManifest.ClusterId)
+	assert.Equal(t, mb.chk.clusterID, sentManifest.ClusterId)
 	assert.Equal(t, []string{"tag:low"}, sentManifest.Tags) // From ExtraTags in test setup
 	assert.Equal(t, int32(1), sentManifest.GroupId)         // MsgGroupRef.Inc() should return 1 for first call
 	assert.Equal(t, int32(1), sentManifest.GroupSize)       // Only one chunk
+	require.NotNil(t, sentManifest.AgentVersion)
+	assert.Equal(t, mb.chk.agentVersion, sentManifest.AgentVersion)
 
 	// Verify manifests are correctly included
 	assert.Len(t, sentManifest.Manifests, 2)

pkg/collector/corechecks/cluster/orchestrator/orchestrator.go

@@ -113,6 +113,7 @@ func newOrchestratorCheck(base core.CheckBase, instance *OrchestratorInstance, c
 			Minor:  agentVersion.Minor,
 			Patch:  agentVersion.Patch,
 			Pre:    agentVersion.Pre,
+			Meta:   agentVersion.Meta,
 			Commit: agentVersion.Commit,
 		},
 	}