diff --git a/control-plane/internal/observability/telemetry.go b/control-plane/internal/observability/telemetry.go index 0398d18f2..51a2612f3 100644 --- a/control-plane/internal/observability/telemetry.go +++ b/control-plane/internal/observability/telemetry.go @@ -28,7 +28,15 @@ import ( const ( defaultTelemetryQueueSize = 256 telemetrySubscriberID = "anonymous-oss-telemetry" - telemetrySchemaVersion = 2 + telemetrySchemaVersion = 3 + + // telemetryReportedCapacity bounds the set of terminal outcomes this + // process remembers having already reported. An execution's outcome is + // reported once and never revisited, so the set only needs to span the + // window in which a duplicate can still arrive (an SDK's bounded callback + // retries, seconds at most). 8192 covers that with room to spare while + // keeping the footprint fixed on a control plane that runs for months. + telemetryReportedCapacity = 8192 ) var telemetryVersionPattern = regexp.MustCompile(`^v?[0-9]+(?:\.[0-9]+){0,3}(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$`) @@ -52,14 +60,17 @@ type telemetrySender func(context.Context, string, time.Duration, TelemetryEvent // TelemetryService subscribes to internal event buses and forwards anonymous, // low-cardinality usage events. It never forwards raw event payloads. type TelemetryService struct { - cfg config.TelemetryConfig - storageMode string - installHash string - eventIDKey []byte - runtimeName string - version string - timeout time.Duration - sender telemetrySender + cfg config.TelemetryConfig + storageMode string + installHash string + eventIDKey []byte + runtimeName string + usageContext string + version string + timeout time.Duration + sender telemetrySender + + reported telemetryReportedSet queue chan TelemetryEvent ctx context.Context @@ -93,16 +104,19 @@ func NewTelemetryService(cfg config.TelemetryConfig, agentfieldHome, storageMode return nil, err } + runtimeName := detectRuntime() + return &TelemetryService{ - cfg: cfg, - storageMode: normalizeStorageMode(storageMode), - installHash: hashInstallID(installID), - eventIDKey: eventIdentityKey(installID), - runtimeName: detectRuntime(), - version: emptyTo(version, "unknown"), - timeout: timeout, - sender: sendTelemetryEvent, - queue: make(chan TelemetryEvent, defaultTelemetryQueueSize), + cfg: cfg, + storageMode: normalizeStorageMode(storageMode), + installHash: hashInstallID(installID), + eventIDKey: eventIdentityKey(installID), + runtimeName: runtimeName, + usageContext: detectUsageContext(runtimeName), + version: emptyTo(version, "unknown"), + timeout: timeout, + sender: sendTelemetryEvent, + queue: make(chan TelemetryEvent, defaultTelemetryQueueSize), }, nil } @@ -195,10 +209,9 @@ func (s *TelemetryService) Start(ctx context.Context) { go s.subscribeExecutionEvents() s.Enqueue("control_plane_started", map[string]interface{}{ - "go_version": runtime.Version(), - "go_os": runtime.GOOS, - "go_arch": runtime.GOARCH, - "usage_context": detectUsageContext(s.runtimeName), + "go_version": runtime.Version(), + "go_os": runtime.GOOS, + "go_arch": runtime.GOARCH, }) logger.Logger.Info().Msg("anonymous OSS telemetry enabled") } @@ -232,7 +245,7 @@ func (s *TelemetryService) enqueue(eventName string, properties map[string]inter AgentFieldVersion: s.version, Runtime: s.runtimeName, StorageMode: normalizeStorageMode(s.storageMode), - Properties: sanitizeProperties(properties), + Properties: s.withUsageContext(sanitizeProperties(properties)), } if identityMaterial != "" && len(s.eventIDKey) != 0 { event.EventID = s.eventIdentity(eventName, identityMaterial) @@ -244,6 +257,26 @@ func (s *TelemetryService) enqueue(eventName string, properties map[string]inter } } +// withUsageContext stamps every event with where this control plane is +// running: "ci", "server", or "dev_or_local". +// +// It belongs on every event, not just control_plane_started. A CI job starts +// the control plane on a fresh volume, so it mints a new install ID and its +// executions are indistinguishable from a real first-time user's — and with +// the context carried only by the startup event, no downstream query could +// separate the two after the fact. Stamping it here is what makes +// execution_completed (and every other event) filterable at the source. +func (s *TelemetryService) withUsageContext(props map[string]interface{}) map[string]interface{} { + if s.usageContext == "" { + return props + } + if props == nil { + props = make(map[string]interface{}, 1) + } + props["usage_context"] = s.usageContext + return props +} + // eventIdentity returns a stable, opaque identifier suitable for idempotent // ingestion. Raw execution and workflow identifiers never leave the process. func (s *TelemetryService) eventIdentity(eventName, identityMaterial string) string { @@ -345,7 +378,8 @@ func (s *TelemetryService) handleExecutionEvent(event events.ExecutionEvent) { props["outcome"] = outcome } identityMaterial := "" - if hasStableCallbackIdentity(event, outcome) { + stable := hasStableCallbackIdentity(event, outcome) + if stable { identityMaterial = event.ExecutionID } else { // Updated/timeout transitions can legitimately recur (for example, @@ -357,9 +391,72 @@ func (s *TelemetryService) handleExecutionEvent(event events.ExecutionEvent) { } } identityMaterial += "\x00" + outcome + + // An execution reaches a given terminal outcome once, so a second event + // for one is a re-publish, not a second execution. The bus has produced + // those before — a status callback re-delivered after a lost 200 used to + // re-run every side effect, publishing another completed event and + // counting the execution again — and it stays one refactor away from + // producing them again. Suppressing here keeps a republish from becoming + // an inflated count even if the ingest side ignores telemetry_event_id. + // + // Only stable identities are eligible: the random ones above belong to + // transitions that are allowed to recur, and collapsing those would lose + // real events. + if stable && s.reported.observe(eventName+"\x00"+identityMaterial) { + logger.Logger.Debug(). + Str("event", eventName). + Msg("anonymous telemetry: outcome already reported for this execution; not sending again") + return + } s.enqueue(eventName, props, identityMaterial) } +// telemetryReportedSet remembers which terminal outcomes have already been +// reported, so a republished lifecycle event cannot count twice. +// +// Capacity is fixed: insertion past the limit evicts the oldest key, which is +// the right trade for a process that may run for months. A duplicate arrives +// within seconds of the original, so an eviction can only ever drop a key long +// past the window in which it could still suppress anything. +// +// The zero value is usable and allocates on first observe, so suppression is +// not something a caller can forget to wire up. +// +// Keys are built from raw execution IDs and never leave the process — only the +// HMAC in eventIdentity is ever sent. +type telemetryReportedSet struct { + mu sync.Mutex + capacity int + seen map[string]struct{} + order []string + next int +} + +// observe records key and reports whether it had already been seen. +func (s *telemetryReportedSet) observe(key string) bool { + s.mu.Lock() + defer s.mu.Unlock() + if s.seen == nil { + capacity := s.capacity + if capacity <= 0 { + capacity = telemetryReportedCapacity + } + s.seen = make(map[string]struct{}, capacity) + s.order = make([]string, capacity) + } + if _, ok := s.seen[key]; ok { + return true + } + if evicted := s.order[s.next]; evicted != "" { + delete(s.seen, evicted) + } + s.order[s.next] = key + s.next = (s.next + 1) % len(s.order) + s.seen[key] = struct{}{} + return false +} + func hasStableCallbackIdentity(event events.ExecutionEvent, outcome string) bool { if event.ExecutionID == "" { return false diff --git a/control-plane/internal/observability/telemetry_test.go b/control-plane/internal/observability/telemetry_test.go index 675fe65eb..8cc4c1a7b 100644 --- a/control-plane/internal/observability/telemetry_test.go +++ b/control-plane/internal/observability/telemetry_test.go @@ -242,11 +242,14 @@ func TestTelemetryExecutionEventIdentityIsStableAndOpaque(t *testing.T) { Status: "failed", } svc.handleExecutionEvent(event) - svc.handleExecutionEvent(event) - first, second := <-svc.queue, <-svc.queue + first := <-svc.queue - if first.EventID != second.EventID { - t.Fatalf("event identity is not stable: %q != %q", first.EventID, second.EventID) + // The identity a re-publish would carry has to match the one already sent, + // so ingest can recognize it as the same outcome. The duplicate itself no + // longer reaches the queue (see TestTelemetryTerminalOutcomeReportedOnce), + // so recompute it rather than sending a second event. + if replay := svc.eventIdentity("execution_failed", "private-execution-id\x00failed"); replay != first.EventID { + t.Fatalf("event identity is not stable: %q != %q", replay, first.EventID) } encoded, err := json.Marshal(first) if err != nil { @@ -259,6 +262,105 @@ func TestTelemetryExecutionEventIdentityIsStableAndOpaque(t *testing.T) { } } +// A terminal outcome is reached once per execution, so a second event for one +// is a re-publish. Counting it again is what inflated execution_completed when +// duplicate status callbacks re-ran the publish path. +func TestTelemetryTerminalOutcomeReportedOnce(t *testing.T) { + for _, tt := range []struct { + name string + event events.ExecutionEvent + }{ + { + name: "completed", + event: events.ExecutionEvent{Type: events.ExecutionCompleted, ExecutionID: "exec-1", Status: "succeeded"}, + }, + { + name: "failed", + event: events.ExecutionEvent{Type: events.ExecutionFailed, ExecutionID: "exec-2", Status: "failed"}, + }, + { + name: "cancelled", + event: events.ExecutionEvent{Type: events.ExecutionCancelledEvent, ExecutionID: "exec-3", Status: "cancelled"}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + svc := &TelemetryService{ + installHash: "install-hash", + eventIDKey: eventIdentityKey("private-install-id"), + runtimeName: "binary", + version: "test", + queue: make(chan TelemetryEvent, 4), + } + for i := 0; i < 4; i++ { + svc.handleExecutionEvent(tt.event) + } + if got := len(svc.queue); got != 1 { + t.Fatalf("four identical terminal events enqueued %d times, want 1", got) + } + }) + } +} + +// Suppression is keyed on the execution, not the event name, so two executions +// completing must both be reported. +func TestTelemetryTerminalSuppressionIsPerExecution(t *testing.T) { + svc := &TelemetryService{ + installHash: "install-hash", + eventIDKey: eventIdentityKey("private-install-id"), + runtimeName: "binary", + version: "test", + queue: make(chan TelemetryEvent, 4), + } + svc.handleExecutionEvent(events.ExecutionEvent{Type: events.ExecutionCompleted, ExecutionID: "exec-1", Status: "succeeded"}) + svc.handleExecutionEvent(events.ExecutionEvent{Type: events.ExecutionCompleted, ExecutionID: "exec-2", Status: "succeeded"}) + + if got := len(svc.queue); got != 2 { + t.Fatalf("two distinct executions enqueued %d events, want 2", got) + } +} + +// Non-terminal lifecycle events recur by design (a retried execution starts +// more than once) and must never be collapsed. +func TestTelemetryNonTerminalEventsAreNotSuppressed(t *testing.T) { + svc := &TelemetryService{ + installHash: "install-hash", + eventIDKey: eventIdentityKey("private-install-id"), + runtimeName: "binary", + version: "test", + queue: make(chan TelemetryEvent, 4), + } + event := events.ExecutionEvent{Type: events.ExecutionStarted, ExecutionID: "exec-1", Status: "running"} + svc.handleExecutionEvent(event) + svc.handleExecutionEvent(event) + + if got := len(svc.queue); got != 2 { + t.Fatalf("repeated execution_started enqueued %d events, want 2", got) + } +} + +// The set is bounded, so a long-lived control plane cannot grow it without +// limit. Eviction is safe because duplicates arrive within seconds. +func TestTelemetryReportedSetEvictsOldestPastCapacity(t *testing.T) { + set := &telemetryReportedSet{capacity: 2} + + if set.observe("a") || set.observe("b") { + t.Fatal("first sighting of a key reported as a duplicate") + } + if !set.observe("a") { + t.Fatal("key still within capacity was not recognized") + } + if set.observe("c") { + t.Fatal("third distinct key reported as a duplicate") + } + // "c" evicted "a", the oldest insertion. + if set.observe("a") { + t.Fatal("evicted key was still recognized") + } + if len(set.seen) != 2 { + t.Fatalf("set holds %d keys, want capacity 2", len(set.seen)) + } +} + func TestTelemetryExecutionEventIdentityDoesNotCollapseMissingIDs(t *testing.T) { svc := &TelemetryService{ installHash: "install-hash", @@ -299,6 +401,74 @@ func TestTelemetryTimeoutEventIdentityDoesNotCollapseRepeatedTransitions(t *test } } +// usage_context has to ride on every event. A CI job mints a fresh install ID, +// so without it a CI execution is indistinguishable from a real user's and no +// downstream query can separate the two after ingestion. +func TestTelemetryUsageContextStampedOnEveryEvent(t *testing.T) { + svc := &TelemetryService{ + installHash: "install-hash", + eventIDKey: eventIdentityKey("private-install-id"), + runtimeName: "binary", + usageContext: "ci", + version: "test", + queue: make(chan TelemetryEvent, 4), + } + + svc.handleExecutionEvent(events.ExecutionEvent{ + Type: events.ExecutionCompleted, + ExecutionID: "exec-1", + Status: "succeeded", + }) + svc.handleNodeEvent(events.NodeEvent{Type: events.NodeRegistered}) + svc.Enqueue("control_plane_stopped", nil) + + for len(svc.queue) > 0 { + event := <-svc.queue + if got := event.Properties["usage_context"]; got != "ci" { + t.Fatalf("%s carried usage_context %v, want \"ci\"", event.EventName, got) + } + } +} + +func TestTelemetryDetectsUsageContext(t *testing.T) { + ciKeys := []string{"CI", "GITHUB_ACTIONS", "GITLAB_CI", "BUILDKITE", "CIRCLECI", "JENKINS_URL"} + + // The suite itself usually runs under CI, so each case has to start from a + // clean slate or every one of them would pass on the ambient CI variable. + clearCIEnv := func(t *testing.T) { + t.Helper() + for _, key := range ciKeys { + t.Setenv(key, "") + } + } + + for _, key := range ciKeys { + t.Run(key, func(t *testing.T) { + clearCIEnv(t) + t.Setenv(key, "1") + if got := detectUsageContext("binary"); got != "ci" { + t.Fatalf("detectUsageContext with %s set = %q, want \"ci\"", key, got) + } + }) + } + + t.Run("container runtimes are servers", func(t *testing.T) { + clearCIEnv(t) + for _, runtimeName := range []string{"docker", "kubernetes"} { + if got := detectUsageContext(runtimeName); got != "server" { + t.Fatalf("detectUsageContext(%q) = %q, want \"server\"", runtimeName, got) + } + } + }) + + t.Run("a plain binary is a developer", func(t *testing.T) { + clearCIEnv(t) + if got := detectUsageContext("binary"); got != "dev_or_local" { + t.Fatalf("detectUsageContext(\"binary\") = %q, want \"dev_or_local\"", got) + } + }) +} + func TestTelemetryFailureCategoriesMatchRuntimeVocabulary(t *testing.T) { for _, category := range []string{ "agent_error", diff --git a/docs/ENVIRONMENT_VARIABLES.md b/docs/ENVIRONMENT_VARIABLES.md index a96c99e77..8aedf2f84 100644 --- a/docs/ENVIRONMENT_VARIABLES.md +++ b/docs/ENVIRONMENT_VARIABLES.md @@ -98,6 +98,10 @@ Anonymous usage telemetry is enabled by default to help us improve AgentField. I The telemetry payload does not include prompts, inputs, outputs, logs, secrets, API keys, IP addresses, hostnames, user IDs, DIDs, or raw error text. Sending is best-effort and does not affect control-plane or execution behavior. +Every event carries a `usage_context` property describing where the control plane is running — `ci` when a CI environment variable is present (`CI`, `GITHUB_ACTIONS`, `GITLAB_CI`, `BUILDKITE`, `CIRCLECI`, or `JENKINS_URL`), `server` for a container or Kubernetes runtime, and `dev_or_local` otherwise. A CI job typically starts on a fresh volume and mints a new install ID, so without this property its activity is indistinguishable from a real first-time user's; filter on it when reading product metrics. Set `AGENTFIELD_TELEMETRY_ENABLED=false` in CI to opt out of reporting entirely. + +An execution's terminal outcome is reported once. If a lifecycle event is republished — for example when an agent SDK retries a status callback after a lost response — the control plane suppresses the repeat rather than counting the execution twice. + - `AGENTFIELD_TELEMETRY_ENABLED` (default: `true`): Set to `false` to disable anonymous usage telemetry. - `AGENTFIELD_TELEMETRY_ENDPOINT` (default: `https://agentfield.ai/api/oss/telemetry`): Hosted anonymous telemetry endpoint. - `AGENTFIELD_TELEMETRY_INSTALL_ID` (optional): Stable externally managed installation ID. Use a random, opaque value—not an email, account name, hostname, or other identifying value. The control plane hashes it before sending.