diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index f0f487aa36f6..02aec172331c 100644 --- a/.gitlab-ci.yml +++ b/.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: diff --git a/.gitlab/test/e2e/e2e.yml b/.gitlab/test/e2e/e2e.yml index 94c37bd35b40..8459c44f2412 100644 --- a/.gitlab/test/e2e/e2e.yml +++ b/.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: diff --git a/test/fakeintake/aggregator/BUILD.bazel b/test/fakeintake/aggregator/BUILD.bazel index 42a757c58e3d..ffcb8b85b4c7 100644 --- a/test/fakeintake/aggregator/BUILD.bazel +++ b/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", diff --git a/test/fakeintake/aggregator/apmTelemetryAggregator.go b/test/fakeintake/aggregator/apmTelemetryAggregator.go new file mode 100644 index 000000000000..57baba28f692 --- /dev/null +++ b/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), + } +} diff --git a/test/fakeintake/aggregator/apmTelemetryAggregator_test.go b/test/fakeintake/aggregator/apmTelemetryAggregator_test.go new file mode 100644 index 000000000000..39784d843096 --- /dev/null +++ b/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()) + }) +} diff --git a/test/fakeintake/aggregator/fixtures/apm_telemetry_bytes b/test/fakeintake/aggregator/fixtures/apm_telemetry_bytes new file mode 100644 index 000000000000..d632e2eb71fc --- /dev/null +++ b/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":""}]}} diff --git a/test/fakeintake/client/client.go b/test/fakeintake/client/client.go index 3c0a77cbe4c9..c89d511e2712 100644 --- a/test/fakeintake/client/client.go +++ b/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() diff --git a/test/fakeintake/client/client_test.go b/test/fakeintake/client/client_test.go index de5902e6505e..9757987507c2 100644 --- a/test/fakeintake/client/client_test.go +++ b/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) + }) + } diff --git a/test/fakeintake/server/serverstore/parser.go b/test/fakeintake/server/serverstore/parser.go index f9f8de4dfb4c..42337e1a1979 100644 --- a/test/fakeintake/server/serverstore/parser.go +++ b/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]