DADP-71 Add ADP point telemetry to Agent telemetry

comp/core/agenttelemetry/impl/agenttelemetry.go

@@ -449,6 +449,38 @@ func (a *atel) transformMetricFamily(p *Profile, mfam *dto.MetricFamily) *agentm
 	}
 }
 
+// coalesceMetricFamilies merges compatible metric families with the same name.
+//
+// The regular and default telemetry registries are gathered separately. Coalescing lets profile aggregation see all
+// time series together instead of later payload writes overwriting earlier ones in the sender's metric map.
+func coalesceMetricFamilies(pms []*telemetry.MetricFamily) []*telemetry.MetricFamily {
+	mergedByName := make(map[string]*telemetry.MetricFamily, len(pms))
+	merged := make([]*telemetry.MetricFamily, 0, len(pms))
+
+	for _, pm := range pms {
+		if pm == nil || pm.Name == nil || pm.Type == nil {
+			merged = append(merged, pm)
+			continue
+		}
+
+		name := pm.GetName()
+		existing := mergedByName[name]
+		if existing == nil {
+			mergedByName[name] = pm
+			merged = append(merged, pm)
+			continue
+		}
+		if existing.GetType() != pm.GetType() {
+			merged = append(merged, pm)
+			continue
+		}
+
+		existing.Metric = append(existing.Metric, pm.Metric...)
+	}
+
+	return merged
+}
+
 func (a *atel) reportAgentMetrics(session *senderSession, pms []*telemetry.MetricFamily, p *Profile) {
 	// If no metrics are configured nothing to report
 	if len(p.metricsMap) == 0 {
@@ -495,6 +527,8 @@ func (a *atel) loadPayloads(profiles []*Profile) (*senderSession, error) {
 		a.logComp.Errorf("failed to get filtered telemetry metrics: %v", err)
 	}
 
+	pms = coalesceMetricFamilies(pms)
+
 	// All metrics stored in the "pms" slice above must follow the format:
 	//    <subsystem>__<metric_name>
 	// The "subsystem" and "name" should be concatenated with a double underscore ("__") separator,

comp/core/agenttelemetry/impl/agenttelemetry_test.go

@@ -2109,6 +2109,48 @@ func TestUsingPayloadCompressionInAgentTelemetrySender(t *testing.T) {
 	assert.True(t, float64(nonCompressBodyLen)/float64(compressBodyLen) > 1.5)
 }
 
+func TestCoalescesDefaultAndNoDefaultMetricFamiliesBeforeAggregation(t *testing.T) {
+	var c = `
+    agent_telemetry:
+      enabled: true
+      profiles:
+        - name: points
+          metric:
+            metrics:
+              - name: point.sent
+                aggregate_tags:
+                  - domain
+                  - remote_agent
+    `
+
+	// setup and initiate atel
+	tel := makeTelMock(t)
+	s := makeSenderImpl(t, nil, c)
+	r := newRunnerMock()
+	a := getTestAtel(t, tel, c, s, nil, r)
+	require.True(t, a.enabled)
+
+	corePointSent := tel.NewGaugeWithOpts("point", "sent", []string{"domain"}, "", telemetry.Options{DefaultMetric: true})
+	adpPointSent := tel.NewGaugeWithOpts("point", "sent", []string{"domain", "remote_agent"}, "", telemetry.Options{DefaultMetric: false})
+	corePointSent.Set(5, "https://api.datadoghq.com")
+	adpPointSent.Set(400, "https://api.datadoghq.com", "agent-data-plane")
+
+	metrics, ok := getPayloadFilteredMetricList(a, "point.sent")
+	require.True(t, ok)
+	require.Len(t, metrics, 2)
+
+	coreMetric, ok := getPayloadMetricByTagValues(metrics, map[string]interface{}{"domain": "https://api.datadoghq.com"})
+	require.True(t, ok)
+	assert.Equal(t, 5.0, coreMetric.Value)
+
+	adpMetric, ok := getPayloadMetricByTagValues(metrics, map[string]interface{}{
+		"domain":       "https://api.datadoghq.com",
+		"remote_agent": "agent-data-plane",
+	})
+	require.True(t, ok)
+	assert.Equal(t, 400.0, adpMetric.Value)
+}
+
 func TestDefaultAndNoDefaultPromRegistries(t *testing.T) {
 	var c = `
     agent_telemetry:

comp/core/agenttelemetry/impl/config.go

@@ -260,7 +260,13 @@ var defaultProfiles = `
         - name: logs.auto_multi_line_default_would_truncate
         - name: logs_destination.destination_workers
         - name: point.sent
+          aggregate_tags:
+            - domain
+            - remote_agent
         - name: point.dropped
+          aggregate_tags:
+            - domain
+            - remote_agent
         - name: transactions.input_count
         - name: transactions.requeued
         - name: transactions.retries

pkg/collector/corechecks/telemetry/check.go

@@ -38,21 +38,37 @@ func (c *checkImpl) Run() error {
 		return err
 	}
 
+	// Remote Agent Registry telemetry lives in the regular registry. Gather it on a best-effort basis so failures there
+	// do not prevent the customer-facing telemetry check from reporting Core Agent default telemetry values.
+	var regularMfs []*dto.MetricFamily
+	if gathered, err := c.telemetry.Gather(false); err != nil {
+		log.Warnf("failed to gather regular telemetry metrics for default telemetry merge: %v", err)
+	} else {
+		regularMfs = gathered
+	}
+
+	mergeLabelsByMetric := discoverMergeLabels(mfs, regularMfs)
+	mergedMetrics := collectMergeMetrics(mfs, false, mergeLabelsByMetric)
+	mergedMetrics.merge(collectMergeMetrics(regularMfs, true, mergeLabelsByMetric))
+
 	sender, err := c.GetSender()
 	if err != nil {
 		return err
 	}
 
 	sender.SetNoIndex(true)
 
+	c.sendMergedMetrics(mergedMetrics, sender)
 	c.handleMetricFamilies(mfs, sender)
 
 	return nil
 }
 
 func (c *checkImpl) handleMetricFamilies(mfs []*dto.MetricFamily, sender sender.Sender) {
 	for _, mf := range mfs {
-		if mf.Name == nil || mf.Type == nil || len(mf.Metric) == 0 {
+		// Merged metrics are emitted explicitly by sendMergedMetrics so overlapping regular-registry values can be included
+		// without changing customer-facing metric names or tags.
+		if mf == nil || mf.Name == nil || mf.Type == nil || len(mf.Metric) == 0 || isMergedMetric(mf.GetName()) {
 			continue
 		}
 

pkg/collector/corechecks/telemetry/check_test.go

@@ -9,13 +9,156 @@ import (
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
 
 	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/integration"
 	"github.com/DataDog/datadog-agent/pkg/aggregator/mocksender"
 	"github.com/DataDog/datadog-agent/pkg/collector/corechecks"
 )
 
+const domainLabel = "domain"
+
+func stringPtr(value string) *string {
+	return &value
+}
+
+func float64Ptr(value float64) *float64 {
+	return &value
+}
+
+func gaugeMetric(labels map[string]string, value float64) *dto.Metric {
+	metric := &dto.Metric{Gauge: &dto.Gauge{Value: float64Ptr(value)}}
+	for name, value := range labels {
+		metric.Label = append(metric.Label, &dto.LabelPair{Name: stringPtr(name), Value: stringPtr(value)})
+	}
+	return metric
+}
+
+func gaugeMetricFamily(name string, metrics ...*dto.Metric) *dto.MetricFamily {
+	metricType := dto.MetricType_GAUGE
+	return &dto.MetricFamily{
+		Name:   stringPtr(name),
+		Type:   &metricType,
+		Metric: metrics,
+	}
+}
+
+func counterMetricFamily(name string, value float64) *dto.MetricFamily {
+	metricType := dto.MetricType_COUNTER
+	return &dto.MetricFamily{
+		Name: stringPtr(name),
+		Type: &metricType,
+		Metric: []*dto.Metric{{
+			Counter: &dto.Counter{Value: float64Ptr(value)},
+		}},
+	}
+}
+
+func TestCollectAndMergeRegularRegistryMetrics(t *testing.T) {
+	defaultMfs := []*dto.MetricFamily{
+		gaugeMetricFamily(
+			"point__sent",
+			gaugeMetric(map[string]string{domainLabel: "https://api.datadoghq.com"}, 10),
+			gaugeMetric(map[string]string{}, 1),
+		),
+		gaugeMetricFamily(
+			"point__dropped",
+			gaugeMetric(map[string]string{domainLabel: "https://api.datadoghq.com"}, 2),
+		),
+	}
+	remoteMfs := []*dto.MetricFamily{
+		gaugeMetricFamily(
+			"point__sent",
+			gaugeMetric(map[string]string{
+				domainLabel:      "https://api.datadoghq.com",
+				remoteAgentLabel: "agent-data-plane",
+			}, 12),
+			gaugeMetric(map[string]string{
+				domainLabel:      "https://api.datadoghq.eu",
+				remoteAgentLabel: "other-remote-agent",
+			}, 5),
+			gaugeMetric(map[string]string{domainLabel: "https://api.datadoghq.com"}, 100),
+		),
+		gaugeMetricFamily(
+			"point__dropped",
+			gaugeMetric(map[string]string{
+				domainLabel:      "https://api.datadoghq.com",
+				remoteAgentLabel: "agent-data-plane",
+			}, 3),
+		),
+	}
+
+	labelsByMetric := discoverMergeLabels(defaultMfs, remoteMfs)
+	values := collectMergeMetrics(defaultMfs, false, labelsByMetric)
+	values.merge(collectMergeMetrics(remoteMfs, true, labelsByMetric))
+
+	require.Equal(t, []string{domainLabel}, labelsByMetric[pointSentMetric])
+	require.Equal(t, []string{domainLabel}, labelsByMetric[pointDroppedMetric])
+
+	sentDefaultDomain := values[pointSentMetric][mergeKey([]string{"domain:https://api.datadoghq.com"})]
+	require.Equal(t, mergeMetricSample{tags: []string{"domain:https://api.datadoghq.com"}, value: 22}, sentDefaultDomain)
+
+	sentEmptyDomain := values[pointSentMetric][mergeKey([]string{"domain:"})]
+	require.Equal(t, mergeMetricSample{tags: []string{"domain:"}, value: 1}, sentEmptyDomain)
+
+	sentRemoteOnlyDomain := values[pointSentMetric][mergeKey([]string{"domain:https://api.datadoghq.eu"})]
+	require.Equal(t, mergeMetricSample{tags: []string{"domain:https://api.datadoghq.eu"}, value: 5}, sentRemoteOnlyDomain)
+
+	droppedDefaultDomain := values[pointDroppedMetric][mergeKey([]string{"domain:https://api.datadoghq.com"})]
+	require.Equal(t, mergeMetricSample{tags: []string{"domain:https://api.datadoghq.com"}, value: 5}, droppedDefaultDomain)
+}
+
+func TestCollectMergeMetricsSkipsNonGaugeMetrics(t *testing.T) {
+	mfs := []*dto.MetricFamily{counterMetricFamily(pointSentMetric, 12)}
+
+	values := collectMergeMetrics(mfs, false, map[string][]string{pointSentMetric: {}})
+
+	require.Empty(t, values)
+}
+
+func TestDiscoverMergeLabelsFallsBackToRegularRegistry(t *testing.T) {
+	defaultMfs := []*dto.MetricFamily{}
+	regularMfs := []*dto.MetricFamily{
+		gaugeMetricFamily(
+			pointSentMetric,
+			gaugeMetric(map[string]string{
+				domainLabel:      "https://api.datadoghq.com",
+				remoteAgentLabel: "agent-data-plane",
+			}, 12),
+		),
+	}
+
+	labelsByMetric := discoverMergeLabels(defaultMfs, regularMfs)
+	values := collectMergeMetrics(regularMfs, true, labelsByMetric)
+
+	require.Equal(t, []string{domainLabel}, labelsByMetric[pointSentMetric])
+	require.Equal(t, mergeMetricSample{
+		tags:  []string{"domain:https://api.datadoghq.com"},
+		value: 12,
+	}, values[pointSentMetric][mergeKey([]string{"domain:https://api.datadoghq.com"})])
+}
+
+func TestSendMergedMetrics(t *testing.T) {
+	sm := mocksender.CreateDefaultDemultiplexer()
+	c := &checkImpl{CheckBase: corechecks.NewCheckBase(CheckName)}
+	c.Configure(sm, integration.FakeConfigHash, nil, nil, "test", "provider")
+
+	s := mocksender.NewMockSenderWithSenderManager(c.ID(), sm)
+	s.On("Gauge", "datadog.agent.point.sent", 22.0, "", []string{"domain:https://api.datadoghq.com"}).Return().Times(1)
+	s.On("Gauge", "datadog.agent.point.sent", 1.0, "", []string{"domain:"}).Return().Times(1)
+	s.On("Gauge", "datadog.agent.point.dropped", 5.0, "", []string{"domain:https://api.datadoghq.com"}).Return().Times(1)
+
+	values := newMergeMetricValues()
+	values.add(pointSentMetric, []string{"domain:"}, 1)
+	values.add(pointSentMetric, []string{"domain:https://api.datadoghq.com"}, 22)
+	values.add(pointDroppedMetric, []string{"domain:https://api.datadoghq.com"}, 5)
+
+	c.sendMergedMetrics(values, s)
+
+	s.AssertExpectations(t)
+}
+
 func TestCheck(t *testing.T) {
 	reg := prometheus.NewRegistry()