[autodiscovery] Advanced Discovery Probe Workqueue (#51890)

### What does this PR do?

Wires Configuration Discovery into Autodiscovery behind a `ConfigDiscoverer` interface. Templates carrying `discovery: {}` route through a workqueue-driven worker pool instead of the synchronous `configresolver.Resolve` path in cfgmgr. The feature is dormant pending the Python-backed bridge from #51501.

### Motivation

Foundation for Configuration Discovery for Agent Integrations (both for Containers + Processes). This delegate config creation to the Integration itself. The Integration can decided which ports telemetry exists on to be monitored.

### Describe how you validated your changes

Unit Tests. There is currently no e2e functionality, however, the unit tests show how this will be hooked in once the real implementation for DiscoveryConfig from the Python Integrations is connected.

### Additional Notes

Thoughts: `onDiscoveryResult` sends the changes through a channel `cm.discoveredCh <- changes` to be picked up by AutoConfig to actually apply the changes onto the Scheduler. During this gap, there could in theory be a race if the service gets deleted but the Discovery result could still apply. To make this atomic, we would need to push down into cfgmgr a callback for autoconfig to apply the changes there too or maybe share the same channel for processing this. Might do this in a separate PR to keep this one simpler.

Co-authored-by: vitkyrka <vincent.whitchurch@datadoghq.com>

Author: gabedos

comp/core/autodiscovery/discoverer/BUILD.bazel

@@ -0,0 +1,42 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "discoverer",
+    srcs = [
+        "discovery_json.go",
+        "python_bridge.go",  # //go:build python
+        "python_bridge_nopython.go",  # //go:build !python
+        "types.go",
+        "worker.go",
+    ],
+    importpath = "github.com/DataDog/datadog-agent/comp/core/autodiscovery/discoverer",
+    tags = ["manual"],
+    visibility = ["//visibility:public"],
+    deps = [
+        "//comp/core/autodiscovery/integration",
+        "//comp/core/workloadmeta/def",
+        "//pkg/util/log",
+        "@//pkg/collector/python",
+        "@io_k8s_apimachinery//pkg/util/wait",
+        "@io_k8s_client_go//util/workqueue",
+    ],
+)
+
+go_test(
+    name = "discoverer_test",
+    srcs = [
+        "discovery_json_test.go",
+        "fake_service_test.go",
+        "worker_test.go",
+    ],
+    embed = [":discoverer"],
+    gotags = ["test"],
+    tags = ["manual"],
+    deps = [
+        "//comp/core/autodiscovery/integration",
+        "//comp/core/workloadmeta/def",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+        "@org_uber_go_atomic//:atomic",
+    ],
+)

comp/core/autodiscovery/discoverer/discovery_json.go

