diff --git a/internal/cli/telemetry_installer_spool.go b/internal/cli/telemetry_installer_spool.go new file mode 100644 index 0000000..3d6f7e0 --- /dev/null +++ b/internal/cli/telemetry_installer_spool.go @@ -0,0 +1,163 @@ +package cli + +// Draining the INSTALLER's spool — backend#2217, option (b). +// +// THE PROBLEM THIS SOLVES. `scripts/lib/telemetry.sh` produces one contract event +// per install run and spools it, but it cannot deliver: the ingest endpoint needs +// a bearer credential and the installer holds only `TRACEBLOC_CLIENT_ID` / +// `TRACEBLOC_CLIENT_PASSWORD`, a provisioning pair with no exchange for a token. +// It also never reads the CLI's config. So its records had no route at all, and +// #1906 cannot help — that is a pod Collector reading container stdout, and these +// are files on the operator's own machine, often written when no cluster exists. +// +// WHY THE CLI AND NOT THE INSTALLER. The CLI already owns the token (device login +// writes it to `~/.tracebloc/config.json`), and since #542 it already owns a +// spool, a drain loop and the OTLP mapping. Teaching it one more file to read +// makes the CLI the single credential holder and leaves the installer with no +// credential handling at all. The alternative — the installer reading the CLI's +// config — delivers nothing for the failures that matter most: `validate_config` +// and `early_data_dir_guard` run BEFORE provisioning, so there is no token on +// disk yet at the moment those events are written. Those are precisely the +// failures the installer's `$TMPDIR` fallback exists to preserve. +// +// THE FALLBACK PATH IS A GLOB, NOT AN INDEX. `_telemetry_fallback_spool` uses +// `mktemp .../tracebloc-telemetry-XXXXXX`, so the exact name is unpredictable but +// the PATTERN is fixed. Globbing it needs no change to the installer and no new +// shared state — an index file would be a second thing to keep in step, and the +// thing it indexed would still have to exist. +// +// EVERY RECORD IS FILTERED BY ITS OWN ENVIRONMENT, and this is not optional. The +// CLI's own spool is partitioned by env in the FILENAME (see telemetrySpoolPath); +// the installer's is not, and its records carry whatever `CLIENT_ENV` that run +// used. Forwarding them blind would post a `prod`-labelled install failure to +// whichever backend this CLI invocation happens to point at — the exact +// label-versus-destination leak #542's second review finding was about. So a +// record is forwarded only when its `deployment.environment` matches this run's, +// and the rest are left where they are for a future invocation against that env. + +import ( + "os" + "path/filepath" + "strings" +) + +const ( + // installerSpoolGlob matches `_telemetry_fallback_spool`'s mktemp template. + installerSpoolGlob = "tracebloc-telemetry-*" + + // installerDrainMax bounds how many installer records one invocation carries, + // on top of the CLI's own. Deliberately small: the installer's spool caps at + // TB_TELEMETRY_SPOOL_MAX=50, and a `tracebloc login` should not turn into the + // largest request this host has ever sent. + installerDrainMax = 10 +) + +// resourceEnvironment is the attribute the filter reads. Resource scope, so it is +// on the resource map rather than the record's own attributes. +const resourceEnvironment = "deployment.environment" + +// installerSpoolFiles returns every file that may hold installer records. +// +// Ordered predictable-first so a run with both delivers the data-dir spool before +// the scratch files, which is the order they were written in the common case. +// Missing files and unreadable directories are simply absent from the result: +// this runs on the exit path of every command and may never report a problem. +func installerSpoolFiles(getenv func(string) string) []string { + var out []string + + // 1. The data-dir spool, which the installer writes once HOST_DATA_DIR exists. + // Same default the installer uses: $HOST_DATA_DIR, else ~/.tracebloc. + base := strings.TrimSpace(getenv("HOST_DATA_DIR")) + if base == "" { + if home, err := os.UserHomeDir(); err == nil { + base = filepath.Join(home, ".tracebloc") + } + } + if base != "" { + out = append(out, filepath.Join(base, "telemetry", "pending.jsonl")) + } + + // 2. The pre-log fallback files. `_telemetry_fallback_dir` picks $TMPDIR, else + // $HOME, else /tmp — and disqualifies $TMPDIR when the installer is running + // from inside it. From here we cannot tell which it chose, so all three are + // candidates; a glob that matches nothing costs one syscall. + seen := map[string]bool{} + for _, dir := range []string{ + strings.TrimSpace(getenv("TMPDIR")), + strings.TrimSpace(getenv("HOME")), + "/tmp", + } { + dir = strings.TrimRight(dir, "/") + if dir == "" || seen[dir] { + continue + } + seen[dir] = true + matches, err := filepath.Glob(filepath.Join(dir, installerSpoolGlob)) + if err != nil { + continue + } + out = append(out, matches...) + } + return out +} + +// installerRecords reads events for `env` out of the installer's spools. +// +// Returns the matching events and, per file, the events that did NOT match so the +// caller can write them back. A file whose records are all foreign is left +// untouched rather than rewritten — rewriting another component's file to change +// nothing is a needless risk on a path that must never disturb an install. +type installerBatch struct { + // Events for this environment, ready to forward. + events []spooledEvent + // Per source file, the events that stay behind. Only files that actually + // contributed a forwarded event appear here. + remainder map[string][]spooledEvent +} + +func installerRecords(files []string, env string, max int) installerBatch { + batch := installerBatch{remainder: map[string][]spooledEvent{}} + for _, path := range files { + if len(batch.events) >= max { + return batch + } + records := readSpool(path) + if len(records) == 0 { + continue + } + var mine, theirs []spooledEvent + for _, rec := range records { + // A record with no environment is NOT forwarded. The contract omits an + // attribute rather than sending it empty, so an absent environment + // means the emitter could not resolve one — and a record no query can + // filter on is the defect the contract exists to remove. Left in place + // rather than dropped: it is still evidence, just not deliverable. + if rec.Resource[resourceEnvironment] == env && len(batch.events)+len(mine) < max { + mine = append(mine, rec) + continue + } + theirs = append(theirs, rec) + } + if len(mine) == 0 { + continue + } + batch.events = append(batch.events, mine...) + batch.remainder[path] = theirs + } + return batch +} + +// clearInstallerRecords rewrites each source file with only the records that were +// left behind, deleting it when nothing remains. +// +// CALLED ONLY AFTER A SUCCESSFUL DELIVERY. On any failure the files are untouched, +// so the next invocation retries them — the same posture as the CLI's own spool. +// Reuses `writeSpool`, so these files inherit 0600 and the atomic temp+rename. +func clearInstallerRecords(remainder map[string][]spooledEvent) { + for path, keep := range remainder { + // writeSpool removes the file when `keep` is empty, which is right for the + // mktemp fallbacks: the installer writes exactly one record per file and + // never returns to them, so an emptied one is litter. + _ = writeSpool(path, keep) + } +} diff --git a/internal/cli/telemetry_installer_spool_test.go b/internal/cli/telemetry_installer_spool_test.go new file mode 100644 index 0000000..9dedcac --- /dev/null +++ b/internal/cli/telemetry_installer_spool_test.go @@ -0,0 +1,319 @@ +package cli + +// Tests for draining the installer's spool — backend#2217 option (b). +// +// The risky property here is not "does it deliver" but "does it deliver the RIGHT +// records to the RIGHT backend, and leave everything else alone". These files +// belong to another component, so every test that asserts something was sent also +// asserts what was NOT sent and what survived on disk. + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// installerEvent mimics what scripts/lib/telemetry.sh writes: the compact +// {resource, attributes} shape, with the environment on the RESOURCE layer. +func installerEvent(env, instance string, exit int) spooledEvent { + return spooledEvent{ + Resource: map[string]string{ + "service.name": "installer", + "tracebloc.component": "install", + "deployment.environment": env, + "service.instance.id": instance, + }, + Attributes: map[string]any{ + "event.name": "install.run.failed", + "error.type": "preflight", + "tracebloc.install.exit_code": exit, + }, + } +} + +func writeInstallerSpool(t *testing.T, path string, events ...spooledEvent) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatalf("mkdir: %v", err) + } + var b strings.Builder + for _, ev := range events { + line, err := json.Marshal(ev) + if err != nil { + t.Fatalf("marshal: %v", err) + } + b.Write(line) + b.WriteString("\n") + } + if err := os.WriteFile(path, []byte(b.String()), 0o600); err != nil { + t.Fatalf("write: %v", err) + } +} + +// ─────────────────────────────────────────────────── locating the spool files + +func TestInstallerSpoolFilesFindsBothShapes(t *testing.T) { + dataDir := t.TempDir() + tmpDir := t.TempDir() + // The predictable data-dir spool. + dataSpool := filepath.Join(dataDir, "telemetry", "pending.jsonl") + writeInstallerSpool(t, dataSpool, installerEvent("prod", "a", 1)) + // A pre-log fallback, whose exact name mktemp chose and we cannot predict. + fallback := filepath.Join(tmpDir, "tracebloc-telemetry-Ab3xY9") + writeInstallerSpool(t, fallback, installerEvent("prod", "b", 2)) + // A file that is NOT ours, in the same directory. + if err := os.WriteFile(filepath.Join(tmpDir, "unrelated.jsonl"), []byte("{}\n"), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + + getenv := func(k string) string { + switch k { + case "HOST_DATA_DIR": + return dataDir + case "TMPDIR": + return tmpDir + } + return "" + } + files := installerSpoolFiles(getenv) + + var sawData, sawFallback, sawUnrelated bool + for _, f := range files { + switch { + case f == dataSpool: + sawData = true + case f == fallback: + sawFallback = true + case strings.HasSuffix(f, "unrelated.jsonl"): + sawUnrelated = true + } + } + if !sawData { + t.Errorf("the data-dir spool was not found; got %v", files) + } + if !sawFallback { + t.Errorf("the mktemp fallback was not found by glob; got %v", files) + } + if sawUnrelated { + t.Errorf("an unrelated file matched the glob; got %v", files) + } +} + +func TestInstallerSpoolFilesDoesNotDuplicateOneDirectory(t *testing.T) { + dir := t.TempDir() + writeInstallerSpool(t, filepath.Join(dir, "tracebloc-telemetry-Zz1"), installerEvent("prod", "a", 1)) + // TMPDIR and HOME pointing at the same place must not yield the file twice — + // a duplicate would send the same install outcome twice in one batch. + getenv := func(k string) string { + if k == "TMPDIR" || k == "HOME" { + return dir + } + return "" + } + files := installerSpoolFiles(getenv) + count := 0 + for _, f := range files { + if strings.Contains(f, "tracebloc-telemetry-Zz1") { + count++ + } + } + if count != 1 { + t.Errorf("the same fallback file appears %d times in %v", count, files) + } +} + +// ─────────────────────────────────────────────── filtering by environment + +func TestInstallerRecordsOnlyTakesThisEnvironment(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "pending.jsonl") + writeInstallerSpool(t, path, + installerEvent("prod", "prod-run", 1), + installerEvent("stg", "stg-run", 2), + installerEvent("prod", "prod-run-2", 3), + ) + + batch := installerRecords([]string{path}, "prod", installerDrainMax) + + if len(batch.events) != 2 { + t.Fatalf("want the 2 prod records, got %d", len(batch.events)) + } + for _, ev := range batch.events { + if got := ev.Resource["deployment.environment"]; got != "prod" { + t.Errorf("forwarded a %q record while draining prod", got) + } + } + // The stg record must be RETAINED, not dropped: it is deliverable by a later + // invocation against stg, and discarding another environment's evidence is + // not this function's call to make. + keep := batch.remainder[path] + if len(keep) != 1 || keep[0].Resource["deployment.environment"] != "stg" { + t.Errorf("the stg record should be left behind for a later stg run; remainder=%v", keep) + } +} + +func TestInstallerRecordsLeavesRecordsWithNoEnvironment(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "pending.jsonl") + noEnv := installerEvent("prod", "x", 1) + delete(noEnv.Resource, "deployment.environment") + writeInstallerSpool(t, path, noEnv) + + batch := installerRecords([]string{path}, "prod", installerDrainMax) + if len(batch.events) != 0 { + t.Errorf("a record with no environment must not be forwarded; got %d", len(batch.events)) + } + if len(batch.remainder) != 0 { + t.Errorf("a file that contributed nothing must not be scheduled for rewrite; got %v", batch.remainder) + } +} + +func TestInstallerRecordsRespectsTheDrainCap(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "pending.jsonl") + var events []spooledEvent + for i := 0; i < installerDrainMax+7; i++ { + events = append(events, installerEvent("prod", "run", i)) + } + writeInstallerSpool(t, path, events...) + + batch := installerRecords([]string{path}, "prod", installerDrainMax) + if len(batch.events) != installerDrainMax { + t.Fatalf("want exactly %d records, got %d", installerDrainMax, len(batch.events)) + } + // The overflow must survive, or a big installer spool loses records silently. + if got := len(batch.remainder[path]); got != 7 { + t.Errorf("want the 7 uncarried records retained, got %d", got) + } +} + +func TestInstallerRecordsSkipsAFileWithNothingForUs(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "pending.jsonl") + writeInstallerSpool(t, path, installerEvent("stg", "s", 1)) + before, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + + batch := installerRecords([]string{path}, "prod", installerDrainMax) + + // THE PRIMARY ASSERTION IS ON `remainder`, not on the file's bytes. A byte + // comparison is satisfied by a rewrite that happens to produce identical + // content — which is exactly what a rewrite of unchanged records does, so the + // first version of this test passed under the mutation "rewrite files we took + // nothing from". `remainder` is the actual contract: only files that + // contributed a forwarded record may appear in it. + if _, scheduled := batch.remainder[path]; scheduled { + t.Errorf("a file we took nothing from was scheduled for rewrite; remainder=%v", batch.remainder) + } + + clearInstallerRecords(batch.remainder) + + after, err := os.ReadFile(path) + if err != nil { + t.Fatalf("the file must not be removed when nothing was taken: %v", err) + } + if string(before) != string(after) { + t.Errorf("a file we took nothing from lost content:\n before %q\n after %q", before, after) + } +} + +// ─────────────────────────────────────────────── end to end through deliver + +func TestDeliverCarriesTheInstallersRecords(t *testing.T) { + ownSpool := withTempConfigDir(t, "prod") + dataDir := t.TempDir() + t.Setenv("HOST_DATA_DIR", dataDir) + t.Setenv("TMPDIR", t.TempDir()) + + installerSpool := filepath.Join(dataDir, "telemetry", "pending.jsonl") + writeInstallerSpool(t, installerSpool, + installerEvent("prod", "installer-prod", 2), + installerEvent("dev", "installer-dev", 3), + ) + + var body string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + body = string(b) + w.WriteHeader(http.StatusAccepted) + })) + defer srv.Close() + + deliver(ownSpool, srv.URL+telemetryIngestPath, "tok", "prod", + event("cli-run", 0), time.Now()) + + if body == "" { + t.Fatal("nothing was POSTed") + } + if !strings.Contains(body, "installer-prod") { + t.Errorf("the installer's prod record was not carried: %s", body) + } + if strings.Contains(body, "installer-dev") { + t.Errorf("a dev-labelled installer record was sent to the prod endpoint: %s", body) + } + if !strings.Contains(body, "cli-run") { + t.Errorf("the CLI's own event was lost: %s", body) + } + // Delivered records are gone; the foreign-env one survives. + left := readSpool(installerSpool) + if len(left) != 1 || left[0].Resource["service.instance.id"] != "installer-dev" { + t.Errorf("after delivery the installer spool should hold only the dev record; got %v", left) + } +} + +func TestDeliverLeavesTheInstallerSpoolAloneOnFailure(t *testing.T) { + ownSpool := withTempConfigDir(t, "prod") + dataDir := t.TempDir() + t.Setenv("HOST_DATA_DIR", dataDir) + t.Setenv("TMPDIR", t.TempDir()) + + installerSpool := filepath.Join(dataDir, "telemetry", "pending.jsonl") + writeInstallerSpool(t, installerSpool, installerEvent("prod", "installer-prod", 2)) + before, err := os.ReadFile(installerSpool) + if err != nil { + t.Fatalf("read: %v", err) + } + + // Unreachable endpoint. + deliver(ownSpool, "http://127.0.0.1:1"+telemetryIngestPath, "tok", "prod", + event("cli-run", 0), time.Now()) + + after, err := os.ReadFile(installerSpool) + if err != nil { + t.Fatalf("the installer spool must survive a failed delivery: %v", err) + } + if string(before) != string(after) { + t.Errorf("a failed delivery modified the installer's spool:\n before %q\n after %q", before, after) + } +} + +func TestDeliverLeavesTheInstallerSpoolAloneWhenNotSignedIn(t *testing.T) { + ownSpool := withTempConfigDir(t, "prod") + dataDir := t.TempDir() + t.Setenv("HOST_DATA_DIR", dataDir) + t.Setenv("TMPDIR", t.TempDir()) + + installerSpool := filepath.Join(dataDir, "telemetry", "pending.jsonl") + writeInstallerSpool(t, installerSpool, installerEvent("prod", "installer-prod", 2)) + before, _ := os.ReadFile(installerSpool) + + // No token: nothing is attempted, so nothing of the installer's is consumed. + deliver(ownSpool, "http://127.0.0.1:1"+telemetryIngestPath, "", "prod", + event("cli-run", 0), time.Now()) + + after, err := os.ReadFile(installerSpool) + if err != nil { + t.Fatalf("the installer spool must survive an unauthenticated run: %v", err) + } + if string(before) != string(after) { + t.Errorf("an unauthenticated run modified the installer's spool") + } +} diff --git a/internal/cli/telemetry_transport.go b/internal/cli/telemetry_transport.go index 6e8ea62..f3041fe 100644 --- a/internal/cli/telemetry_transport.go +++ b/internal/cli/telemetry_transport.go @@ -286,7 +286,7 @@ func postBatch(ctx context.Context, url, token string, batch []spooledEvent) pos // computed api.BaseURL() internally had no seam — its own test posted to // PRODUCTION; and one that computed the spool path internally drained another // environment's queue into this one's endpoint (Bugbot on #542). -func deliver(spool, url, token string, ev spooledEvent, now time.Time) { +func deliver(spool, url, token, env string, ev spooledEvent, now time.Time) { path := spool pending := readSpool(path) @@ -301,11 +301,24 @@ func deliver(spool, url, token string, ev spooledEvent, now time.Time) { batch = append(batch, drained...) batch = append(batch, ev) + // THE INSTALLER'S RECORDS RIDE ALONG (backend#2217 option b). It produces + // contract events and cannot deliver them — it holds a provisioning pair, not + // a bearer token — so the CLI, which does hold one, carries them. Filtered to + // THIS environment by each record's own `deployment.environment`, because the + // installer's spool is not partitioned by env the way ours is. + // + // Appended AFTER our own, so a full installer spool can never crowd out the + // event this invocation just produced. + installer := installerRecords(installerSpoolFiles(os.Getenv), env, installerDrainMax) + batch = append(batch, installer.events...) + // No token means not signed in. Spool rather than attempt: the events are // still worth sending after the next login, and a POST with no credential // would spend the budget earning a 401. if token == "" { _ = writeSpool(path, append(pending, ev)) + // The installer's files are deliberately NOT touched here. Not signed in + // means they were never sent, and they are not ours to discard. return } @@ -317,6 +330,11 @@ func deliver(spool, url, token string, ev spooledEvent, now time.Time) { // Both consume the batch. The difference is only whether the server // stored it, and neither is a reason to carry it again. _ = writeSpool(path, pending[len(drained):]) + // The installer's files are only ever touched on a path that CONSUMED + // them. A discard clears them too, for the same reason it clears ours: a + // permanently unparseable batch carried forever wedges every later send, + // and these records are as unparseable as the rest of it. + clearInstallerRecords(installer.remainder) case postRetry: _ = writeSpool(path, append(pending, ev)) } @@ -354,7 +372,7 @@ func pendingSink(env string) telemetry.Sink { return nil } return func(resource map[string]string, record map[string]any) { - deliver(spool, url, telemetryToken(env), spooledEvent{ + deliver(spool, url, telemetryToken(env), env, spooledEvent{ Resource: resource, Attributes: record, }, time.Now()) diff --git a/internal/cli/telemetry_transport_test.go b/internal/cli/telemetry_transport_test.go index 5858d20..34b9dbe 100644 --- a/internal/cli/telemetry_transport_test.go +++ b/internal/cli/telemetry_transport_test.go @@ -338,7 +338,7 @@ func TestDeliverSpoolsWhenTheServerIsUnreachable(t *testing.T) { url := srv.URL + telemetryIngestPath srv.Close() - deliver(path, url, "tok", event("run-partition", 3), time.Now()) + deliver(path, url, "tok", "prod", event("run-partition", 3), time.Now()) got := readSpool(path) if len(got) != 1 { @@ -351,7 +351,7 @@ func TestDeliverSpoolsWhenTheServerIsUnreachable(t *testing.T) { func TestDeliverSpoolsWhenNotSignedIn(t *testing.T) { path := withTempConfigDir(t, "prod") - deliver(path, "http://127.0.0.1:1"+telemetryIngestPath, "", event("run-anon", 0), time.Now()) + deliver(path, "http://127.0.0.1:1"+telemetryIngestPath, "", "prod", event("run-anon", 0), time.Now()) got := readSpool(path) if len(got) != 1 { t.Fatalf("no token must spool rather than post; got %d records", len(got)) @@ -364,7 +364,7 @@ func TestDeliverSpoolsWhenNotSignedIn(t *testing.T) { func TestDeliverKeepsTheSpoolBoundedAcrossManyFailures(t *testing.T) { path := withTempConfigDir(t, "prod") for i := 0; i < telemetrySpoolMax+10; i++ { - deliver(path, "http://127.0.0.1:1"+telemetryIngestPath, "", event(fmt.Sprintf("run-%02d", i), i), time.Now()) + deliver(path, "http://127.0.0.1:1"+telemetryIngestPath, "", "prod", event(fmt.Sprintf("run-%02d", i), i), time.Now()) } got := readSpool(path) if len(got) != telemetrySpoolMax { @@ -523,7 +523,7 @@ func TestTheSpoolDoesNotLeakAcrossEnvironments(t *testing.T) { Attributes: map[string]any{"event.name": "cli.command.failed", "error.type": "usage"}, } // Unreachable endpoint, so it spools against prod. - deliver(prodSpool, "http://127.0.0.1:1"+telemetryIngestPath, "prod-token", prodEvent, time.Now()) + deliver(prodSpool, "http://127.0.0.1:1"+telemetryIngestPath, "prod-token", "prod", prodEvent, time.Now()) if got := readSpool(prodSpool); len(got) != 1 { t.Fatalf("setup: the prod event should be spooled; got %d records", len(got)) } @@ -540,7 +540,7 @@ func TestTheSpoolDoesNotLeakAcrossEnvironments(t *testing.T) { Resource: map[string]string{"service.name": "cli", "deployment.environment": "dev"}, Attributes: map[string]any{"event.name": "cli.command.succeeded"}, } - deliver(devSpool, srv.URL+telemetryIngestPath, "dev-token", devEvent, time.Now()) + deliver(devSpool, srv.URL+telemetryIngestPath, "dev-token", "dev", devEvent, time.Now()) if received == "" { t.Fatal("the dev endpoint received nothing; the dev delivery did not happen")