diff --git a/internal/cli/data.go b/internal/cli/data.go index 98421b8..cc7f8b8 100644 --- a/internal/cli/data.go +++ b/internal/cli/data.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "strings" "github.com/spf13/cobra" @@ -165,7 +166,7 @@ func runDataIngest(ctx context.Context, out, errOut io.Writer, a runDataIngestAr // DROP/rm and then "succeed". plan := push.PlanTeardown(existingTable) rmSpin := a.Printer.Spinner(fmt.Sprintf("Removing the existing %q first", existingTable), "") - _, terr := push.Teardown(ctx, cs, &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, resolved.Namespace, plan, push.PodSpecOptions{ + tres, terr := push.Teardown(ctx, cs, &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, resolved.Namespace, plan, push.PodSpecOptions{ Namespace: resolved.Namespace, PVCClaimName: pvc.ClaimName, PVCMountPath: pvc.MountPath, @@ -185,6 +186,13 @@ func runDataIngest(ctx context.Context, out, errOut io.Writer, a runDataIngestAr "first, then re-run this ingest. Nothing new was staged. (%w)", existingTable, existingTable, terr)} } + if !tres.BookkeepingCleaned { + // Same surfacing `data delete` does (Bugbot on the PR): the + // overwrite pre-clean runs the identical teardown, and a silent + // bookkeeping failure here would hide the same schema-drift + // regression on this path. + a.Printer.Warnf("Bookkeeping cleanup incomplete — the old table is gone, but its run-journal/salt rows may remain: %s", strings.Join(tres.BookkeepingErrs, "; ")) + } a.Printer.Successf("Removed the old %q — ingesting the new data.", existingTable) } diff --git a/internal/cli/data_delete.go b/internal/cli/data_delete.go index 7b1e6d3..9961491 100644 --- a/internal/cli/data_delete.go +++ b/internal/cli/data_delete.go @@ -212,7 +212,7 @@ undone — re-ingesting the data is the only way back.`) p.Newline() p.Successf("Dry-run — nothing was deleted.") if a.OutputJSON { - writeDataDeleteJSON(a.JSONOut, "dry-run", resolved.Namespace, release.ReleaseName, plan, nil) + writeDataDeleteJSON(a.JSONOut, "dry-run", resolved.Namespace, release.ReleaseName, plan, nil, false) jsonEmitted = true } return nil @@ -234,7 +234,7 @@ undone — re-ingesting the data is the only way back.`) // exit 0. One closure so the pair can't drift apart. declined := func() error { if a.OutputJSON { - writeDataDeleteJSON(a.JSONOut, "declined", resolved.Namespace, release.ReleaseName, plan, nil) + writeDataDeleteJSON(a.JSONOut, "declined", resolved.Namespace, release.ReleaseName, plan, nil, false) jsonEmitted = true } return cleanCancel(p, "nothing was deleted.") @@ -283,9 +283,15 @@ undone — re-ingesting the data is the only way back.`) p.Newline() p.Successf("Deleted %s.%s and %d PVC path(s).", plan.Database, plan.Table, len(res.RemovedPaths)) + if !res.BookkeepingCleaned { + // Best-effort cleanup failed — say so, or a schema-drift regression + // (a renamed keying column) is indistinguishable from a legacy + // cluster without the bookkeeping tables (review, Saqlain). + p.Warnf("Bookkeeping cleanup incomplete — the table is gone, but its run-journal/salt rows may remain: %s", strings.Join(res.BookkeepingErrs, "; ")) + } p.Infof("The dataset's catalog metadata is kept as a record on tracebloc, marked unavailable — never removed.") if a.OutputJSON { - writeDataDeleteJSON(a.JSONOut, "deleted", resolved.Namespace, release.ReleaseName, plan, res.RemovedPaths) + writeDataDeleteJSON(a.JSONOut, "deleted", resolved.Namespace, release.ReleaseName, plan, res.RemovedPaths, res.BookkeepingCleaned) jsonEmitted = true } return nil @@ -302,12 +308,17 @@ type dataDeleteJSON struct { Table string `json:"table"` // the REAL (case-resolved) spelling, not the raw argument PVCPaths []string `json:"pvc_paths"` RemovedPaths []string `json:"removed_paths"` + // BookkeepingCleaned mirrors push.TeardownResult: whether the + // run-journal/salt rows were removed with the table. Always false for + // dry-run/declined — nothing was attempted, and a strict consumer must + // never read "cleanup happened" out of a run that deleted nothing. + BookkeepingCleaned bool `json:"bookkeeping_cleaned"` } // writeDataDeleteJSON serializes the delete result to w (stdout in // --output-json mode). Marshal errors are dropped: marshaling our own // struct can't fail in practice, and the exit code remains the contract. -func writeDataDeleteJSON(w io.Writer, status, namespace, release string, plan push.TeardownPlan, removed []string) { +func writeDataDeleteJSON(w io.Writer, status, namespace, release string, plan push.TeardownPlan, removed []string, bookkeepingCleaned bool) { pvcPaths := plan.PVCPaths if pvcPaths == nil { pvcPaths = []string{} // emit [] not null @@ -316,13 +327,14 @@ func writeDataDeleteJSON(w io.Writer, status, namespace, release string, plan pu removed = []string{} // emit [] not null } res := dataDeleteJSON{ - Status: status, - Namespace: namespace, - Release: release, - Database: plan.Database, - Table: plan.Table, - PVCPaths: pvcPaths, - RemovedPaths: removed, + Status: status, + Namespace: namespace, + Release: release, + Database: plan.Database, + Table: plan.Table, + PVCPaths: pvcPaths, + RemovedPaths: removed, + BookkeepingCleaned: bookkeepingCleaned, } b, err := json.MarshalIndent(res, "", " ") if err != nil { diff --git a/internal/cli/testdata/golden/zz-all-strings.golden b/internal/cli/testdata/golden/zz-all-strings.golden index f2341e5..46dd8d4 100644 --- a/internal/cli/testdata/golden/zz-all-strings.golden +++ b/internal/cli/testdata/golden/zz-all-strings.golden @@ -60,6 +60,7 @@ screen. %s/%d are runtime placeholders. "%s — %s" "%s, … and %d more" "%s/%s" +"%s: %v" "%s: %w" "%s=%s,%s=%s" "%v (policy: %v)" @@ -96,6 +97,8 @@ screen. %s/%d are runtime placeholders. "Already signed out." "Applies to your next training run; a run already going keeps its size." "Ask one of these admins (or ask them to grant you access)" +"Bookkeeping cleanup incomplete — the old table is gone, but its run-journal/salt rows may remain: %s" +"Bookkeeping cleanup incomplete — the table is gone, but its run-journal/salt rows may remain: %s" "CPU cores for one run (1–%d)" "CSV %s has no columns" "Can't reach tracebloc from here." @@ -140,6 +143,7 @@ screen. %s/%d are runtime placeholders. "Ctrl-C to cancel" "Ctrl-C to stop watching — the run keeps going on the cluster" "DB failures" +"DELETE FROM `%s`.`%s` WHERE table_name='%s'" "DROP TABLE IF EXISTS `%s`.`%s`" "Datasets in %s (0)" "Datasets in %s — %d" @@ -553,6 +557,7 @@ screen. %s/%d are runtime placeholders. "resource env" "restarted ≥%d times — check logs: %v" "root" +"running mysql query: %w%s" "scanning the cluster for tracebloc clients: %w" "schema" "schema entry %q must be col:TYPE (e.g. age:INT,price:FLOAT)" diff --git a/internal/push/list_detailed.go b/internal/push/list_detailed.go index 2e0cd4f..1ce5811 100644 --- a/internal/push/list_detailed.go +++ b/internal/push/list_detailed.go @@ -288,7 +288,7 @@ func runMySQLQuery(ctx context.Context, exec Executor, namespace, pod, container script := `mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N` if err := exec.Exec(ctx, namespace, pod, container, []string{"sh", "-c", script}, strings.NewReader(query), &stdout, &stderr); err != nil { - return "", fmt.Errorf("querying datasets: %w%s", err, stderrSuffix(&stderr)) + return "", fmt.Errorf("running mysql query: %w%s", err, stderrSuffix(&stderr)) } return stdout.String(), nil } diff --git a/internal/push/teardown.go b/internal/push/teardown.go index b917d6f..e81487f 100644 --- a/internal/push/teardown.go +++ b/internal/push/teardown.go @@ -66,6 +66,18 @@ func PlanTeardown(table string) TeardownPlan { type TeardownResult struct { DroppedTable bool RemovedPaths []string + // BookkeepingCleaned reports whether the ingestor's bookkeeping rows + // for the table (run-journal + pseudonymization salt) were deleted + // alongside it. Best-effort: false on clusters whose ingestor never + // created those tables — the teardown itself still succeeds. + BookkeepingCleaned bool + // BookkeepingErrs carries the per-table failure detail (which + // bookkeeping table, mysql's stderr folded into the error) so callers + // can SURFACE it: a silent false is indistinguishable from the + // schema-drift regression this cleanup exists to prevent — e.g. a + // renamed keying column would otherwise no-op invisibly, reopening + // the husk-row leak (review, Saqlain). + BookkeepingErrs []string } // Teardown performs the in-cluster teardown described by plan: @@ -107,6 +119,39 @@ func Teardown(ctx context.Context, cs kubernetes.Interface, exec Executor, names } res.DroppedTable = true + // 1b. Best-effort bookkeeping cleanup (RFC-0003 I6 — tracebloc/backend#1209): + // the ingestor keeps one run-journal row per ingest and one + // pseudonymization-salt row per table; dropping the table alone + // strands them. Under per-ingestion tables (data-ingestors#408) + // every dataset is its own table, so every delete would leave one + // husk row of each kind — an unbounded slow leak. plan.Table passed + // ValidateTableName ([A-Za-z_][A-Za-z0-9_]*), so it cannot escape + // the single-quoted literal. Each DELETE runs separately and + // best-effort: either bookkeeping table may be absent on clusters + // that never ran a journal-aware ingestor, and these are metadata + // rows, not data — never fail a teardown whose DROP succeeded. + // The SQL rides runMySQLQuery — stdin, never a shell -e argument: the + // string literal's single quotes would terminate a single-quoted shell + // string and mysql would see an unquoted identifier, silently no-oping + // the DELETEs forever (Bugbot on the PR). Column contract, pinned + // against data-ingestors tracebloc_ingestor/database.py: BOTH + // bookkeeping tables key these rows by `table_name` — + // RUNS_TABLE tracebloc_ingest_runs (ingestor_id PK, table_name + // indexed via ix_tracebloc_ingest_runs_table) + // SALT_TABLE tracebloc_ingest_meta (table_name PK, salt) + // Each DELETE stays a separate best-effort call: batched on one stdin, + // a missing first table would abort the second (mysql stops on error). + res.BookkeepingCleaned = true + for _, bookkeeping := range []string{ingestRunsTable, ingestMetaTable} { + cleanupSQL := fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE table_name='%s'", + plan.Database, bookkeeping, plan.Table) + if _, err := runMySQLQuery(ctx, exec, namespace, mysqlPod, mysqlContainer, cleanupSQL); err != nil { + res.BookkeepingCleaned = false + res.BookkeepingErrs = append(res.BookkeepingErrs, + fmt.Sprintf("%s: %v", bookkeeping, err)) + } + } + // 2. rm the PVC dirs from an ephemeral stage-identity pod (see the // doc note above + #259). The pod owns the staging files it // deletes, so this works on hostPath and CSI. diff --git a/internal/push/teardown_test.go b/internal/push/teardown_test.go index df514ed..1b2bc50 100644 --- a/internal/push/teardown_test.go +++ b/internal/push/teardown_test.go @@ -3,6 +3,8 @@ package push import ( "context" "errors" + "fmt" + "io" "strings" "testing" @@ -183,3 +185,103 @@ func TestCleanStaging_PodCreateFailureReturnsError(t *testing.T) { t.Errorf("rm ran (%v) despite the pod never being created", fe.gotCmd) } } + +// TestTeardown_CleansBookkeepingRows pins the RFC-0003 I6 half of teardown +// (tracebloc/backend#1209): after the DROP, the ingestor's run-journal and +// salt rows for the table are deleted best-effort — and a failure there +// never fails a teardown whose DROP already succeeded. +func TestTeardown_CleansBookkeepingRows(t *testing.T) { + newCS := func() *fake.Clientset { + cs := fake.NewClientset(&corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "mysql-0", Namespace: "tracebloc"}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "mysql"}}}, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + }) + readyOnNextGet(cs) + return cs + } + opts := PodSpecOptions{ + Namespace: "tracebloc", + PVCClaimName: "client-pvc", + PVCMountPath: "/data/shared", + Table: "ds_0f2ab1de3c444e558f66778899aabbcc", + } + plan := PlanTeardown("ds_0f2ab1de3c444e558f66778899aabbcc") + + t.Run("journal and salt rows are deleted for the dropped table", func(t *testing.T) { + rec := &recordingExecutor{} + res, err := Teardown(context.Background(), newCS(), rec, "tracebloc", plan, opts) + if err != nil { + t.Fatalf("Teardown: %v", err) + } + if !res.BookkeepingCleaned { + t.Error("BookkeepingCleaned = false, want true") + } + // The SQL must arrive on STDIN with its quoted literal intact — + // never through a shell -e argument, where the literal's single + // quotes would be eaten by the shell (Bugbot: the DELETEs would + // silently no-op forever). + var journal, salt bool + for _, call := range rec.calls { + if strings.Contains(strings.Join(call.cmd, " "), "DELETE FROM") { + t.Errorf("DELETE passed as a shell argument (%q) — must be fed on stdin", call.cmd) + } + stdin := string(call.stdin) + want := "WHERE table_name='" + plan.Table + "'" + if strings.Contains(stdin, "DELETE FROM") && strings.Contains(stdin, ingestRunsTable) && strings.Contains(stdin, want) { + journal = true + } + if strings.Contains(stdin, "DELETE FROM") && strings.Contains(stdin, ingestMetaTable) && strings.Contains(stdin, want) { + salt = true + } + } + if !journal { + t.Errorf("no stdin DELETE against %s with a quoted literal for %s observed", ingestRunsTable, plan.Table) + } + if !salt { + t.Errorf("no stdin DELETE against %s with a quoted literal for %s observed", ingestMetaTable, plan.Table) + } + }) + + t.Run("bookkeeping failure never fails the teardown", func(t *testing.T) { + rec := &recordingExecutor{failWhenStdinContains: "DELETE FROM"} + res, err := Teardown(context.Background(), newCS(), rec, "tracebloc", plan, opts) + if err != nil { + t.Fatalf("Teardown should tolerate bookkeeping failures, got: %v", err) + } + if !res.DroppedTable { + t.Error("DroppedTable = false, want true") + } + if res.BookkeepingCleaned { + t.Error("BookkeepingCleaned = true, want false when the DELETEs fail") + } + if len(res.RemovedPaths) == 0 { + t.Error("PVC rm did not run — bookkeeping failure must not short-circuit step 2") + } + }) +} + +// recordingExecutor records every Exec call (command AND stdin) and can +// fail calls whose stdin matches a marker. +type recordingExecutor struct { + calls []execCall + failWhenStdinContains string +} + +type execCall struct { + pod, container string + cmd []string + stdin []byte +} + +func (r *recordingExecutor) Exec(ctx context.Context, namespace, pod, container string, cmd []string, stdin io.Reader, stdout, stderr io.Writer) error { + var in []byte + if stdin != nil { + in, _ = io.ReadAll(stdin) + } + r.calls = append(r.calls, execCall{pod: pod, container: container, cmd: cmd, stdin: in}) + if r.failWhenStdinContains != "" && strings.Contains(string(in), r.failWhenStdinContains) { + return fmt.Errorf("simulated bookkeeping failure") + } + return nil +}