diff --git a/internal/cli/coverage_test.go b/internal/cli/coverage_test.go index 76cfde7..046a59e 100644 --- a/internal/cli/coverage_test.go +++ b/internal/cli/coverage_test.go @@ -102,8 +102,15 @@ func TestClassifyPushOutcome(t *testing.T) { {"clean", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 10}}}, nil, "succeeded", 0}, {"partial", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 7, FailedRecords: 3}}}, nil, "completed_with_failures", 9}, {"failed", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeFailed}}, nil, "failed", 9}, + {"unknown", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeUnknown}}, nil, "unknown", 9}, {"detached", &submit.Result{Submit: resp}, nil, "detached", 0}, + {"nil result", nil, nil, "detached", 0}, {"watch error", &submit.Result{Submit: resp}, &submit.WatchError{Err: errors.New("stream broke")}, "watch_error", 9}, + // The submit-side error buckets (exit 5 vs 8) the original matrix missed. + {"auth 401", nil, &submit.SubmitError{StatusCode: 401}, "auth_error", 5}, + {"auth 403", nil, &submit.SubmitError{StatusCode: 403}, "auth_error", 5}, + {"submit 500", nil, &submit.SubmitError{StatusCode: 500}, "submit_error", 8}, + {"submit 422", nil, &submit.SubmitError{StatusCode: 422}, "submit_error", 8}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { diff --git a/internal/cli/data.go b/internal/cli/data.go index b392123..9bf1b27 100644 --- a/internal/cli/data.go +++ b/internal/cli/data.go @@ -758,6 +758,37 @@ other collaborators train against it without ever seeing the raw files.`)) return &exitError{code: 7, err: stageErr} } + // 10–12. The ingestion-run tail: mint token → port-forward → submit → + // classify → emit JSON → reclaim staging. Extracted so its + // outcome matrix (exit 5/8/9, JSON emission, and the + // must-NOT-reclaim-on-partial gate) is table-testable via the + // injected seams without a cluster (#1009). jsonEmitted flows + // back so the --output-json error defer above stays correct. + je, runErr := runIngestionRun(ctx, out, a, target, specBytes, spec) + jsonEmitted = je + return runErr +} + +// runIngestionRun is the money path's outcome tail. It mints the ingestor +// token, port-forwards to jobs-manager, POSTs the run, classifies the result +// into a status + process exit code (kept in lockstep by classifyPushOutcome), +// emits the machine-readable JSON in --output-json mode, and reclaims the +// staged source copy on a clean success only. +// +// Split out of runDataIngest purely for testability: the four cluster-touching +// steps go through package-level seams (mintIngestorTokenFn / +// portForwardJobsManagerFn / submitRunFn / cleanStagingFn), so a table test can +// drive the full classify → exit-code → JSON → reclaim matrix — including the +// "must NOT reclaim on partial failure" gate — without standing up a cluster +// (#1009). +// +// Returns jsonEmitted so runDataIngest's --output-json error defer knows +// whether a result object already reached stdout: the mint / port-forward +// failures return before the emit and rely on that defer; the submit path +// always emits. +func runIngestionRun(ctx context.Context, out io.Writer, a runDataIngestArgs, target *clusterTarget, specBytes []byte, spec map[string]any) (jsonEmitted bool, err error) { + resolved, cs, release, pvc := target.Resolved, target.Clientset, target.Release, target.PVC + // 10. Mint the SA token Phase 4 uses to authenticate the POST // to jobs-manager. Expiry is 1 hour (vs cluster info's 10 // min) because the full Phase 4 lifecycle — submit + watch @@ -770,10 +801,10 @@ other collaborators train against it without ever seeing the raw files.`)) a.Printer.Hintf("Submitting the run, then following along as tracebloc validates your data and loads it into the table — progress streams below.") a.Printer.Hintf("This follows the run for up to an hour; a longer run keeps going on its own (or start it with --detach and check back later).") } - tok, err := cluster.MintIngestorToken(ctx, cs, resolved.Namespace, + tok, err := mintIngestorTokenFn(ctx, cs, resolved.Namespace, release.IngestorSAName, 3600, nil) if err != nil { - return &exitError{code: 5, err: err} + return false, &exitError{code: 5, err: err} } // 11. Open a port-forward to a Pod backing the jobs-manager @@ -784,10 +815,10 @@ other collaborators train against it without ever seeing the raw files.`)) // `kubectl port-forward`. Bugbot PR #10 r3 caught the // original broken-by-design direct-URL POST. a.Printer.Infof("Connecting to your workspace to submit the run…") - pf, err := submit.PortForwardJobsManager(ctx, cs, resolved.RestConfig, + pf, err := portForwardJobsManagerFn(ctx, cs, resolved.RestConfig, resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort) if err != nil { - return &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} + return false, &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} } defer pf.Close() @@ -810,7 +841,7 @@ other collaborators train against it without ever seeing the raw files.`)) // WatchResult Detached → 0 (cluster keeps running) // WatchResult Succeeded clean → 0 localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort) - submitRes, err := submit.Run(ctx, submit.Options{ + submitRes, err := submitRunFn(ctx, submit.Options{ Submitter: submit.NewHTTPSubmitter(localEndpoint, tok.Token), Client: cs, IngestConfigYAML: string(specBytes), @@ -845,21 +876,13 @@ other collaborators train against it without ever seeing the raw files.`)) jsonEmitted = true } - // Reclaim the staged source copy on a CLEAN success only. The - // ingestor copies (not moves) the staged files into the table, so - // leaving .tracebloc-staging/ behind doubles PVC use for - // file-bearing datasets until the next --overwrite or `data delete` - // (the staging-leak found by the ingest UX audit; cli#166 / epic #67). - // Gated on status=="succeeded" so we never touch the source on a - // - detached run (status "detached"): the Job is still reading it; - // - partial (status "completed_with_failures") or failed run: the - // user may want the source to inspect/retry. - // Best-effort and time-bounded (push.StagingCleanupTimeout): a failed - // or slow reclaim must not fail — or noticeably delay — a successful - // ingest. - if status == "succeeded" { + // Reclaim the staged source copy on a CLEAN success only (see + // shouldReclaimStaging). Best-effort and time-bounded + // (push.StagingCleanupTimeout): a failed or slow reclaim must not + // fail — or noticeably delay — a successful ingest. + if shouldReclaimStaging(status) { a.Printer.Infof("Reclaiming the temporary staging copy on the cluster…") - if cerr := push.CleanStaging(ctx, cs, + if cerr := cleanStagingFn(ctx, cs, &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, resolved.Namespace, a.Spec.Table, push.PodSpecOptions{ Namespace: resolved.Namespace, @@ -875,9 +898,25 @@ other collaborators train against it without ever seeing the raw files.`)) } if exitErr != nil { - return exitErr + return jsonEmitted, exitErr } - return nil + return jsonEmitted, nil +} + +// shouldReclaimStaging reports whether the staged source copy should be +// reclaimed after the run. ONLY on a clean success: the ingestor copies (not +// moves) the staged files into the table, so leaving .tracebloc-staging/
+// behind doubles PVC use for file-bearing datasets until the next --overwrite +// or `data delete` (the staging-leak found by the ingest UX audit; cli#166 / +// epic #67). Everything else keeps the source: +// - a detached run ("detached") — the Job is still reading it; +// - a partial ("completed_with_failures") or failed/errored run — the user +// may want the source to inspect or retry. +// +// This is the "must NOT reclaim on partial failure" gate (#1009), named so the +// invariant is table-testable in isolation. +func shouldReclaimStaging(status string) bool { + return status == "succeeded" } // classifyPushOutcome maps the result of submit.Run to a machine- @@ -1072,6 +1111,18 @@ func writePushErrorJSON(w io.Writer, sp push.SpecArgs, e error, code int) { // listDatasetsFn is a test seam over push.ListDatasets. var listDatasetsFn = push.ListDatasets +// Test seams over the cluster-touching steps of runIngestionRun (#1009). +// Production wires them to the real functions; a table test overrides them to +// drive the classify → exit-code → JSON → reclaim matrix without a cluster +// (mirrors the listDatasetsFn seam). cleanStagingFn is here too so a test can +// observe whether the staging reclaim ran (the must-NOT-reclaim gate). +var ( + mintIngestorTokenFn = cluster.MintIngestorToken + portForwardJobsManagerFn = submit.PortForwardJobsManager + submitRunFn = submit.Run + cleanStagingFn = push.CleanStaging +) + // destTableExists reports whether the destination table already holds an // ingested dataset, via the same query `data list` uses. It fails OPEN: a // broken check returns (false, note) so the ingest proceeds — the in-cluster diff --git a/internal/cli/ingestion_run_test.go b/internal/cli/ingestion_run_test.go new file mode 100644 index 0000000..94b7bfa --- /dev/null +++ b/internal/cli/ingestion_run_test.go @@ -0,0 +1,166 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "testing" + + "github.com/tracebloc/cli/internal/cluster" + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/submit" + "github.com/tracebloc/cli/internal/ui" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" +) + +// The money path (#1009): submit → classify → exit-code → JSON → reclaim. +// These tests pin the outcome matrix — including the "must NOT reclaim on +// partial failure" gate — without standing up a cluster, via the seams +// (mintIngestorTokenFn / portForwardJobsManagerFn / submitRunFn / +// cleanStagingFn) that runIngestionRun goes through. + +func succeededResult() *submit.Result { + return &submit.Result{ + Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "ingestor-x"}, + Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 2, InsertedRecords: 2}}, + } +} + +func partialResult() *submit.Result { + return &submit.Result{ + Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "ingestor-x"}, + Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 2, InsertedRecords: 1, FailedRecords: 1}}, + } +} + +// TestShouldReclaimStaging pins the must-NOT-reclaim-on-partial gate: the +// staged source is reclaimed ONLY on a clean success. +func TestShouldReclaimStaging(t *testing.T) { + if !shouldReclaimStaging("succeeded") { + t.Error(`shouldReclaimStaging("succeeded") = false, want true`) + } + for _, st := range []string{ + "completed_with_failures", "failed", "unknown", "detached", + "auth_error", "submit_error", "watch_error", "dry-run", "error", "", + } { + if shouldReclaimStaging(st) { + t.Errorf("shouldReclaimStaging(%q) = true, want false — a non-clean run must keep the source", st) + } + } +} + +// TestRunIngestionRun_Matrix drives the whole outcome tail through the seams: +// per row it asserts the exit code, whether the staging reclaim ran, and the +// emitted --output-json status — all in lockstep. +func TestRunIngestionRun_Matrix(t *testing.T) { + origMint, origPF, origRun, origClean := mintIngestorTokenFn, portForwardJobsManagerFn, submitRunFn, cleanStagingFn + defer func() { + mintIngestorTokenFn, portForwardJobsManagerFn, submitRunFn, cleanStagingFn = origMint, origPF, origRun, origClean + }() + + target := &clusterTarget{ + Resolved: &cluster.ResolvedConfig{Namespace: "tracebloc"}, + Clientset: nil, // the seams ignore it; the reclaim SPDYExecutor literal doesn't deref + Release: &cluster.ParentRelease{IngestorSAName: "ingestor", JobsManagerServiceName: "jm", JobsManagerPort: 8080}, + PVC: &cluster.SharedPVC{ClaimName: "pvc", MountPath: "/data/shared"}, + } + spec := map[string]any{"table": "t", "category": "image_classification", "intent": "train", "label": "label"} + + cases := []struct { + name string + mintErr error + pfErr error + submitRes *submit.Result + submitErr error + wantCode int // 0 == success (nil err) + wantStatus string + wantReclaim bool + wantJSON bool + }{ + {"succeeded", nil, nil, succeededResult(), nil, 0, "succeeded", true, true}, + {"partial", nil, nil, partialResult(), nil, 9, "completed_with_failures", false, true}, + {"failed", nil, nil, &submit.Result{Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "j"}, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeFailed}}, nil, 9, "failed", false, true}, + {"detached", nil, nil, &submit.Result{Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "j"}, Watch: nil}, nil, 0, "detached", false, true}, + {"submit-auth", nil, nil, nil, &submit.SubmitError{StatusCode: 401}, 5, "auth_error", false, true}, + {"submit-5xx", nil, nil, nil, &submit.SubmitError{StatusCode: 500}, 8, "submit_error", false, true}, + {"watch-err", nil, nil, nil, &submit.WatchError{Err: errors.New("x")}, 9, "watch_error", false, true}, + // mint / port-forward failures return BEFORE the JSON emit and the + // reclaim; jsonEmitted is false (runDataIngest's error defer covers it). + {"mint-fail", errors.New("mint boom"), nil, nil, nil, 5, "", false, false}, + {"pf-fail", nil, errors.New("pf boom"), nil, nil, 8, "", false, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + reclaimCalled := false + mintIngestorTokenFn = func(_ context.Context, _ kubernetes.Interface, _, _ string, _ int64, _ []string) (*cluster.IngestorToken, error) { + if c.mintErr != nil { + return nil, c.mintErr + } + return &cluster.IngestorToken{Token: "tok"}, nil + } + portForwardJobsManagerFn = func(_ context.Context, _ kubernetes.Interface, _ *rest.Config, _, _ string, _ int) (*submit.ForwardedConnection, error) { + if c.pfErr != nil { + return nil, c.pfErr + } + return &submit.ForwardedConnection{LocalPort: 12345}, nil + } + submitRunFn = func(_ context.Context, _ submit.Options) (*submit.Result, error) { + return c.submitRes, c.submitErr + } + cleanStagingFn = func(_ context.Context, _ kubernetes.Interface, _ push.Executor, _, _ string, _ push.PodSpecOptions) error { + reclaimCalled = true + return nil + } + + var jsonBuf bytes.Buffer + a := runDataIngestArgs{ + Spec: push.SpecArgs{Table: "t"}, + Printer: ui.New(io.Discard, ui.WithColor(false)), + OutputJSON: true, + JSONOut: &jsonBuf, + } + je, err := runIngestionRun(context.Background(), io.Discard, a, target, []byte("yaml"), spec) + + code := 0 + if err != nil { + var ee *exitError + if !errors.As(err, &ee) { + t.Fatalf("err is not *exitError: %v", err) + } + code = ee.Code() + } + if code != c.wantCode { + t.Errorf("exit code = %d, want %d", code, c.wantCode) + } + if reclaimCalled != c.wantReclaim { + t.Errorf("reclaim called = %v, want %v (only a clean success reclaims)", reclaimCalled, c.wantReclaim) + } + if je != c.wantJSON { + t.Errorf("jsonEmitted = %v, want %v", je, c.wantJSON) + } + if c.wantJSON { + var got pushJSONResult + if err := json.Unmarshal(jsonBuf.Bytes(), &got); err != nil { + t.Fatalf("emitted JSON invalid: %v (%q)", err, jsonBuf.String()) + } + if got.Status != c.wantStatus { + t.Errorf("emitted JSON status = %q, want %q", got.Status, c.wantStatus) + } + } else if jsonBuf.Len() != 0 { + t.Errorf("expected no JSON on the pre-submit failure path, got %q", jsonBuf.String()) + } + }) + } +} + +// TestSeamsWiredToRealFns guards that the indirection didn't accidentally +// leave a seam nil (a nil seam would panic the money path in production). +func TestSeamsWiredToRealFns(t *testing.T) { + if mintIngestorTokenFn == nil || portForwardJobsManagerFn == nil || + submitRunFn == nil || cleanStagingFn == nil { + t.Fatal("a money-path seam is nil — production would panic") + } +} diff --git a/internal/submit/portforward.go b/internal/submit/portforward.go index 20c6155..1dc2cef 100644 --- a/internal/submit/portforward.go +++ b/internal/submit/portforward.go @@ -30,8 +30,13 @@ type ForwardedConnection struct { done chan struct{} } -// Close tears down the port-forward. Safe to call multiple times. +// Close tears down the port-forward. Safe to call multiple times, and +// safe on a zero-value connection that was never started (stopCh nil) — +// e.g. a test fake handed back by an injected PortForwardJobsManager. func (f *ForwardedConnection) Close() { + if f.stopCh == nil { + return // never started; nothing to tear down + } select { case <-f.stopCh: return // already closed