From 7bba5b086b2f8ac01a6c0409cc60eb4bfade6943 Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Thu, 21 May 2026 23:59:15 +0500 Subject: [PATCH 1/5] feat(dataset): pre-flight for `dataset push` (PR-a of #151) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 (tracebloc/client#151) lands in two PRs to keep diffs reviewable. PR-a (this one) is the no-op-safe path: synthesize the ingest spec from flags, validate against the embedded ingest.v1 schema, walk the local directory, discover the cluster's parent release + shared PVC. Print a summary, then either stop (--dry-run) or fail cleanly with "wait for PR-b" so a customer who pulled the PR-a binary doesn't see "0 files transferred" confusion. PR-b adds the novel-engineering piece: ephemeral stage Pod (alpine 3.20 pinned by digest), client-go remotecommand SPDY executor + tar stream, schollz/progressbar/v3, SIGINT-safe cleanup. It plugs into the "TODO: PR-b" branch at the end of runDatasetPush() without touching anything upstream. Surface area added: internal/push/ (new package) spec.go SpecArgs -> ingest.v1.json-conforming map. Path is leave-validation-to-schema; the schema is the single source of truth. walk.go Discover() walks /labels.csv + /images/, enforces v0.1 size caps (1 GiB total, 500 MiB per file), accepts {.jpg,.jpeg,.png,.webp} case-insensitively. internal/cluster/pvc.go (new file) DiscoverSharedPVC reads the chart's `client-pvc` (hardcoded by _helpers.tpl, not parameterized) in the namespace, verifies it is Bound, returns access modes. IsReadWriteMany() drives the stage-Pod scheduling warning in PR-b. internal/cli/dataset.go (new file) `tracebloc dataset push ` command. Order: 1. flag->spec synthesize + schema-validate (stderr diagnostic, exit 2 — same wording style as ingest validate) 2. walk local layout + size caps (exit 3) 3. load kubeconfig + clientset (exit 3) 4. discover parent release (exit 4) 5. discover shared PVC (exit 4) 6. print pre-flight summary 7. --dry-run stop, or exit 6 "wait for PR-b" internal/cli/root.go (1-line wire-up) Register the new command. Why image_classification only: the epic (tracebloc/client#147) v0.1 non-goals defer other categories to v0.2 as one-PR additions. The --category flag does NOT pre-validate so the schema's enum check produces the canonical error (rather than CLI-side drift). Why exit code 6 for "not implemented yet": distinct from the existing schema (2), local (3), discovery (4), and Phase-2's token-mint (5) codes. Tests assert the exact code so the contract sticks across PR-b's refactor. Tests: spec_test.go Build() ↔ embedded schema contract (the core "this must produce something the validator accepts" pin) walk_test.go Layout + size caps, all happy + sad paths (case-insensitive ext, skip .DS_Store, missing files, byte-sum sanity) pvc_test.go Fake-clientset coverage: happy path, not-found diagnostic, Pending-phase diagnostic, mixed-mode IsReadWriteMany dataset_test.go CLI integration: schema-fail exit 2, walk-fail exit 3, kubeconfig-fail exit 3, cobra args check Locally: vet, test -race -cover, gofmt -s, errcheck — all green. Coverage: push 81.0%, cluster 83.2%, schema 80.7%, cli 52.0%. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 361 +++++++++++++++++++++++++++++++++++ internal/cli/dataset_test.go | 225 ++++++++++++++++++++++ internal/cli/root.go | 1 + internal/cluster/pvc.go | 146 ++++++++++++++ internal/cluster/pvc_test.go | 117 ++++++++++++ internal/push/spec.go | 111 +++++++++++ internal/push/spec_test.go | 107 +++++++++++ internal/push/walk.go | 227 ++++++++++++++++++++++ internal/push/walk_test.go | 221 +++++++++++++++++++++ 9 files changed, 1516 insertions(+) create mode 100644 internal/cli/dataset.go create mode 100644 internal/cli/dataset_test.go create mode 100644 internal/cluster/pvc.go create mode 100644 internal/cluster/pvc_test.go create mode 100644 internal/push/spec.go create mode 100644 internal/push/spec_test.go create mode 100644 internal/push/walk.go create mode 100644 internal/push/walk_test.go diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go new file mode 100644 index 00000000..4795daec --- /dev/null +++ b/internal/cli/dataset.go @@ -0,0 +1,361 @@ +package cli + +import ( + "context" + "errors" + "fmt" + "io" + + "github.com/spf13/cobra" + "gopkg.in/yaml.v3" + + "github.com/tracebloc/cli/internal/cluster" + "github.com/tracebloc/cli/internal/push" + "github.com/tracebloc/cli/internal/schema" +) + +// newDatasetCmd wires the `tracebloc dataset` subtree. The dominant +// verb is `push`, introduced in Phase 3 (tracebloc/client#151) and +// landing across two PRs: PR-a (this one) implements the +// no-op-safe pre-flight; PR-b adds the actual file streaming via +// an ephemeral Pod + tar-over-exec. Future verbs (`dataset list`, +// `dataset rm`) hang off this parent in v0.2. +func newDatasetCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "dataset", + Short: "Manage datasets in the parent client release", + Long: `Commands for staging and managing datasets on the cluster's +shared PVC. + +Today: ` + "`dataset push`" + ` stages a local directory and (in PR-b) +submits an ingestion run. ` + "`tracebloc cluster info`" + ` is the +pre-flight you'd typically run before the first push.`, + } + cmd.AddCommand(newDatasetPushCmd()) + return cmd +} + +// newDatasetPushCmd implements `tracebloc dataset push `. +// +// PR-a scope (what this implements today): +// +// - 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 "ready to stage" summary +// - --dry-run stops here (and so does this PR — actual staging +// errors out with "coming in PR-b" until #151 PR-b merges) +// +// PR-b will add the ephemeral stage Pod, tar-over-exec stream, +// progress bar, and SIGINT-safe cleanup — slotting into the +// "TODO: PR-b" branch below without touching anything above it. +func newDatasetPushCmd() *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. The set is intentionally + // image_classification-only for v0.1 per epic #147 + // non-goals; other categories are one-PR additions in v0.2. + table string + category string + intent string + labelColumn string + + // Operations flags. + dryRun bool + + // Ingestor SA name override (only matters once PR-b mints + // a token to talk to the future stage-pod-creation hook). + // Plumbed today so PR-b doesn't have to touch flag wiring. + ingestorSAName string + ) + + cmd := &cobra.Command{ + Use: "push ", + Short: "Stage a local dataset to the cluster's shared PVC", + Long: `Stages a local image_classification dataset to the parent client +release's shared PVC, then (in PR-b) submits an ingestion run. + +Expected local layout: + + / + labels.csv (required) + images/ (required) + 001.jpg + 002.jpg + ... + +Accepted image extensions: .jpg, .jpeg, .png, .webp (case-insensitive). + +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 staged + (in PR-b) submitted successfully + 2 schema validation failed (synthesized spec rejected) + 3 local-layout or kubeconfig error + 4 cluster reachable but parent release missing + 6 pre-flight succeeded but the actual stage step isn't + implemented yet (PR-b for #151 will deliver it)`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + return runDatasetPush(cmd.Context(), cmd.OutOrStdout(), cmd.ErrOrStderr(), + runDatasetPushArgs{ + LocalPath: args[0], + Kubeconfig: kubeconfigPath, + Context: contextOverride, + Namespace: nsOverride, + Spec: push.SpecArgs{Table: table, Category: category, Intent: intent, LabelColumn: labelColumn}, + DryRun: dryRun, + IngestorSAName: ingestorSAName, + }) + }, + } + + cmd.Flags().StringVar(&kubeconfigPath, "kubeconfig", "", + "path to the kubeconfig file (default: $KUBECONFIG, then ~/.kube/config)") + cmd.Flags().StringVar(&contextOverride, "context", "", + "name of the kubeconfig context to use (default: kubeconfig's current-context)") + cmd.Flags().StringVarP(&nsOverride, "namespace", "n", "", + "namespace where the parent tracebloc/client release is installed") + + // 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(&table, "table", "", + "destination table name (MySQL identifier; matches /data/shared// on the PVC)") + cmd.Flags().StringVar(&category, "category", "image_classification", + "task category (v0.1 only supports image_classification; see tracebloc/client#147 non-goals)") + cmd.Flags().StringVar(&intent, "intent", "", + "intent: train|test") + cmd.Flags().StringVar(&labelColumn, "label-column", "", + "column name in labels.csv that holds the label") + + cmd.Flags().BoolVar(&dryRun, "dry-run", false, + "validate + discover + walk, but don't create any cluster resources") + cmd.Flags().StringVar(&ingestorSAName, "ingestor-sa", "", + "override the ingestor ServiceAccount name (default: \"ingestor\"); "+ + "set this if you customized ingestionAuthz.serviceAccountName in the parent client chart") + + return cmd +} + +// runDatasetPushArgs collects every parameter runDatasetPush 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 runDatasetPushArgs struct { + LocalPath string + Kubeconfig string + Context string + Namespace string + Spec push.SpecArgs + DryRun bool + IngestorSAName string +} + +// runDatasetPush is the PR-a slim implementation. It performs every +// pre-flight check and prints a summary; the actual file staging +// is gated behind a clear "not yet implemented" error so PR-a +// merging doesn't silently advertise a feature it can't deliver. +// +// Step order is "fail fast, fail local" — every step that doesn't +// need the cluster runs before any that does, so a customer with +// a bad label-column or oversized dataset gets the diagnostic in +// milliseconds without a kubeconfig round-trip. +func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPushArgs) error { + // 1. Synthesize the spec from flags + validate against schema. + // Catches "no table", "bad category", "missing intent" etc. + // BEFORE we touch the filesystem or the cluster. The error + // formatter is the same one ingest validate uses, so a + // customer who YAML'd manually first sees identical wording. + spec := a.Spec.Build() + specBytes, err := yaml.Marshal(spec) + if err != nil { + return &exitError{code: 3, err: fmt.Errorf("marshaling synthesized spec: %w", err)} + } + v, err := schema.NewV1Validator() + if err != nil { + return &exitError{code: 3, err: fmt.Errorf("loading embedded schema: %w", err)} + } + _, errs, parseErr := v.ValidateYAML(specBytes) + if parseErr != nil { + // "Parse" failing on a spec we marshaled ourselves is a + // programming error, not a customer error — surface it + // with the bytes so we can diagnose. Exit 3 (the + // "internal" bucket) matches the marshal-failure branch + // above. + return &exitError{code: 3, err: fmt.Errorf("internal: re-parsing synthesized spec: %w\n%s", parseErr, specBytes)} + } + if len(errs) > 0 { + // Use the SAME formatter `ingest validate` uses, so the + // experience is identical whether the customer authored + // YAML by hand or via flags. Diagnostics go to stderr + // (matching ingest validate) so a downstream pipe of + // stdout (e.g. piping the summary to jq once that's a + // JSON output mode) isn't polluted by error text. Exit 2 + // is reserved for schema violations across the CLI. + _, _ = fmt.Fprintf(errOut, "synthesized spec failed schema validation (%d issue%s):\n", + len(errs), pluralS(len(errs))) + _, _ = fmt.Fprintln(errOut, schema.FormatErrors(errs)) + return &exitError{code: 2, err: errors.New("synthesized spec failed schema validation; check the flag values above")} + } + + // 2. Walk the local directory. Enforces layout + size caps; + // customer sees a clear pointer to expected layout if they + // pass the wrong directory. + layout, err := push.Discover(a.LocalPath) + if err != nil { + return &exitError{code: 3, err: err} + } + + // 3. Cluster discovery — same kubeconfig path as `cluster info`. + // Errors mirror that command's exit-code contract (3 for + // kubeconfig, 4 for missing release) so behaviour is + // consistent across pre-flight commands. + resolved, err := cluster.Load(cluster.KubeconfigOptions{ + Path: a.Kubeconfig, + Context: a.Context, + Namespace: a.Namespace, + }) + if err != nil { + return &exitError{code: 3, err: fmt.Errorf("loading kubeconfig: %w", err)} + } + cs, err := cluster.NewClientset(resolved) + if err != nil { + return &exitError{code: 3, err: err} + } + release, err := cluster.DiscoverParentRelease(ctx, cs, resolved.Namespace) + if err != nil { + return &exitError{code: 4, err: err} + } + if a.IngestorSAName != "" { + release.IngestorSAName = a.IngestorSAName + } + + // 4. PVC discovery. New in this PR — confirms the chart's + // shared-data PVC is Bound before we waste time provisioning + // a Pod that can't mount it. + pvc, err := cluster.DiscoverSharedPVC(ctx, cs, resolved.Namespace) + if err != nil { + return &exitError{code: 4, err: err} + } + + // 5. Print the pre-flight summary. The output is the same in + // dry-run and (eventually) live mode — only the "what + // happens next" line differs. Customers iterating on a + // bad layout see this every attempt, so it's worth keeping + // skimmable: one fact per line, aligned by column. + printPushPreflight(out, layout, release, pvc, spec, a.DryRun) + + // 6. Dry-run stop. Acknowledged success. + if a.DryRun { + _, _ = fmt.Fprintln(out, "Dry-run complete — no cluster resources were created.") + return nil + } + + // 7. The actual staging branch lands in PR-b. Failing here + // rather than silently returning success means a customer + // who pulled PR-a's binary and ran without --dry-run gets + // a clear "wait for PR-b" signal instead of "0 files + // transferred" confusion in Phase 4. + return &exitError{code: 6, err: errors.New( + "pre-flight succeeded but the actual file staging step isn't " + + "implemented yet — wait for tracebloc/client#151 PR-b. " + + "Re-run with --dry-run to validate without this error.")} +} + +// printPushPreflight is the customer-facing summary. Mirrors +// `cluster info`'s layout for consistency: section header, +// indented key:value rows. Kept here (not on the layout/release/pvc +// types) because the formatting is policy and lives with the CLI, +// not the data. +func printPushPreflight( + out io.Writer, + layout *push.LocalLayout, + release *cluster.ParentRelease, + pvc *cluster.SharedPVC, + spec map[string]any, + dryRun bool, +) { + // Explicit-discard the writer errors throughout — same rationale + // as cli/cluster.go and cli/ingest.go: a pipe-write failure + // shouldn't convert success into failure. The exit code is + // the contract. + _, _ = fmt.Fprintf(out, "Local dataset:\n") + _, _ = fmt.Fprintf(out, " root: %s\n", layout.Root) + _, _ = fmt.Fprintf(out, " labels.csv: %s\n", layout.LabelsCSV) + _, _ = fmt.Fprintf(out, " images: %d files\n", len(layout.Images)) + _, _ = fmt.Fprintf(out, " total size: %s\n", humanBytesForSummary(layout.TotalBytes)) + _, _ = fmt.Fprintln(out) + + _, _ = fmt.Fprintf(out, "Target cluster:\n") + _, _ = fmt.Fprintf(out, " release: %s (chart %s)\n", release.ReleaseName, release.ChartVersion) + _, _ = fmt.Fprintf(out, " jobs-manager: %s\n", release.JobsManagerService) + _, _ = fmt.Fprintf(out, " shared PVC: %s (%s)\n", pvc.ClaimName, pvc.Phase) + if !pvc.IsReadWriteMany() { + // Warn but don't block — RWO clusters still work, the + // scheduler will co-locate the stage Pod with the existing + // mounter. Phase 3 PR-b will surface the same warning at + // pod-create time too. + _, _ = fmt.Fprintf(out, " access: %v (warn: not ReadWriteMany — stage Pod will co-locate)\n", pvc.AccessModes) + } + _, _ = fmt.Fprintln(out) + + _, _ = fmt.Fprintf(out, "Synthesized ingest spec:\n") + _, _ = fmt.Fprintf(out, " table: %s\n", spec["table"]) + _, _ = fmt.Fprintf(out, " category: %s\n", spec["category"]) + _, _ = fmt.Fprintf(out, " intent: %s\n", spec["intent"]) + _, _ = fmt.Fprintf(out, " label column: %s\n", spec["label"]) + _, _ = fmt.Fprintf(out, " destination: %s\n", push.StagedPrefix(spec["table"].(string))) + _, _ = fmt.Fprintln(out) + + if !dryRun { + _, _ = fmt.Fprintf(out, "Next: stage %d files (%s) → %s (coming in PR-b for #151)\n", + 1+len(layout.Images), humanBytesForSummary(layout.TotalBytes), + push.StagedPrefix(spec["table"].(string))) + _, _ = fmt.Fprintln(out) + } +} + +// pluralS returns "s" for n != 1, else "". Tiny helper that keeps +// the "1 issue" / "3 issues" diagnostic readable without an inline +// ternary. +func pluralS(n int) string { + if n == 1 { + return "" + } + return "s" +} + +// humanBytesForSummary mirrors push.humanBytes but lives here to +// keep internal/push's API surface narrow (the internal helper is +// unexported). Yes, this is a tiny duplication; if a third caller +// shows up, we promote it to a shared util in v0.2. +func humanBytesForSummary(n int64) string { + const ( + KiB = 1024 + MiB = 1024 * KiB + GiB = 1024 * MiB + ) + switch { + case n >= GiB: + return fmt.Sprintf("%.2f GiB", float64(n)/float64(GiB)) + case n >= MiB: + return fmt.Sprintf("%.2f MiB", float64(n)/float64(MiB)) + case n >= KiB: + return fmt.Sprintf("%.2f KiB", float64(n)/float64(KiB)) + default: + return fmt.Sprintf("%d B", n) + } +} diff --git a/internal/cli/dataset_test.go b/internal/cli/dataset_test.go new file mode 100644 index 00000000..c9a35fc1 --- /dev/null +++ b/internal/cli/dataset_test.go @@ -0,0 +1,225 @@ +package cli + +import ( + "bytes" + "os" + "path/filepath" + "strings" + "testing" +) + +// imgcLayout drops a minimum-viable image_classification directory +// under t.TempDir() and returns its path. Mirrors push.imgcDir +// (tests can't import test helpers across packages, so we duplicate +// the few lines). +func imgcLayout(t *testing.T) string { + t.Helper() + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, "labels.csv"), + []byte("image_id,label\n001.jpg,cat\n"), 0o644); err != nil { + t.Fatalf("write labels.csv: %v", err) + } + imagesDir := filepath.Join(root, "images") + if err := os.MkdirAll(imagesDir, 0o755); err != nil { + t.Fatalf("mkdir images: %v", err) + } + if err := os.WriteFile(filepath.Join(imagesDir, "001.jpg"), + make([]byte, 100), 0o644); err != nil { + t.Fatalf("write image: %v", err) + } + return root +} + +// execDatasetPush drives the full cobra dispatch for the push +// command and returns the exit code + captured stdout/stderr. +// Mirrors the execIngestValidate helper from ingest_test.go — same +// rationale about not sharing *cobra.Command across cases (cobra +// holds flag state on the command tree). +// +// kubeconfigPath is required because every push invocation tries +// kubeconfig load before any cluster work; tests that want to +// stop EARLIER (at schema validation or layout walk) still need a +// kubeconfig path that resolves predictably. We feed in a path +// that's guaranteed to fail os.Stat so the kubeconfig branch +// errors out consistently when reached — and tests assert on the +// EARLIER stage's exit code, which fires before kubeconfig. +func execDatasetPush(t *testing.T, args []string) (exitCode int, stdout, stderr string) { + t.Helper() + root := NewRootCmd(BuildInfo{Version: "test"}) + var so, se bytes.Buffer + root.SetOut(&so) + root.SetErr(&se) + + // Always inject a guaranteed-bad kubeconfig path so tests that + // "fall through" the local pre-checks into kubeconfig load + // get a deterministic exit 3 (not a flaky "depends on whether + // you have a real kubeconfig" outcome). + cmdArgs := append([]string{"dataset", "push", + "--kubeconfig=/tmp/tracebloc-cli-test-nonexistent-" + t.Name()}, + args...) + root.SetArgs(cmdArgs) + + err := root.Execute() + return ExitCodeFromError(err), so.String(), se.String() +} + +// TestDatasetPush_BadCategory_ExitsTwo: the synthesized spec gets +// fed through the embedded schema before any cluster work; an +// invalid category surfaces with exit 2 (the CLI-wide "schema +// violation" code). +func TestDatasetPush_BadCategory_ExitsTwo(t *testing.T) { + root := imgcLayout(t) + code, _, stderr := execDatasetPush(t, []string{ + root, + "--table=t1", + "--category=definitely-not-a-category", + "--intent=train", + "--label-column=label", + }) + if code != 2 { + t.Fatalf("expected exit 2 for schema failure, got %d", code) + } + // The same FormatErrors output ingest validate uses, routed + // to stderr so downstream pipes of stdout aren't polluted. + // Checking for "category" + "image_classification" surfacing + // means we're getting the JSON-pointer-anchored diagnostic + + // the enum-list expansion, not a generic message. + // Check three things: (1) the JSON-pointer-style "category" + // anchor surfaces, (2) the enum-list expansion happened (so + // "image_classification" appears as a valid option), (3) our + // "synthesized spec" framing is on the right output stream. + // The wording "synthesized spec" is intentionally distinct + // from ingest validate's "schema validation failed" — it tells + // the customer the CLI synthesized the YAML; they didn't + // author it. + for _, want := range []string{"category", "image_classification", "synthesized spec failed schema validation"} { + if !strings.Contains(stderr, want) { + t.Errorf("expected stderr to mention %q, got:\n%s", want, stderr) + } + } +} + +// TestDatasetPush_MissingIntent_ExitsTwo: pins the "intent is +// required" diagnostic path — different schema violation but the +// same exit-code class. +func TestDatasetPush_MissingIntent_ExitsTwo(t *testing.T) { + root := imgcLayout(t) + code, _, stderr := execDatasetPush(t, []string{ + root, + "--table=t1", + "--category=image_classification", + // intent omitted + "--label-column=label", + }) + if code != 2 { + t.Fatalf("expected exit 2 for missing intent, got %d", code) + } + if !strings.Contains(stderr, "intent") { + t.Errorf("expected stderr to mention 'intent', got:\n%s", stderr) + } +} + +// TestDatasetPush_NonexistentLocalPath_ExitsThree: the layout walk +// runs AFTER schema validation, so an invalid local path with +// otherwise-valid flags surfaces at the walk stage with exit 3 +// (the "local input or kubeconfig" code). +// +// We only assert on the exit code, not the error text — cobra's +// SilenceErrors=true means root.Execute() doesn't surface the +// returned error to stderr (main.go does that). Mirrors +// ingest_test.go's TestIngestValidate_UnreadableFileExitsThree +// pattern; the error-content surface is exercised at the package +// level (internal/push.Discover's own tests). +func TestDatasetPush_NonexistentLocalPath_ExitsThree(t *testing.T) { + code, _, _ := execDatasetPush(t, []string{ + "/tmp/tracebloc-cli-test-no-such-dir-" + t.Name(), + "--table=t1", + "--category=image_classification", + "--intent=train", + "--label-column=label", + }) + if code != 3 { + t.Fatalf("expected exit 3 for missing local path, got %d", code) + } +} + +// TestDatasetPush_MissingLabelsCSV_ExitsThree: most likely "real +// world" wrong-layout case — customer has images but forgot +// labels.csv. Pins the exit-code contract for the common failure +// mode; the diagnostic-text content is covered by +// internal/push.TestDiscover_MissingLabelsCSV. +func TestDatasetPush_MissingLabelsCSV_ExitsThree(t *testing.T) { + root := t.TempDir() + imagesDir := filepath.Join(root, "images") + if err := os.MkdirAll(imagesDir, 0o755); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := os.WriteFile(filepath.Join(imagesDir, "a.jpg"), + make([]byte, 100), 0o644); err != nil { + t.Fatalf("write img: %v", err) + } + + code, _, _ := execDatasetPush(t, []string{ + root, + "--table=t1", + "--category=image_classification", + "--intent=train", + "--label-column=label", + }) + if code != 3 { + t.Fatalf("expected exit 3 for missing labels.csv, got %d", code) + } +} + +// TestDatasetPush_BadKubeconfig_ExitsThree: schema + layout both +// pass; kubeconfig load fails because the injected path doesn't +// exist. The exit-code contract matches `cluster info`'s — same +// class of failure (3 = local input problem) surfaces with the +// same code regardless of which command tripped it. +func TestDatasetPush_BadKubeconfig_ExitsThree(t *testing.T) { + root := imgcLayout(t) + code, _, _ := execDatasetPush(t, []string{ + root, + "--table=t1", + "--category=image_classification", + "--intent=train", + "--label-column=label", + }) + if code != 3 { + t.Fatalf("expected exit 3 for bad kubeconfig, got %d", code) + } +} + +// TestDatasetPush_RequiresExactlyOneArg: cobra-level Args check +// pins the command signature. Two positional args, or zero, should +// fail before the runner even fires. +func TestDatasetPush_RequiresExactlyOneArg(t *testing.T) { + cases := []struct { + name string + args []string + }{ + { + name: "no positional", + args: []string{ + "--table=t1", "--category=image_classification", + "--intent=train", "--label-column=label", + }, + }, + { + name: "two positionals", + args: []string{ + "./a", "./b", + "--table=t1", "--category=image_classification", + "--intent=train", "--label-column=label", + }, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + code, _, _ := execDatasetPush(t, c.args) + if code == 0 { + t.Errorf("expected non-zero exit for %s, got 0", c.name) + } + }) + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 0d790aeb..1dc41842 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -66,6 +66,7 @@ roadmap. Subsequent phases land subcommands incrementally.`, root.AddCommand(newVersionCmd(info)) root.AddCommand(newIngestCmd()) root.AddCommand(newClusterCmd()) + root.AddCommand(newDatasetCmd()) return root } diff --git a/internal/cluster/pvc.go b/internal/cluster/pvc.go new file mode 100644 index 00000000..fd744d01 --- /dev/null +++ b/internal/cluster/pvc.go @@ -0,0 +1,146 @@ +package cluster + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" +) + +// SharedPVCClaimName is the chart's hardcoded shared-data PVC name. +// +// From tracebloc/client's templates/_helpers.tpl: +// +// {{- define "tracebloc.clientDataPvc" -}} +// client-pvc +// {{- end }} +// +// The helper isn't parameterized by release name (yet) — every +// installation of the chart creates a PVC literally named +// "client-pvc". We probe by claim name rather than by labels here +// because the chart's labels include the helm release name and we +// already discovered the release via DiscoverParentRelease — once +// we have the namespace, the name is unambiguous. +// +// If a customer renames the PVC (out-of-band patch, mostly), they +// need the v0.2 follow-up that reads the name from +// jobs-manager's volume-mount spec instead of hardcoding. Tracked +// as a future ticket alongside #7 (the ingestor-SA-name discovery). +const SharedPVCClaimName = "client-pvc" + +// SharedPVCMountPath is where jobs-manager mounts the shared PVC +// inside its container. The CLI's staging Pod (Phase 3 PR-b) uses +// the same mount path so any tooling that introspects "where files +// live" in either context agrees. From jobs-manager-deployment.yaml: +// +// volumeMounts: +// - name: shared-volume +// mountPath: "/data/shared" +const SharedPVCMountPath = "/data/shared" + +// SharedPVC describes the chart's shared-data PVC after discovery. +// Carries enough metadata for Phase 3 PR-b to construct a stage Pod +// that can mount the same claim. +type SharedPVC struct { + // ClaimName is the metadata.name of the PVC, always + // SharedPVCClaimName today. Wrapped in a field rather than + // re-using the constant so a future "discovered via labels" + // implementation can vary the name without changing callers. + ClaimName string + + // MountPath is the in-container directory where the chart's + // pods mount this claim — SharedPVCMountPath today. Same + // rationale as ClaimName for being a field. + MountPath string + + // AccessModes is the resolved access-mode list on the PVC's + // spec. Surfaces in `tracebloc cluster info` (eventually) and + // drives the stage-Pod scheduling story: + // + // - ReadWriteMany: stage Pod can schedule on any node + // - ReadWriteOnce: stage Pod must land on the same node + // as whatever else is using the PVC (jobs-manager, + // mysql-client). PR-b will surface a warning here so + // RWO clusters get diagnostic guidance up front. + AccessModes []corev1.PersistentVolumeAccessMode + + // Phase is the PVC's current bind phase. We check Bound — an + // Unbound PVC means the cluster never provisioned storage + // (e.g. missing StorageClass), and the stage Pod would hang + // indefinitely waiting for a volume. + Phase corev1.PersistentVolumeClaimPhase +} + +// DiscoverSharedPVC verifies the chart's shared-data PVC exists in +// the given namespace and returns its metadata. Returns a friendly +// error if the PVC isn't there or isn't Bound — both are real +// situations Phase 3's pre-flight catches before we waste time +// constructing a Pod that can't mount anything. +func DiscoverSharedPVC(ctx context.Context, cs kubernetes.Interface, namespace string) (*SharedPVC, error) { + pvc, err := cs.CoreV1().PersistentVolumeClaims(namespace). + Get(ctx, SharedPVCClaimName, metav1.GetOptions{}) + if err != nil { + if apierrors.IsNotFound(err) { + // The 80% case for "the CLI's pre-flight failed against + // a cluster that has SOME tracebloc release installed + // but not the parent client chart" — the parent-release + // check (DiscoverParentRelease) catches the no-release + // case first, so reaching here usually means the chart + // was installed with a customized PVC name. + return nil, fmt.Errorf( + "no PersistentVolumeClaim named %q found in namespace %q. "+ + "The chart's _helpers.tpl pins this name; if your install "+ + "renamed it out-of-band, the CLI doesn't yet support that "+ + "(read-name-from-jobs-manager is a v0.2 follow-up). "+ + "Verify with: kubectl get pvc -n %s", + SharedPVCClaimName, namespace, namespace) + } + // Forbidden / network / other — surface as-is so the + // customer can RBAC-debug. Wrapping rather than substituting + // because the underlying %w already carries the useful info. + return nil, fmt.Errorf("reading PVC %s/%s: %w", + namespace, SharedPVCClaimName, err) + } + + if pvc.Status.Phase != corev1.ClaimBound { + // Unbound PVCs are a real cluster-config issue — pre-flight + // surfacing of this is much more helpful than the stage Pod + // silently pending forever. The most common cause is a + // missing StorageClass (the chart's default is the cluster + // default, which may not exist on EKS without + // gp2/gp3-csi configured). + return nil, fmt.Errorf( + "PVC %s/%s is in phase %q, not Bound. "+ + "The shared volume hasn't been provisioned — "+ + "check that the cluster has a usable StorageClass "+ + "(kubectl get sc) and that the PVC's storageClassName matches.", + namespace, SharedPVCClaimName, pvc.Status.Phase) + } + + return &SharedPVC{ + ClaimName: pvc.Name, + MountPath: SharedPVCMountPath, + AccessModes: pvc.Spec.AccessModes, + Phase: pvc.Status.Phase, + }, nil +} + +// IsReadWriteMany reports whether the PVC accepts simultaneous +// mounts from multiple nodes. The Phase 3 stage Pod cares about +// this because RWO claims force same-node scheduling — and if the +// existing mounter (jobs-manager) is on a different node than +// where the scheduler wants to put our stage Pod, the Pod will +// pend indefinitely. PR-b will surface a pre-flight warning when +// this returns false; the dataset push still proceeds (eventually +// succeeds when the scheduler co-locates), it just takes longer. +func (p *SharedPVC) IsReadWriteMany() bool { + for _, m := range p.AccessModes { + if m == corev1.ReadWriteMany { + return true + } + } + return false +} diff --git a/internal/cluster/pvc_test.go b/internal/cluster/pvc_test.go new file mode 100644 index 00000000..3260ed75 --- /dev/null +++ b/internal/cluster/pvc_test.go @@ -0,0 +1,117 @@ +package cluster + +import ( + "context" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +// seedPVC returns a corev1.PersistentVolumeClaim wired to look like +// the chart's shared-data claim. Lets each test mutate one field +// (phase, access mode) to exercise its specific branch. +func seedPVC(phase corev1.PersistentVolumeClaimPhase, modes ...corev1.PersistentVolumeAccessMode) *corev1.PersistentVolumeClaim { + if len(modes) == 0 { + modes = []corev1.PersistentVolumeAccessMode{corev1.ReadWriteMany} + } + return &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: SharedPVCClaimName, + Namespace: "tracebloc", + }, + Spec: corev1.PersistentVolumeClaimSpec{ + AccessModes: modes, + }, + Status: corev1.PersistentVolumeClaimStatus{ + Phase: phase, + }, + } +} + +func TestDiscoverSharedPVC_HappyPath(t *testing.T) { + cs := fake.NewClientset(seedPVC(corev1.ClaimBound)) + got, err := DiscoverSharedPVC(context.Background(), cs, "tracebloc") + if err != nil { + t.Fatalf("DiscoverSharedPVC: %v", err) + } + if got.ClaimName != SharedPVCClaimName { + t.Errorf("ClaimName = %q, want %q", got.ClaimName, SharedPVCClaimName) + } + if got.MountPath != SharedPVCMountPath { + t.Errorf("MountPath = %q, want %q", got.MountPath, SharedPVCMountPath) + } + if got.Phase != corev1.ClaimBound { + t.Errorf("Phase = %v, want Bound", got.Phase) + } + if !got.IsReadWriteMany() { + t.Errorf("IsReadWriteMany() = false on a RWX-seeded PVC") + } +} + +func TestDiscoverSharedPVC_NotFound(t *testing.T) { + // Empty cluster — the parent-release check should normally + // have failed first, but we still pin the diagnostic for the + // case where customers renamed the PVC out-of-band. + cs := fake.NewClientset() + _, err := DiscoverSharedPVC(context.Background(), cs, "tracebloc") + if err == nil { + t.Fatal("DiscoverSharedPVC returned nil error on empty cluster") + } + // Diagnostic must point at the v0.2 follow-up and `kubectl + // get pvc` command — those are the actionable bits. + for _, want := range []string{SharedPVCClaimName, "v0.2", "kubectl get pvc"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error missing %q: %s", want, err) + } + } +} + +func TestDiscoverSharedPVC_UnboundPhase(t *testing.T) { + // Most common cause: missing StorageClass on EKS where the + // admin removed gp2-default before installing the chart. The + // pre-flight diagnostic must call out StorageClass explicitly + // so customers know what to fix. + cs := fake.NewClientset(seedPVC(corev1.ClaimPending)) + _, err := DiscoverSharedPVC(context.Background(), cs, "tracebloc") + if err == nil { + t.Fatal("DiscoverSharedPVC returned nil error on Pending PVC") + } + for _, want := range []string{"Pending", "not Bound", "StorageClass"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error missing %q: %s", want, err) + } + } +} + +func TestIsReadWriteMany_RWO(t *testing.T) { + // ReadWriteOnce — common on cheap-storage clusters (single-node + // minikube, single-zone EBS). Phase 3 still works against RWO + // (the scheduler co-locates), but PR-b will print a warning. + // Pin the API surface that the warning logic will key off. + cs := fake.NewClientset(seedPVC(corev1.ClaimBound, corev1.ReadWriteOnce)) + got, err := DiscoverSharedPVC(context.Background(), cs, "tracebloc") + if err != nil { + t.Fatalf("DiscoverSharedPVC: %v", err) + } + if got.IsReadWriteMany() { + t.Errorf("IsReadWriteMany() = true on a RWO-seeded PVC") + } +} + +func TestIsReadWriteMany_RWXMixedWithOther(t *testing.T) { + // A few cloud providers (specifically EFS-on-EKS) advertise + // both RWX and ROX. As long as RWX is in the list, our stage + // Pod can schedule freely. + cs := fake.NewClientset(seedPVC(corev1.ClaimBound, + corev1.ReadOnlyMany, corev1.ReadWriteMany)) + got, err := DiscoverSharedPVC(context.Background(), cs, "tracebloc") + if err != nil { + t.Fatalf("DiscoverSharedPVC: %v", err) + } + if !got.IsReadWriteMany() { + t.Errorf("IsReadWriteMany() = false on a mixed-mode PVC including RWX") + } +} diff --git a/internal/push/spec.go b/internal/push/spec.go new file mode 100644 index 00000000..483c5f53 --- /dev/null +++ b/internal/push/spec.go @@ -0,0 +1,111 @@ +// Package push owns the `tracebloc dataset push` flow: synthesizing +// an ingest spec from CLI flags, walking the customer's local data +// directory, and (in a follow-up PR) staging files into the cluster's +// shared PVC via an ephemeral Pod + tar-over-exec stream. +// +// Phase 3 lands in two PRs (per tracebloc/client#151): +// +// - PR-a (this one): the no-op-safe path — spec synthesis, local +// layout discovery, PVC discovery, --dry-run. Everything up to +// "ready to copy files" without actually copying. +// - PR-b (next): the novel-engineering core — ephemeral stage Pod, +// client-go remotecommand executor, tar stream, progress bar, +// SIGINT-safe cleanup. +// +// The split keeps each diff reviewable. PR-a's purpose is "fail fast +// before we touch the cluster"; PR-b is "now actually push the bytes". +package push + +import "path" + +// SpecArgs is the user-facing flag set for `tracebloc dataset push`. +// +// It's intentionally narrower than the full ingest.v1.json schema — +// for v0.1, only the image_classification minimum is exposed. The +// epic (#147) defers all other categories to v0.2 as one-PR +// additions; the flag set will grow when those land. +// +// Validation is NOT enforced here. Build() produces an +// ingest.v1.json-conforming map, and the caller pipes it through +// internal/schema's V1 validator. Duplicating "intent must be +// train|test" or similar in Go-side code would drift from the +// embedded schema — the schema is the single source of truth. +type SpecArgs struct { + // Table is the destination MySQL table name in the cluster. Also + // used as the per-table subdirectory under /data/shared/ so + // pushes of multiple tables don't collide on the PVC. + Table string + + // Category pins the task family. For v0.1, only + // "image_classification" is supported end-to-end — the epic + // non-goals defer other categories to v0.2. The flag accepts + // other values so the schema's enum check produces the canonical + // error message (rather than a CLI-side "unknown category" + // that drifts from the schema). + Category string + + // Intent is "train" or "test" per the schema's enum. Same + // rationale as Category for not pre-validating here. + Intent string + + // LabelColumn is the column name in labels.csv that holds the + // label. The schema accepts either a shorthand string (this + // field) or a {column, policy} object; v0.1 only emits the + // shorthand because passthrough is the only policy + // image_classification cares about. + LabelColumn string +} + +// Build produces the ingest.v1.json-conforming spec map. The +// returned map is YAML-marshalable and ready to feed to +// internal/schema's V1 validator. +// +// PVC paths (csv, images) are constructed under +// /data/shared/
/ to match the chart's mount convention +// surfaced by Phase 2's cluster discovery — jobs-manager mounts +// client-pvc at /data/shared, and the per-table subdir prevents +// cross-table collisions when a customer pushes multiple datasets +// to the same release. +// +// File-name conventions inside that subdir (labels.csv, +// images/) are dictated by the local layout that internal/push.Discover +// requires. Keeping the layout convention in lock-step on both +// sides — the CLI's view of "what local files we expect" and the +// spec's view of "where they'll live in the cluster" — means a +// successful Discover guarantees a runnable spec. +func (a SpecArgs) Build() map[string]any { + prefix := StagedPrefix(a.Table) + return map[string]any{ + "apiVersion": "tracebloc.io/v1", + "kind": "IngestConfig", + "category": a.Category, + "table": a.Table, + "intent": a.Intent, + // Trailing slash on `images` matches the schema example + // (data-ingestors/examples/yaml/image_classification.yaml, + // line 14). The ingestor treats it as a directory glob. + "csv": path.Join(prefix, "labels.csv"), + "images": path.Join(prefix, "images") + "/", + "label": a.LabelColumn, + } +} + +// StagedPrefix returns the in-cluster destination directory the CLI +// writes files into for a given table. Used in two places that +// MUST agree: +// +// 1. Phase 3 (this PR + PR-b): the path the ephemeral stage Pod +// creates and tars files into. +// 2. The csv/images fields in Build() above, which jobs-manager +// reads to know where the ingestor Job will find them. +// +// Exported because Phase 3's PR-b (stage Pod construction) needs +// it from the same place, and Phase 4 (submit) might want to print +// it as part of "what we pushed." +func StagedPrefix(table string) string { + // path.Join collapses redundant slashes but doesn't preserve + // trailing slashes — fine here because callers either append a + // filename (labels.csv) or add the trailing slash explicitly + // (images/). + return path.Join("/data/shared", table) +} diff --git a/internal/push/spec_test.go b/internal/push/spec_test.go new file mode 100644 index 00000000..674aad1c --- /dev/null +++ b/internal/push/spec_test.go @@ -0,0 +1,107 @@ +package push + +import ( + "testing" + + "gopkg.in/yaml.v3" + + "github.com/tracebloc/cli/internal/schema" +) + +// TestBuild_ImageClassificationMinimum_PassesSchema is the contract +// test that pins Phase 3's flag → spec synthesis to the embedded +// schema. The whole point of Build() is "produce something the +// canonical validator accepts" — if a refactor breaks that, every +// `dataset push` invocation fails after kubeconfig load but before +// the user sees a useful error, so we want this caught in CI. +func TestBuild_ImageClassificationMinimum_PassesSchema(t *testing.T) { + args := SpecArgs{ + Table: "chest_xrays_train", + Category: "image_classification", + Intent: "train", + LabelColumn: "image_label", + } + spec := args.Build() + + // Round-trip via YAML because the validator's public API is + // YAML-input. The map → JSON → YAML → parse-back chain is a + // microscopic cost per `dataset push` invocation; not worth + // adding a Validate(parsed) method to internal/schema for v0.1. + specBytes, err := yaml.Marshal(spec) + if err != nil { + t.Fatalf("yaml.Marshal: %v", err) + } + + v, err := schema.NewV1Validator() + if err != nil { + t.Fatalf("NewV1Validator: %v", err) + } + _, errs, parseErr := v.ValidateYAML(specBytes) + if parseErr != nil { + t.Fatalf("ValidateYAML returned parse error on our own output: %v\n%s", parseErr, specBytes) + } + if len(errs) != 0 { + t.Fatalf("synthesized spec failed schema validation: %s\nspec:\n%s", + schema.FormatErrors(errs), specBytes) + } +} + +// TestBuild_PathsMatchStagedPrefix pins the contract between Phase 3 +// (where the CLI puts files on the PVC) and Phase 4 (what paths the +// submitted spec tells jobs-manager to look at). If these ever +// drift, the ingestor Job spawned by jobs-manager won't find the +// files we just staged — a silent "0 rows ingested" outcome that's +// hard to debug. +func TestBuild_PathsMatchStagedPrefix(t *testing.T) { + const table = "cats_dogs" + spec := SpecArgs{ + Table: table, + Category: "image_classification", + Intent: "train", + LabelColumn: "label", + }.Build() + + prefix := StagedPrefix(table) + wantCSV := prefix + "/labels.csv" + wantImages := prefix + "/images/" + + if got := spec["csv"].(string); got != wantCSV { + t.Errorf("spec.csv = %q, want %q", got, wantCSV) + } + if got := spec["images"].(string); got != wantImages { + t.Errorf("spec.images = %q, want %q", got, wantImages) + } +} + +// TestBuild_LeavesValidationToSchema asserts that Build() does NOT +// pre-validate. A garbage category goes through unchanged so the +// schema's enum check produces the canonical error message. The +// alternative — duplicating the schema's enum in Go — would drift +// the moment data-ingestors adds a new category and we forget to +// mirror it here. +func TestBuild_LeavesValidationToSchema(t *testing.T) { + spec := SpecArgs{ + Table: "x", + Category: "definitely-not-a-real-category", + Intent: "train", + LabelColumn: "label", + }.Build() + + if got := spec["category"].(string); got != "definitely-not-a-real-category" { + t.Errorf("Build() pre-validated category; want raw passthrough, got %q", got) + } +} + +// TestStagedPrefix_PerTableIsolation pins the contract that two +// concurrent pushes for different tables don't write to the same +// PVC subdirectory. If this ever returns the same path for +// different tables, parallel `dataset push` calls would race on +// labels.csv overwrites. +func TestStagedPrefix_PerTableIsolation(t *testing.T) { + if a, b := StagedPrefix("cats"), StagedPrefix("dogs"); a == b { + t.Errorf("StagedPrefix(%q) == StagedPrefix(%q) = %q, want distinct", "cats", "dogs", a) + } + if got := StagedPrefix("table_a"); got != "/data/shared/table_a" { + t.Errorf("StagedPrefix(%q) = %q, want /data/shared/table_a", "table_a", got) + } +} diff --git a/internal/push/walk.go b/internal/push/walk.go new file mode 100644 index 00000000..c82e4f6b --- /dev/null +++ b/internal/push/walk.go @@ -0,0 +1,227 @@ +package push + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "strings" +) + +// Size limits enforced before we touch the cluster. Both caps are +// soft engineering choices, not protocol limits — they exist +// because tar-over-exec via client-go's remotecommand executor has +// a memory profile that degrades steeply past ~1 GB total transfer. +// Customers hitting these get pointed at the v0.2 cloud-source +// story (S3/GCS/HTTPS sources, currently in epic non-goals). +// +// The single-file cap is stricter than the total cap because the +// streaming buffer for one file lives in memory longer than the +// inter-file overhead — a 1 GB single file is worse than ten 100 MB +// files for the executor's working-set. +const ( + // MaxTotalBytes is the v0.1 ceiling on the sum of all files in + // a single `dataset push`. Picked from the epic's stated + // "anything above ~1GB needs the cloud-source story (v0.2)." + MaxTotalBytes int64 = 1 * 1024 * 1024 * 1024 + + // MaxSingleFileBytes caps any single file. Tuned via the + // data-ingestors templates' largest sample image (~30 KB) and + // a 10000x safety margin for typical user uploads. Files in the + // hundreds-of-MB range work in testing but degrade noticeably. + MaxSingleFileBytes int64 = 500 * 1024 * 1024 +) + +// LocalLayout describes a validated local directory ready to stage. +// All paths are absolute, resolved against the customer's working +// directory before this struct is returned. +type LocalLayout struct { + // Root is the absolute path the customer passed (after cleanup). + Root string + + // LabelsCSV is the absolute path to labels.csv inside Root. + // Required for image_classification. + LabelsCSV string + + // Images is the list of absolute paths to image files under + // Root/images/. Order is filesystem-walk order — Discover + // doesn't sort, so callers that need determinism (e.g. + // reproducible-build tests) sort before use. + Images []string + + // TotalBytes is the sum of all files Discover will stage — + // labels.csv plus every entry in Images. Pre-computed during + // the walk so the size-cap check + the progress bar (PR-b) + // can read it without re-stat'ing. + TotalBytes int64 +} + +// imageExtensions accepts the file types the chart's +// image_classification ingestor processes by default. From +// data-ingestors' FileTypeValidator(images) defaults: .jpg, .jpeg, +// .png. The chart's defaults file (see chartversion 1.3.5+) also +// accepts .webp; we mirror that here so customers on a recent +// chart can stage webp files without hitting "no images found." +// +// Comparison is case-insensitive — filesystems vary (case-sensitive +// on Linux, case-preserving-but-insensitive on macOS default APFS). +var imageExtensions = map[string]struct{}{ + ".jpg": {}, + ".jpeg": {}, + ".png": {}, + ".webp": {}, +} + +// Discover walks rootDir and validates it matches the layout Phase 3 +// expects for image_classification: +// +// - /labels.csv (required) +// - /images/*.{jpg,jpeg,png,webp} (at least one file) +// +// Returns specific errors keyed to the layout mistakes a customer +// is most likely to hit — these surface as the CLI's diagnostic +// output before any cluster work, so they're a primary UX surface. +// +// Enforces both v0.1 size caps (MaxTotalBytes, MaxSingleFileBytes); +// over-cap returns ErrTooBig with a pointer to the cloud-source +// story. +func Discover(rootDir string) (*LocalLayout, error) { + abs, err := filepath.Abs(rootDir) + if err != nil { + return nil, fmt.Errorf("resolving %q: %w", rootDir, err) + } + + st, err := os.Stat(abs) + if err != nil { + // Stat covers both "path doesn't exist" and "permission + // denied" via the wrapped fs.PathError; the customer sees + // the underlying message which is already clear. + return nil, fmt.Errorf("reading dataset directory %q: %w", abs, err) + } + if !st.IsDir() { + return nil, fmt.Errorf( + "%q is not a directory; pass the directory containing labels.csv + images/", + abs) + } + + layout := &LocalLayout{Root: abs} + + // labels.csv (required). + labelsPath := filepath.Join(abs, "labels.csv") + labelsStat, err := os.Stat(labelsPath) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf( + "missing labels.csv in %q. The CLI expects "+ + "/labels.csv + /images/ for image_classification; "+ + "see https://github.com/tracebloc/client/issues/147 for the "+ + "v0.1 layout contract.", + abs) + } + return nil, fmt.Errorf("stat labels.csv: %w", err) + } + if labelsStat.Size() > MaxSingleFileBytes { + return nil, sizeError("labels.csv", labelsStat.Size(), MaxSingleFileBytes) + } + layout.LabelsCSV = labelsPath + layout.TotalBytes += labelsStat.Size() + + // images/ subdir (required, must contain at least one + // image-extension file). + imagesDir := filepath.Join(abs, "images") + imagesStat, err := os.Stat(imagesDir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf( + "missing images/ subdirectory in %q. The CLI expects "+ + "/labels.csv + /images/*.{jpg,jpeg,png,webp}.", + abs) + } + return nil, fmt.Errorf("stat images/: %w", err) + } + if !imagesStat.IsDir() { + return nil, fmt.Errorf("%q exists but is not a directory", imagesDir) + } + + // Walk just the images/ directory — we don't recurse, image + // classification's layout is flat. If a customer has nested + // subdirs (e.g. images/cats/ + images/dogs/), that's a + // different category convention and out of scope for v0.1. + entries, err := os.ReadDir(imagesDir) + if err != nil { + return nil, fmt.Errorf("reading images/: %w", err) + } + for _, entry := range entries { + if entry.IsDir() { + // Silently skip subdirectories so a stray .DS_Store or + // thumbnails dir doesn't error out the whole walk. + // We DO surface the count of accepted images at the + // end, so a customer with all-nested-subdirs gets + // "0 images found" which is the right diagnostic. + continue + } + ext := strings.ToLower(filepath.Ext(entry.Name())) + if _, ok := imageExtensions[ext]; !ok { + continue + } + info, err := entry.Info() + if err != nil { + return nil, fmt.Errorf("stat %q: %w", entry.Name(), err) + } + if info.Size() > MaxSingleFileBytes { + return nil, sizeError(filepath.Join("images", entry.Name()), + info.Size(), MaxSingleFileBytes) + } + layout.Images = append(layout.Images, filepath.Join(imagesDir, entry.Name())) + layout.TotalBytes += info.Size() + } + + if len(layout.Images) == 0 { + return nil, fmt.Errorf( + "no image files found in %q. Expected .jpg, .jpeg, .png, or .webp; "+ + "got %d non-image entries.", + imagesDir, len(entries)) + } + + if layout.TotalBytes > MaxTotalBytes { + return nil, fmt.Errorf( + "dataset is %s, exceeds v0.1 cap of %s. "+ + "For larger datasets, the cloud-source path (S3/GCS/HTTPS) "+ + "is on the v0.2 roadmap — see tracebloc/client#147 non-goals. "+ + "Workaround for v0.1: split the push into multiple smaller "+ + "tables, or stage directly via the existing helm flow.", + humanBytes(layout.TotalBytes), humanBytes(MaxTotalBytes)) + } + + return layout, nil +} + +// sizeError builds the over-the-single-file-cap error with the same +// human-readable framing as the total-cap branch above. Centralized +// so the message stays consistent if we tune the wording later. +func sizeError(relPath string, got, cap int64) error { + return fmt.Errorf( + "file %q is %s, exceeds v0.1 single-file cap of %s. "+ + "For larger files, see tracebloc/client#147's v0.2 cloud-source story.", + relPath, humanBytes(got), humanBytes(cap)) +} + +// humanBytes renders a byte count in the shortest readable unit. +// Not internationalized — the CLI is English-only for v0.1. +func humanBytes(n int64) string { + const ( + KiB = 1024 + MiB = 1024 * KiB + GiB = 1024 * MiB + ) + switch { + case n >= GiB: + return fmt.Sprintf("%.2f GiB", float64(n)/float64(GiB)) + case n >= MiB: + return fmt.Sprintf("%.2f MiB", float64(n)/float64(MiB)) + case n >= KiB: + return fmt.Sprintf("%.2f KiB", float64(n)/float64(KiB)) + default: + return fmt.Sprintf("%d B", n) + } +} diff --git a/internal/push/walk_test.go b/internal/push/walk_test.go new file mode 100644 index 00000000..79597dd9 --- /dev/null +++ b/internal/push/walk_test.go @@ -0,0 +1,221 @@ +package push + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// imgcDir builds a valid image_classification layout under t.TempDir() +// and returns its absolute path. Used as the happy-path baseline that +// individual negative-case tests mutate. +func imgcDir(t *testing.T, withImages ...string) string { + t.Helper() + root := t.TempDir() + + if err := os.WriteFile(filepath.Join(root, "labels.csv"), + []byte("image_id,label\n001.jpg,cat\n002.jpg,dog\n"), 0o644); err != nil { + t.Fatalf("write labels.csv: %v", err) + } + imagesDir := filepath.Join(root, "images") + if err := os.MkdirAll(imagesDir, 0o755); err != nil { + t.Fatalf("mkdir images/: %v", err) + } + if len(withImages) == 0 { + withImages = []string{"001.jpg", "002.jpg"} + } + for _, name := range withImages { + // 100 bytes of stub data per image — keeps the total-bytes + // math predictable in TestDiscover_TotalBytesSum without + // generating real image headers. + if err := os.WriteFile(filepath.Join(imagesDir, name), + make([]byte, 100), 0o644); err != nil { + t.Fatalf("write image %s: %v", name, err) + } + } + return root +} + +func TestDiscover_HappyPath(t *testing.T) { + root := imgcDir(t) + got, err := Discover(root) + if err != nil { + t.Fatalf("Discover: %v", err) + } + if got.Root == "" || !filepath.IsAbs(got.Root) { + t.Errorf("Root = %q, want non-empty absolute path", got.Root) + } + if filepath.Base(got.LabelsCSV) != "labels.csv" { + t.Errorf("LabelsCSV basename = %q, want labels.csv", filepath.Base(got.LabelsCSV)) + } + if len(got.Images) != 2 { + t.Errorf("len(Images) = %d, want 2", len(got.Images)) + } +} + +func TestDiscover_TotalBytesSum(t *testing.T) { + // Two 100-byte images + the inline labels.csv string (39 bytes: + // "image_id,label\n001.jpg,cat\n002.jpg,dog\n"). 100+100+39 = 239. + // This pins the pre-cluster size summary the dry-run output + // prints — if we ever undercount, customers see "0 bytes" + // pre-push and panic. + root := imgcDir(t) + got, err := Discover(root) + if err != nil { + t.Fatalf("Discover: %v", err) + } + const want = int64(100 + 100 + 39) + if got.TotalBytes != want { + t.Errorf("TotalBytes = %d, want %d", got.TotalBytes, want) + } +} + +func TestDiscover_AcceptsAllImageExtensions(t *testing.T) { + // Mirror the chart's FileTypeValidator(images) defaults — if a + // customer's image-set has .png + .webp, both should stage. + root := imgcDir(t, "a.jpg", "b.jpeg", "c.png", "d.webp", "e.JPG") + got, err := Discover(root) + if err != nil { + t.Fatalf("Discover: %v", err) + } + if len(got.Images) != 5 { + t.Errorf("len(Images) = %d, want 5 (case-insensitive); names=%v", + len(got.Images), got.Images) + } +} + +func TestDiscover_SkipsNonImageFiles(t *testing.T) { + // .DS_Store, thumbnails.db, sibling .yaml etc. should be + // silently skipped — not error-out, not stage. The "no image + // files" diagnostic only fires when *zero* accepted files + // remain after filtering. + root := imgcDir(t, "real.jpg") + if err := os.WriteFile(filepath.Join(root, "images", ".DS_Store"), + make([]byte, 50), 0o644); err != nil { + t.Fatalf("write .DS_Store: %v", err) + } + got, err := Discover(root) + if err != nil { + t.Fatalf("Discover: %v", err) + } + if len(got.Images) != 1 { + t.Errorf("len(Images) = %d, want 1; .DS_Store should be filtered", len(got.Images)) + } +} + +func TestDiscover_MissingLabelsCSV(t *testing.T) { + root := t.TempDir() + if err := os.MkdirAll(filepath.Join(root, "images"), 0o755); err != nil { + t.Fatalf("mkdir images: %v", err) + } + if err := os.WriteFile(filepath.Join(root, "images", "a.jpg"), + make([]byte, 100), 0o644); err != nil { + t.Fatalf("write image: %v", err) + } + _, err := Discover(root) + if err == nil { + t.Fatal("Discover returned nil error; expected missing-labels error") + } + if !strings.Contains(err.Error(), "missing labels.csv") { + t.Errorf("error = %q, want it to mention missing labels.csv", err) + } +} + +func TestDiscover_MissingImagesDir(t *testing.T) { + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, "labels.csv"), + []byte("x"), 0o644); err != nil { + t.Fatalf("write labels.csv: %v", err) + } + _, err := Discover(root) + if err == nil { + t.Fatal("Discover returned nil error; expected missing-images-dir error") + } + if !strings.Contains(err.Error(), "missing images/") { + t.Errorf("error = %q, want it to mention missing images/", err) + } +} + +func TestDiscover_NoAcceptedImageExtensions(t *testing.T) { + // images/ exists but only contains .gif and .bmp — neither + // in our accept-set. Customer should see "no image files" + // pointing at the accepted extensions, not a successful walk + // with len(Images)==0 that then succeeds the dry-run. + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, "labels.csv"), + []byte("x"), 0o644); err != nil { + t.Fatalf("write labels.csv: %v", err) + } + imagesDir := filepath.Join(root, "images") + if err := os.MkdirAll(imagesDir, 0o755); err != nil { + t.Fatalf("mkdir images: %v", err) + } + for _, n := range []string{"old.gif", "old.bmp"} { + if err := os.WriteFile(filepath.Join(imagesDir, n), []byte("x"), 0o644); err != nil { + t.Fatalf("write %s: %v", n, err) + } + } + _, err := Discover(root) + if err == nil { + t.Fatal("Discover returned nil error; expected no-images error") + } + if !strings.Contains(err.Error(), "no image files") { + t.Errorf("error = %q, want it to mention no image files", err) + } +} + +func TestDiscover_NotADirectory(t *testing.T) { + // Customer passes a path to a single file instead of a dir. + // This is a common autocomplete-mistake (tab-completing + // into the file). Should be caught early with a clear error. + root := t.TempDir() + filePath := filepath.Join(root, "looks-like-data.tar") + if err := os.WriteFile(filePath, []byte("x"), 0o644); err != nil { + t.Fatalf("write file: %v", err) + } + _, err := Discover(filePath) + if err == nil { + t.Fatal("Discover returned nil error; expected not-a-directory error") + } + if !strings.Contains(err.Error(), "not a directory") { + t.Errorf("error = %q, want it to mention not a directory", err) + } +} + +func TestDiscover_OverSingleFileCap(t *testing.T) { + // Use a fake-size pattern: create a real small file but assert + // the cap logic by exercising the human-readable error format + // at the boundary. We can't easily generate a 500MB+ file in + // CI without slowing the suite — instead pin the human-bytes + // formatter (which is what the customer sees) via its own + // boundary test below, and exercise sizeError() directly. + got := sizeError("images/big.jpg", 600*1024*1024, MaxSingleFileBytes).Error() + for _, want := range []string{"images/big.jpg", "600.00 MiB", "500.00 MiB", "v0.2", "cloud-source"} { + if !strings.Contains(got, want) { + t.Errorf("sizeError missing %q in: %s", want, got) + } + } +} + +func TestHumanBytes(t *testing.T) { + // Boundary check: the formatter is what surfaces in every + // diagnostic, so a regression here makes the error messages + // unreadable for the customer. Pin a few representative values. + cases := []struct { + in int64 + want string + }{ + {0, "0 B"}, + {1023, "1023 B"}, + {1024, "1.00 KiB"}, + {1024 * 1024, "1.00 MiB"}, + {1024 * 1024 * 1024, "1.00 GiB"}, + {500 * 1024 * 1024, "500.00 MiB"}, + } + for _, c := range cases { + if got := humanBytes(c.in); got != c.want { + t.Errorf("humanBytes(%d) = %q, want %q", c.in, got, c.want) + } + } +} From 42400977acf650dc71d560fdaac1798ba7bda59b Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 10:09:27 +0500 Subject: [PATCH 2/5] fix(dataset): drop duplicate pluralS, reuse existing plural helper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cursor Bugbot (Low severity) on PR #8: pluralS() in dataset.go was an exact duplicate of plural() already defined in ingest.go — same cli package, identical body. Removed pluralS, the schema-error diagnostic now calls plural() directly. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 4795daec..1436ca3c 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -206,7 +206,7 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush // JSON output mode) isn't polluted by error text. Exit 2 // is reserved for schema violations across the CLI. _, _ = fmt.Fprintf(errOut, "synthesized spec failed schema validation (%d issue%s):\n", - len(errs), pluralS(len(errs))) + len(errs), plural(len(errs))) _, _ = fmt.Fprintln(errOut, schema.FormatErrors(errs)) return &exitError{code: 2, err: errors.New("synthesized spec failed schema validation; check the flag values above")} } @@ -328,16 +328,6 @@ func printPushPreflight( } } -// pluralS returns "s" for n != 1, else "". Tiny helper that keeps -// the "1 issue" / "3 issues" diagnostic readable without an inline -// ternary. -func pluralS(n int) string { - if n == 1 { - return "" - } - return "s" -} - // humanBytesForSummary mirrors push.humanBytes but lives here to // keep internal/push's API surface narrow (the internal helper is // unexported). Yes, this is a tiny duplication; if a third caller From 837157f5ed54bfc1d4ec98d5112b509845d3f770 Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 10:26:29 +0500 Subject: [PATCH 3/5] fix(push): reject path-traversal table names (Bugbot, Medium) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cursor Bugbot on PR #8 (commit 4240097), Medium severity: the `table` value flows unsanitized into the /data/shared/
/ PVC path. The embedded ingest.v1 schema only enforces minLength:1 on `table` — no pattern — so --table=../../etc would resolve (via path-cleaning) to /etc, and PR-b's stage Pod would write outside the intended subtree, potentially clobbering another table's staged data. Fix: - push.ValidateTableName(table): rejects anything not matching ^[A-Za-z0-9_]+$ — the intersection of "valid unquoted MySQL identifier" and "safe single path segment". Called as step 1 of runDatasetPush, before SpecArgs.Build(). - StagedPrefix hardened: drops path.Join (whose ".."-cleaning is the silent footgun) for plain concatenation, and panics if handed an unsafe name. Callers MUST ValidateTableName first; the panic is a defense-in-depth backstop so a PR-b call site that forgets validation fails loudly in tests instead of silently escaping the prefix. - Build() doc gains the precondition note. The upstream half — adding a `pattern` constraint to the schema's `table` field so the helm flow + jobs-manager get the same guard — is filed as tracebloc/data-ingestors#116. Once that lands and the CLI re-syncs the schema, ValidateTableName collapses to a thin wrapper. Tests: - TestValidateTableName_Accepts / _RejectsUnsafe: the security regression pin (../../etc, ../foo, slashes, dots, etc.) - TestStagedPrefix_PanicsOnUnsafeName: the backstop - TestDatasetPush_TraversalTableName_ExitsTwo: CLI-layer, --table=../../etc → exit 2 before any cluster work Locally: vet, test -race -cover, gofmt -s, errcheck — all green. Coverage: push 83.3%, cluster 83.2%, schema 80.7%, cli 52.2%. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 34 +++++++++----- internal/cli/dataset_test.go | 24 ++++++++++ internal/push/spec.go | 87 +++++++++++++++++++++++++++++++++--- internal/push/spec_test.go | 66 +++++++++++++++++++++++++++ 4 files changed, 194 insertions(+), 17 deletions(-) diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 1436ca3c..30e4331f 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -174,11 +174,23 @@ type runDatasetPushArgs struct { // a bad label-column or oversized dataset gets the diagnostic in // milliseconds without a kubeconfig round-trip. func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPushArgs) error { - // 1. Synthesize the spec from flags + validate against schema. - // Catches "no table", "bad category", "missing intent" etc. - // BEFORE we touch the filesystem or the cluster. The error - // formatter is the same one ingest validate uses, so a - // customer who YAML'd manually first sees identical wording. + // 1. Validate the table name BEFORE anything else. It's both + // the MySQL identifier and the /data/shared/
/ PVC + // subdirectory — an unsanitized traversal name (../../etc) + // would escape that subtree once PR-b's stage Pod writes to + // it. The embedded schema only checks minLength on `table`, + // so this CLI-side guard is the real fix. SpecArgs.Build() + // below calls StagedPrefix, which panics on an unsafe name — + // so this check MUST come first. + if err := push.ValidateTableName(a.Spec.Table); err != nil { + return &exitError{code: 2, err: err} + } + + // 2. Synthesize the spec from flags + validate against schema. + // Catches "bad category", "missing intent" etc. BEFORE we + // touch the filesystem or the cluster. The error formatter + // is the same one ingest validate uses, so a customer who + // YAML'd manually first sees identical wording. spec := a.Spec.Build() specBytes, err := yaml.Marshal(spec) if err != nil { @@ -211,7 +223,7 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return &exitError{code: 2, err: errors.New("synthesized spec failed schema validation; check the flag values above")} } - // 2. Walk the local directory. Enforces layout + size caps; + // 3. Walk the local directory. Enforces layout + size caps; // customer sees a clear pointer to expected layout if they // pass the wrong directory. layout, err := push.Discover(a.LocalPath) @@ -219,7 +231,7 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return &exitError{code: 3, err: err} } - // 3. Cluster discovery — same kubeconfig path as `cluster info`. + // 4. Cluster discovery — same kubeconfig path as `cluster info`. // Errors mirror that command's exit-code contract (3 for // kubeconfig, 4 for missing release) so behaviour is // consistent across pre-flight commands. @@ -243,7 +255,7 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush release.IngestorSAName = a.IngestorSAName } - // 4. PVC discovery. New in this PR — confirms the chart's + // 5. PVC discovery. New in this PR — confirms the chart's // shared-data PVC is Bound before we waste time provisioning // a Pod that can't mount it. pvc, err := cluster.DiscoverSharedPVC(ctx, cs, resolved.Namespace) @@ -251,20 +263,20 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return &exitError{code: 4, err: err} } - // 5. Print the pre-flight summary. The output is the same in + // 6. Print the pre-flight summary. The output is the same in // dry-run and (eventually) live mode — only the "what // happens next" line differs. Customers iterating on a // bad layout see this every attempt, so it's worth keeping // skimmable: one fact per line, aligned by column. printPushPreflight(out, layout, release, pvc, spec, a.DryRun) - // 6. Dry-run stop. Acknowledged success. + // 7. Dry-run stop. Acknowledged success. if a.DryRun { _, _ = fmt.Fprintln(out, "Dry-run complete — no cluster resources were created.") return nil } - // 7. The actual staging branch lands in PR-b. Failing here + // 8. The actual staging branch lands in PR-b. Failing here // rather than silently returning success means a customer // who pulled PR-a's binary and ran without --dry-run gets // a clear "wait for PR-b" signal instead of "0 files diff --git a/internal/cli/dataset_test.go b/internal/cli/dataset_test.go index c9a35fc1..5a4ef69e 100644 --- a/internal/cli/dataset_test.go +++ b/internal/cli/dataset_test.go @@ -99,6 +99,30 @@ func TestDatasetPush_BadCategory_ExitsTwo(t *testing.T) { } } +// TestDatasetPush_TraversalTableName_ExitsTwo is the security +// regression pin at the CLI layer. --table=../../etc must be +// rejected with exit 2 BEFORE any spec synthesis or cluster work — +// the table name flows into the /data/shared/
/ PVC path, +// and a traversal value would let PR-b's stage Pod escape that +// subtree. Bugbot flagged this on PR #8 commit 4240097. +func TestDatasetPush_TraversalTableName_ExitsTwo(t *testing.T) { + root := imgcLayout(t) + for _, bad := range []string{"../../etc", "../foo", "foo/bar"} { + t.Run(bad, func(t *testing.T) { + code, _, _ := execDatasetPush(t, []string{ + root, + "--table=" + bad, + "--category=image_classification", + "--intent=train", + "--label-column=label", + }) + if code != 2 { + t.Fatalf("expected exit 2 for traversal table name %q, got %d", bad, code) + } + }) + } +} + // TestDatasetPush_MissingIntent_ExitsTwo: pins the "intent is // required" diagnostic path — different schema violation but the // same exit-code class. diff --git a/internal/push/spec.go b/internal/push/spec.go index 483c5f53..ec727b05 100644 --- a/internal/push/spec.go +++ b/internal/push/spec.go @@ -16,7 +16,63 @@ // before we touch the cluster"; PR-b is "now actually push the bytes". package push -import "path" +import ( + "fmt" + "path" + "regexp" +) + +// tableNamePattern is the safe character set for a table name. It +// must satisfy TWO independent constraints simultaneously: +// +// 1. A valid unquoted MySQL identifier — the chart's ingestor +// CREATEs a table with this exact name. +// 2. A safe single path segment — the name becomes the +// /data/shared/
/ subdirectory on the PVC. +// +// The intersection of "MySQL identifier" and "single safe path +// component" is [A-Za-z0-9_]: letters, digits, underscore. No +// slashes, no dots — which is what closes the path-traversal hole +// (see ValidateTableName). +// +// All the real-world example tables (chest_xrays_train, +// cats_dogs_train) match this; it's the conventional snake_case +// table-naming style anyway. +var tableNamePattern = regexp.MustCompile(`^[A-Za-z0-9_]+$`) + +// ValidateTableName rejects table names that aren't safe as both a +// MySQL identifier and a single PVC path segment. +// +// Why a CLI-side check rather than the schema: the embedded +// ingest.v1.json only enforces `minLength: 1` on `table` — no +// `pattern`. Without this guard, --table=../../etc would flow into +// the /data/shared/
/ PVC path; PR-b's stage Pod would then +// write outside the intended subtree and could clobber another +// table's data. Tightening the upstream schema with a `pattern` +// is the proper long-term fix (it would protect the helm flow + +// jobs-manager too) but needs a change to tracebloc/data-ingestors' +// schema, which the schema-drift CI check pins — filed as +// tracebloc/data-ingestors#116. Once that lands and we re-sync, +// this guard can collapse to a thin "schema says so" wrapper. +// +// Callers MUST run this before SpecArgs.Build() or StagedPrefix(), +// both of which assume a validated name. +func ValidateTableName(table string) error { + if table == "" { + return fmt.Errorf("table name is required (set --table)") + } + if !tableNamePattern.MatchString(table) { + return fmt.Errorf( + "table name %q is invalid: must match [A-Za-z0-9_]+ "+ + "(letters, digits, underscore only). The table name is "+ + "used both as the MySQL table identifier and as the "+ + "/data/shared/
/ subdirectory on the cluster PVC, "+ + "so slashes, dots, and path-traversal sequences are "+ + "rejected.", + table) + } + return nil +} // SpecArgs is the user-facing flag set for `tracebloc dataset push`. // @@ -73,6 +129,9 @@ type SpecArgs struct { // sides — the CLI's view of "what local files we expect" and the // spec's view of "where they'll live in the cluster" — means a // successful Discover guarantees a runnable spec. +// +// PRECONDITION: a.Table must have passed ValidateTableName. Build +// calls StagedPrefix, which panics on an unsafe name. func (a SpecArgs) Build() map[string]any { prefix := StagedPrefix(a.Table) return map[string]any{ @@ -102,10 +161,26 @@ func (a SpecArgs) Build() map[string]any { // Exported because Phase 3's PR-b (stage Pod construction) needs // it from the same place, and Phase 4 (submit) might want to print // it as part of "what we pushed." +// +// PRECONDITION: table must already have passed ValidateTableName. +// This function panics on an unsafe name rather than returning an +// escape path — a name that escapes /data/shared is a caller bug +// (validation was skipped), and a panic surfaces it loudly in +// tests instead of silently letting PR-b's stage Pod write to, +// say, /etc. Every production call path runs ValidateTableName +// first (see cli.runDatasetPush), so the panic is unreachable in +// correct code. func StagedPrefix(table string) string { - // path.Join collapses redundant slashes but doesn't preserve - // trailing slashes — fine here because callers either append a - // filename (labels.csv) or add the trailing slash explicitly - // (images/). - return path.Join("/data/shared", table) + // Deliberately NOT path.Join here: path.Join cleans ".." + // segments, which is exactly the silent traversal we're + // guarding against. Plain concatenation keeps the name as a + // literal segment so the assertion below can detect a bad one. + prefix := "/data/shared/" + table + if !tableNamePattern.MatchString(table) { + panic(fmt.Sprintf( + "push.StagedPrefix: unsafe table name %q — caller must "+ + "ValidateTableName before constructing a PVC path", + table)) + } + return prefix } diff --git a/internal/push/spec_test.go b/internal/push/spec_test.go index 674aad1c..8f19716f 100644 --- a/internal/push/spec_test.go +++ b/internal/push/spec_test.go @@ -105,3 +105,69 @@ func TestStagedPrefix_PerTableIsolation(t *testing.T) { t.Errorf("StagedPrefix(%q) = %q, want /data/shared/table_a", "table_a", got) } } + +// TestValidateTableName_Accepts pins the names that MUST pass — +// the real-world example tables plus a few edge shapes (single +// char, leading underscore, mixed case, digits). A regression +// that rejects a valid name would break legitimate pushes. +func TestValidateTableName_Accepts(t *testing.T) { + for _, name := range []string{ + "cats_dogs", + "chest_xrays_train", + "t1", + "ABC", + "table_123", + "_leading_underscore", + "9starts_with_digit", // valid MySQL identifier + safe path segment + } { + if err := ValidateTableName(name); err != nil { + t.Errorf("ValidateTableName(%q) = %v, want nil", name, err) + } + } +} + +// TestValidateTableName_RejectsUnsafe is the security-regression +// pin. The path-traversal cases (../../etc, ../foo) are the ones +// Bugbot flagged on PR #8 — if this test ever goes green with +// those removed, the traversal hole is back open. +func TestValidateTableName_RejectsUnsafe(t *testing.T) { + cases := map[string]string{ + "empty": "", + "parent traversal": "../../etc", + "single parent": "../foo", + "embedded slash": "foo/bar", + "embedded dot": "foo.bar", + "bare dot": ".", + "absolute": "/etc/passwd", + "space": "my table", + "dash": "cats-dogs", // not a valid unquoted MySQL identifier + "trailing newline": "foo\n", + "shell metachar": "foo;rm", + "null-ish unicode": "foo\x00bar", + } + for desc, table := range cases { + if err := ValidateTableName(table); err == nil { + t.Errorf("%s: ValidateTableName(%q) = nil, want a rejection error", desc, table) + } + } +} + +// TestStagedPrefix_PanicsOnUnsafeName pins the defense-in-depth +// backstop: if a caller skips ValidateTableName and hands a +// traversal name straight to StagedPrefix, it must panic rather +// than silently return an escape path. PR-b adds new call sites +// for StagedPrefix — this test guards against one of them +// forgetting the validation step. +func TestStagedPrefix_PanicsOnUnsafeName(t *testing.T) { + for _, unsafe := range []string{"../../etc", "foo/bar", ""} { + t.Run(unsafe, func(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Errorf("StagedPrefix(%q) did not panic; an unsafe "+ + "name must panic, not return an escape path", unsafe) + } + }() + _ = StagedPrefix(unsafe) + }) + } +} From 99158662b960c60bf043895fa2b7d000561c6f1a Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 10:46:30 +0500 Subject: [PATCH 4/5] refactor(push): export HumanBytes, drop the copy in dataset.go MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cursor Bugbot on PR #8 (commit 837157f), Low severity: humanBytesForSummary in dataset.go was a line-for-line copy of humanBytes in walk.go — both new in this PR. A future tweak (TiB support, precision change) would likely patch one and not the other, drifting the size shown in an over-cap error from the size shown in the dry-run summary. Fix: export push.HumanBytes as the single implementation; delete the dataset.go copy. Both the over-cap diagnostics and the pre-flight summary now format through the same function. Locally: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 26 ++------------------------ internal/push/walk.go | 14 ++++++++++---- internal/push/walk_test.go | 4 ++-- 3 files changed, 14 insertions(+), 30 deletions(-) diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 30e4331f..d740e0fa 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -308,7 +308,7 @@ func printPushPreflight( _, _ = fmt.Fprintf(out, " root: %s\n", layout.Root) _, _ = fmt.Fprintf(out, " labels.csv: %s\n", layout.LabelsCSV) _, _ = fmt.Fprintf(out, " images: %d files\n", len(layout.Images)) - _, _ = fmt.Fprintf(out, " total size: %s\n", humanBytesForSummary(layout.TotalBytes)) + _, _ = fmt.Fprintf(out, " total size: %s\n", push.HumanBytes(layout.TotalBytes)) _, _ = fmt.Fprintln(out) _, _ = fmt.Fprintf(out, "Target cluster:\n") @@ -334,30 +334,8 @@ func printPushPreflight( if !dryRun { _, _ = fmt.Fprintf(out, "Next: stage %d files (%s) → %s (coming in PR-b for #151)\n", - 1+len(layout.Images), humanBytesForSummary(layout.TotalBytes), + 1+len(layout.Images), push.HumanBytes(layout.TotalBytes), push.StagedPrefix(spec["table"].(string))) _, _ = fmt.Fprintln(out) } } - -// humanBytesForSummary mirrors push.humanBytes but lives here to -// keep internal/push's API surface narrow (the internal helper is -// unexported). Yes, this is a tiny duplication; if a third caller -// shows up, we promote it to a shared util in v0.2. -func humanBytesForSummary(n int64) string { - const ( - KiB = 1024 - MiB = 1024 * KiB - GiB = 1024 * MiB - ) - switch { - case n >= GiB: - return fmt.Sprintf("%.2f GiB", float64(n)/float64(GiB)) - case n >= MiB: - return fmt.Sprintf("%.2f MiB", float64(n)/float64(MiB)) - case n >= KiB: - return fmt.Sprintf("%.2f KiB", float64(n)/float64(KiB)) - default: - return fmt.Sprintf("%d B", n) - } -} diff --git a/internal/push/walk.go b/internal/push/walk.go index c82e4f6b..4b1c54d8 100644 --- a/internal/push/walk.go +++ b/internal/push/walk.go @@ -190,7 +190,7 @@ func Discover(rootDir string) (*LocalLayout, error) { "is on the v0.2 roadmap — see tracebloc/client#147 non-goals. "+ "Workaround for v0.1: split the push into multiple smaller "+ "tables, or stage directly via the existing helm flow.", - humanBytes(layout.TotalBytes), humanBytes(MaxTotalBytes)) + HumanBytes(layout.TotalBytes), HumanBytes(MaxTotalBytes)) } return layout, nil @@ -203,12 +203,18 @@ func sizeError(relPath string, got, cap int64) error { return fmt.Errorf( "file %q is %s, exceeds v0.1 single-file cap of %s. "+ "For larger files, see tracebloc/client#147's v0.2 cloud-source story.", - relPath, humanBytes(got), humanBytes(cap)) + relPath, HumanBytes(got), HumanBytes(cap)) } -// humanBytes renders a byte count in the shortest readable unit. +// HumanBytes renders a byte count in the shortest readable unit. // Not internationalized — the CLI is English-only for v0.1. -func humanBytes(n int64) string { +// +// Exported because the CLI's pre-flight summary (internal/cli) needs +// the identical formatting — keeping one implementation here means +// the size shown in an over-cap error and the size shown in the +// dry-run summary can never drift (Bugbot flagged the earlier +// copy-pasted variant on PR #8). +func HumanBytes(n int64) string { const ( KiB = 1024 MiB = 1024 * KiB diff --git a/internal/push/walk_test.go b/internal/push/walk_test.go index 79597dd9..967bce47 100644 --- a/internal/push/walk_test.go +++ b/internal/push/walk_test.go @@ -214,8 +214,8 @@ func TestHumanBytes(t *testing.T) { {500 * 1024 * 1024, "500.00 MiB"}, } for _, c := range cases { - if got := humanBytes(c.in); got != c.want { - t.Errorf("humanBytes(%d) = %q, want %q", c.in, got, c.want) + if got := HumanBytes(c.in); got != c.want { + t.Errorf("HumanBytes(%d) = %q, want %q", c.in, got, c.want) } } } From b1419f2220631dedee0f75892a09382ddd27fb14 Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 10:55:28 +0500 Subject: [PATCH 5/5] fix(push): reject a directory named labels.csv in Discover MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cursor Bugbot on PR #8 (commit 9915866), Low severity: Discover checked imagesStat.IsDir() for the images/ entry but never the reverse for labels.csv. A directory literally named "labels.csv" passes os.Stat, so the pre-flight would accept it and PR-b's tar stream would later fail confusingly trying to read a directory as a CSV. Added the symmetric labelsStat.IsDir() guard with a clear error. TestDiscover_LabelsCSVIsDirectory pins it. Locally: vet, test -race -cover, gofmt -s, errcheck — all green. Coverage: push 83.8%. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/push/walk.go | 11 +++++++++++ internal/push/walk_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/internal/push/walk.go b/internal/push/walk.go index 4b1c54d8..dd955716 100644 --- a/internal/push/walk.go +++ b/internal/push/walk.go @@ -120,6 +120,17 @@ func Discover(rootDir string) (*LocalLayout, error) { } return nil, fmt.Errorf("stat labels.csv: %w", err) } + if labelsStat.IsDir() { + // A directory literally named "labels.csv" passes the + // os.Stat above — without this check the pre-flight would + // accept it, and PR-b's tar stream would fail confusingly + // trying to read a directory as a CSV. Symmetric with the + // imagesStat.IsDir() check below. + return nil, fmt.Errorf( + "%q is a directory, not a file. labels.csv must be the "+ + "CSV file holding the image_id,label rows.", + labelsPath) + } if labelsStat.Size() > MaxSingleFileBytes { return nil, sizeError("labels.csv", labelsStat.Size(), MaxSingleFileBytes) } diff --git a/internal/push/walk_test.go b/internal/push/walk_test.go index 967bce47..80954152 100644 --- a/internal/push/walk_test.go +++ b/internal/push/walk_test.go @@ -165,6 +165,32 @@ func TestDiscover_NoAcceptedImageExtensions(t *testing.T) { } } +func TestDiscover_LabelsCSVIsDirectory(t *testing.T) { + // A directory literally named "labels.csv" — os.Stat succeeds, + // so without the IsDir guard the pre-flight would accept it and + // PR-b's tar stream would fail confusingly. Symmetric with the + // images/ check. Bugbot flagged the missing guard on PR #8. + root := t.TempDir() + if err := os.MkdirAll(filepath.Join(root, "labels.csv"), 0o755); err != nil { + t.Fatalf("mkdir labels.csv/: %v", err) + } + imagesDir := filepath.Join(root, "images") + if err := os.MkdirAll(imagesDir, 0o755); err != nil { + t.Fatalf("mkdir images/: %v", err) + } + if err := os.WriteFile(filepath.Join(imagesDir, "a.jpg"), + make([]byte, 100), 0o644); err != nil { + t.Fatalf("write image: %v", err) + } + _, err := Discover(root) + if err == nil { + t.Fatal("Discover returned nil error; expected labels.csv-is-a-directory error") + } + if !strings.Contains(err.Error(), "is a directory") { + t.Errorf("error = %q, want it to mention 'is a directory'", err) + } +} + func TestDiscover_NotADirectory(t *testing.T) { // Customer passes a path to a single file instead of a dir. // This is a common autocomplete-mistake (tab-completing