@@ -0,0 +1,120 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package discoverer
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/integration"
+)
+
+// discoveryService is the JSON payload sent to the integration when asking it
+// to discover its config for a given service.
+type discoveryService struct {
+	ID    string          `json:"id"`
+	Host  string          `json:"host"`
+	Ports []discoveryPort `json:"ports"`
+}
+
+type discoveryPort struct {
+	Number int    `json:"number"`
+	Name   string `json:"name"`
+}
+
+// discoveredConfig is the JSON shape returned by the integration.
+type discoveredConfig struct {
+	Instances               []json.RawMessage `json:"instances"`
+	InitConfig              json.RawMessage   `json:"init_config"`
+	MetricConfig            json.RawMessage   `json:"metric_config"`
+	LogsConfig              json.RawMessage   `json:"logs"`
+	IgnoreAutodiscoveryTags bool              `json:"ignore_autodiscovery_tags"`
+	CheckTagCardinality     string            `json:"check_tag_cardinality"`
+}
+
+// marshalService builds the JSON payload sent to the integration for the
+// given live service. Returns ("", false, nil) when the service has no
+// usable host yet — typical during container startup, treated by callers as
+// a transient failure that warrants a retry.
+func marshalService(svc ServiceInfo) (string, bool, error) {
+	hosts, err := svc.GetHosts()
+	if err != nil {
+		return "", false, nil
+	}
+	host, ok := pickHost(hosts)
+	if !ok {
+		return "", false, nil
+	}
+	exposed, err := svc.GetPorts()
+	if err != nil {
+		return "", false, fmt.Errorf("GetPorts: %w", err)
+	}
+	payload := discoveryService{
+		ID:    svc.GetServiceID(),
+		Host:  host,
+		Ports: make([]discoveryPort, 0, len(exposed)),
+	}
+	for _, p := range exposed {
+		payload.Ports = append(payload.Ports, discoveryPort{Number: p.Port, Name: p.Name})
+	}
+	b, err := json.Marshal(payload)
+	if err != nil {
+		return "", false, fmt.Errorf("marshal: %w", err)
+	}
+	return string(b), true, nil
+}
+
+// pickHost applies the same fallback policy as %%host%%: single network →
+// use it; multiple networks with a "bridge" → use bridge; otherwise no
+// deterministic choice is possible and we return false.
+func pickHost(hosts map[string]string) (string, bool) {
+	if len(hosts) == 0 {
+		return "", false
+	}
+	if len(hosts) == 1 {
+		for _, ip := range hosts {
+			return ip, true
+		}
+	}
+	if ip, ok := hosts["bridge"]; ok {
+		return ip, true
+	}
+	return "", false
+}
+
+// parseDiscoveryResult turns the raw JSON returned by ConfigDiscoverer into a
+// slice of integration.Config. The configs returned here are not yet resolved
+// through configresolver — the caller is expected to run them through the
+// normal substitution + secret-decryption path before scheduling.
+func parseDiscoveryResult(integrationName, resultJSON string) ([]integration.Config, error) {
+	var raws []discoveredConfig
+	if err := json.Unmarshal([]byte(resultJSON), &raws); err != nil {
+		return nil, fmt.Errorf("decode discovery payload for %s: %w", integrationName, err)
+	}
+	if len(raws) == 0 {
+		return nil, nil
+	}
+	configs := make([]integration.Config, 0, len(raws))
+	for _, raw := range raws {
+		initConfig := raw.InitConfig
+		if len(initConfig) == 0 {
+			initConfig = json.RawMessage("{}")
+		}
+		cfg := integration.Config{
+			Name:                    integrationName,
+			InitConfig:              integration.Data(initConfig),
+			MetricConfig:            integration.Data(raw.MetricConfig),
+			LogsConfig:              integration.Data(raw.LogsConfig),
+			IgnoreAutodiscoveryTags: raw.IgnoreAutodiscoveryTags,
+			CheckTagCardinality:     raw.CheckTagCardinality,
+		}
+		for _, inst := range raw.Instances {
+			cfg.Instances = append(cfg.Instances, integration.Data(inst))
+		}
+		configs = append(configs, cfg)
+	}
+	return configs, nil
+}

comp/core/autodiscovery/discoverer/discovery_json_test.go

