From 2c5020d2bdd0c7576063a4fb078509a40582ea33 Mon Sep 17 00:00:00 2001 From: Nguyen Manh <0xmanhnv@gmail.com> Date: Tue, 4 Aug 2026 02:39:08 +0000 Subject: [PATCH] fix(telemetry): report unpaired events, and retract migration 000155's reconciler promise MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Migration 000155 says endpoint_asset_id is nullable because "during onboarding the agent may not yet know its asset UUID. 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. Checked every promised background job in every migration comment (15 of them); this is the only one with no implementation. The other 14 all exist and are started. 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, at any point, by any means. So a NULL endpoint_asset_id is permanent, not a pending state — and the consequence is invisible. The event is stored, the response says accepted, and the IOC correlator still matches it because it keys on values inside the event. What silently does not happen is every asset-scoped read: Stage-4 detection correlation's heuristic fallback (CountNearTarget, by asset + window) and the per-asset Stage-6 dashboards. Half the feature does not apply, and nothing says so. Two changes, no new background job — building a reaper with no join key, for a stream that currently has no producer at all, would be adding to the silently-inert class rather than fixing it: 1. The migration comment retracts the promise and records why it cannot be kept, so the next person does not go looking for a job that was never possible. 2. The ingest response gains `unpaired`, counting ACCEPTED events that arrived without an asset link, plus a WARN naming the impact. A producer integrating today sees the degradation on the response it already reads instead of discovering months later that asset-scoped correlation never applied to its data. Rejected counts are excluded: those events were never stored, and counting them would send a producer looking for a configuration problem that is really a validation error. --- .../http/handler/runtime_telemetry_handler.go | 39 +++- .../runtime_telemetry_unpaired_test.go | 212 ++++++++++++++++++ migrations/000155_runtime_telemetry.up.sql | 24 +- 3 files changed, 270 insertions(+), 5 deletions(-) create mode 100644 internal/infra/http/handler/runtime_telemetry_unpaired_test.go 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.