feat(errortracking): add core error-tracking package (#52485)

> **Stacked PR 1/4** — part of [#50607](#50607). Review this slice before stack 2.

### What does this PR do?

Introduces `pkg/util/log/errortracking/`, the foundational package for forwarding Agent `ERROR` logs to internal telemetry. This package is self-contained (stdlib-only) and defines the vocabulary used by the rest of the stack:

- `ErrorLog` — value-typed snapshot of an error log record (time, level, message, PC, attrs)
- `Submitter` — atomically-loaded callback type; non-blocking on the hot path
- `Handler` — `slog.Handler` that filters `Level >= Error`, builds an `ErrorLog`, and calls the installed `Submitter`. No queueing, no batching, no retry.
- `Bouncer` — deduplication/windowing logic that collapses repeated error stacks within a configurable time window

### Motivation

Part of AGTHEAL-15: Datadog Engineers need to monitor Agent error patterns at scale. Error log payloads drop the message to avoid leaking sensitive data and collect the full Go call stack.

### Stack
1. **This PR** — core `pkg/util/log/errortracking/` package
2. [stack-2] Agent telemetry component extensions + sender refactor
3. [stack-3] Config, log-setup wiring, agent run command
4. [stack-4] FakeIntake infrastructure + E2E tests + CI

Co-authored-by: paola.ducolin <paola.ducolin@datadoghq.com>

Author: pducolin

pkg/util/log/errortracking/BUILD.bazel

@@ -0,0 +1,27 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "errortracking",
+    srcs = [
+        "bouncer.go",
+        "doc.go",
+        "handler.go",
+        "types.go",
+    ],
+    importpath = "github.com/DataDog/datadog-agent/pkg/util/log/errortracking",
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "errortracking_test",
+    srcs = [
+        "bouncer_test.go",
+        "handler_test.go",
+    ],
+    embed = [":errortracking"],
+    gotags = ["test"],
+    deps = [
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
+)

pkg/util/log/errortracking/bouncer.go

@@ -0,0 +1,155 @@
+// 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 errortracking
+
+import (
+	"sync"
+	"time"
+)
+
+// Bouncer is a per-key first-sighting deduplicator with a sliding time
+// window. The first time a given key is observed inside a window,
+// Observe returns suppressed=false with count=1 and the observation
+// time as firstSeen. Subsequent observations of the same key inside the
+// same window are suppressed (return suppressed=true) and the count is
+// incremented. After the window elapses, the next observation returns
+// suppressed=false carrying the suppressed count of the prior window
+// (priorTotal-1, since the first sighting was already delivered), then
+// resets the entry to a fresh count=1. If no sightings were suppressed
+// (priorTotal==1), the rollover is silent and the next observation is
+// treated as a fresh first sighting (count=1).
+//
+// The key is an opaque uint64 — the caller is responsible for choosing
+// it. Today's caller (Handler.Handle) hashes the captured stack PCs
+// with FNV-1a so two distinct stacks reaching the same terminal
+// function are NOT collapsed into the same bouncer entry.
+//
+// The 15-minute default at the agent call site
+// (agent_telemetry.errortracking.bouncer_window_seconds) was chosen so a hot
+// bug path collapses to one record per quarter-hour — long enough to avoid
+// flooding the wire, short enough that operators see new error patterns
+// promptly. A hot bug path with N sightings per window ships Count=1 at
+// first sighting, then Count=N-1 on rollover (the suppressed portion).
+// Summing both gives N without double-counting the first sighting.
+//
+// Bouncer is purpose-built rather than reusing the global
+// rate.Sometimes wrapper in pkg/util/log/log_limit.go — that primitive
+// is keyless (one Limit per call site), whereas we need per-key state
+// and need to expose the per-key count. The implementation is a small
+// mutex-protected map with a periodic prune to bound memory; total
+// entries are capped so a pathological input cannot blow up memory.
+//
+// The zero value is NOT usable; construct via NewBouncer. Observe is
+// safe for concurrent use.
+type Bouncer struct {
+	window     time.Duration
+	maxEntries int
+
+	mu      sync.Mutex
+	entries map[uint64]*bouncerEntry
+}
+
+type bouncerEntry struct {
+	firstSeen time.Time
+	lastSeen  time.Time
+	count     uint32
+}
+
+// NewBouncer returns a Bouncer with the given sliding-window duration
+// and a soft cap on tracked entries. A non-positive window disables
+// dedup (Observe always returns suppressed=false with count=1); a
+// non-positive maxEntries falls back to a sane default (4096).
+func NewBouncer(window time.Duration, maxEntries int) *Bouncer {
+	if maxEntries <= 0 {
+		maxEntries = 4096
+	}
+	return &Bouncer{
+		window:     window,
+		maxEntries: maxEntries,
+		entries:    make(map[uint64]*bouncerEntry),
+	}
+}
+
+// Observe records a sighting of the given key at now and returns
+// whether the caller should suppress the record, the count to attach to
+// the next delivered record (≥ 1), and the firstSeen time of the
+// current window.
+//
+// The first sighting of a key in a window returns suppressed=false,
+// count=1, firstSeen=now. Subsequent sightings inside the same window
+// return suppressed=true with an incrementing count and the original
+// firstSeen. When the window elapses since firstSeen, the next
+// sighting returns suppressed=false with count=priorSuppressed (the
+// number of sightings that were suppressed in the elapsed window, i.e.
+// priorTotal-1), then resets the entry to a fresh count=1. If no
+// sightings were suppressed (priorTotal==1), the rollover is silent:
+// Observe returns suppressed=false, count=1, as though it were a fresh
+// first sighting, avoiding a Count=0 delivery.
+//
+// This design ensures consumers can sum Count fields without
+// double-counting: the first delivery carries Count=1, and the rollover
+// carries the remainder.
+//
+// When window is non-positive, Observe is a pass-through: returns
+// suppressed=false, count=1, firstSeen=now.
+func (b *Bouncer) Observe(key uint64, now time.Time) (suppressed bool, count uint32, firstSeen time.Time) {
+	if b.window <= 0 {
+		return false, 1, now
+	}
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if e, ok := b.entries[key]; ok {
+		if now.Sub(e.firstSeen) > b.window {
+			// Window elapsed. Deliver the suppressed portion of the prior
+			// window (priorTotal-1) so callers can sum without
+			// double-counting the first sighting already delivered.
+			// When nothing was suppressed (priorTotal==1), skip the
+			// rollover and treat this as a fresh first sighting.
+			priorCount := e.count
+			e.firstSeen = now
+			e.lastSeen = now
+			e.count = 1
+			if priorCount <= 1 {
+				return false, 1, now
+			}
+			return false, priorCount - 1, now
+		}
+		e.lastSeen = now
+		e.count++
+		return true, e.count, e.firstSeen
+	}
+
+	// New key. Prune opportunistically if at the cap so the map doesn't
+	// grow unboundedly under pathological churn (many unique keys, each
+	// appearing once). Re-check after pruning: if all entries are still
+	// within the window, pruneLocked removes nothing and the cap must be
+	// enforced by dropping the new key (pass-through, not suppressed).
+	if len(b.entries) >= b.maxEntries {
+		b.pruneLocked(now)
+		if len(b.entries) >= b.maxEntries {
+			return false, 1, now
+		}
+	}
+	b.entries[key] = &bouncerEntry{
+		firstSeen: now,
+		lastSeen:  now,
+		count:     1,
+	}
+	return false, 1, now
+}
+
+// pruneLocked drops entries whose firstSeen is older than the window.
+// Caller MUST hold b.mu. Linear in entry count; runs only on
+// near-capacity insert, so amortized cost is bounded.
+func (b *Bouncer) pruneLocked(now time.Time) {
+	for key, e := range b.entries {
+		if now.Sub(e.firstSeen) > b.window {
+			delete(b.entries, key)
+		}
+	}
+}

pkg/util/log/errortracking/bouncer_test.go

@@ -0,0 +1,177 @@
+// 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.
+
+//go:build test
+
+package errortracking
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBouncer_FirstSightingNotSuppressed(t *testing.T) {
+	b := NewBouncer(15*time.Minute, 0)
+	now := time.Now()
+	suppressed, count, firstSeen := b.Observe(0xCAFE, now)
+	assert.False(t, suppressed, "first sighting must NOT be suppressed")
+	assert.Equal(t, uint32(1), count, "first sighting count")
+	assert.True(t, firstSeen.Equal(now), "first sighting firstSeen = %v, want %v", firstSeen, now)
+}
+
+func TestBouncer_SecondSightingSuppressedInsideWindow(t *testing.T) {
+	b := NewBouncer(15*time.Minute, 0)
+	t0 := time.Now()
+	b.Observe(0xCAFE, t0)
+
+	t1 := t0.Add(time.Minute)
+	suppressed, count, firstSeen := b.Observe(0xCAFE, t1)
+	assert.True(t, suppressed, "second sighting inside window MUST be suppressed")
+	assert.Equal(t, uint32(2), count, "second sighting count")
+	assert.True(t, firstSeen.Equal(t0), "firstSeen drifted: got %v, want original %v", firstSeen, t0)
+}
+
+func TestBouncer_ResetsAfterWindow(t *testing.T) {
+	b := NewBouncer(15*time.Minute, 0)
+	t0 := time.Now()
+	b.Observe(0xCAFE, t0)                  // count=1
+	b.Observe(0xCAFE, t0.Add(time.Minute)) // count=2 (suppressed)
+
+	// Step past the window — next sighting MUST NOT be suppressed and
+	// MUST carry the suppressed count of the prior window (1 = total-1)
+	// rather than the full total, to avoid double-counting the first
+	// sighting already delivered. The entry is then reset to a fresh
+	// window internally.
+	t2 := t0.Add(16 * time.Minute)
+	suppressed, count, firstSeen := b.Observe(0xCAFE, t2)
+	assert.False(t, suppressed, "sighting after window MUST NOT be suppressed")
+	assert.Equal(t, uint32(1), count, "post-window count (want suppressed-only: total-1)")
+	assert.True(t, firstSeen.Equal(t2), "post-window firstSeen = %v, want %v", firstSeen, t2)
+}
+
+// TestBouncer_WindowElapseCarriesPriorCount exercises the
+// suppressed-count carry-forward contract end-to-end: a hot bug path
+// with N sightings per window must deliver Count=1 at first sighting
+// and Count=N-1 on rollover (the suppressed sightings), so consumers
+// summing both get N without double-counting the first sighting.
+func TestBouncer_WindowElapseCarriesPriorCount(t *testing.T) {
+	b := NewBouncer(15*time.Minute, 0)
+	var key uint64 = 0xABCDEF
+
+	t0 := time.Now()
+
+	// First sighting — delivered, count=1.
+	suppressed, count, _ := b.Observe(key, t0)
+	assert.False(t, suppressed, "first sighting must NOT be suppressed")
+	assert.Equal(t, uint32(1), count, "first sighting count")
+
+	// Two more sightings within the window — both suppressed.
+	suppressed, count, _ = b.Observe(key, t0.Add(5*time.Minute))
+	assert.True(t, suppressed, "sighting inside window MUST be suppressed")
+	assert.Equal(t, uint32(2), count, "second sighting count")
+
+	suppressed, count, _ = b.Observe(key, t0.Add(10*time.Minute))
+	assert.True(t, suppressed, "sighting inside window MUST be suppressed")
+	assert.Equal(t, uint32(3), count, "third sighting count")
+
+	// Window elapses; next sighting delivers the suppressed count of the
+	// prior window (2 = total(3) - first-already-delivered(1)).
+	suppressed, count, _ = b.Observe(key, t0.Add(20*time.Minute))
+	assert.False(t, suppressed, "window elapsed; sighting must be delivered")
+	assert.Equal(t, uint32(2), count, "rollover count (want suppressed-only: total-1)")
+
+	// Subsequent sighting in the new window — suppressed, count starts
+	// from the post-reset 1 and increments to 2.
+	suppressed, count, _ = b.Observe(key, t0.Add(21*time.Minute))
+	assert.True(t, suppressed, "sighting inside fresh window MUST be suppressed")
+	assert.Equal(t, uint32(2), count, "post-reset sighting count")
+}
+
+func TestBouncer_PerPCIndependent(t *testing.T) {
+	b := NewBouncer(15*time.Minute, 0)
+	now := time.Now()
+	b.Observe(0xCAFE, now)
+	// A different PC must NOT see the suppression state of 0xCAFE.
+	suppressed, count, _ := b.Observe(0xBEEF, now)
+	assert.False(t, suppressed, "different PC must NOT be suppressed by another PC's sighting")
+	assert.Equal(t, uint32(1), count, "different PC count")
+}
+
+func TestBouncer_DisabledWindowPassesThrough(t *testing.T) {
+	b := NewBouncer(0, 0)
+	now := time.Now()
+	for i := 0; i < 5; i++ {
+		suppressed, count, _ := b.Observe(0xCAFE, now.Add(time.Duration(i)*time.Second))
+		assert.False(t, suppressed, "disabled window must never suppress (i=%d)", i)
+		assert.Equal(t, uint32(1), count, "disabled window count (i=%d)", i)
+	}
+}
+
+func TestBouncer_RaceFree_ConcurrentObserve(t *testing.T) {
+	b := NewBouncer(15*time.Minute, 1024)
+	const goroutines = 32
+	const perGoroutine = 1000
+
+	var wg sync.WaitGroup
+	wg.Add(goroutines)
+	for g := 0; g < goroutines; g++ {
+		g := g
+		go func() {
+			defer wg.Done()
+			now := time.Now()
+			for i := 0; i < perGoroutine; i++ {
+				pc := uint64(g*1000 + i%50) // some overlap to exercise the suppress path
+				b.Observe(pc, now)
+			}
+		}()
+	}
+	wg.Wait()
+	// Survived without -race detecting anything; this is the
+	// observable for the test (the race detector flags concurrent
+	// map access or counter races at the failure site, not here).
+	t.Log("concurrent observe completed without race violation")
+}
+
+func TestBouncer_PrunesNearCap(t *testing.T) {
+	b := NewBouncer(time.Minute, 4)
+	t0 := time.Now()
+	// Fill the cap with entries that are about to expire.
+	for pc := uint64(1); pc <= 4; pc++ {
+		b.Observe(pc, t0)
+	}
+	// All entries are well past their window when we insert one more.
+	t1 := t0.Add(2 * time.Minute)
+	b.Observe(uint64(5), t1)
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	assert.LessOrEqual(t, len(b.entries), b.maxEntries, "entries exceeded cap after prune")
+}
+
+// TestBouncer_CapEnforced_WhenAllWithinWindow verifies that when the cap
+// is reached and pruneLocked removes nothing (all entries still within the
+// window), new keys are dropped rather than growing the map past maxEntries.
+// The dropped observation is returned as a pass-through (suppressed=false,
+// count=1) so the record still reaches the consumer without dedup tracking.
+func TestBouncer_CapEnforced_WhenAllWithinWindow(t *testing.T) {
+	b := NewBouncer(time.Hour, 4) // large window: no entries expire
+	t0 := time.Now()
+	for pc := uint64(1); pc <= 4; pc++ {
+		b.Observe(pc, t0)
+	}
+
+	// New unique key inserted while all existing entries are still within window.
+	t1 := t0.Add(time.Minute)
+	suppressed, count, _ := b.Observe(uint64(5), t1)
+	assert.False(t, suppressed, "dropped key must not be reported as suppressed")
+	assert.Equal(t, uint32(1), count, "dropped key must return count=1 (pass-through)")
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	assert.LessOrEqual(t, len(b.entries), b.maxEntries, "map must not exceed maxEntries")
+}