diff --git a/internal/infra/http/handler/runtime_telemetry_handler.go b/internal/infra/http/handler/runtime_telemetry_handler.go index d5efd6a9..f79d1cd9 100644 --- a/internal/infra/http/handler/runtime_telemetry_handler.go +++ b/internal/infra/http/handler/runtime_telemetry_handler.go @@ -61,8 +61,26 @@ type ingestRequest struct { } type ingestResponse struct { - Accepted int `json:"accepted"` - Rejected int `json:"rejected"` + Accepted int `json:"accepted"` + Rejected int `json:"rejected"` + + // Unpaired counts ACCEPTED events that carried no endpoint_asset_id. + // They are stored and the IOC correlator still matches them, because it + // keys on values inside the event. They are invisible to every + // asset-scoped read: Stage-4 detection correlation's heuristic fallback + // and the per-asset Stage-6 dashboards. + // + // This is permanent, not a pending state. There is no server-side way to + // fill it in later — `agents` has no asset column and `assets` has no + // agent column, and only the producer knows which endpoint an event + // describes anyway (a forwarder reports on many hosts). Migration 000155 + // once promised a nightly reconciler; it was never written and could not + // have been. + // + // Reported so a producer sees the degradation on the response it already + // reads, rather than discovering months later that half the feature never + // applied to its data. + Unpaired int `json:"unpaired"` Errors []string `json:"errors,omitempty"` } @@ -188,6 +206,9 @@ func (h *RuntimeTelemetryHandler) Ingest(w http.ResponseWriter, r *http.Request) continue } resp.Accepted++ + if ev.EndpointAssetID == "" { + resp.Unpaired++ + } if eventID, parseErr := shared.IDFromString(eventIDStr); parseErr == nil { accepted = append(accepted, iocapp.TelemetryEvent{ @@ -198,6 +219,20 @@ func (h *RuntimeTelemetryHandler) Ingest(w http.ResponseWriter, r *http.Request) } } + // Surface the degradation in the logs too. A producer that never sends + // endpoint_asset_id gets a fully successful 200 with a healthy accepted + // count, and would have no reason to suspect that asset-scoped correlation + // silently does not apply to any of its data. + if resp.Unpaired > 0 { + h.logger.Warn("runtime telemetry accepted without an endpoint asset link", + "tenant_id", agt.TenantID.String(), + "agent_id", agt.ID.String(), + "unpaired", resp.Unpaired, + "accepted", resp.Accepted, + "impact", "invisible to asset-scoped detection correlation and per-asset dashboards; "+ + "the producer must supply endpoint_asset_id, the server cannot infer it") + } + // B6 wire: ONE batch correlate call for the whole accepted slice. // Correlator dedups candidates internally and runs a single // FindActiveByValues query; errors here are logged but never block diff --git a/internal/infra/http/handler/runtime_telemetry_unpaired_test.go b/internal/infra/http/handler/runtime_telemetry_unpaired_test.go new file mode 100644 index 00000000..35ab66a3 --- /dev/null +++ b/internal/infra/http/handler/runtime_telemetry_unpaired_test.go @@ -0,0 +1,212 @@ +package handler + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + _ "github.com/lib/pq" + + "github.com/openctemio/api/pkg/domain/agent" + "github.com/openctemio/api/pkg/domain/shared" + "github.com/openctemio/api/pkg/logger" +) + +// endpoint_asset_id is nullable, and nothing will ever fill it in later. +// +// Migration 000155 promised "a nightly reconciler job pairs events with assets +// by agent_id". That job was never written and could not be: `agents` has no +// asset column, `assets` has no agent column, and there is no join table — so +// there is no key to pair BY. It is also the wrong idea, because only the +// producer knows which endpoint an event describes; an EDR/XDR forwarder +// reports on many hosts. +// +// So an event without an asset link is permanently invisible to every +// asset-scoped read — Stage-4 detection correlation's heuristic fallback and +// the per-asset Stage-6 dashboards — while the response says "accepted" and the +// IOC correlator, which keys on values inside the event, still works. That is a +// half-working feature that looks fully working. +// +// These tests pin the counter that makes it visible. + +func openTelemetryDB(t *testing.T) *sql.DB { + t.Helper() + + dbURL := os.Getenv("DATABASE_URL") + if dbURL == "" { + t.Skip("DATABASE_URL not set; skipping telemetry ingest tests") + } + db, err := sql.Open("postgres", dbURL) + if err != nil { + t.Skipf("open db: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := db.PingContext(context.Background()); err != nil { + t.Skipf("cannot reach DATABASE_URL: %v", err) + } + return db +} + +func seedTelemetryTenant(t *testing.T, db *sql.DB) shared.ID { + t.Helper() + + id := shared.NewID() + if _, err := db.ExecContext(context.Background(), + `INSERT INTO tenants (id, name, slug) VALUES ($1, $2, $3)`, + id.String(), "telemetry unpaired test", "tel-"+id.String()); err != nil { + t.Fatalf("seed tenant: %v", err) + } + t.Cleanup(func() { + _, _ = db.ExecContext(context.Background(), `DELETE FROM tenants WHERE id = $1`, id.String()) + }) + return id +} + +func seedTelemetryAsset(t *testing.T, db *sql.DB, tenantID shared.ID) shared.ID { + t.Helper() + + id := shared.NewID() + if _, err := db.ExecContext(context.Background(), + `INSERT INTO assets (id, tenant_id, name, asset_type) VALUES ($1, $2, $3, 'host')`, + id.String(), tenantID.String(), "host-"+id.String()); err != nil { + t.Fatalf("seed asset: %v", err) + } + return id +} + +// ingestEvents posts a batch as an authenticated tenant agent and returns the +// decoded response. +func ingestEvents(t *testing.T, db *sql.DB, tenantID shared.ID, events []map[string]any) ingestResponse { + t.Helper() + + body, err := json.Marshal(map[string]any{"events": events}) + if err != nil { + t.Fatalf("marshal body: %v", err) + } + + h := NewRuntimeTelemetryHandler(db, logger.NewNop()) + + r := httptest.NewRequest(http.MethodPost, "/api/v1/telemetry-events", bytes.NewReader(body)) + tid := tenantID + agt := &agent.Agent{ID: shared.NewID(), TenantID: &tid, Status: agent.AgentStatusActive} + r = r.WithContext(context.WithValue(r.Context(), agentContextKey, agt)) + + w := httptest.NewRecorder() + h.Ingest(w, r) + + // 202 on a fully-accepted batch, 207 when some events were rejected. + if w.Code != http.StatusAccepted && w.Code != http.StatusMultiStatus { + t.Fatalf("ingest returned %d: %s", w.Code, w.Body.String()) + } + + var resp ingestResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + t.Cleanup(func() { + _, _ = db.ExecContext(context.Background(), + `DELETE FROM runtime_telemetry_events WHERE tenant_id = $1`, tenantID.String()) + }) + return resp +} + +func event(assetID string) map[string]any { + e := map[string]any{ + "event_type": "network_connect", + "observed_at": time.Now().UTC().Format(time.RFC3339), + } + if assetID != "" { + e["endpoint_asset_id"] = assetID + } + return e +} + +// The core case: a producer that never sends endpoint_asset_id gets a fully +// successful response. Without the counter there is nothing in that response to +// tell it half the feature does not apply. +func TestIngest_ReportsUnpairedEvents(t *testing.T) { + db := openTelemetryDB(t) + tenantID := seedTelemetryTenant(t, db) + + resp := ingestEvents(t, db, tenantID, []map[string]any{ + event(""), event(""), event(""), + }) + + if resp.Accepted != 3 { + t.Fatalf("accepted = %d, want 3 (errors: %v)", resp.Accepted, resp.Errors) + } + if resp.Rejected != 0 { + t.Fatalf("rejected = %d, want 0: unpaired events are stored, not refused", resp.Rejected) + } + if resp.Unpaired != 3 { + t.Fatalf("unpaired = %d, want 3: the response claims full success while every "+ + "event is invisible to asset-scoped correlation", resp.Unpaired) + } +} + +// A producer doing it right must not be told it has a problem. +func TestIngest_PairedEventsAreNotCountedUnpaired(t *testing.T) { + db := openTelemetryDB(t) + tenantID := seedTelemetryTenant(t, db) + assetID := seedTelemetryAsset(t, db, tenantID) + + resp := ingestEvents(t, db, tenantID, []map[string]any{ + event(assetID.String()), event(assetID.String()), + }) + + if resp.Accepted != 2 { + t.Fatalf("accepted = %d, want 2 (errors: %v)", resp.Accepted, resp.Errors) + } + if resp.Unpaired != 0 { + t.Fatalf("unpaired = %d, want 0: these events carry a valid asset link", resp.Unpaired) + } +} + +// A mixed batch is the realistic case — the count must be per-event, not a +// boolean about the batch. +func TestIngest_CountsUnpairedPerEvent(t *testing.T) { + db := openTelemetryDB(t) + tenantID := seedTelemetryTenant(t, db) + assetID := seedTelemetryAsset(t, db, tenantID) + + resp := ingestEvents(t, db, tenantID, []map[string]any{ + event(assetID.String()), event(""), event(assetID.String()), event(""), + }) + + if resp.Accepted != 4 { + t.Fatalf("accepted = %d, want 4 (errors: %v)", resp.Accepted, resp.Errors) + } + if resp.Unpaired != 2 { + t.Fatalf("unpaired = %d, want 2", resp.Unpaired) + } +} + +// Unpaired counts ACCEPTED events only. A rejected event was never stored, so +// counting it here would overstate the gap and send a producer looking for a +// configuration problem that is really a validation error. +func TestIngest_RejectedEventsAreNotCountedUnpaired(t *testing.T) { + db := openTelemetryDB(t) + tenantID := seedTelemetryTenant(t, db) + + resp := ingestEvents(t, db, tenantID, []map[string]any{ + {"observed_at": time.Now().UTC().Format(time.RFC3339)}, // no event_type -> rejected + event(""), // accepted, unpaired + }) + + if resp.Rejected != 1 { + t.Fatalf("rejected = %d, want 1", resp.Rejected) + } + if resp.Accepted != 1 { + t.Fatalf("accepted = %d, want 1", resp.Accepted) + } + if resp.Unpaired != 1 { + t.Fatalf("unpaired = %d, want 1: a rejected event was never stored and must "+ + "not be reported as an unpaired one", resp.Unpaired) + } +} diff --git a/migrations/000155_runtime_telemetry.up.sql b/migrations/000155_runtime_telemetry.up.sql index a3b26bc0..73a92b2f 100644 --- a/migrations/000155_runtime_telemetry.up.sql +++ b/migrations/000155_runtime_telemetry.up.sql @@ -14,9 +14,27 @@ -- - properties JSONB holds event-specific fields. Kept intentionally -- schemaless so agents on different OSes (Windows EDR, Linux -- osquery, …) can emit without a wire-format migration every time. --- - endpoint_asset_id is nullable — during onboarding the agent may --- not yet know its asset UUID. A nightly reconciler job pairs --- events with assets by agent_id. +-- - endpoint_asset_id is nullable — during onboarding the producer may +-- not yet know its asset UUID. +-- +-- CORRECTION (2026-08-04): an earlier version of this comment promised +-- "a nightly reconciler job pairs events with assets by agent_id". +-- No such job was ever written, and it cannot be: there is no join +-- key. `agents` has no asset column, `assets` has no agent column, +-- and there is no join table — so there is nothing to pair BY. +-- +-- It is also the wrong idea. Only the producer knows which endpoint +-- an event describes. An EDR/XDR forwarder reports on MANY hosts, so +-- even the emitting agent's own hostname is not the answer. The +-- server cannot infer this after the fact. +-- +-- So a NULL here is permanent. Such an event is still stored and +-- still matched by the IOC correlator (which keys on values inside +-- the event, not on the asset), but it is invisible to every +-- asset-scoped read: Stage-4 detection correlation's heuristic +-- fallback and the per-asset Stage-6 dashboards. The ingest response +-- reports these as `unpaired` so a producer sees the degradation +-- instead of silently losing half the feature. -- - (tenant_id, observed_at) compound index — all downstream reads -- are per-tenant, time-ordered.