@@ -0,0 +1,156 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package discoverer
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseDiscoveryResult(t *testing.T) {
+	tests := []struct {
+		name            string
+		integration     string
+		payload         string
+		wantErr         bool
+		wantNil         bool
+		wantLen         int
+		wantNames       []string
+		wantInitConfigs []string // per-config; "" means default "{}"
+		wantInstLens    []int
+	}{
+		{
+			name:        "two configs use integration name",
+			integration: "redis",
+			payload: `[
+				{"instances": [{"host": "10.0.0.1"}], "init_config": {"foo": 1}},
+				{"instances": [{"host": "10.0.0.2"}]}
+			]`,
+			wantLen:         2,
+			wantNames:       []string{"redis", "redis"},
+			wantInitConfigs: []string{`{"foo":1}`, `{}`},
+			wantInstLens:    []int{1, 1},
+		},
+		{
+			name:            "integration name used when no name field",
+			integration:     "krakend",
+			payload:         `[{"instances":[{"host":"x"}]}]`,
+			wantLen:         1,
+			wantNames:       []string{"krakend"},
+			wantInitConfigs: []string{`{}`},
+			wantInstLens:    []int{1},
+		},
+		{
+			name:        "empty array returns nil",
+			integration: "redis",
+			payload:     `[]`,
+			wantNil:     true,
+		},
+		{
+			name:        "invalid JSON returns error",
+			integration: "redis",
+			payload:     `not-json`,
+			wantErr:     true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			configs, err := parseDiscoveryResult(tc.integration, tc.payload)
+			if tc.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			if tc.wantNil {
+				assert.Nil(t, configs)
+				return
+			}
+			require.Len(t, configs, tc.wantLen)
+			for i, cfg := range configs {
+				assert.Equal(t, tc.wantNames[i], cfg.Name)
+				assert.JSONEq(t, tc.wantInitConfigs[i], string(cfg.InitConfig))
+				assert.Len(t, cfg.Instances, tc.wantInstLens[i])
+			}
+		})
+	}
+}
+
+// TestMarshalService_PrefersBridgeOverOtherNetworks: with multiple networks
+// the bridge IP wins (matches %%host%%'s getFallbackHost).
+func TestMarshalService_PrefersBridgeOverOtherNetworks(t *testing.T) {
+	svc := &fakeService{
+		id:    "docker://abc",
+		hosts: map[string]string{"main": "1.2.3.4", "bridge": "10.0.0.1"},
+		ports: nil,
+	}
+	jsonStr, ok, err := marshalService(svc)
+	require.NoError(t, err)
+	require.True(t, ok)
+	var got discoveryService
+	require.NoError(t, json.Unmarshal([]byte(jsonStr), &got))
+	assert.Equal(t, "docker://abc", got.ID)
+	assert.Equal(t, "10.0.0.1", got.Host)
+	// Empty port list still serializes as an empty array, not null, so the
+	// Python side gets a stable shape.
+	assert.NotNil(t, got.Ports)
+	assert.Empty(t, got.Ports)
+}
+
+// TestMarshalService_SingleNetworkUsesIt: a service with exactly one network
+// uses that network's IP, matching the %%host%% single-network fallback.
+func TestMarshalService_SingleNetworkUsesIt(t *testing.T) {
+	svc := &fakeService{
+		id:    "docker://abc",
+		hosts: map[string]string{"main": "1.2.3.4"},
+		ports: nil,
+	}
+	jsonStr, ok, err := marshalService(svc)
+	require.NoError(t, err)
+	require.True(t, ok)
+	var got discoveryService
+	require.NoError(t, json.Unmarshal([]byte(jsonStr), &got))
+	assert.Equal(t, "1.2.3.4", got.Host)
+}
+
+// TestMarshalService_MultiNetworkWithoutBridge_NotOK: %%host%%'s policy
+// refuses to guess between equally-valid IPs when no bridge is present, so
+// we treat it as a transient failure (returns ok=false, no error).
+func TestMarshalService_MultiNetworkWithoutBridge_NotOK(t *testing.T) {
+	svc := &fakeService{
+		id:    "docker://abc",
+		hosts: map[string]string{"netA": "1.2.3.4", "netB": "5.6.7.8"},
+	}
+	_, ok, err := marshalService(svc)
+	require.NoError(t, err)
+	assert.False(t, ok)
+}
+
+func TestMarshalService_NoHostReturnsNotOK(t *testing.T) {
+	svc := &fakeService{id: "docker://abc", hosts: map[string]string{}}
+	_, ok, err := marshalService(svc)
+	require.NoError(t, err)
+	assert.False(t, ok)
+}
+
+func TestMarshalService_PortsRoundTrip(t *testing.T) {
+	svc := &fakeService{
+		id:    "docker://abc",
+		hosts: map[string]string{"main": "1.2.3.4"},
+		ports: []servicePort{{Name: "http", Port: 8080}, {Name: "metrics", Port: 9090}},
+	}
+	jsonStr, ok, err := marshalService(svc)
+	require.NoError(t, err)
+	require.True(t, ok)
+	var got discoveryService
+	require.NoError(t, json.Unmarshal([]byte(jsonStr), &got))
+	expected := []discoveryPort{{Number: 8080, Name: "http"}, {Number: 9090, Name: "metrics"}}
+	assert.True(t, reflect.DeepEqual(expected, got.Ports))
+}

comp/core/autodiscovery/discoverer/fake_service_test.go

@@ -0,0 +1,70 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package discoverer
+
+import (
+	"sync"
+
+	workloadmeta "github.com/DataDog/datadog-agent/comp/core/workloadmeta/def"
+)
+
+// servicePort is a tiny port struct used by fakeService so tests don't depend
+// on the workloadmeta package internals beyond the public ContainerPort.
+type servicePort struct {
+	Name string
+	Port int
+}
+
+// fakeService is a minimal ServiceInfo implementation for the
+// discoverer package's unit tests.
+type fakeService struct {
+	id       string
+	hosts    map[string]string
+	hostsErr error
+	ports    []servicePort
+	portsErr error
+}
+
+var _ ServiceInfo = (*fakeService)(nil)
+
+func (f *fakeService) GetServiceID() string { return f.id }
+func (f *fakeService) GetHosts() (map[string]string, error) {
+	return f.hosts, f.hostsErr
+}
+func (f *fakeService) GetPorts() ([]workloadmeta.ContainerPort, error) {
+	if f.portsErr != nil {
+		return nil, f.portsErr
+	}
+	out := make([]workloadmeta.ContainerPort, 0, len(f.ports))
+	for _, p := range f.ports {
+		out = append(out, workloadmeta.ContainerPort{Name: p.Name, Port: p.Port})
+	}
+	return out, nil
+}
+
+// fixedLookup is a ServiceLookup mock implementation.
+type fixedLookup struct {
+	mu       sync.Mutex
+	services map[string]ServiceInfo
+}
+
+func (l *fixedLookup) LookupService(svcID string) (ServiceInfo, bool) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.services == nil {
+		return nil, false
+	}
+	svc, ok := l.services[svcID]
+	return svc, ok
+}
+
+// remove drops the given svcID from the lookup, simulating a service
+// deletion observed by the worker on its next pop.
+func (l *fixedLookup) remove(svcID string) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	delete(l.services, svcID)
+}