From 87db2e1118056cc0da026235fb6d478f809bb99e Mon Sep 17 00:00:00 2001 From: Lukas Wuttke Date: Tue, 7 Jul 2026 17:24:29 +0200 Subject: [PATCH] feat(data ingest): honest waits + plain-language run output (#172) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `tracebloc data ingest` run worked but spoke Kubernetes and went silent at the worst moment. Three UX fixes surfaced by the ingest-flow audit: 1. Silent hang after submit. WatchJob's waitForJobPod polls up to 5 min (PodReadyTimeout) for the ingestor Pod to schedule + pull its image, printing nothing — the CLI looked hung right after the key step. Now a live spinner ("Waiting for the ingestion to start…", with a Ctrl-C hint) covers that wait, then a clean "Ingestion started — live progress:" precedes the log stream. WatchJob takes a nil-safe *ui.Printer for this; nil (tests) stays silent as before. 2. Undisclosed 1-hour watch cap. JobWatchTimeout is 1h; on expiry the CLI silently "detached". Step 4 now discloses the follow window up front — and only when NOT --detach (the hints used to promise streaming even under --detach, which was itself dishonest). 3. Kubernetes jargon on the happy path. Reworded the run narrative to plain language: upload-channel wording instead of "stage Pod", "Submitted — tracebloc is validating your data…", "Connecting to your workspace to submit the run…", and plain detach/timeout messages. The raw `kubectl logs` reconnect stays (there is no CLI re-attach verb yet) but as a labelled optional follow, not jargon in a sentence. From the pre-PR adversarial review: the green "✔ Ingestion started" now only shows for a live/completed pod — a Pod already Failed (immediate crash, or a prior backoffLimit retry) gets a neutral "streaming logs:" line instead of a success checkmark before its crash output (waitForJobPod now returns the selected pod's phase). The pod-wait-timeout detach message no longer claims "you don't need to do anything" (false for the PSA-rejection / unschedulable subcases the same path covers). No control-flow or exit-code changes — output copy + one nil-safe progress-reporter param. Tests updated for the reworded strings + the new waitForJobPod phase return. Closes #172. Part of the data ingest UX sweep (epic #67). Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/cli/data.go | 17 +++++---- internal/push/stage.go | 10 +++--- internal/push/stage_test.go | 2 +- internal/push/stream.go | 3 +- internal/submit/submit.go | 65 ++++++++++++++++------------------ internal/submit/submit_test.go | 14 ++++---- internal/submit/summary.go | 2 +- internal/submit/watch.go | 50 +++++++++++++++++++++----- internal/submit/watch_test.go | 27 +++++++++----- 9 files changed, 117 insertions(+), 73 deletions(-) diff --git a/internal/cli/data.go b/internal/cli/data.go index cca5bdc1..2a681770 100644 --- a/internal/cli/data.go +++ b/internal/cli/data.go @@ -641,8 +641,8 @@ other collaborators train against it without ever seeing the raw files.`)) // Errors mirror that command's exit-code contract (3 for // kubeconfig, 4 for missing release) so behaviour is // consistent across pre-flight commands. - a.Printer.Step(2, 4, "Connect to your workspace's cluster") - a.Printer.Hintf("Using your kubeconfig to find the tracebloc release in your workspace and the shared storage your dataset will live on.") + a.Printer.Step(2, 4, "Connect to your workspace") + a.Printer.Hintf("Finding your tracebloc workspace and the shared storage your dataset will live on.") // 6. PVC discovery (needPVC) confirms the chart's shared-data PVC is // Bound before we waste time provisioning a Pod that can't mount it. opts := cluster.KubeconfigOptions{Path: a.Kubeconfig, Context: a.Context, Namespace: a.Namespace} @@ -734,7 +734,7 @@ other collaborators train against it without ever seeing the raw files.`)) // pre-flight codes so customers can branch on whether the // failure was their environment vs the actual data transfer. a.Printer.Step(3, 4, "Stage your files") - a.Printer.Hintf("A short-lived helper pod mounts the shared storage and your files stream into it — like `kubectl cp`, but set up and cleaned up for you.") + a.Printer.Hintf("Your files upload securely into your workspace's storage — set up and cleaned up for you.") progress := push.NewProgress(out, layout.TotalBytes, fmt.Sprintf("Staging %s", a.Spec.Table)) // Defer Finish so a failure path that returns BEFORE @@ -771,7 +771,12 @@ other collaborators train against it without ever seeing the raw files.`)) // + log stream — can run that long for large ingestions. // The chart's helm flow uses the same token-mint code path. a.Printer.Step(4, 4, "Run the ingestion") - a.Printer.Hintf("Submitting the run to your workspace, then watching as it validates your data and loads it into the table — progress streams below.") + 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 := cluster.MintIngestorToken(ctx, cs, resolved.Namespace, release.IngestorSAName, 3600, nil) if err != nil { @@ -785,7 +790,7 @@ other collaborators train against it without ever seeing the raw files.`)) // through the kubeconfig-authenticated apiserver, same as // `kubectl port-forward`. Bugbot PR #10 r3 caught the // original broken-by-design direct-URL POST. - _, _ = fmt.Fprintln(out, "Opening port-forward to jobs-manager...") + a.Printer.Infof("Connecting to your workspace to submit the run…") pf, err := submit.PortForwardJobsManager(ctx, cs, resolved.RestConfig, resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort) if err != nil { @@ -981,7 +986,7 @@ func printLocalSummary(p *ui.Printer, layout *push.LocalLayout, spec map[string] } // printClusterSummary shows the discovered workspace cluster target — -// the detail under step 2 ("Connect to your workspace's cluster"). +// the detail under step 2 ("Connect to your workspace"). func printClusterSummary(p *ui.Printer, release *cluster.ParentRelease, pvc *cluster.SharedPVC) { p.Section("Target cluster") p.Field("release", fmt.Sprintf("%s (chart %s)", release.ReleaseName, release.ChartVersion)) diff --git a/internal/push/stage.go b/internal/push/stage.go index eed58850..d52812b9 100644 --- a/internal/push/stage.go +++ b/internal/push/stage.go @@ -50,7 +50,7 @@ type StageOptions struct { Progress Progress // Out is where Stage prints non-progress diagnostic output - // (orphan warnings, "Created stage Pod X" etc.). Nil = io.Discard. + // (orphan warnings, upload-channel status, etc.). Nil = io.Discard. // Progress bar output is separate — schollz writes to its own // configured writer (typically the same one). Out io.Writer @@ -106,7 +106,7 @@ func Stage(ctx context.Context, opts StageOptions) error { if err != nil { return err } - _, _ = fmt.Fprintf(opts.Out, "Created stage Pod %s/%s\n", opts.Namespace, podName) + _, _ = fmt.Fprintf(opts.Out, "Opened a secure upload channel to your workspace.\n") // 3. Defer cleanup. The deferred call uses a FRESH context with // its own deadline — if the parent ctx is cancelled (SIGINT, @@ -130,14 +130,14 @@ func Stage(ctx context.Context, opts StageOptions) error { // accept exec). Times out at StagePodReadyTimeout with // diagnostic hints from container statuses (image pull, // scheduling) when we have them. - _, _ = fmt.Fprintf(opts.Out, "Waiting for stage Pod to be Ready (timeout %s)...\n", StagePodReadyTimeout) + _, _ = fmt.Fprintf(opts.Out, "Preparing the upload channel (up to %s)…\n", StagePodReadyTimeout) if _, err := WaitForStagePodReady(ctx, opts.Client, opts.Namespace, podName); err != nil { return err } // 5. Stream the tar. This is where actual bytes flow. The // progress bar (if TTY) renders during this call. - _, _ = fmt.Fprintf(opts.Out, "Streaming %d files (%s) for table %q...\n", + _, _ = fmt.Fprintf(opts.Out, "Uploading %d files (%s) into %q…\n", opts.Layout.FileCount(), HumanBytes(opts.Layout.TotalBytes), opts.Table) if err := StreamLayout(ctx, opts.Executor, @@ -147,7 +147,7 @@ func Stage(ctx context.Context, opts StageOptions) error { } // 6. Print "done" message. The deferred cleanup runs after this. - _, _ = fmt.Fprintf(opts.Out, "Staged %d files for table %q\n", + _, _ = fmt.Fprintf(opts.Out, "Uploaded %d files into %q.\n", opts.Layout.FileCount(), opts.Table) return nil } diff --git a/internal/push/stage_test.go b/internal/push/stage_test.go index b46e0cbb..0393249a 100644 --- a/internal/push/stage_test.go +++ b/internal/push/stage_test.go @@ -82,7 +82,7 @@ func TestStage_HappyPath(t *testing.T) { // Diagnostic output: customer should see the lifecycle // breadcrumbs. Pin a few key phrases so a regression that // silences them is caught. - for _, want := range []string{"Created stage Pod", "Waiting for stage Pod", "Streaming", "Staged"} { + for _, want := range []string{"Opened a secure upload channel", "Preparing the upload channel", "Uploading", "Uploaded"} { if !strings.Contains(out.String(), want) { t.Errorf("output missing %q:\n%s", want, out.String()) } diff --git a/internal/push/stream.go b/internal/push/stream.go index bdeff331..d0b194f4 100644 --- a/internal/push/stream.go +++ b/internal/push/stream.go @@ -192,7 +192,8 @@ func StreamLayout( // next command regardless of prior status, and `|| true` then // forces exit 0. A failed tar mid-script would silently // return success to the exec subprocess, and the CLI would - // report "Staged N files" on what was actually a failed push. + // report the "Uploaded N files" success line on what was + // actually a failed push. // // set -e fixes that: any unguarded non-zero exits the script // with that status. The find's `|| true` is still fine diff --git a/internal/submit/submit.go b/internal/submit/submit.go index 7611d90c..adc03c68 100644 --- a/internal/submit/submit.go +++ b/internal/submit/submit.go @@ -101,34 +101,38 @@ func Run(ctx context.Context, opts Options) (*Result, error) { return nil, err } - // 201 announcement. Customer sees this whether --detach is - // set or not, so they have the Job name for kubectl-poke - // follow-up. + // A Printer for the human-facing status lines (spinner, detach + // notes, summary). nil-safe fallback so callers that didn't thread + // the --plain decision still get sensible auto-detected rendering. + p := opts.Printer + if p == nil { + p = ui.New(opts.Out) + } + + // Submission announcement. Customers see this whether or not + // --detach is set. The raw namespace/job identifiers are kept only + // where they're actionable (detach + replay, which need them to + // reconnect) — not on the happy streaming path. if resp.Replay { - _, _ = fmt.Fprintf(opts.Out, - "Replayed: idempotency key matches a previous run; attaching to existing Job %s/%s\n", - resp.Namespace, resp.JobName) + p.Infof("This matches a previous run (same idempotency key) — attaching to the run already in progress.") } else { - _, _ = fmt.Fprintf(opts.Out, - "Submitted: jobs-manager spawned ingestor Job %s/%s\n", - resp.Namespace, resp.JobName) + p.Successf("Submitted — tracebloc is validating your data and loading it into the table.") } if opts.Detach { - // --detach: print the reconnect hint and bail. The - // cluster continues without us; the customer can come - // back with `kubectl logs -f -n job/`. - _, _ = fmt.Fprintf(opts.Out, - "Detached (no log streaming). Reconnect with: kubectl logs -f -n %s job/%s\n", - resp.Namespace, resp.JobName) + // --detach: report and bail. The run continues in the cluster; + // there is no CLI re-attach verb yet, so the honest way back is + // the raw log follow, offered as a labelled command. + p.Infof("Detached — the ingestion runs in the background on your workspace.") + p.Hintf("Follow it later with: kubectl logs -f -n %s job/%s", resp.Namespace, resp.JobName) return &Result{Submit: resp}, nil } // Watch loop. ctx propagates SIGINT cancellation (main.go's // signal.NotifyContext); a Ctrl-C during the watch produces - // Outcome=Detached + the reconnect hint below. - _, _ = fmt.Fprintf(opts.Out, "Streaming logs from Job %s/%s:\n", resp.Namespace, resp.JobName) - wr, err := WatchJob(ctx, opts.Client, resp.Namespace, resp.JobName, opts.Out) + // Outcome=Detached + the reconnect hint below. WatchJob renders its + // own start spinner + "Ingestion started" header via p. + wr, err := WatchJob(ctx, opts.Client, resp.Namespace, resp.JobName, opts.Out, p) if err != nil { // Tag as WatchError so the orchestrator picks the // ingest-flavored exit code (9), not the submit-flavored @@ -144,24 +148,19 @@ func Run(ctx context.Context, opts Options) (*Result, error) { // PodReadyTimeout or 1-hour JobWatchTimeout aren't pressing // Ctrl-C, so attributing it to "signal" was wrong. if wr.Outcome == JobOutcomeDetached { - _, _ = fmt.Fprintln(opts.Out) + p.Newline() switch wr.DetachReason { case DetachReasonSignal: - _, _ = fmt.Fprintln(opts.Out, "Detached on signal.") + p.Infof("Stopped watching — the ingestion keeps running on your workspace.") case DetachReasonPodWaitTimeout: - _, _ = fmt.Fprintln(opts.Out, - "Detached: the ingestor Pod didn't reach Ready within the observation window. "+ - "This usually means slow image pull or scheduling backlog — the run is in "+ - "jobs-manager's queue and will execute when the Pod starts.") + p.Infof("The ingestion hasn't started yet (usually a slow image pull or a busy cluster). " + + "It's queued to run once the cluster can schedule it — check on it with the command below.") case DetachReasonWatchCap: - _, _ = fmt.Fprintln(opts.Out, - "Detached: the watch window (1 hour) elapsed while the ingestion was still running.") + p.Infof("Stopped following after 1 hour — the ingestion is still running and will finish on its own.") default: - _, _ = fmt.Fprintln(opts.Out, "Detached.") + p.Infof("Stopped watching — the ingestion keeps running on your workspace.") } - _, _ = fmt.Fprintf(opts.Out, - "Ingestion continues in the cluster. Reconnect with: kubectl logs -f -n %s job/%s\n", - resp.Namespace, resp.JobName) + p.Hintf("Check on it later with: kubectl logs -f -n %s job/%s", resp.Namespace, resp.JobName) return &Result{Submit: resp, Watch: wr}, nil } @@ -169,11 +168,7 @@ func Run(ctx context.Context, opts Options) (*Result, error) { // Both Succeeded and Failed paths print it — on Failed, the // banner tells the customer what got partially through. if wr.Summary != nil { - p := opts.Printer - if p == nil { - p = ui.New(opts.Out) - } - _, _ = fmt.Fprintln(opts.Out) + p.Newline() RenderSummary(p, wr.Summary) } diff --git a/internal/submit/submit_test.go b/internal/submit/submit_test.go index cf34126d..0eeadf6e 100644 --- a/internal/submit/submit_test.go +++ b/internal/submit/submit_test.go @@ -54,8 +54,8 @@ func TestRun_DetachPath_HappyPath(t *testing.T) { t.Errorf("Result.Watch = %+v, want nil (detach skips watch)", res.Watch) } for _, want := range []string{ - "Submitted: jobs-manager spawned ingestor Job tracebloc/ingestor-abc", - "Detached (no log streaming)", + "Submitted — tracebloc is validating your data", + "Detached — the ingestion runs in the background", "kubectl logs -f -n tracebloc job/ingestor-abc", } { if !strings.Contains(out.String(), want) { @@ -65,8 +65,8 @@ func TestRun_DetachPath_HappyPath(t *testing.T) { } // TestRun_ReplayPath: replay=true changes the announcement -// wording — "attaching to existing Job" instead of "spawned" — -// because the cluster is already doing the work. +// wording — "attaching to the run already in progress" — because the +// cluster is already doing the work. func TestRun_ReplayPath(t *testing.T) { sub := &fakeSubmitter{ resp: &SubmitResponse{ @@ -87,10 +87,10 @@ func TestRun_ReplayPath(t *testing.T) { if err != nil { t.Fatalf("Run: %v", err) } - if !strings.Contains(out.String(), "Replayed:") { - t.Errorf("output missing Replayed framing:\n%s", out.String()) + if !strings.Contains(out.String(), "matches a previous run") { + t.Errorf("output missing replay framing:\n%s", out.String()) } - if !strings.Contains(out.String(), "attaching to existing Job") { + if !strings.Contains(out.String(), "attaching to the run already in progress") { t.Errorf("output missing replay-specific wording:\n%s", out.String()) } } diff --git a/internal/submit/summary.go b/internal/submit/summary.go index 2be77720..eb0a40b7 100644 --- a/internal/submit/summary.go +++ b/internal/submit/summary.go @@ -359,7 +359,7 @@ func RenderSummary(p *ui.Printer, s *Summary) { p.Section("What's next") p.Infof("View it in the dashboard: https://ai.tracebloc.io/metadata") - p.Hintf("The table is staged and ready for training jobs.") + p.Hintf("Your dataset is ready for training jobs.") } // commaSep formats an int64 with thousands-separator commas to diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 3ef8d477..755d1cb5 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -14,6 +14,8 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes" + + "github.com/tracebloc/cli/internal/ui" ) // Watch-loop tunables. Both deliberately conservative — Phase 4's @@ -107,9 +109,9 @@ type WatchResult struct { // DetachReason qualifies the Detached outcome — set only // when Outcome == JobOutcomeDetached. Lets the orchestrator - // print accurate diagnostics ("Detached on signal" vs - // "Pod didn't become Ready within timeout" vs "Watch cap - // exceeded") instead of the previous one-size-fits-all + // print an accurate per-reason diagnostic (signal → stopped + // watching, pod-wait timeout → not started yet, watch cap → + // stopped following after 1 hour) instead of one one-size-fits-all // message. Bugbot PR #10 r7 flagged the misleading "signal" // framing for the timeout-detach paths. DetachReason DetachReason @@ -157,11 +159,18 @@ const ( // out is the customer-facing log stream (typically os.Stdout). // Logs are written verbatim — no prefix, no munging — so the // stream looks identical to `kubectl logs -f `. +// +// p (may be nil) renders a live spinner during the otherwise-silent +// wait for the ingestor Pod to schedule + pull its image — without it +// the CLI printed one line and then went quiet for up to PodReadyTimeout +// (5 min), looking hung. When p is nil (tests, --output-json-to-a-pipe) +// the wait is silent as before. func WatchJob( ctx context.Context, cs kubernetes.Interface, namespace, jobName string, out io.Writer, + p *ui.Printer, ) (*WatchResult, error) { // Keep the customer's original ctx separately so finalJobStatus // can derive a FRESH 30s context from it (rather than inheriting @@ -180,8 +189,19 @@ func WatchJob( // 1. Wait for the ingestor Job's Pod to exist + reach Running. // jobs-manager creates the Job and Kubernetes spawns the // Pod asynchronously, so the Pod usually isn't there the - // moment after the 201 comes back. - podName, err := waitForJobPod(watchCtx, cs, namespace, jobName) + // moment after the 201 comes back — and scheduling + image + // pull can take minutes. A spinner keeps that wait honest + // instead of looking hung (the pre-spinner behaviour). + var startSpin *ui.Spinner + if p != nil { + startSpin = p.Spinner( + "Waiting for the ingestion to start (scheduling + pulling the image)", + "Ctrl-C to stop watching — the run keeps going on the cluster") + } + podName, podPhase, err := waitForJobPod(watchCtx, cs, namespace, jobName) + if startSpin != nil { + startSpin.Stop() + } if err != nil { if errors.Is(err, context.Canceled) { // SIGINT before the Pod even appeared. jobs-manager @@ -219,6 +239,18 @@ func WatchJob( // a side-channel to the summary parser so we end up with // a structured representation of the banner without // requiring a second log fetch post-completion. + if p != nil { + // Don't success-frame a pod that's already Failed (immediate + // crash, or a Failed pod left by a prior backoffLimit retry that + // bestPod selected) — its crash logs are about to stream and + // finalJobStatus will report the failure. A neutral line for that + // case; the green ✔ only for a live/completed pod. + if podPhase == corev1.PodFailed { + p.Infof("Ingestion started — streaming logs:") + } else { + p.Successf("Ingestion started — live progress:") + } + } summary, logErr := streamPodLogsAndParse(watchCtx, cs, namespace, podName, out) // 3. Detach branches — checked FIRST, since the customer's SIGINT @@ -326,8 +358,9 @@ func WatchJob( // Pod has reached Phase=Running. The selection key is the // `job-name=` label that batch/v1 controllers attach to // every Pod they create. -func waitForJobPod(ctx context.Context, cs kubernetes.Interface, namespace, jobName string) (string, error) { +func waitForJobPod(ctx context.Context, cs kubernetes.Interface, namespace, jobName string) (string, corev1.PodPhase, error) { var podName string + var podPhase corev1.PodPhase err := wait.PollUntilContextTimeout(ctx, PodPollInterval, PodReadyTimeout, true, func(ctx context.Context) (bool, error) { pods, err := cs.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ @@ -374,12 +407,13 @@ func waitForJobPod(ctx context.Context, cs kubernetes.Interface, namespace, jobN return false, nil // all Pods still Pending } podName = bestPod.Name + podPhase = bestPod.Status.Phase return true, nil }) if err != nil { - return "", err + return "", "", err } - return podName, nil + return podName, podPhase, nil } // streamPodLogsAndParse opens a streaming log read on the Pod and diff --git a/internal/submit/watch_test.go b/internal/submit/watch_test.go index 4da6fb9a..5903c793 100644 --- a/internal/submit/watch_test.go +++ b/internal/submit/watch_test.go @@ -40,13 +40,16 @@ func TestWaitForJobPod_RunningPodSurfaces(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - name, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor-abc") + name, phase, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor-abc") if err != nil { t.Fatalf("waitForJobPod: %v", err) } if name != "ingestor-abc-xyz" { t.Errorf("name = %q, want ingestor-abc-xyz", name) } + if phase != corev1.PodRunning { + t.Errorf("phase = %q, want Running (drives the success-vs-neutral start line)", phase) + } } // TestWaitForJobPod_PicksMostRecentNotFirst: Jobs with retries @@ -68,7 +71,7 @@ func TestWaitForJobPod_PicksMostRecentNotFirst(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - name, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") + name, phase, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") if err != nil { t.Fatalf("waitForJobPod: %v", err) } @@ -76,6 +79,9 @@ func TestWaitForJobPod_PicksMostRecentNotFirst(t *testing.T) { t.Errorf("name = %q, want ingestor-new-running "+ "(most-recent useful-phase Pod, not items[0])", name) } + if phase != corev1.PodRunning { + t.Errorf("phase = %q, want Running (the newer pod's phase, not the old Failed one)", phase) + } } // TestWaitForJobPod_AllPendingKeepsPolling: if every Pod is still @@ -88,7 +94,7 @@ func TestWaitForJobPod_AllPendingKeepsPolling(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - _, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") + _, _, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") if err == nil { t.Fatal("waitForJobPod returned nil on all-Pending; expected DeadlineExceeded") } @@ -107,13 +113,16 @@ func TestWaitForJobPod_FastCompletionPath(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - name, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") + name, phase, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") if err != nil { t.Fatalf("waitForJobPod on Succeeded: %v", err) } if name != "ingestor-fast" { t.Errorf("name = %q, want ingestor-fast", name) } + if phase != corev1.PodSucceeded { + t.Errorf("phase = %q, want Succeeded", phase) + } } // TestWaitForJobPod_ForbiddenIsTerminal: an RBAC denial on @@ -133,7 +142,7 @@ func TestWaitForJobPod_ForbiddenIsTerminal(t *testing.T) { defer cancel() start := time.Now() - _, err := waitForJobPod(ctx, cs, "tracebloc", "j") + _, _, err := waitForJobPod(ctx, cs, "tracebloc", "j") elapsed := time.Since(start) if err == nil { t.Fatal("waitForJobPod returned nil on Forbidden") @@ -153,7 +162,7 @@ func TestWaitForJobPod_NoPodEverShows(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - _, err := waitForJobPod(ctx, cs, "tracebloc", "missing-job") + _, _, err := waitForJobPod(ctx, cs, "tracebloc", "missing-job") if err == nil { t.Fatal("waitForJobPod returned nil when no Pod ever appeared") } @@ -251,7 +260,7 @@ func TestWatchJob_PodWaitTimeoutMapsToDetach(t *testing.T) { defer cancel() var out bytes.Buffer - wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor-stuck", &out) + wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor-stuck", &out, nil) if err != nil { t.Fatalf("WatchJob returned error on Pod-wait timeout; want nil + Detached: %v", err) } @@ -282,7 +291,7 @@ func TestWatchJob_TerminalJobStatusWins(t *testing.T) { defer cancel() var out bytes.Buffer - wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor", &out) + wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor", &out, nil) if err != nil { t.Fatalf("WatchJob returned error; want nil + Succeeded: %v", err) } @@ -307,7 +316,7 @@ func TestWatchJob_TerminalFailedJobReported(t *testing.T) { defer cancel() var out bytes.Buffer - wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor", &out) + wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor", &out, nil) if err != nil { t.Fatalf("WatchJob returned error; want nil + Failed: %v", err) }