test(errortracking): add FakeIntake support for agent telemetry logs

.gitlab-ci.yml

@@ -1387,6 +1387,15 @@ workflow:
       compare_to: $COMPARE_TO_BRANCH
     when: on_success
 
+.on_agenttelemetry_or_e2e_changes:
+  - !reference [.on_e2e_main_release_or_rc]
+  - changes:
+      paths:
+        - comp/core/agenttelemetry/**/*
+        - pkg/util/log/errortracking/**/*
+      compare_to: $COMPARE_TO_BRANCH
+    when: on_success
+
 .on_privateactionrunner_or_e2e_changes:
   - !reference [.on_e2e_main_release_or_rc]
   - changes:

.gitlab/test/e2e/e2e.yml

@@ -435,6 +435,16 @@ new-e2e-agent-health:
     TARGETS: ./tests/agent-health
     TEAM: agent-health
 
+new-e2e-agent-telemetry:
+  extends: .new_e2e_template_needs_deb_x64
+  rules:
+    # only manual, will enable on changes in follow up pr
+    # - !reference [.on_agenttelemetry_or_e2e_changes]
+    - !reference [.manual]
+  variables:
+    TARGETS: ./tests/agent-telemetry
+    TEAM: agent-runtimes
+
 new-e2e-privateactionrunner:
   extends: .new_e2e_template
   needs:

test/fakeintake/aggregator/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "agenthealthAggregator.go",
         "apmStatsAggregator.go",
+        "apmTelemetryAggregator.go",
         "checkRunAggregator.go",
         "common.go",
         "connectionsAggregator.go",
