From 9f75c1de733d61b373da6e6df7938bc8ee297e94 Mon Sep 17 00:00:00 2001 From: Lukas Wuttke Date: Tue, 14 Jul 2026 14:22:56 +0200 Subject: [PATCH] refactor(cli): split data.go into per-concern ingest files (mechanical) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure moves of whole top-level declarations out of the 1539-line data.go, same package cli, zero behavior change: - data_ingest_cmd.go — newDataIngestCmd (flag surface incl. hidden deprecated aliases) + runDataIngestArgs - data_ingest_local.go — sortedKeys / expandHome / statDatasetPath + printLocalSummary + runLocalPreflight - data_ingest_cluster.go — runIngestionRun + shouldReclaimStaging + printClusterSummary + test seams + destTableExists + existingTableAction - data_ingest_output.go — pushJSONResult/pushJSONSummary + writePushJSON/writePushErrorJSON + classifyPushOutcome (kept together with writePushJSON — mutation-hardened lockstep) data.go keeps the data group command, the deprecation-alias notice, and runDataIngest itself (extracted separately in cli#283). Only new text is each file's header comment + package/import clauses; every moved section was verified verbatim against the pre-split file. data_test.go is untouched and compiles unchanged. Co-Authored-By: Claude Fable 5 --- internal/cli/data.go | 846 ---------------------------- internal/cli/data_ingest_cluster.go | 275 +++++++++ internal/cli/data_ingest_cmd.go | 367 ++++++++++++ internal/cli/data_ingest_local.go | 129 +++++ internal/cli/data_ingest_output.go | 136 +++++ 5 files changed, 907 insertions(+), 846 deletions(-) create mode 100644 internal/cli/data_ingest_cluster.go create mode 100644 internal/cli/data_ingest_cmd.go create mode 100644 internal/cli/data_ingest_local.go create mode 100644 internal/cli/data_ingest_output.go diff --git a/internal/cli/data.go b/internal/cli/data.go index 6699c9f3..cb14b3a5 100644 --- a/internal/cli/data.go +++ b/internal/cli/data.go @@ -2,25 +2,18 @@ package cli import ( "context" - "encoding/json" "errors" "fmt" "io" - "k8s.io/client-go/kubernetes" - "os" "path/filepath" - "sort" "strings" "github.com/spf13/cobra" "gopkg.in/yaml.v3" "github.com/tracebloc/cli/internal/cluster" - "github.com/tracebloc/cli/internal/pathutil" "github.com/tracebloc/cli/internal/push" "github.com/tracebloc/cli/internal/schema" - "github.com/tracebloc/cli/internal/submit" - "github.com/tracebloc/cli/internal/ui" ) // newDataCmd wires the `tracebloc data` subtree. The dominant @@ -103,397 +96,6 @@ func warnDeprecatedAlias(cmd *cobra.Command, w io.Writer) { } } -// newDataIngestCmd implements `tracebloc data ingest `. -// -// Phase 3 scope (now complete across PR-a + PR-b): -// -// - Synthesize the ingest spec from flags (`internal/push.SpecArgs.Build`) -// - Validate it against the embedded ingest.v1 schema -// - Walk the local directory + enforce v0.1 size caps -// - Discover cluster, parent release, and shared PVC -// - Print a single-screen pre-flight summary -// - Either --dry-run stop, OR create an ephemeral stage Pod -// (alpine 3.20 pinned by digest, PSA-restricted security -// context), tar local files into it via an SPDY exec stream -// with a progress bar, then defer-delete the Pod -// -// Phase 4 (`tracebloc/client#152`) hooks the submit-to-jobs-manager -// step into the bottom of this command, replacing the "manually -// kick off helm ingestor" workaround in the success message. -// -// Aliases: "push" is kept for one deprecation cycle so existing -// scripts continue to work. -func newDataIngestCmd() *cobra.Command { - var ( - // Kubeconfig flags — same conventions as `cluster info`. - // Promoting these to persistent on the root is a v0.2 - // follow-up (tracebloc/cli#3); for now they live on each - // command that needs them. - kubeconfigPath string - contextOverride string - nsOverride string - - // Ingest-spec flags. All schema task categories are CLI-supported now - // (image classification / detection / segmentation / keypoint, the full - // text family, and the tabular / time-series family). - // - // --name/--task are the canonical flags (#180); --table/--category - // stay on as hidden deprecated aliases so existing scripts keep - // working. --intent is unchanged. The wire/spec field names don't - // change — this is a CLI-surface rename only. - name string - tableAlias string - task string - categoryAlias string - intent string - labelColumn string - targetSize string - minSize string - schemaFlag string - labelPolicy string - timeColumn string - numberOfKeypoints int - - // Operations flags. - dryRun bool - overwrite bool - noInput bool - outputJSON bool - - // Stage Pod image override. Defaults to the digest-pinned - // alpine that ships with the CLI; air-gapped customers - // override this to an image their registry mirror serves. - // Pin by digest in your override too — tag-only references - // drift silently and break "all my pushes worked yesterday." - stagePodImage string - - // Phase 4 flags. --detach exits immediately after the 201 - // from jobs-manager; --idempotency-key plumbs through to - // the submit body for retry-safety across CLI invocations - // (default: fresh per call); --image-digest pins the - // ingestor image (default: jobs-manager picks the - // cluster-configured one). - detach bool - idempotencyKey string - imageDigest string - ) - - cmd := &cobra.Command{ - Use: "ingest ", - Aliases: []string{"push"}, - Short: "Ingest a local dataset into your workspace", - // The task COUNT and the text-family subdir names are derived from the - // registry / vendored layout contract (push.SupportedCategoryIDs + - // push.TextSidecarDir) rather than hardcoded, so the help can't drift - // from what the CLI actually supports (cli#215): the count used to read - // a stale "9", and the text example showed texts/ for every text task - // even though masked_language_modeling stages into sequences/. - Long: fmt.Sprintf(`Ingests a local dataset into your workspace's storage, -submits the ingestion run, and follows it to completion (streaming -progress + the final summary). Your data never leaves your own -infrastructure. Supports %[1]d tasks across the image, text, and -tabular / time-series families; pick one with --task. - - is the data itself. What it looks like depends on the task: - - tabular / time-series — the dataset is a single CSV. Pass the .csv - file directly, or a folder holding exactly one .csv: - - churn.csv (the .csv file itself) - or - churn/ - data.csv (the one .csv in the folder) - - image (classification, object/keypoint detection) — a folder with - labels.csv + an images/ subfolder: - - cats_dogs/ - labels.csv (required) - images/ (required) - 001.jpg - ... - - text (classification, masked language modeling) — a folder with - labels.csv + a %[2]s/ subfolder (masked language modeling uses %[3]s/): - - reviews/ - labels.csv (required) - %[2]s/ (required — %[3]s/ for masked language modeling) - 001.txt - ... - -A bare .csv file is accepted only for the tabular / time-series family; -image and text datasets must be a folder. - -Accepted image extensions: .jpg, .jpeg, or .png (case-insensitive). -All images in one dataset must share a single type — the cluster -validates the type it was told to expect. - -v0.1 caps the dataset at 1 GiB total + 500 MiB per file. Larger -datasets need the v0.2 cloud-source story (S3/GCS/HTTPS sources) — -see tracebloc/client#147 non-goals. - -Exit codes: - 0 files staged + ingested successfully (or --detach: just staged + submitted) - 2 schema validation failed (synthesized spec rejected) or - v0.1-unsupported task passed - 3 local-layout or kubeconfig error - 4 cluster reachable but no tracebloc client / shared storage missing - 5 ingestor SA token couldn't be obtained, or jobs-manager - rejected the token (401/403) - 6 destination table already exists (re-run with --overwrite to - replace it, or pick a different --name) - 7 pre-flight succeeded but staging the files failed - (Pod creation, image pull, exec stream, or remote tar error) — - or, with --overwrite, removing the old table failed - 8 jobs-manager rejected the submit (4xx/5xx other than auth) - 9 ingestion Job exited non-zero, or completed with row-level - failures the summary panel reports`, - len(push.SupportedCategoryIDs()), - push.TextSidecarDir("text_classification"), - push.TextSidecarDir("masked_language_modeling")), - Args: cobra.MaximumNArgs(1), - RunE: func(cmd *cobra.Command, args []string) error { - var localPath string - if len(args) > 0 { - localPath = args[0] - } - // Resolve the deprecated flag aliases (#180): the canonical - // flag wins; a hidden legacy alias fills in only when the new - // flag wasn't passed, so old scripts keep working without the - // new surface silently shadowing them. The wire/spec field - // names are unchanged — this is a CLI rename only. - nameVal := name - if cmd.Flags().Changed("table") && !cmd.Flags().Changed("name") { - nameVal = tableAlias - } - taskVal := task - if cmd.Flags().Changed("category") && !cmd.Flags().Changed("task") { - taskVal = categoryAlias - } - // Whether the task was chosen at all (via either spelling). - // Dropping --task's old image_classification default means an - // unset task now drives the picker (TTY) or a clear error - // (non-interactive), never a silent image assumption. - taskSet := cmd.Flags().Changed("task") || cmd.Flags().Changed("category") - // Record whether --number-of-keypoints was explicitly passed, so - // the keypoint set-vs-unset message (#76b) can distinguish an - // explicit zero value from an unset flag (both look like the Go - // zero value in the spec). - changedFlags := map[string]bool{} - if cmd.Flags().Changed("number-of-keypoints") { - changedFlags["number-of-keypoints"] = true - } - // Guided mode: on a terminal (and unless --no-input), prompt - // for whatever's still missing. Off a TTY / with --no-input, - // prompter stays nil and runDataIngest keeps flag-only - // behavior. - interactive := !noInput && !outputJSON && isInteractiveTTY() - var pr prompter - if interactive { - pr = surveyPrompter{} - } - // In --output-json mode, human output goes to stderr so - // stdout carries only the JSON result. - humanOut := cmd.OutOrStdout() - printer := printerFor(cmd) - var jsonOut io.Writer - if outputJSON { - humanOut = cmd.ErrOrStderr() - printer = printerForWriter(cmd, cmd.ErrOrStderr()) - jsonOut = cmd.OutOrStdout() - } - return runDataIngest(cmd.Context(), humanOut, cmd.ErrOrStderr(), - runDataIngestArgs{ - LocalPath: localPath, - Kubeconfig: kubeconfigPath, - Context: contextOverride, - Namespace: nsOverride, - Spec: push.SpecArgs{ - Table: nameVal, Category: taskVal, Intent: intent, - LabelColumn: labelColumn, LabelPolicy: labelPolicy, TimeColumn: timeColumn, - NumberOfKeypoints: numberOfKeypoints, - }, - TargetSizeFlag: targetSize, - MinSizeFlag: minSize, - SchemaFlag: schemaFlag, - DryRun: dryRun, - Overwrite: overwrite, - StagePodImage: stagePodImage, - Detach: detach, - IdempotencyKey: idempotencyKey, - ImageDigest: imageDigest, - Printer: printer, - Interactive: interactive, - Prompter: pr, - TaskSet: taskSet, - ChangedFlags: changedFlags, - OutputJSON: outputJSON, - JSONOut: jsonOut, - }) - }, - } - - addKubeconfigFlags(cmd, &kubeconfigPath, &contextOverride, kubeconfigFlagUsage, contextFlagUsage) - addNamespaceFlag(cmd, &nsOverride, namespaceFlagUsage) - - // Required spec flags. We DON'T mark them required-at-cobra-level - // because cobra's "required flag" error message is terse and - // pre-empts our richer schema-driven diagnostic. Instead, the - // schema validator catches missing/empty values with the canonical - // JSON-pointer-anchored error. - cmd.Flags().StringVar(&name, "name", "", - "a name for this dataset — start with a letter or underscore, then letters/digits/underscores — you'll reference it by this name when you start a training run") - cmd.Flags().StringVar(&tableAlias, "table", "", - "deprecated alias for --name") - _ = cmd.Flags().MarkHidden("table") - cmd.Flags().StringVar(&task, "task", "", - "the task this data is for, one of: "+push.SupportedCategoriesList()+ - ". Omit it on a terminal to pick interactively.") - cmd.Flags().StringVar(&categoryAlias, "category", "", - "deprecated alias for --task") - _ = cmd.Flags().MarkHidden("category") - cmd.Flags().StringVar(&intent, "intent", "", - "is this training or test data? train|test (default train)") - cmd.Flags().StringVar(&labelColumn, "label-column", "", - "name of the label/target column (in labels.csv for image tasks, in the data CSV for tabular)") - cmd.Flags().StringVar(&targetSize, "target-size", "", - "image tasks only: the resolution your images already are, as WxH (e.g. 512x512). tracebloc never "+ - "resizes — it checks every image is exactly this size and rejects any that differ. Default: "+ - "read from your first image.") - cmd.Flags().StringVar(&minSize, "min-size", "", - "image tasks only: reject images smaller than WxH before the ingest (e.g. 64x64). Set it to the "+ - "smallest size your model can train on — raise or lower it freely. Default: unset (no local "+ - "size check).") - cmd.Flags().StringVar(&schemaFlag, "schema", "", - "tabular/time-series only: column types as col:TYPE,col:TYPE (e.g. age:INT,price:FLOAT). "+ - "Default: inferred from the CSV (INT/BIGINT/FLOAT/BOOLEAN/DATE/DATETIME/VARCHAR(n)).") - cmd.Flags().StringVar(&labelPolicy, "label-policy", "", - "regression-class only (tabular_regression, time_series_forecasting, time_to_event_prediction): "+ - "passthrough|bucket (default bucket — bins the target so the raw value never leaves the cluster)") - cmd.Flags().StringVar(&timeColumn, "time-column", "", - "time_to_event_prediction only: name of the time/duration column (default: a column named \"time\")") - cmd.Flags().IntVar(&numberOfKeypoints, "number-of-keypoints", 0, - "keypoint_detection only: number of keypoints per sample (required; e.g. 17 for COCO pose)") - - cmd.Flags().BoolVar(&overwrite, "overwrite", false, - "replace the destination table if it already exists: its current table + files are removed first (same as `tracebloc data delete`), then the new data is ingested. Not combinable with --idempotency-key") - cmd.Flags().BoolVar(&dryRun, "dry-run", false, - "validate + discover + walk, but don't create any cluster resources") - cmd.Flags().BoolVar(&noInput, "no-input", false, - "disable interactive prompts; fail on missing required values (for CI/scripts)") - cmd.Flags().BoolVar(&outputJSON, "output-json", false, - "emit a machine-readable JSON result on stdout (human output → stderr; implies --no-input)") - cmd.Flags().StringVar(&stagePodImage, "stage-pod-image", "", - "override the ephemeral stage Pod's image (default: digest-pinned alpine 3.20 baked into the CLI). "+ - "Pin by digest in your override too — tag-only refs drift silently.") - - cmd.Flags().BoolVar(&detach, "detach", false, - "exit immediately after jobs-manager accepts the run (no log streaming, no summary panel). "+ - "Use for CI scenarios; reconnect later with `kubectl logs -f -n job/`.") - cmd.Flags().StringVar(&idempotencyKey, "idempotency-key", "", - "reuse this idempotency key across retry attempts (default: fresh per invocation). "+ - "jobs-manager treats a duplicate key as a replay and attaches to the existing Job "+ - "rather than spawning a new one — useful for at-most-once-across-attempts semantics.") - cmd.Flags().StringVar(&imageDigest, "image-digest", "", - "pin the ingestor container image to a specific digest (default: jobs-manager picks the "+ - "cluster-configured `images.ingestor.digest`). Format: sha256:.") - - return cmd -} - -// runDataIngestArgs collects every parameter runDataIngest needs, -// so the body stays testable without going through cobra. The cobra -// RunE wrapper above is the ONLY caller in production; tests -// construct one of these directly. -type runDataIngestArgs struct { - LocalPath string - Kubeconfig string - Context string - Namespace string - Spec push.SpecArgs - TargetSizeFlag string // raw --target-size; resolved after Discover (image) - MinSizeFlag string // raw --min-size; resolved after Discover (image) — #348 floor override - SchemaFlag string // raw --schema; resolved or inferred after Discover (tabular) - DryRun bool - Overwrite bool - StagePodImage string - - // Printer renders the pre-flight summary + status output. Built in - // the RunE from the persistent --plain flag (see printerFor). - Printer *ui.Printer - - // Interactive guided mode (#28). When Interactive is true, - // runDataIngest prompts (via Prompter) for any missing core inputs - // before validation. TaskSet records whether the task was passed - // explicitly (via --task or the hidden --category alias); an unset - // task drives the picker rather than assuming a default. Prompter is - // nil off a TTY / --no-input. - Interactive bool - Prompter prompter - TaskSet bool - - // ChangedFlags records which CLI flags were EXPLICITLY set - // (cmd.Flags().Changed), decoupling "was it passed" from "is its value - // non-zero" — the value alone can't tell `--number-of-keypoints 0` (an - // explicit, invalid value) from an unset flag (also 0). The RunE - // populates it for --number-of-keypoints; the keypoint set-vs-unset - // diagnostic (#76b) reads it. Nil in direct-construction tests that - // don't exercise that path. - ChangedFlags map[string]bool - - // OutputJSON routes human output to stderr and emits a JSON result - // to JSONOut (stdout); set together by the RunE in --output-json - // mode (which also forces non-interactive). - OutputJSON bool - JSONOut io.Writer - - // Phase 4 (#152) fields. See the flag declarations for the - // per-knob rationale; all three are optional. - Detach bool - IdempotencyKey string - ImageDigest string -} - -// sortedKeys returns m's keys in sorted order — used to list a CSV's inferred -// columns in the friendly missing-label message (#214) deterministically. -func sortedKeys(m map[string]string) []string { - keys := make([]string, 0, len(m)) - for k := range m { - keys = append(keys, k) - } - sort.Strings(keys) - return keys -} - -// expandHome expands a leading ~ (current user or ~user) to a home -// directory, leaving every other path untouched. It's the CLI-local -// name for the shared pathutil.ExpandHome; cluster.expandPath resolves -// to the same helper, so ~-expansion is identical across subcommands -// (a --kubeconfig ~alice/... resolves alice's home just like a data -// ingest path does). See pathutil.ExpandHome for the full contract. (#181) -func expandHome(path string) string { - return pathutil.ExpandHome(path) -} - -// statDatasetPath is the "path existence FIRST" guard (#181): a typo'd -// path fails plainly on the path — a clean "no such file or directory" — -// before any family sniff, label preview, or schema work touches it. -// Both entry points call it: the flag-only path from runDataIngest's 0b -// step, and the guided path from runInteractive (before the family sniff), -// so the invariant holds on every route rather than only the flag path. -func statDatasetPath(path string) error { - if _, serr := os.Stat(path); serr != nil { - if errors.Is(serr, os.ErrNotExist) { - return &exitError{code: 3, err: fmt.Errorf( - "no such file or directory: %q — check the path to your dataset", path)} - } - return &exitError{code: 3, err: fmt.Errorf( - "can't read %q: %w", path, serr)} - } - return nil -} - // runDataIngest is the full Phase 3 implementation: pre-flight // checks, then either --dry-run stop or stage Pod + tar stream + // cleanup. Phase 4 (#152) will hook submit-to-jobs-manager after @@ -1089,451 +691,3 @@ collaborators can train against that table without ever seeing the raw files.`)) 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 - // + log stream — can run that long for large ingestions. - // The chart's helm flow uses the same token-mint code path. - a.Printer.Step(3, 3, "Validate and load") - if a.Detach { - a.Printer.Hintf("Submitting the run — with --detach it keeps running on your workspace after this command returns; the reconnect command is shown below.") - } else { - 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 := mintIngestorTokenFn(ctx, cs, resolved.Namespace, - release.IngestorSAName, 3600, nil) - if err != nil { - return false, &exitError{code: 5, err: err} - } - - // 11. Open a port-forward to a Pod backing the jobs-manager - // Service. The CLI runs off-cluster (on a laptop, in CI - // runners outside the cluster network), so the discovered - // *.svc.cluster.local URL isn't reachable — we tunnel - // through the kubeconfig-authenticated apiserver, same as - // `kubectl port-forward`. Bugbot PR #10 r3 caught the - // original broken-by-design direct-URL POST. - // Opening the port-forward is a blocking wait (tunnel setup through the - // apiserver), so it runs under a spinner — no wait on the happy path stays - // silent (RFC-0002 "progress on every wait"). The submit POST itself is a - // separate ~30s synchronous wait; its spinner lives in submit.Run, next to - // the POST it covers. - connectSpin := a.Printer.Spinner("Connecting to your workspace to submit the run", "") - pf, err := portForwardJobsManagerFn(ctx, cs, resolved.RestConfig, - resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort) - connectSpin.Stop() - if err != nil { - return false, &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} - } - defer pf.Close() - - // 12. Phase 4: POST to jobs-manager via the local port, - // watch the spawned ingestor Job, render the parsed - // INGESTION SUMMARY panel. - // - // Exit-code mapping: - // SubmitError 401/403 → 5 (auth — same bucket as - // token-mint, shared - // "your SA can't do this" - // diagnostic class) - // SubmitError other 4xx/5xx → 8 (submit failed) - // WatchResult Failed → 9 (ingest failed) - // WatchResult Succeeded + - // summary.HasFailures() → 9 (some rows failed - // even though Job exited 0; - // the ingestor surfaces - // partial-failure summaries) - // WatchResult Detached → 0 (cluster keeps running) - // WatchResult Succeeded clean → 0 - localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort) - submitRes, err := submitRunFn(ctx, submit.Options{ - Submitter: submit.NewHTTPSubmitter(localEndpoint, tok.Token), - Client: cs, - IngestConfigYAML: string(specBytes), - IdempotencyKey: a.IdempotencyKey, - ImageDigest: a.ImageDigest, - Detach: a.Detach, - Out: out, - Printer: a.Printer, - }) - // Classify once: a machine-readable status + the process exit error - // in lockstep, so --output-json emits exactly one result object on - // EVERY path (success / partial / failure / submit-or-watch error) - // whose status matches the exit code. (Bugbot #38.) - status, exitErr := classifyPushOutcome(submitRes, err) - - // Emit the machine-readable result BEFORE the best-effort staging - // reclaim below, so a scripted --output-json consumer gets its result - // object at ingest-completion latency and never waits on a slow - // cluster-side cleanup that has no bearing on the ingest outcome. - if a.OutputJSON { - var summary *submit.Summary - var ns, jobName string - if submitRes != nil { - if submitRes.Watch != nil { - summary = submitRes.Watch.Summary - } - if submitRes.Submit != nil { - ns, jobName = submitRes.Submit.Namespace, submitRes.Submit.JobName - } - } - writePushJSON(a.JSONOut, status, spec, summary, ns, jobName) - jsonEmitted = true - } - - // 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) { - reclaimSpin := a.Printer.Spinner("Reclaiming the temporary copy", "") - cerr := cleanStagingFn(ctx, cs, - &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, - resolved.Namespace, a.Spec.Table, push.PodSpecOptions{ - Namespace: resolved.Namespace, - PVCClaimName: pvc.ClaimName, - PVCMountPath: pvc.MountPath, - Table: a.Spec.Table, - ServiceAccountName: release.IngestorSAName, - Image: a.StagePodImage, - }) - reclaimSpin.Stop() - if cerr != nil { - a.Printer.Warnf("Couldn't reclaim the temporary copy (%v). It's harmless — the next re-ingest of %q or a `tracebloc data delete %s` will clear it.", - cerr, a.Spec.Table, a.Spec.Table) - } - } - - if exitErr != nil { - return jsonEmitted, exitErr - } - 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- -// readable status string + the process exit error, kept in lockstep so -// --output-json's status always agrees with the exit code (a nil -// *exitError = success, exit 0). It also covers the error paths -// (auth/submit/watch) so --output-json can still emit a result object -// when submit.Run returns an error. (Bugbot #38.) -func classifyPushOutcome(res *submit.Result, err error) (string, *exitError) { - if err != nil { - switch { - case submit.IsAuthError(err): - return "auth_error", &exitError{code: 5, err: err} - case submit.IsWatchError(err): - // jobs-manager accepted the run; the cluster is doing the - // work, the CLI just couldn't follow along — ingest-side - // (exit 9), not submit-side (8). - return "watch_error", &exitError{code: 9, err: err} - default: - return "submit_error", &exitError{code: 8, err: err} - } - } - // --detach (no watch) or SIGINT-mid-watch: success; cluster runs on. - if res == nil || res.Watch == nil || res.Watch.Outcome == submit.JobOutcomeDetached { - return "detached", nil - } - switch res.Watch.Outcome { - case submit.JobOutcomeFailed: - return "failed", &exitError{code: 9, err: errors.New("ingestion Job exited non-zero — see logs above")} - case submit.JobOutcomeUnknown: - return "unknown", &exitError{code: 9, err: errors.New( - "ingestion Job's final status couldn't be determined within the watch window — " + - "check `kubectl get job -n " + res.Submit.Namespace + " " + res.Submit.JobName + "` for the outcome")} - case submit.JobOutcomeSucceeded: - // Job exited 0, but rows can still have failed — exit 9, and the - // JSON status must say so, NOT "succeeded". (Bugbot #38.) - if res.Watch.Summary != nil && res.Watch.Summary.HasFailures() { - return "completed_with_failures", &exitError{code: 9, err: errors.New( - "ingestion Job completed but the summary reports failures — see panel above")} - } - return "succeeded", nil - } - return "unknown", nil -} - -// printLocalSummary shows what the CLI found on disk plus the ingest -// settings it assembled — the detail under step 1 ("Check your data"). -// Mirrors `cluster info`'s section/Field layout. -func printLocalSummary(p *ui.Printer, layout *push.LocalLayout, spec map[string]any) { - cat, _ := spec["category"].(string) - - p.Section("Local dataset") - p.Field("root", layout.Root) - switch { - case push.IsTabular(cat): - p.Field("data CSV", layout.LabelsCSV) - if sch, ok := spec["schema"].(map[string]string); ok { - p.Field("columns", fmt.Sprintf("%d", len(sch))) - } - case push.IsText(cat): - dir := push.TextSidecarDir(cat) - p.Field("labels.csv", layout.LabelsCSV) - p.Field(dir, fmt.Sprintf("%d files", len(layout.Sidecars[dir]))) - default: - p.Field("labels.csv", layout.LabelsCSV) - imagesVal := fmt.Sprintf("%d files", len(layout.Images)) - if ext, _ := spec["spec"].(map[string]any); ext != nil { - if fo, _ := ext["file_options"].(map[string]any); fo != nil { - if e, _ := fo["extension"].(string); e != "" { - imagesVal = fmt.Sprintf("%d files (%s)", len(layout.Images), e) - } - } - } - p.Field("images", imagesVal) - if anns := layout.Sidecars["annotations"]; len(anns) > 0 { - p.Field("annotations", fmt.Sprintf("%d files", len(anns))) - } - if masks := layout.Sidecars["masks"]; len(masks) > 0 { - p.Field("masks", fmt.Sprintf("%d files", len(masks))) - } - } - p.Field("total size", push.HumanBytes(layout.TotalBytes)) - - p.Section("Ingest settings") - p.Field("name", fmt.Sprintf("%v", spec["table"])) - p.Field("task", fmt.Sprintf("%v", spec["category"])) - p.Field("intent", fmt.Sprintf("%v", spec["intent"])) - switch lbl := spec["label"].(type) { - case string: - p.Field("label column", lbl) - case map[string]any: - p.Field("label column", fmt.Sprintf("%v (policy: %v)", lbl["column"], lbl["policy"])) - } - if tc, ok := spec["time_column"].(string); ok && tc != "" { - p.Field("time column", tc) - } - p.Field("destination", push.FinalDestPrefix(spec["table"].(string))) -} - -// printClusterSummary shows the discovered workspace target. It's Kubernetes -// plumbing (release / jobs-manager / shared PVC) the happy path hides, so the -// whole block — header, fields, and the RWO-PVC note — prints only under -// --verbose (RFC-0002 §6). Discovery + guards are unchanged; this is -// presentation only. -func printClusterSummary(p *ui.Printer, release *cluster.ParentRelease, pvc *cluster.SharedPVC) { - if !p.Verbose() { - return - } - p.Section("Target cluster") - p.Detailf("release: %s (chart %s)", release.ReleaseName, release.ChartVersion) - p.Detailf("jobs-manager: %s", release.JobsManagerService) - p.Detailf("shared PVC: %s (%s)", pvc.ClaimName, pvc.Phase) - if !pvc.IsReadWriteMany() { - // Note but don't block — RWO clusters still work; the scheduler - // co-locates the stage Pod with the existing mounter. - p.Detailf("PVC is %v, not ReadWriteMany — the stage Pod will co-locate with the existing mounter", pvc.AccessModes) - } -} - -// pushJSONResult is the machine-readable shape emitted by --output-json. -// It's a presentation type owned by the CLI layer, so submit.Summary -// stays json-tag-free and this wire format can evolve independently. -type pushJSONResult struct { - Status string `json:"status"` // dry-run|succeeded|completed_with_failures|failed|detached|unknown|auth_error|submit_error|watch_error|error - Table string `json:"table"` - Category string `json:"category"` - Intent string `json:"intent"` - Namespace string `json:"namespace,omitempty"` - JobName string `json:"job_name,omitempty"` - Summary *pushJSONSummary `json:"summary,omitempty"` - Error string `json:"error,omitempty"` - ExitCode int `json:"exit_code,omitempty"` -} - -type pushJSONSummary struct { - IngestorID string `json:"ingestor_id,omitempty"` - TotalRecords int64 `json:"total_records"` - InsertedRecords int64 `json:"inserted_records"` - SentToAPI int64 `json:"sent_to_api"` - SkippedRecords int64 `json:"skipped_records"` - FileTransferFailures int64 `json:"file_transfer_failures"` - DBInsertFailures int64 `json:"db_insert_failures"` - SuccessRate float64 `json:"success_rate"` -} - -// writePushJSON serializes the push result to w (stdout in -// --output-json mode). Errors are dropped: marshaling our own struct -// can't fail in practice, and the exit code remains the contract. -func writePushJSON(w io.Writer, status string, spec map[string]any, s *submit.Summary, ns, jobName string) { - res := pushJSONResult{ - Status: status, - Table: fmt.Sprintf("%v", spec["table"]), - Category: fmt.Sprintf("%v", spec["category"]), - Intent: fmt.Sprintf("%v", spec["intent"]), - Namespace: ns, - JobName: jobName, - } - if s != nil { - res.Summary = &pushJSONSummary{ - IngestorID: s.IngestorID, - TotalRecords: s.TotalRecords, - InsertedRecords: s.InsertedRecords, - SentToAPI: s.APISentRecords, - SkippedRecords: s.SkippedRecords, - FileTransferFailures: s.FileTransferFailures, - DBInsertFailures: s.FailedRecords, - SuccessRate: s.SuccessRate(), - } - } - b, err := json.MarshalIndent(res, "", " ") - if err != nil { - return - } - _, _ = fmt.Fprintln(w, string(b)) -} - -// writePushErrorJSON emits a JSON error object for --output-json runs -// that fail before a result is produced (validation, discovery, -// staging, token, port-forward). Keeps the stdout-always-JSON contract -// so a script parsing it never hits empty output on failure. (Bugbot #49) -func writePushErrorJSON(w io.Writer, sp push.SpecArgs, e error, code int) { - res := pushJSONResult{ - Status: "error", - Table: sp.Table, - Category: sp.Category, - Intent: sp.Intent, - Error: e.Error(), - ExitCode: code, - } - b, err := json.MarshalIndent(res, "", " ") - if err != nil { - return - } - _, _ = fmt.Fprintln(w, string(b)) -} - -// listDatasetsFn is a test seam over push.ListDatasets. -var listDatasetsFn = push.ListDatasets - -// teardownFn is a test seam over push.Teardown (the destructive DROP TABLE + -// file removal). Production points at the real Teardown; a test overrides it to -// drive the clean and the partial-failure (table dropped, files remain → exit -// 7) paths without a cluster. -var teardownFn = push.Teardown - -// 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 -// duplicate check still backstops it — but the note tells the user the guard -// didn't run rather than silently skipping it. -// The first return is the EXISTING table's exact name ("" when absent): -// matching is case-insensitive (mysql's catalog may be), but any teardown -// must act on the real spelling — DROP/rm against the flag's casing would -// silently no-op on case-sensitive systems and then claim success. -func destTableExists(ctx context.Context, cs kubernetes.Interface, resolved *cluster.ResolvedConfig, table string) (string, string) { - names, err := listDatasetsFn(ctx, cs, resolved.RestConfig, resolved.Namespace) - if err != nil { - return "", fmt.Sprintf("(couldn't check whether %q already exists — continuing; the cluster still refuses duplicates: %v)", table, err) - } - for _, n := range names { - if strings.EqualFold(n, table) { - return n, "" - } - } - return "", "" -} - -// existingTableAction resolves what to do when the destination table -// already exists and --overwrite was NOT passed on the command line. -// -// - proceed=true, err=nil → replace it: the caller sets Overwrite and -// runs the same teardown `data delete` does. -// - proceed=false, err=nil → the user declined the replace prompt; a -// clean cancel (exit 0), nothing ingested. -// - err != nil (exit 6) → non-interactive and no --overwrite: refuse, -// same hard contract scripts have always had. -// -// Interactive mode prompts to replace UNLESS a --idempotency-key was -// reused: a reused key + a replace is the data-loss trap the top-of-func -// guard forbids (the teardown removes the data, then the cluster replays -// the old run and ingests nothing), so that combination falls through to -// the exit-6 refusal rather than being offered as a prompt. -func existingTableAction(a *runDataIngestArgs, existingTable string) (proceed bool, err error) { - if a.Interactive && a.Prompter != nil && a.IdempotencyKey == "" { - ok, perr := a.Prompter.Confirm(fmt.Sprintf( - "A dataset named %q already exists — replace it?", existingTable), false) - if perr != nil { - if errors.Is(perr, errInteractiveCancelled) { - return false, nil - } - return false, &exitError{code: 3, err: fmt.Errorf("overwrite prompt: %w", perr)} - } - return ok, nil - } - return false, &exitError{code: 6, err: fmt.Errorf( - "table %q already exists in this workspace. Re-ingesting the same table doesn't merge or replace — "+ - "the run would fail after uploading everything. Re-run with --overwrite to replace it, "+ - "or pick a different --name. (`tracebloc data delete %s` also removes it.)", - existingTable, existingTable)} -} - -// runLocalPreflight maps push.PreflightDataset — THE shared preview -// dispatch, also exercised verbatim by the parity harness — onto the CLI's -// conventions: notes print dim to errOut, a BadFlag problem exits 2 (fix a -// flag), anything else exits 3 (fix the data). -func runLocalPreflight(a runDataIngestArgs, layout *push.LocalLayout, errOut io.Writer) error { - notes, problem := push.PreflightDataset(a.Spec, layout) - for _, n := range notes { - _, _ = fmt.Fprintln(errOut, n) - } - if problem == nil { - return nil - } - code := 3 - if problem.BadFlag { - code = 2 - } - return &exitError{code: code, err: problem.Err} -} diff --git a/internal/cli/data_ingest_cluster.go b/internal/cli/data_ingest_cluster.go new file mode 100644 index 00000000..f7718dfe --- /dev/null +++ b/internal/cli/data_ingest_cluster.go @@ -0,0 +1,275 @@ +// The cluster-touching half of `data ingest`: the ingestion-run tail +// (mint token -> port-forward -> submit -> classify -> reclaim), the +// destination-table guard, and the test seams over the cluster steps. +// Moved verbatim from data.go (cli#282) — behavior unchanged. +package cli + +import ( + "context" + "errors" + "fmt" + "io" + "strings" + + "k8s.io/client-go/kubernetes" + + "github.com/tracebloc/cli/internal/cluster" + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/submit" + "github.com/tracebloc/cli/internal/ui" +) + +// 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 + // + log stream — can run that long for large ingestions. + // The chart's helm flow uses the same token-mint code path. + a.Printer.Step(3, 3, "Validate and load") + if a.Detach { + a.Printer.Hintf("Submitting the run — with --detach it keeps running on your workspace after this command returns; the reconnect command is shown below.") + } else { + 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 := mintIngestorTokenFn(ctx, cs, resolved.Namespace, + release.IngestorSAName, 3600, nil) + if err != nil { + return false, &exitError{code: 5, err: err} + } + + // 11. Open a port-forward to a Pod backing the jobs-manager + // Service. The CLI runs off-cluster (on a laptop, in CI + // runners outside the cluster network), so the discovered + // *.svc.cluster.local URL isn't reachable — we tunnel + // through the kubeconfig-authenticated apiserver, same as + // `kubectl port-forward`. Bugbot PR #10 r3 caught the + // original broken-by-design direct-URL POST. + // Opening the port-forward is a blocking wait (tunnel setup through the + // apiserver), so it runs under a spinner — no wait on the happy path stays + // silent (RFC-0002 "progress on every wait"). The submit POST itself is a + // separate ~30s synchronous wait; its spinner lives in submit.Run, next to + // the POST it covers. + connectSpin := a.Printer.Spinner("Connecting to your workspace to submit the run", "") + pf, err := portForwardJobsManagerFn(ctx, cs, resolved.RestConfig, + resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort) + connectSpin.Stop() + if err != nil { + return false, &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} + } + defer pf.Close() + + // 12. Phase 4: POST to jobs-manager via the local port, + // watch the spawned ingestor Job, render the parsed + // INGESTION SUMMARY panel. + // + // Exit-code mapping: + // SubmitError 401/403 → 5 (auth — same bucket as + // token-mint, shared + // "your SA can't do this" + // diagnostic class) + // SubmitError other 4xx/5xx → 8 (submit failed) + // WatchResult Failed → 9 (ingest failed) + // WatchResult Succeeded + + // summary.HasFailures() → 9 (some rows failed + // even though Job exited 0; + // the ingestor surfaces + // partial-failure summaries) + // WatchResult Detached → 0 (cluster keeps running) + // WatchResult Succeeded clean → 0 + localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort) + submitRes, err := submitRunFn(ctx, submit.Options{ + Submitter: submit.NewHTTPSubmitter(localEndpoint, tok.Token), + Client: cs, + IngestConfigYAML: string(specBytes), + IdempotencyKey: a.IdempotencyKey, + ImageDigest: a.ImageDigest, + Detach: a.Detach, + Out: out, + Printer: a.Printer, + }) + // Classify once: a machine-readable status + the process exit error + // in lockstep, so --output-json emits exactly one result object on + // EVERY path (success / partial / failure / submit-or-watch error) + // whose status matches the exit code. (Bugbot #38.) + status, exitErr := classifyPushOutcome(submitRes, err) + + // Emit the machine-readable result BEFORE the best-effort staging + // reclaim below, so a scripted --output-json consumer gets its result + // object at ingest-completion latency and never waits on a slow + // cluster-side cleanup that has no bearing on the ingest outcome. + if a.OutputJSON { + var summary *submit.Summary + var ns, jobName string + if submitRes != nil { + if submitRes.Watch != nil { + summary = submitRes.Watch.Summary + } + if submitRes.Submit != nil { + ns, jobName = submitRes.Submit.Namespace, submitRes.Submit.JobName + } + } + writePushJSON(a.JSONOut, status, spec, summary, ns, jobName) + jsonEmitted = true + } + + // 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) { + reclaimSpin := a.Printer.Spinner("Reclaiming the temporary copy", "") + cerr := cleanStagingFn(ctx, cs, + &push.SPDYExecutor{Config: resolved.RestConfig, Client: cs}, + resolved.Namespace, a.Spec.Table, push.PodSpecOptions{ + Namespace: resolved.Namespace, + PVCClaimName: pvc.ClaimName, + PVCMountPath: pvc.MountPath, + Table: a.Spec.Table, + ServiceAccountName: release.IngestorSAName, + Image: a.StagePodImage, + }) + reclaimSpin.Stop() + if cerr != nil { + a.Printer.Warnf("Couldn't reclaim the temporary copy (%v). It's harmless — the next re-ingest of %q or a `tracebloc data delete %s` will clear it.", + cerr, a.Spec.Table, a.Spec.Table) + } + } + + if exitErr != nil { + return jsonEmitted, exitErr + } + 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" +} + +// printClusterSummary shows the discovered workspace target. It's Kubernetes +// plumbing (release / jobs-manager / shared PVC) the happy path hides, so the +// whole block — header, fields, and the RWO-PVC note — prints only under +// --verbose (RFC-0002 §6). Discovery + guards are unchanged; this is +// presentation only. +func printClusterSummary(p *ui.Printer, release *cluster.ParentRelease, pvc *cluster.SharedPVC) { + if !p.Verbose() { + return + } + p.Section("Target cluster") + p.Detailf("release: %s (chart %s)", release.ReleaseName, release.ChartVersion) + p.Detailf("jobs-manager: %s", release.JobsManagerService) + p.Detailf("shared PVC: %s (%s)", pvc.ClaimName, pvc.Phase) + if !pvc.IsReadWriteMany() { + // Note but don't block — RWO clusters still work; the scheduler + // co-locates the stage Pod with the existing mounter. + p.Detailf("PVC is %v, not ReadWriteMany — the stage Pod will co-locate with the existing mounter", pvc.AccessModes) + } +} + +// listDatasetsFn is a test seam over push.ListDatasets. +var listDatasetsFn = push.ListDatasets + +// teardownFn is a test seam over push.Teardown (the destructive DROP TABLE + +// file removal). Production points at the real Teardown; a test overrides it to +// drive the clean and the partial-failure (table dropped, files remain → exit +// 7) paths without a cluster. +var teardownFn = push.Teardown + +// 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 +// duplicate check still backstops it — but the note tells the user the guard +// didn't run rather than silently skipping it. +// The first return is the EXISTING table's exact name ("" when absent): +// matching is case-insensitive (mysql's catalog may be), but any teardown +// must act on the real spelling — DROP/rm against the flag's casing would +// silently no-op on case-sensitive systems and then claim success. +func destTableExists(ctx context.Context, cs kubernetes.Interface, resolved *cluster.ResolvedConfig, table string) (string, string) { + names, err := listDatasetsFn(ctx, cs, resolved.RestConfig, resolved.Namespace) + if err != nil { + return "", fmt.Sprintf("(couldn't check whether %q already exists — continuing; the cluster still refuses duplicates: %v)", table, err) + } + for _, n := range names { + if strings.EqualFold(n, table) { + return n, "" + } + } + return "", "" +} + +// existingTableAction resolves what to do when the destination table +// already exists and --overwrite was NOT passed on the command line. +// +// - proceed=true, err=nil → replace it: the caller sets Overwrite and +// runs the same teardown `data delete` does. +// - proceed=false, err=nil → the user declined the replace prompt; a +// clean cancel (exit 0), nothing ingested. +// - err != nil (exit 6) → non-interactive and no --overwrite: refuse, +// same hard contract scripts have always had. +// +// Interactive mode prompts to replace UNLESS a --idempotency-key was +// reused: a reused key + a replace is the data-loss trap the top-of-func +// guard forbids (the teardown removes the data, then the cluster replays +// the old run and ingests nothing), so that combination falls through to +// the exit-6 refusal rather than being offered as a prompt. +func existingTableAction(a *runDataIngestArgs, existingTable string) (proceed bool, err error) { + if a.Interactive && a.Prompter != nil && a.IdempotencyKey == "" { + ok, perr := a.Prompter.Confirm(fmt.Sprintf( + "A dataset named %q already exists — replace it?", existingTable), false) + if perr != nil { + if errors.Is(perr, errInteractiveCancelled) { + return false, nil + } + return false, &exitError{code: 3, err: fmt.Errorf("overwrite prompt: %w", perr)} + } + return ok, nil + } + return false, &exitError{code: 6, err: fmt.Errorf( + "table %q already exists in this workspace. Re-ingesting the same table doesn't merge or replace — "+ + "the run would fail after uploading everything. Re-run with --overwrite to replace it, "+ + "or pick a different --name. (`tracebloc data delete %s` also removes it.)", + existingTable, existingTable)} +} diff --git a/internal/cli/data_ingest_cmd.go b/internal/cli/data_ingest_cmd.go new file mode 100644 index 00000000..fd98cf04 --- /dev/null +++ b/internal/cli/data_ingest_cmd.go @@ -0,0 +1,367 @@ +// The `data ingest` command surface: cobra wiring, the full flag set +// (canonical + hidden deprecated aliases), and the args struct the +// testable runDataIngest body consumes. Moved verbatim from data.go +// (cli#282) — behavior unchanged. +package cli + +import ( + "fmt" + "io" + + "github.com/spf13/cobra" + + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/ui" +) + +// newDataIngestCmd implements `tracebloc data ingest `. +// +// Phase 3 scope (now complete across PR-a + PR-b): +// +// - Synthesize the ingest spec from flags (`internal/push.SpecArgs.Build`) +// - Validate it against the embedded ingest.v1 schema +// - Walk the local directory + enforce v0.1 size caps +// - Discover cluster, parent release, and shared PVC +// - Print a single-screen pre-flight summary +// - Either --dry-run stop, OR create an ephemeral stage Pod +// (alpine 3.20 pinned by digest, PSA-restricted security +// context), tar local files into it via an SPDY exec stream +// with a progress bar, then defer-delete the Pod +// +// Phase 4 (`tracebloc/client#152`) hooks the submit-to-jobs-manager +// step into the bottom of this command, replacing the "manually +// kick off helm ingestor" workaround in the success message. +// +// Aliases: "push" is kept for one deprecation cycle so existing +// scripts continue to work. +func newDataIngestCmd() *cobra.Command { + var ( + // Kubeconfig flags — same conventions as `cluster info`. + // Promoting these to persistent on the root is a v0.2 + // follow-up (tracebloc/cli#3); for now they live on each + // command that needs them. + kubeconfigPath string + contextOverride string + nsOverride string + + // Ingest-spec flags. All schema task categories are CLI-supported now + // (image classification / detection / segmentation / keypoint, the full + // text family, and the tabular / time-series family). + // + // --name/--task are the canonical flags (#180); --table/--category + // stay on as hidden deprecated aliases so existing scripts keep + // working. --intent is unchanged. The wire/spec field names don't + // change — this is a CLI-surface rename only. + name string + tableAlias string + task string + categoryAlias string + intent string + labelColumn string + targetSize string + minSize string + schemaFlag string + labelPolicy string + timeColumn string + numberOfKeypoints int + + // Operations flags. + dryRun bool + overwrite bool + noInput bool + outputJSON bool + + // Stage Pod image override. Defaults to the digest-pinned + // alpine that ships with the CLI; air-gapped customers + // override this to an image their registry mirror serves. + // Pin by digest in your override too — tag-only references + // drift silently and break "all my pushes worked yesterday." + stagePodImage string + + // Phase 4 flags. --detach exits immediately after the 201 + // from jobs-manager; --idempotency-key plumbs through to + // the submit body for retry-safety across CLI invocations + // (default: fresh per call); --image-digest pins the + // ingestor image (default: jobs-manager picks the + // cluster-configured one). + detach bool + idempotencyKey string + imageDigest string + ) + + cmd := &cobra.Command{ + Use: "ingest ", + Aliases: []string{"push"}, + Short: "Ingest a local dataset into your workspace", + // The task COUNT and the text-family subdir names are derived from the + // registry / vendored layout contract (push.SupportedCategoryIDs + + // push.TextSidecarDir) rather than hardcoded, so the help can't drift + // from what the CLI actually supports (cli#215): the count used to read + // a stale "9", and the text example showed texts/ for every text task + // even though masked_language_modeling stages into sequences/. + Long: fmt.Sprintf(`Ingests a local dataset into your workspace's storage, +submits the ingestion run, and follows it to completion (streaming +progress + the final summary). Your data never leaves your own +infrastructure. Supports %[1]d tasks across the image, text, and +tabular / time-series families; pick one with --task. + + is the data itself. What it looks like depends on the task: + + tabular / time-series — the dataset is a single CSV. Pass the .csv + file directly, or a folder holding exactly one .csv: + + churn.csv (the .csv file itself) + or + churn/ + data.csv (the one .csv in the folder) + + image (classification, object/keypoint detection) — a folder with + labels.csv + an images/ subfolder: + + cats_dogs/ + labels.csv (required) + images/ (required) + 001.jpg + ... + + text (classification, masked language modeling) — a folder with + labels.csv + a %[2]s/ subfolder (masked language modeling uses %[3]s/): + + reviews/ + labels.csv (required) + %[2]s/ (required — %[3]s/ for masked language modeling) + 001.txt + ... + +A bare .csv file is accepted only for the tabular / time-series family; +image and text datasets must be a folder. + +Accepted image extensions: .jpg, .jpeg, or .png (case-insensitive). +All images in one dataset must share a single type — the cluster +validates the type it was told to expect. + +v0.1 caps the dataset at 1 GiB total + 500 MiB per file. Larger +datasets need the v0.2 cloud-source story (S3/GCS/HTTPS sources) — +see tracebloc/client#147 non-goals. + +Exit codes: + 0 files staged + ingested successfully (or --detach: just staged + submitted) + 2 schema validation failed (synthesized spec rejected) or + v0.1-unsupported task passed + 3 local-layout or kubeconfig error + 4 cluster reachable but no tracebloc client / shared storage missing + 5 ingestor SA token couldn't be obtained, or jobs-manager + rejected the token (401/403) + 6 destination table already exists (re-run with --overwrite to + replace it, or pick a different --name) + 7 pre-flight succeeded but staging the files failed + (Pod creation, image pull, exec stream, or remote tar error) — + or, with --overwrite, removing the old table failed + 8 jobs-manager rejected the submit (4xx/5xx other than auth) + 9 ingestion Job exited non-zero, or completed with row-level + failures the summary panel reports`, + len(push.SupportedCategoryIDs()), + push.TextSidecarDir("text_classification"), + push.TextSidecarDir("masked_language_modeling")), + Args: cobra.MaximumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var localPath string + if len(args) > 0 { + localPath = args[0] + } + // Resolve the deprecated flag aliases (#180): the canonical + // flag wins; a hidden legacy alias fills in only when the new + // flag wasn't passed, so old scripts keep working without the + // new surface silently shadowing them. The wire/spec field + // names are unchanged — this is a CLI rename only. + nameVal := name + if cmd.Flags().Changed("table") && !cmd.Flags().Changed("name") { + nameVal = tableAlias + } + taskVal := task + if cmd.Flags().Changed("category") && !cmd.Flags().Changed("task") { + taskVal = categoryAlias + } + // Whether the task was chosen at all (via either spelling). + // Dropping --task's old image_classification default means an + // unset task now drives the picker (TTY) or a clear error + // (non-interactive), never a silent image assumption. + taskSet := cmd.Flags().Changed("task") || cmd.Flags().Changed("category") + // Record whether --number-of-keypoints was explicitly passed, so + // the keypoint set-vs-unset message (#76b) can distinguish an + // explicit zero value from an unset flag (both look like the Go + // zero value in the spec). + changedFlags := map[string]bool{} + if cmd.Flags().Changed("number-of-keypoints") { + changedFlags["number-of-keypoints"] = true + } + // Guided mode: on a terminal (and unless --no-input), prompt + // for whatever's still missing. Off a TTY / with --no-input, + // prompter stays nil and runDataIngest keeps flag-only + // behavior. + interactive := !noInput && !outputJSON && isInteractiveTTY() + var pr prompter + if interactive { + pr = surveyPrompter{} + } + // In --output-json mode, human output goes to stderr so + // stdout carries only the JSON result. + humanOut := cmd.OutOrStdout() + printer := printerFor(cmd) + var jsonOut io.Writer + if outputJSON { + humanOut = cmd.ErrOrStderr() + printer = printerForWriter(cmd, cmd.ErrOrStderr()) + jsonOut = cmd.OutOrStdout() + } + return runDataIngest(cmd.Context(), humanOut, cmd.ErrOrStderr(), + runDataIngestArgs{ + LocalPath: localPath, + Kubeconfig: kubeconfigPath, + Context: contextOverride, + Namespace: nsOverride, + Spec: push.SpecArgs{ + Table: nameVal, Category: taskVal, Intent: intent, + LabelColumn: labelColumn, LabelPolicy: labelPolicy, TimeColumn: timeColumn, + NumberOfKeypoints: numberOfKeypoints, + }, + TargetSizeFlag: targetSize, + MinSizeFlag: minSize, + SchemaFlag: schemaFlag, + DryRun: dryRun, + Overwrite: overwrite, + StagePodImage: stagePodImage, + Detach: detach, + IdempotencyKey: idempotencyKey, + ImageDigest: imageDigest, + Printer: printer, + Interactive: interactive, + Prompter: pr, + TaskSet: taskSet, + ChangedFlags: changedFlags, + OutputJSON: outputJSON, + JSONOut: jsonOut, + }) + }, + } + + addKubeconfigFlags(cmd, &kubeconfigPath, &contextOverride, kubeconfigFlagUsage, contextFlagUsage) + addNamespaceFlag(cmd, &nsOverride, namespaceFlagUsage) + + // Required spec flags. We DON'T mark them required-at-cobra-level + // because cobra's "required flag" error message is terse and + // pre-empts our richer schema-driven diagnostic. Instead, the + // schema validator catches missing/empty values with the canonical + // JSON-pointer-anchored error. + cmd.Flags().StringVar(&name, "name", "", + "a name for this dataset — start with a letter or underscore, then letters/digits/underscores — you'll reference it by this name when you start a training run") + cmd.Flags().StringVar(&tableAlias, "table", "", + "deprecated alias for --name") + _ = cmd.Flags().MarkHidden("table") + cmd.Flags().StringVar(&task, "task", "", + "the task this data is for, one of: "+push.SupportedCategoriesList()+ + ". Omit it on a terminal to pick interactively.") + cmd.Flags().StringVar(&categoryAlias, "category", "", + "deprecated alias for --task") + _ = cmd.Flags().MarkHidden("category") + cmd.Flags().StringVar(&intent, "intent", "", + "is this training or test data? train|test (default train)") + cmd.Flags().StringVar(&labelColumn, "label-column", "", + "name of the label/target column (in labels.csv for image tasks, in the data CSV for tabular)") + cmd.Flags().StringVar(&targetSize, "target-size", "", + "image tasks only: the resolution your images already are, as WxH (e.g. 512x512). tracebloc never "+ + "resizes — it checks every image is exactly this size and rejects any that differ. Default: "+ + "read from your first image.") + cmd.Flags().StringVar(&minSize, "min-size", "", + "image tasks only: reject images smaller than WxH before the ingest (e.g. 64x64). Set it to the "+ + "smallest size your model can train on — raise or lower it freely. Default: unset (no local "+ + "size check).") + cmd.Flags().StringVar(&schemaFlag, "schema", "", + "tabular/time-series only: column types as col:TYPE,col:TYPE (e.g. age:INT,price:FLOAT). "+ + "Default: inferred from the CSV (INT/BIGINT/FLOAT/BOOLEAN/DATE/DATETIME/VARCHAR(n)).") + cmd.Flags().StringVar(&labelPolicy, "label-policy", "", + "regression-class only (tabular_regression, time_series_forecasting, time_to_event_prediction): "+ + "passthrough|bucket (default bucket — bins the target so the raw value never leaves the cluster)") + cmd.Flags().StringVar(&timeColumn, "time-column", "", + "time_to_event_prediction only: name of the time/duration column (default: a column named \"time\")") + cmd.Flags().IntVar(&numberOfKeypoints, "number-of-keypoints", 0, + "keypoint_detection only: number of keypoints per sample (required; e.g. 17 for COCO pose)") + + cmd.Flags().BoolVar(&overwrite, "overwrite", false, + "replace the destination table if it already exists: its current table + files are removed first (same as `tracebloc data delete`), then the new data is ingested. Not combinable with --idempotency-key") + cmd.Flags().BoolVar(&dryRun, "dry-run", false, + "validate + discover + walk, but don't create any cluster resources") + cmd.Flags().BoolVar(&noInput, "no-input", false, + "disable interactive prompts; fail on missing required values (for CI/scripts)") + cmd.Flags().BoolVar(&outputJSON, "output-json", false, + "emit a machine-readable JSON result on stdout (human output → stderr; implies --no-input)") + cmd.Flags().StringVar(&stagePodImage, "stage-pod-image", "", + "override the ephemeral stage Pod's image (default: digest-pinned alpine 3.20 baked into the CLI). "+ + "Pin by digest in your override too — tag-only refs drift silently.") + + cmd.Flags().BoolVar(&detach, "detach", false, + "exit immediately after jobs-manager accepts the run (no log streaming, no summary panel). "+ + "Use for CI scenarios; reconnect later with `kubectl logs -f -n job/`.") + cmd.Flags().StringVar(&idempotencyKey, "idempotency-key", "", + "reuse this idempotency key across retry attempts (default: fresh per invocation). "+ + "jobs-manager treats a duplicate key as a replay and attaches to the existing Job "+ + "rather than spawning a new one — useful for at-most-once-across-attempts semantics.") + cmd.Flags().StringVar(&imageDigest, "image-digest", "", + "pin the ingestor container image to a specific digest (default: jobs-manager picks the "+ + "cluster-configured `images.ingestor.digest`). Format: sha256:.") + + return cmd +} + +// runDataIngestArgs collects every parameter runDataIngest needs, +// so the body stays testable without going through cobra. The cobra +// RunE wrapper above is the ONLY caller in production; tests +// construct one of these directly. +type runDataIngestArgs struct { + LocalPath string + Kubeconfig string + Context string + Namespace string + Spec push.SpecArgs + TargetSizeFlag string // raw --target-size; resolved after Discover (image) + MinSizeFlag string // raw --min-size; resolved after Discover (image) — #348 floor override + SchemaFlag string // raw --schema; resolved or inferred after Discover (tabular) + DryRun bool + Overwrite bool + StagePodImage string + + // Printer renders the pre-flight summary + status output. Built in + // the RunE from the persistent --plain flag (see printerFor). + Printer *ui.Printer + + // Interactive guided mode (#28). When Interactive is true, + // runDataIngest prompts (via Prompter) for any missing core inputs + // before validation. TaskSet records whether the task was passed + // explicitly (via --task or the hidden --category alias); an unset + // task drives the picker rather than assuming a default. Prompter is + // nil off a TTY / --no-input. + Interactive bool + Prompter prompter + TaskSet bool + + // ChangedFlags records which CLI flags were EXPLICITLY set + // (cmd.Flags().Changed), decoupling "was it passed" from "is its value + // non-zero" — the value alone can't tell `--number-of-keypoints 0` (an + // explicit, invalid value) from an unset flag (also 0). The RunE + // populates it for --number-of-keypoints; the keypoint set-vs-unset + // diagnostic (#76b) reads it. Nil in direct-construction tests that + // don't exercise that path. + ChangedFlags map[string]bool + + // OutputJSON routes human output to stderr and emits a JSON result + // to JSONOut (stdout); set together by the RunE in --output-json + // mode (which also forces non-interactive). + OutputJSON bool + JSONOut io.Writer + + // Phase 4 (#152) fields. See the flag declarations for the + // per-knob rationale; all three are optional. + Detach bool + IdempotencyKey string + ImageDigest string +} diff --git a/internal/cli/data_ingest_local.go b/internal/cli/data_ingest_local.go new file mode 100644 index 00000000..7a382ac3 --- /dev/null +++ b/internal/cli/data_ingest_local.go @@ -0,0 +1,129 @@ +// The local, cluster-free half of `data ingest`: path expansion + the +// path-existence-first guard, the local dataset summary, and the +// preflight that previews the ingestor's validators on the local data. +// Moved verbatim from data.go (cli#282) — behavior unchanged. +package cli + +import ( + "errors" + "fmt" + "io" + "os" + "sort" + + "github.com/tracebloc/cli/internal/pathutil" + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/ui" +) + +// sortedKeys returns m's keys in sorted order — used to list a CSV's inferred +// columns in the friendly missing-label message (#214) deterministically. +func sortedKeys(m map[string]string) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +// expandHome expands a leading ~ (current user or ~user) to a home +// directory, leaving every other path untouched. It's the CLI-local +// name for the shared pathutil.ExpandHome; cluster.expandPath resolves +// to the same helper, so ~-expansion is identical across subcommands +// (a --kubeconfig ~alice/... resolves alice's home just like a data +// ingest path does). See pathutil.ExpandHome for the full contract. (#181) +func expandHome(path string) string { + return pathutil.ExpandHome(path) +} + +// statDatasetPath is the "path existence FIRST" guard (#181): a typo'd +// path fails plainly on the path — a clean "no such file or directory" — +// before any family sniff, label preview, or schema work touches it. +// Both entry points call it: the flag-only path from runDataIngest's 0b +// step, and the guided path from runInteractive (before the family sniff), +// so the invariant holds on every route rather than only the flag path. +func statDatasetPath(path string) error { + if _, serr := os.Stat(path); serr != nil { + if errors.Is(serr, os.ErrNotExist) { + return &exitError{code: 3, err: fmt.Errorf( + "no such file or directory: %q — check the path to your dataset", path)} + } + return &exitError{code: 3, err: fmt.Errorf( + "can't read %q: %w", path, serr)} + } + return nil +} + +// printLocalSummary shows what the CLI found on disk plus the ingest +// settings it assembled — the detail under step 1 ("Check your data"). +// Mirrors `cluster info`'s section/Field layout. +func printLocalSummary(p *ui.Printer, layout *push.LocalLayout, spec map[string]any) { + cat, _ := spec["category"].(string) + + p.Section("Local dataset") + p.Field("root", layout.Root) + switch { + case push.IsTabular(cat): + p.Field("data CSV", layout.LabelsCSV) + if sch, ok := spec["schema"].(map[string]string); ok { + p.Field("columns", fmt.Sprintf("%d", len(sch))) + } + case push.IsText(cat): + dir := push.TextSidecarDir(cat) + p.Field("labels.csv", layout.LabelsCSV) + p.Field(dir, fmt.Sprintf("%d files", len(layout.Sidecars[dir]))) + default: + p.Field("labels.csv", layout.LabelsCSV) + imagesVal := fmt.Sprintf("%d files", len(layout.Images)) + if ext, _ := spec["spec"].(map[string]any); ext != nil { + if fo, _ := ext["file_options"].(map[string]any); fo != nil { + if e, _ := fo["extension"].(string); e != "" { + imagesVal = fmt.Sprintf("%d files (%s)", len(layout.Images), e) + } + } + } + p.Field("images", imagesVal) + if anns := layout.Sidecars["annotations"]; len(anns) > 0 { + p.Field("annotations", fmt.Sprintf("%d files", len(anns))) + } + if masks := layout.Sidecars["masks"]; len(masks) > 0 { + p.Field("masks", fmt.Sprintf("%d files", len(masks))) + } + } + p.Field("total size", push.HumanBytes(layout.TotalBytes)) + + p.Section("Ingest settings") + p.Field("name", fmt.Sprintf("%v", spec["table"])) + p.Field("task", fmt.Sprintf("%v", spec["category"])) + p.Field("intent", fmt.Sprintf("%v", spec["intent"])) + switch lbl := spec["label"].(type) { + case string: + p.Field("label column", lbl) + case map[string]any: + p.Field("label column", fmt.Sprintf("%v (policy: %v)", lbl["column"], lbl["policy"])) + } + if tc, ok := spec["time_column"].(string); ok && tc != "" { + p.Field("time column", tc) + } + p.Field("destination", push.FinalDestPrefix(spec["table"].(string))) +} + +// runLocalPreflight maps push.PreflightDataset — THE shared preview +// dispatch, also exercised verbatim by the parity harness — onto the CLI's +// conventions: notes print dim to errOut, a BadFlag problem exits 2 (fix a +// flag), anything else exits 3 (fix the data). +func runLocalPreflight(a runDataIngestArgs, layout *push.LocalLayout, errOut io.Writer) error { + notes, problem := push.PreflightDataset(a.Spec, layout) + for _, n := range notes { + _, _ = fmt.Fprintln(errOut, n) + } + if problem == nil { + return nil + } + code := 3 + if problem.BadFlag { + code = 2 + } + return &exitError{code: code, err: problem.Err} +} diff --git a/internal/cli/data_ingest_output.go b/internal/cli/data_ingest_output.go new file mode 100644 index 00000000..03c0a234 --- /dev/null +++ b/internal/cli/data_ingest_output.go @@ -0,0 +1,136 @@ +// The machine-readable output surface of `data ingest`: the --output-json +// wire types + writers, and classifyPushOutcome, which keeps the JSON +// status string and the process exit code in lockstep (kept in one file +// with writePushJSON deliberately — they are mutation-hardened together). +// Moved verbatim from data.go (cli#282) — behavior unchanged. +package cli + +import ( + "encoding/json" + "errors" + "fmt" + "io" + + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/submit" +) + +// classifyPushOutcome maps the result of submit.Run to a machine- +// readable status string + the process exit error, kept in lockstep so +// --output-json's status always agrees with the exit code (a nil +// *exitError = success, exit 0). It also covers the error paths +// (auth/submit/watch) so --output-json can still emit a result object +// when submit.Run returns an error. (Bugbot #38.) +func classifyPushOutcome(res *submit.Result, err error) (string, *exitError) { + if err != nil { + switch { + case submit.IsAuthError(err): + return "auth_error", &exitError{code: 5, err: err} + case submit.IsWatchError(err): + // jobs-manager accepted the run; the cluster is doing the + // work, the CLI just couldn't follow along — ingest-side + // (exit 9), not submit-side (8). + return "watch_error", &exitError{code: 9, err: err} + default: + return "submit_error", &exitError{code: 8, err: err} + } + } + // --detach (no watch) or SIGINT-mid-watch: success; cluster runs on. + if res == nil || res.Watch == nil || res.Watch.Outcome == submit.JobOutcomeDetached { + return "detached", nil + } + switch res.Watch.Outcome { + case submit.JobOutcomeFailed: + return "failed", &exitError{code: 9, err: errors.New("ingestion Job exited non-zero — see logs above")} + case submit.JobOutcomeUnknown: + return "unknown", &exitError{code: 9, err: errors.New( + "ingestion Job's final status couldn't be determined within the watch window — " + + "check `kubectl get job -n " + res.Submit.Namespace + " " + res.Submit.JobName + "` for the outcome")} + case submit.JobOutcomeSucceeded: + // Job exited 0, but rows can still have failed — exit 9, and the + // JSON status must say so, NOT "succeeded". (Bugbot #38.) + if res.Watch.Summary != nil && res.Watch.Summary.HasFailures() { + return "completed_with_failures", &exitError{code: 9, err: errors.New( + "ingestion Job completed but the summary reports failures — see panel above")} + } + return "succeeded", nil + } + return "unknown", nil +} + +// pushJSONResult is the machine-readable shape emitted by --output-json. +// It's a presentation type owned by the CLI layer, so submit.Summary +// stays json-tag-free and this wire format can evolve independently. +type pushJSONResult struct { + Status string `json:"status"` // dry-run|succeeded|completed_with_failures|failed|detached|unknown|auth_error|submit_error|watch_error|error + Table string `json:"table"` + Category string `json:"category"` + Intent string `json:"intent"` + Namespace string `json:"namespace,omitempty"` + JobName string `json:"job_name,omitempty"` + Summary *pushJSONSummary `json:"summary,omitempty"` + Error string `json:"error,omitempty"` + ExitCode int `json:"exit_code,omitempty"` +} + +type pushJSONSummary struct { + IngestorID string `json:"ingestor_id,omitempty"` + TotalRecords int64 `json:"total_records"` + InsertedRecords int64 `json:"inserted_records"` + SentToAPI int64 `json:"sent_to_api"` + SkippedRecords int64 `json:"skipped_records"` + FileTransferFailures int64 `json:"file_transfer_failures"` + DBInsertFailures int64 `json:"db_insert_failures"` + SuccessRate float64 `json:"success_rate"` +} + +// writePushJSON serializes the push result to w (stdout in +// --output-json mode). Errors are dropped: marshaling our own struct +// can't fail in practice, and the exit code remains the contract. +func writePushJSON(w io.Writer, status string, spec map[string]any, s *submit.Summary, ns, jobName string) { + res := pushJSONResult{ + Status: status, + Table: fmt.Sprintf("%v", spec["table"]), + Category: fmt.Sprintf("%v", spec["category"]), + Intent: fmt.Sprintf("%v", spec["intent"]), + Namespace: ns, + JobName: jobName, + } + if s != nil { + res.Summary = &pushJSONSummary{ + IngestorID: s.IngestorID, + TotalRecords: s.TotalRecords, + InsertedRecords: s.InsertedRecords, + SentToAPI: s.APISentRecords, + SkippedRecords: s.SkippedRecords, + FileTransferFailures: s.FileTransferFailures, + DBInsertFailures: s.FailedRecords, + SuccessRate: s.SuccessRate(), + } + } + b, err := json.MarshalIndent(res, "", " ") + if err != nil { + return + } + _, _ = fmt.Fprintln(w, string(b)) +} + +// writePushErrorJSON emits a JSON error object for --output-json runs +// that fail before a result is produced (validation, discovery, +// staging, token, port-forward). Keeps the stdout-always-JSON contract +// so a script parsing it never hits empty output on failure. (Bugbot #49) +func writePushErrorJSON(w io.Writer, sp push.SpecArgs, e error, code int) { + res := pushJSONResult{ + Status: "error", + Table: sp.Table, + Category: sp.Category, + Intent: sp.Intent, + Error: e.Error(), + ExitCode: code, + } + b, err := json.MarshalIndent(res, "", " ") + if err != nil { + return + } + _, _ = fmt.Fprintln(w, string(b)) +}