@@ -58,6 +59,7 @@ go_test(
     srcs = [
         "agenthealthAggregator_test.go",
         "apmStatsAggregator_test.go",
+        "apmTelemetryAggregator_test.go",
         "checkRunAggregator_test.go",
         "common_test.go",
         "connectionsAggregator_test.go",
@@ -87,6 +89,7 @@ go_test(
         "fixtures/agenthealth_multi_bytes",
         "fixtures/agenthealth_noissues_bytes",
         "fixtures/apm_stats_bytes",
+        "fixtures/apm_telemetry_bytes",
         "fixtures/checkrun_bytes",
         "fixtures/connections_bytes",
         "fixtures/contimage_bytes",

test/fakeintake/aggregator/apmTelemetryAggregator.go

@@ -0,0 +1,94 @@
+// 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 aggregator
+
+import (
+	"bytes"
+	"encoding/json"
+	"time"
+
+	"github.com/DataDog/datadog-agent/test/fakeintake/api"
+)
+
+// AgentTelemetryLog represents a single log record from the agent's internal
+// telemetry pipeline. It mirrors comp/core/agenttelemetry/impl/logs_payload.go::Log
+// and is populated only for payloads whose request_type is "agent-logs".
+type AgentTelemetryLog struct {
+	Level         string `json:"level"`
+	StackTrace    string `json:"stack_trace"`
+	TracerTime    int64  `json:"tracer_time"`
+	Count         int    `json:"count"`
+	IsCrash       bool   `json:"is_crash"`
+	Message       string `json:"message"`
+	ErrorKind     string `json:"error_kind"`
+	Tags          string `json:"tags"`
+	collectedTime time.Time
+}
+
+// name returns the grouping key used by the aggregator. All agent-logs records
+// share one key; per-hostname grouping is not needed for single-agent test VMs.
+func (l *AgentTelemetryLog) name() string { return "agent-errortracking" }
+
+// GetTags returns no tags — agent telemetry logs carry none on the wire.
+func (l *AgentTelemetryLog) GetTags() []string { return []string{} }
+
+// GetCollectedTime returns when the fakeintake server received this payload.
+func (l *AgentTelemetryLog) GetCollectedTime() time.Time { return l.collectedTime }
+
+// apmTelemetryEnvelope is the outer POST body sent to /api/v2/apmtelemetry.
+// The same endpoint receives metrics, heartbeats, and other request types;
+// request_type discriminates agent-logs records from the rest.
+type apmTelemetryEnvelope struct {
+	RequestType string `json:"request_type"`
+	Payload     struct {
+		Logs []*AgentTelemetryLog `json:"logs"`
+	} `json:"payload"`
+}
+
+// ParseAgentTelemetryLogs parses one api.Payload into zero-or-more
+// AgentTelemetryLog items. Payloads whose request_type is not "agent-logs"
+// are silently skipped — the /api/v2/apmtelemetry endpoint is shared with
+// agent metrics, heartbeats, and other telemetry types.
+func ParseAgentTelemetryLogs(payload api.Payload) ([]*AgentTelemetryLog, error) {
+	if bytes.Equal(payload.Data, []byte("{}")) {
+		return []*AgentTelemetryLog{}, nil
+	}
+
+	inflated, err := inflate(payload.Data, payload.Encoding)
+	if err != nil {
+		return nil, err
+	}
+
+	// The /api/v2/apmtelemetry endpoint is shared: agent metrics, heartbeats,
+	// and other payload types arrive here too, some encoded as protobuf. If
+	// JSON unmarshal fails, skip silently — this payload is not agent-logs.
+	var env apmTelemetryEnvelope
+	if err := json.Unmarshal(inflated, &env); err != nil {
+		return []*AgentTelemetryLog{}, nil
+	}
+
+	if env.RequestType != "agent-logs" {
+		return []*AgentTelemetryLog{}, nil
+	}
+
+	for _, l := range env.Payload.Logs {
+		l.collectedTime = payload.Timestamp
+	}
+	return env.Payload.Logs, nil
+}
+
+// AgentTelemetryLogAggregator aggregates AgentTelemetryLog payloads received
+// on the /api/v2/apmtelemetry endpoint.
+type AgentTelemetryLogAggregator struct {
+	Aggregator[*AgentTelemetryLog]
+}
+
+// NewAgentTelemetryLogAggregator returns a new AgentTelemetryLogAggregator.
+func NewAgentTelemetryLogAggregator() AgentTelemetryLogAggregator {
+	return AgentTelemetryLogAggregator{
+		Aggregator: newAggregator(ParseAgentTelemetryLogs),
+	}
+}

test/fakeintake/aggregator/apmTelemetryAggregator_test.go

@@ -0,0 +1,94 @@
+// 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 aggregator
+
+import (
+	_ "embed"
+	"testing"
+	"time"
+
+	"github.com/DataDog/datadog-agent/test/fakeintake/api"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+//go:embed fixtures/apm_telemetry_bytes
+var apmTelemetryData []byte
+
+func TestParseAgentTelemetryLogs(t *testing.T) {
+	t.Run("empty JSON object is silently skipped", func(t *testing.T) {
+		logs, err := ParseAgentTelemetryLogs(api.Payload{
+			Data:     []byte("{}"),
+			Encoding: encodingJSON,
+		})
+		require.NoError(t, err)
+		assert.Empty(t, logs)
+	})
+
+	t.Run("non-agent-logs request_type is silently skipped", func(t *testing.T) {
+		logs, err := ParseAgentTelemetryLogs(api.Payload{
+			Data:     []byte(`{"request_type":"agent-metrics","payload":{}}`),
+			Encoding: encodingJSON,
+		})
+		require.NoError(t, err)
+		assert.Empty(t, logs)
+	})
+
+	t.Run("non-JSON payload (e.g. protobuf) is silently skipped", func(t *testing.T) {
+		logs, err := ParseAgentTelemetryLogs(api.Payload{
+			Data:     []byte("not valid json"),
+			Encoding: encodingJSON,
+		})
+		require.NoError(t, err)
+		assert.Empty(t, logs)
+	})
+
+	t.Run("valid agent-logs payload parses all fields and stamps collected time", func(t *testing.T) {
+		ts := time.Date(2025, 6, 1, 12, 0, 0, 0, time.UTC)
+		logs, err := ParseAgentTelemetryLogs(api.Payload{
+			Data:      apmTelemetryData,
+			Encoding:  encodingJSON,
+			Timestamp: ts,
+		})
+		require.NoError(t, err)
+		require.Len(t, logs, 1)
+
+		l := logs[0]
+		assert.Equal(t, "ERROR", l.Level)
+		assert.Contains(t, l.StackTrace, "main.main()")
+		assert.Equal(t, int64(1234567890), l.TracerTime)
+		assert.Equal(t, 3, l.Count)
+		assert.False(t, l.IsCrash)
+		assert.Empty(t, l.Message)
+		assert.Equal(t, ts, l.GetCollectedTime())
+	})
+
+	t.Run("multiple logs in one payload are all returned", func(t *testing.T) {
+		data := []byte(`{"request_type":"agent-logs","payload":{"logs":[` +
+			`{"level":"ERROR","stack_trace":"foo","tracer_time":1,"count":1,"is_crash":false},` +
+			`{"level":"ERROR","stack_trace":"bar","tracer_time":2,"count":2,"is_crash":true}` +
+			`]}}`)
+		logs, err := ParseAgentTelemetryLogs(api.Payload{
+			Data:     data,
+			Encoding: encodingJSON,
+		})
+		require.NoError(t, err)
+		require.Len(t, logs, 2)
+		assert.Equal(t, int64(1), logs[0].TracerTime)
+		assert.Equal(t, int64(2), logs[1].TracerTime)
+		assert.True(t, logs[1].IsCrash)
+	})
+
+	t.Run("GetTags returns empty slice", func(t *testing.T) {
+		l := &AgentTelemetryLog{}
+		assert.Empty(t, l.GetTags())
+	})
+
+	t.Run("name returns agent-errortracking", func(t *testing.T) {
+		l := &AgentTelemetryLog{}
+		assert.Equal(t, "agent-errortracking", l.name())
+	})
+}

test/fakeintake/aggregator/fixtures/apm_telemetry_bytes

@@ -0,0 +1 @@
+{"request_type":"agent-logs","payload":{"logs":[{"level":"ERROR","stack_trace":"goroutine 1 [running]:\nmain.main()\n\t/app/main.go:10 +0x58","tracer_time":1234567890,"count":3,"is_crash":false,"message":""}]}}

test/fakeintake/client/client.go

@@ -154,6 +154,7 @@ type Client struct {
 	ncmAggregator                  aggregator.NCMAggregator
 	hostAggregator                 aggregator.HostTagsAggregator
 	agentHealthAggregator          aggregator.AgentHealthAggregator
+	agentTelemetryLogAggregator    aggregator.AgentTelemetryLogAggregator
 }
 
 // NewClient creates a new fake intake client
@@ -189,6 +190,7 @@ func NewClient(fakeIntakeURL string, opts ...Option) *Client {
 		ncmAggregator:                  aggregator.NewNCMAggregator(),
 		hostAggregator:                 aggregator.NewHostTagsAggregator(),
 		agentHealthAggregator:          aggregator.NewAgentHealthAggregator(),
+		agentTelemetryLogAggregator:    aggregator.NewAgentTelemetryLogAggregator(),
 	}
 	for _, opt := range opts {
 		opt(client)
@@ -380,6 +382,14 @@ func (c *Client) getAgentHealth() error {
 	return c.agentHealthAggregator.UnmarshallPayloads(payloads)
 }
 
+func (c *Client) getAgentTelemetryLogs() error {
+	payloads, err := c.getFakePayloads(apmTelemetryEndpoint)
+	if err != nil {
+		return err
+	}
+	return c.agentTelemetryLogAggregator.UnmarshallPayloads(payloads)
+}
+
 // FilterMetrics fetches fakeintake on both `/api/v2/series` and `/api/intake/metrics/v3/series`
 // and returns metrics matching `name` and any [MatchOpt](#MatchOpt) options.
 // Results from both endpoints are merged.
@@ -742,6 +752,7 @@ func (c *Client) FlushServerAndResetAggregators() error {
 	c.logAggregator.Reset()
 	c.apmStatsAggregator.Reset()
 	c.traceAggregator.Reset()
+	c.agentTelemetryLogAggregator.Reset()
 	return nil
 }
 
@@ -1187,6 +1198,21 @@ func (c *Client) GetHosts() ([]string, error) {
 	return c.hostAggregator.GetNames(), nil
 }
 
+// GetAgentTelemetryLogs fetches fakeintake on `/api/v2/apmtelemetry` and returns
+// all agent-logs telemetry records received since the last flush. Each call
+// fetches the server's full accumulated state (UnmarshallPayloads replaces, not
+// appends). Payloads whose request_type is not "agent-logs" are silently skipped.
+func (c *Client) GetAgentTelemetryLogs() ([]*aggregator.AgentTelemetryLog, error) {
+	if err := c.getAgentTelemetryLogs(); err != nil {
+		return nil, err
+	}
+	var logs []*aggregator.AgentTelemetryLog
+	for _, name := range c.agentTelemetryLogAggregator.GetNames() {
+		logs = append(logs, c.agentTelemetryLogAggregator.GetPayloadsByName(name)...)
+	}
+	return logs, nil
+}
+
 // GetAgentHealth fetches fakeintake on `/api/v2/agenthealth` endpoint and returns all received agent health payloads
 func (c *Client) GetAgentHealth() ([]*aggregator.AgentHealthPayload, error) {
 	err := c.getAgentHealth()

test/fakeintake/client/client_test.go

@@ -732,4 +732,51 @@ func TestClient(t *testing.T) {
 		assert.Contains(t, issue.Tags, "docker:installed")
 	})
 
+	t.Run("getAgentTelemetryLogs", func(t *testing.T) {
+		payload := `{"request_type":"agent-logs","payload":{"logs":[{"level":"ERROR","stack_trace":"main.main()","tracer_time":1234567890,"count":3,"is_crash":false,"message":""}]}}`
+		response, err := json.Marshal(api.APIFakeIntakePayloadsRawGETResponse{
+			Payloads: []api.Payload{
+				{Data: []byte(payload), Encoding: "application/json"},
+			},
+		})
+		require.NoError(t, err)
+
+		ts := NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			w.Write(response)
+		}))
+		defer ts.Close()
+
+		client := NewClient(ts.URL)
+		err = client.getAgentTelemetryLogs()
+		require.NoError(t, err)
+		assert.True(t, client.agentTelemetryLogAggregator.ContainsPayloadName("agent-errortracking"))
+		assert.False(t, client.agentTelemetryLogAggregator.ContainsPayloadName("totoro"))
+	})
+
+	t.Run("GetAgentTelemetryLogs", func(t *testing.T) {
+		payload := `{"request_type":"agent-logs","payload":{"logs":[{"level":"ERROR","stack_trace":"main.main()","tracer_time":1234567890,"count":3,"is_crash":false,"message":""}]}}`
+		response, err := json.Marshal(api.APIFakeIntakePayloadsRawGETResponse{
+			Payloads: []api.Payload{
+				{Data: []byte(payload), Encoding: "application/json"},
+			},
+		})
+		require.NoError(t, err)
+
+		ts := NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			w.Write(response)
+		}))
+		defer ts.Close()
+
+		client := NewClient(ts.URL)
+		logs, err := client.GetAgentTelemetryLogs()
+		require.NoError(t, err)
+		require.Len(t, logs, 1)
+		assert.Equal(t, "ERROR", logs[0].Level)
+		assert.Equal(t, "main.main()", logs[0].StackTrace)
+		assert.Equal(t, int64(1234567890), logs[0].TracerTime)
+		assert.Equal(t, 3, logs[0].Count)
+		assert.False(t, logs[0].IsCrash)
+		assert.Empty(t, logs[0].Message)
+	})
+
 }

test/fakeintake/server/serverstore/parser.go

@@ -20,6 +20,7 @@ var parserMap = map[string]parserFunc{
 	"/api/v1/connections":           getConnectionsPayLoadProtobuf,
 	"/api/beta/sketches":            getSketchPayloadProtobuf,
 	"/api/intake/metrics/v3/series": getMetricV3SeriesPayload,
+	"/api/v2/apmtelemetry":          getAgentTelemetryLogsJSON,
 }
 
 func getLogPayLoadJSON(payload api.Payload) (interface{}, error) {
@@ -50,6 +51,10 @@ func getMetricV3SeriesPayload(payload api.Payload) (interface{}, error) {
 	return aggregator.ParseMetricSeriesV3(payload)
 }
 
+func getAgentTelemetryLogsJSON(payload api.Payload) (interface{}, error) {
+	return aggregator.ParseAgentTelemetryLogs(payload)
+}
+
 // IsRouteHandled checks if a route is handled by the Datadog parsed store
 func IsRouteHandled(route string) bool {
 	_, ok := parserMap[route]