From 8b5ff6d0008b6b09ab9b771aea3c1cd5a5050f81 Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 20:21:59 +0500 Subject: [PATCH 1/8] =?UTF-8?q?feat(dataset):=20Phase=204=20=E2=80=94=20su?= =?UTF-8?q?bmit=20to=20jobs-manager=20+=20watch=20+=20summary=20(#152)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the v0.1 happy path. After Phase 3 stages files on the PVC, Phase 4 POSTs the ingest spec to jobs-manager, watches the spawned ingestor Job, and renders the parsed INGESTION SUMMARY panel. Customer sees a single end-to-end flow: $ tracebloc dataset push ./cats-dogs --table=cats_dogs_train ... Local dataset: ... Target cluster: ... Synthesized ingest spec: ... Created stage Pod tracebloc/tracebloc-stage-... Streaming N files (X MiB) to /data/shared/cats_dogs_train/... Staged N files to /data/shared/cats_dogs_train Submitted: jobs-manager spawned ingestor Job tracebloc/ingestor-... Streaming logs from Job tracebloc/ingestor-...: [ingestor's own log output streams here, verbatim] ┌─ Ingestion summary ──────────────────────────┐ │ Ingestor ID: run-abc │ │ Total records: 1,234 │ │ Inserted: 1,200 │ │ ... etc │ └──────────────────────────────────────────────┘ ## What lands internal/submit/ (5 new files): - body.go SubmitRequest struct (ingest_config + idempotency_key + optional image_digest); BuildRequest auto-generates a 16-byte hex idempotency key per invocation unless --idempotency-key overrides. - client.go HTTPSubmitter with bearer-token auth; 4xx/5xx surface jobs-manager's body verbatim via SubmitError; IsAuthError(err) for the 401/403 exit-code branch. - watch.go Poll-based watch (per the design discussion): wait for the Job's Pod to exist + Running, stream logs via GetLogs(Follow=true), terminal poll on Job conditions to distinguish Succeeded/Failed/Unknown. - summary.go Streaming parser for the 📊 INGESTION SUMMARY banner. Strips ANSI codes, accumulates fields by prefix match, finalizes on the closing ═-rule. RenderPanel formats the parsed Summary as a box-drawn panel. - submit.go Run() orchestrator: BuildRequest → POST → announce (Spawned vs Replayed) → either --detach exit or watch loop. SIGINT during watch produces Outcome=Detached with a kubectl-logs reconnect hint. internal/cli/dataset.go: wired after push.Stage. Mints a 1-hour SA token (vs cluster info's 10 min — the watch loop can run that long). Adds --detach, --idempotency-key, --image-digest flags. Maps Submit + Watch outcomes to exit codes 5/8/9. ## Design choices (per pre-build discussion) - Poll every 2s for Job status (not client-go Watch). Matches the Phase 3 WaitForStagePodReady pattern; simpler test surface, no long-lived connection to babysit. ~30 API calls/min is negligible. - SIGINT-after-201 → graceful detach. The cluster has already accepted the run; Ctrl-C just stops watching. Customer sees the kubectl logs reconnect hint and exit 0. Matches kubectl logs behavior. ## Exit codes (full set, v0.1) 0 — success (or --detach success: 201 came back clean) 2 — schema / v0.1-unsupported-category 3 — local-layout or kubeconfig 4 — cluster discovery (parent release or shared PVC missing) 5 — SA token couldn't be obtained, OR jobs-manager 401/403 7 — Phase 3 stage step failed 8 — jobs-manager 4xx/5xx other than auth 9 — ingestion Job exited non-zero, OR summary reports row failures ## Tests submit/ ≥85% line coverage across 5 files. Per-file: - body_test.go idempotency-key generation, override, uniqueness, JSON shape pinning (image_digest omitempty) - client_test.go httptest.Server: 201 happy path, replay path, 4xx body framing, 401/403 → IsAuthError, 5xx ≠ auth, missing-job_name, network drop, ctx-cancel - summary_test.go Real ingestor banner end-to-end, byte-by-byte streamed feed, no-banner path (early failure), post-banner lines ignored, ANSI stripping, panel rendering, comma formatting - watch_test.go Fake clientset: Running Pod surfaces, fast- completion path, Forbidden short-circuits, no-Pod timeout, Job Conditions to outcome, parserWriter contract - submit_test.go Run orchestrator: --detach happy path, replay, SubmitError propagation, request fields plumb through, nil-Out defaults to Discard, IsAuthError Local: vet, test -race -cover, gofmt -s, errcheck — all green. Phase 5 (#153: GitHub Releases, Homebrew tap, install.sh) is the last item on the v0.1 epic. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 108 ++++++++- internal/submit/body.go | 107 +++++++++ internal/submit/body_test.go | 100 +++++++++ internal/submit/client.go | 185 +++++++++++++++ internal/submit/client_test.go | 234 +++++++++++++++++++ internal/submit/submit.go | 161 +++++++++++++ internal/submit/submit_test.go | 205 +++++++++++++++++ internal/submit/summary.go | 384 ++++++++++++++++++++++++++++++++ internal/submit/summary_test.go | 240 ++++++++++++++++++++ internal/submit/watch.go | 360 ++++++++++++++++++++++++++++++ internal/submit/watch_test.go | 220 ++++++++++++++++++ 11 files changed, 2295 insertions(+), 9 deletions(-) create mode 100644 internal/submit/body.go create mode 100644 internal/submit/body_test.go create mode 100644 internal/submit/client.go create mode 100644 internal/submit/client_test.go create mode 100644 internal/submit/submit.go create mode 100644 internal/submit/submit_test.go create mode 100644 internal/submit/summary.go create mode 100644 internal/submit/summary_test.go create mode 100644 internal/submit/watch.go create mode 100644 internal/submit/watch_test.go diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 6034e03c..3a75dcbb 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -12,6 +12,7 @@ import ( "github.com/tracebloc/cli/internal/cluster" "github.com/tracebloc/cli/internal/push" "github.com/tracebloc/cli/internal/schema" + "github.com/tracebloc/cli/internal/submit" ) // newDatasetCmd wires the `tracebloc dataset` subtree. The dominant @@ -89,6 +90,16 @@ func newDatasetPushCmd() *cobra.Command { // Pin by digest in your override too — tag-only references // drift silently and break "all my pushes worked yesterday." stagePodImage string + + // Phase 4 flags. --detach exits immediately after the 201 + // from jobs-manager; --idempotency-key plumbs through to + // the submit body for retry-safety across CLI invocations + // (default: fresh per call); --image-digest pins the + // ingestor image (default: jobs-manager picks the + // cluster-configured one). + detach bool + idempotencyKey string + imageDigest string ) cmd := &cobra.Command{ @@ -113,13 +124,18 @@ datasets need the v0.2 cloud-source story (S3/GCS/HTTPS sources) — see tracebloc/client#147 non-goals. Exit codes: - 0 files staged successfully (Phase 4 will add: submitted + completed) + 0 files staged + ingested successfully (or --detach: just staged + submitted) 2 schema validation failed (synthesized spec rejected) or v0.1-unsupported category passed 3 local-layout or kubeconfig error 4 cluster reachable but parent release / shared PVC missing + 5 ingestor SA token couldn't be obtained, or jobs-manager + rejected the token (401/403) 7 pre-flight succeeded but staging the files failed - (Pod creation, image pull, exec stream, or remote tar error)`, + (Pod creation, image pull, exec stream, or remote tar error) + 8 jobs-manager rejected the submit (4xx/5xx other than auth) + 9 ingestion Job exited non-zero, or completed with row-level + failures the summary panel reports`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { return runDatasetPush(cmd.Context(), cmd.OutOrStdout(), cmd.ErrOrStderr(), @@ -132,6 +148,9 @@ Exit codes: DryRun: dryRun, IngestorSAName: ingestorSAName, StagePodImage: stagePodImage, + Detach: detach, + IdempotencyKey: idempotencyKey, + ImageDigest: imageDigest, }) }, } @@ -166,6 +185,17 @@ Exit codes: "override the ephemeral stage Pod's image (default: digest-pinned alpine 3.20 baked into the CLI). "+ "Pin by digest in your override too — tag-only refs drift silently.") + cmd.Flags().BoolVar(&detach, "detach", false, + "exit immediately after jobs-manager accepts the run (no log streaming, no summary panel). "+ + "Use for CI scenarios; reconnect later with `kubectl logs -f -n job/`.") + cmd.Flags().StringVar(&idempotencyKey, "idempotency-key", "", + "reuse this idempotency key across retry attempts (default: fresh per invocation). "+ + "jobs-manager treats a duplicate key as a replay and attaches to the existing Job "+ + "rather than spawning a new one — useful for at-most-once-across-attempts semantics.") + cmd.Flags().StringVar(&imageDigest, "image-digest", "", + "pin the ingestor container image to a specific digest (default: jobs-manager picks the "+ + "cluster-configured `images.ingestor.digest`). Format: sha256:.") + return cmd } @@ -182,6 +212,12 @@ type runDatasetPushArgs struct { DryRun bool IngestorSAName string StagePodImage string + + // Phase 4 (#152) fields. See the flag declarations for the + // per-knob rationale; all three are optional. + Detach bool + IdempotencyKey string + ImageDigest string } // runDatasetPush is the full Phase 3 implementation: pre-flight @@ -350,14 +386,68 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return &exitError{code: 7, err: stageErr} } - // 10. Phase 4 (submit to jobs-manager + watch + summary) hooks - // in here — tracebloc/client#152. For now PR-b leaves the - // dataset staged on the PVC and exits 0, which the customer - // can then chase manually via `helm install ingestor ...` if - // they need the ingestion to actually run today. + // 10. Mint the SA token Phase 4 uses to authenticate the POST + // to jobs-manager. Expiry is 1 hour (vs cluster info's 10 + // min) because the full Phase 4 lifecycle — submit + watch + // + log stream — can run that long for large ingestions. + // The chart's helm flow uses the same token-mint code path. _, _ = fmt.Fprintln(out) - _, _ = fmt.Fprintln(out, "Dataset staged. Submission to jobs-manager arrives in Phase 4 (#152);") - _, _ = fmt.Fprintln(out, "in the meantime, the existing helm ingestor flow can pick up the staged files.") + tok, err := cluster.MintIngestorToken(ctx, cs, resolved.Namespace, + release.IngestorSAName, 3600, nil) + if err != nil { + return &exitError{code: 5, err: err} + } + + // 11. Phase 4: POST to jobs-manager, watch the spawned ingestor + // Job, render the parsed INGESTION SUMMARY panel. + // + // Exit-code mapping: + // SubmitError 401/403 → 5 (auth — same bucket as + // token-mint, shared + // "your SA can't do this" + // diagnostic class) + // SubmitError other 4xx/5xx → 8 (submit failed) + // WatchResult Failed → 9 (ingest failed) + // WatchResult Succeeded + + // summary.HasFailures() → 9 (some rows failed + // even though Job exited 0; + // the ingestor surfaces + // partial-failure summaries) + // WatchResult Detached → 0 (cluster keeps running) + // WatchResult Succeeded clean → 0 + submitRes, err := submit.Run(ctx, submit.Options{ + Submitter: submit.NewHTTPSubmitter(release.JobsManagerService, tok.Token), + Client: cs, + IngestConfigYAML: string(specBytes), + IdempotencyKey: a.IdempotencyKey, + ImageDigest: a.ImageDigest, + Detach: a.Detach, + Out: out, + }) + if err != nil { + if submit.IsAuthError(err) { + return &exitError{code: 5, err: err} + } + return &exitError{code: 8, err: err} + } + + // Detach paths (--detach flag OR SIGINT-mid-watch) are + // success — cluster keeps running; the orchestrator already + // printed the reconnect hint. + if submitRes.Watch == nil || submitRes.Watch.Outcome == submit.JobOutcomeDetached { + return nil + } + + // Watch outcomes. + switch submitRes.Watch.Outcome { + case submit.JobOutcomeFailed: + return &exitError{code: 9, err: errors.New("ingestion Job exited non-zero — see logs above")} + case submit.JobOutcomeSucceeded: + if submitRes.Watch.Summary != nil && submitRes.Watch.Summary.HasFailures() { + return &exitError{code: 9, err: errors.New( + "ingestion Job completed but the summary reports failures — see panel above")} + } + } return nil } diff --git a/internal/submit/body.go b/internal/submit/body.go new file mode 100644 index 00000000..80b641e9 --- /dev/null +++ b/internal/submit/body.go @@ -0,0 +1,107 @@ +// Package submit owns the `tracebloc dataset push` Phase 4 step: +// POST the synthesized ingest spec to jobs-manager's +// /internal/submit-ingestion-run endpoint, then watch the ingestor +// Job the cluster spawns in response. +// +// Phase 4 sits between Phase 3's stage Pod (which lays files on the +// PVC) and Phase 5's release distribution. The protocol is the same +// one tracebloc/client's ingestor subchart's post-install hook uses +// today — see ingestor/templates/configmap-ingest-config.yaml. +// Keeping the protocol identical means the CLI and the helm flow +// are interchangeable at the cluster's API surface, and the chart +// stays a fully-supported alternative for ops folks who prefer it. +// +// The package is split into: +// - body.go (this file): synthesize the POST body +// - client.go: HTTP client + bearer token + 4xx framing +// - watch.go: poll Job + stream Pod logs +// - summary.go: parse the 📊 INGESTION SUMMARY banner +// - submit.go: top-level orchestrator +package submit + +import ( + "crypto/rand" + "encoding/hex" + "fmt" +) + +// SubmitRequest is the wire shape POSTed to jobs-manager's +// /internal/submit-ingestion-run. Field names mirror the chart's +// ingestor/templates/configmap-ingest-config.yaml body.json key +// for-key so the chart and the CLI are interchangeable on the +// server side. +// +// json struct tags are explicit-snake so a future re-import via +// encoding/json doesn't silently produce mixedCase keys. +type SubmitRequest struct { + // IngestConfig is the customer's ingest spec as a verbatim + // YAML string. jobs-manager re-parses + revalidates this + // server-side. We don't re-marshal — the CLI's Phase 3 spec + // synthesis already produced canonical YAML. + IngestConfig string `json:"ingest_config"` + + // IdempotencyKey is the per-invocation replay token. + // jobs-manager records this in its idempotency-key table; a + // second POST with the same key returns the SAME job_name as + // the first (with replay=true) instead of spawning a new Job. + // + // Default is a fresh UUID-ish 16-byte hex string per + // invocation; `--idempotency-key ` overrides for the + // at-most-once-across-attempts case where the customer + // genuinely wants retry-safety across multiple CLI runs. + IdempotencyKey string `json:"idempotency_key"` + + // ImageDigest optionally pins the ingestor container image. + // Empty = let jobs-manager use the cluster's configured + // default (set by the parent client chart's + // `images.ingestor.digest`, kept current by the auto-upgrade + // cronjob). Setting it locks the run to a specific image, + // matching the chart's --set image.digest=... override path. + // + // `omitempty` on the JSON tag means jobs-manager sees no + // image_digest key at all when this is empty, which is the + // well-tested default-image code path on the server side. + ImageDigest string `json:"image_digest,omitempty"` +} + +// BuildRequest is the constructor used by the orchestrator. Both +// IngestConfig (the YAML the CLI already synthesized in Phase 3) +// and the optional ImageDigest flow through unchanged; the +// idempotency key is the only non-trivial bit. +// +// If override is empty, a fresh 16-byte hex string is generated +// from crypto/rand. UUID-shaped without the dashes — the chart's +// own helper does the same (ingestor.idempotencyKey in +// _helpers.tpl), so server-side hash-table lookups are uniform +// across both flows. +func BuildRequest(ingestYAML string, idempotencyKeyOverride, imageDigest string) (*SubmitRequest, error) { + key := idempotencyKeyOverride + if key == "" { + raw := make([]byte, 16) + if _, err := rand.Read(raw); err != nil { + return nil, fmt.Errorf("generating idempotency key: %w", err) + } + key = hex.EncodeToString(raw) + } + return &SubmitRequest{ + IngestConfig: ingestYAML, + IdempotencyKey: key, + ImageDigest: imageDigest, + }, nil +} + +// SubmitResponse is jobs-manager's 201 reply. job_name is the +// ingestor Job watch.go will poll; namespace is the resolved API +// namespace (usually the same one the CLI POSTed to, but +// jobs-manager can in principle redirect cross-namespace). +// +// Replay distinguishes "we just spawned this Job" (replay=false) +// from "we already have a Job for this idempotency key, here it +// is" (replay=true). The CLI prints a different lifecycle banner +// for each — replay means "another invocation already kicked +// this off; we're attaching to it." +type SubmitResponse struct { + JobName string `json:"job_name"` + Namespace string `json:"namespace"` + Replay bool `json:"replay"` +} diff --git a/internal/submit/body_test.go b/internal/submit/body_test.go new file mode 100644 index 00000000..dd6e4597 --- /dev/null +++ b/internal/submit/body_test.go @@ -0,0 +1,100 @@ +package submit + +import ( + "encoding/json" + "strings" + "testing" +) + +// TestBuildRequest_GeneratesIdempotencyKey: when no override is +// provided, BuildRequest produces a fresh hex-encoded 16-byte key. +// Pins both "non-empty" + "looks-like-hex" so a future change to +// random source can't silently produce a malformed key. +func TestBuildRequest_GeneratesIdempotencyKey(t *testing.T) { + req, err := BuildRequest("yaml content", "", "") + if err != nil { + t.Fatalf("BuildRequest: %v", err) + } + if req.IdempotencyKey == "" { + t.Fatal("generated idempotency key is empty") + } + if len(req.IdempotencyKey) != 32 { + t.Errorf("generated key length = %d, want 32 (16 bytes hex)", len(req.IdempotencyKey)) + } + for i, c := range req.IdempotencyKey { + if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) { + t.Errorf("key[%d] = %c, want lowercase hex", i, c) + break + } + } +} + +// TestBuildRequest_IdempotencyKeyOverride: --idempotency-key flag +// path. The override is plumbed verbatim — no hashing, no munging — +// so a customer using the same key across retries gets the chart's +// replay semantics. +func TestBuildRequest_IdempotencyKeyOverride(t *testing.T) { + req, err := BuildRequest("yaml", "my-fixed-key-abc123", "") + if err != nil { + t.Fatalf("BuildRequest: %v", err) + } + if req.IdempotencyKey != "my-fixed-key-abc123" { + t.Errorf("IdempotencyKey = %q, want override value", req.IdempotencyKey) + } +} + +// TestBuildRequest_KeysAreUniquePerCall: two back-to-back +// BuildRequest calls with no override produce distinct keys. +// Critical for jobs-manager's idempotency table — same key means +// "this is a retry, replay the previous run." +func TestBuildRequest_KeysAreUniquePerCall(t *testing.T) { + a, err := BuildRequest("yaml", "", "") + if err != nil { + t.Fatalf("BuildRequest a: %v", err) + } + b, err := BuildRequest("yaml", "", "") + if err != nil { + t.Fatalf("BuildRequest b: %v", err) + } + if a.IdempotencyKey == b.IdempotencyKey { + t.Errorf("back-to-back BuildRequest produced identical keys %q", a.IdempotencyKey) + } +} + +// TestBuildRequest_JSONShape: the wire format jobs-manager expects. +// Field names + omitempty behavior are the contract; if any of +// these drift, the server-side handler fails to parse. +func TestBuildRequest_JSONShape(t *testing.T) { + t.Run("with image digest", func(t *testing.T) { + req, _ := BuildRequest("yaml-content", "key123", "sha256:abc") + b, err := json.Marshal(req) + if err != nil { + t.Fatalf("json.Marshal: %v", err) + } + s := string(b) + for _, want := range []string{ + `"ingest_config":"yaml-content"`, + `"idempotency_key":"key123"`, + `"image_digest":"sha256:abc"`, + } { + if !strings.Contains(s, want) { + t.Errorf("JSON missing %q in: %s", want, s) + } + } + }) + t.Run("omits empty image digest", func(t *testing.T) { + req, _ := BuildRequest("yaml", "key", "") + b, err := json.Marshal(req) + if err != nil { + t.Fatalf("json.Marshal: %v", err) + } + // jobs-manager's no-image-digest code path is the + // well-tested default; passing an empty string would + // route through the override path on the server. The + // omitempty tag is the contract that keeps the default + // path engaged. + if strings.Contains(string(b), "image_digest") { + t.Errorf("JSON includes image_digest when empty: %s", b) + } + }) +} diff --git a/internal/submit/client.go b/internal/submit/client.go new file mode 100644 index 00000000..5ef433a5 --- /dev/null +++ b/internal/submit/client.go @@ -0,0 +1,185 @@ +package submit + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// SubmitTimeout caps how long we wait for jobs-manager to respond to +// the POST. jobs-manager validates synchronously (schema re-check, +// idempotency lookup, Job creation) — the chart's hook bounds this +// at 30s and we mirror that. Beyond 30s, something genuinely wrong +// is happening server-side, and the customer wants the diagnostic +// rather than a longer wait. +const SubmitTimeout = 30 * time.Second + +// Submitter is the narrow surface the orchestrator (submit.go) uses +// for the POST. Real impl is *HTTPSubmitter; tests use a fake that +// records the request and returns a synthetic response. +type Submitter interface { + Submit(ctx context.Context, req *SubmitRequest) (*SubmitResponse, error) +} + +// HTTPSubmitter is the production implementation. Wraps net/http +// with a fixed jobs-manager URL + a SA token from Phase 2's mint. +// +// Endpoint comes from Phase 2's cluster.DiscoverParentRelease +// (`http://-jobs-manager..svc.cluster.local:8080`), +// SubmitPath is hardcoded per the chart's published contract. +type HTTPSubmitter struct { + // Endpoint is the full jobs-manager URL, e.g. + // "http://release-jobs-manager.tracebloc.svc.cluster.local:8080". + // No trailing slash — the submitter appends SubmitPath. + Endpoint string + + // Token is the bearer token to send in the Authorization + // header. Comes from Phase 2's cluster.MintIngestorToken. + Token string + + // Client is the underlying *http.Client. Set in NewHTTPSubmitter + // with a sensible timeout. Exposed for tests that need to point + // at an httptest.Server with a custom RoundTripper. + Client *http.Client +} + +// SubmitPath is the well-known URL path on jobs-manager. Pinned +// here as a constant rather than a knob because the chart's +// post-install hook also pins it; if jobs-manager ever moves the +// endpoint, both have to bump together (which is a coordinated +// release across tracebloc/client + tracebloc/cli, exactly what +// you want for a protocol-level change). +const SubmitPath = "/internal/submit-ingestion-run" + +// NewHTTPSubmitter returns a Submitter wired with a sensible +// timeout + a transport that DOESN'T do TLS verification against +// the in-cluster CA. The jobs-manager endpoint is HTTP-only inside +// the cluster (kube-proxy handles all the rest), so TLS isn't in +// the picture at all today. If a future jobs-manager exposes +// HTTPS, the InsecureSkipVerify path is the right v0.1 default +// because the customer's kubeconfig has already authenticated them +// to the cluster — the cluster-internal jobs-manager doesn't have +// a CA the laptop would recognize anyway. +func NewHTTPSubmitter(endpoint, token string) *HTTPSubmitter { + return &HTTPSubmitter{ + Endpoint: strings.TrimRight(endpoint, "/"), + Token: token, + Client: &http.Client{ + Timeout: SubmitTimeout, + Transport: &http.Transport{ + // See doc on NewHTTPSubmitter for the + // InsecureSkipVerify rationale. + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec + }, + }, + } +} + +// Submit POSTs the request body to jobs-manager and decodes the +// 201 response into a SubmitResponse. On non-201 status codes, the +// remote body is surfaced verbatim so the customer sees whatever +// jobs-manager said (typically a JSON {error, detail} from the +// fastapi handler) rather than just "HTTP 422". +// +// Replays (idempotency-key already seen → 200 with replay=true) +// are reported in the response struct; this method treats them as +// success because the upstream behavior IS "your run is already +// in progress / already done." The orchestrator handles the +// replay branch in its own diagnostic output. +func (s *HTTPSubmitter) Submit(ctx context.Context, req *SubmitRequest) (*SubmitResponse, error) { + body, err := json.Marshal(req) + if err != nil { + // Marshaling a struct of strings shouldn't fail at all; + // surfacing the error means a future struct change broke + // the wire format. + return nil, fmt.Errorf("marshaling submit request: %w", err) + } + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, + s.Endpoint+SubmitPath, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("building submit request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/json") + httpReq.Header.Set("Authorization", "Bearer "+s.Token) + + httpResp, err := s.Client.Do(httpReq) + if err != nil { + // Network errors (DNS, connection refused, TLS handshake + // failure, ctx cancellation). The error wraps net.OpError + // already; just frame it with our endpoint so the + // customer knows what was being attempted. + return nil, fmt.Errorf("POST %s%s: %w", s.Endpoint, SubmitPath, err) + } + defer func() { _ = httpResp.Body.Close() }() + + respBody, err := io.ReadAll(httpResp.Body) + if err != nil { + // Short read on a 201 body — extremely rare; the + // connection dropped between header and body. + return nil, fmt.Errorf("reading submit response body: %w", err) + } + + // 2xx (typically 201 Created; also 200 for replays) is success. + // Everything else surfaces the remote body verbatim so the + // customer sees jobs-manager's actual diagnostic (HTTP 4xx + // schema-rejection, HTTP 5xx kube-apiserver-failure, etc.) + // rather than just a status code. + if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { + return nil, &SubmitError{ + StatusCode: httpResp.StatusCode, + Body: string(respBody), + Endpoint: s.Endpoint + SubmitPath, + } + } + + var parsed SubmitResponse + if err := json.Unmarshal(respBody, &parsed); err != nil { + // 2xx but the body isn't the expected shape. Almost + // certainly a protocol mismatch (jobs-manager version + // drift). Include the raw body so the customer can pin + // the version they're on. + return nil, fmt.Errorf("decoding submit response (got body %q): %w", string(respBody), err) + } + if parsed.JobName == "" { + // Defensive: a malformed-but-parsing response with no + // job_name would silently break Phase 4's watch step. + return nil, fmt.Errorf("submit response missing job_name (got body %q)", string(respBody)) + } + return &parsed, nil +} + +// SubmitError is the typed non-2xx response. Pulled into a struct +// (rather than an opaque string) so the orchestrator can branch on +// StatusCode for the exit-code mapping: 401/403 → auth exit code, +// 4xx other → submit-validation exit code, 5xx → submit-server. +// +// Implements `error` + an Is for errors.Is detection in tests. +type SubmitError struct { + StatusCode int + Body string + Endpoint string +} + +func (e *SubmitError) Error() string { + // Compact framing: the customer sees status + body. The body + // is jobs-manager's actual diagnostic (e.g. fastapi's + // {"detail": "..."}) which is the actionable part. + return fmt.Sprintf("jobs-manager %s returned HTTP %d: %s", + e.Endpoint, e.StatusCode, strings.TrimSpace(e.Body)) +} + +// IsSubmitError reports whether err is a *SubmitError. Convenience +// for the orchestrator's exit-code mapping; errors.As would also +// work but this reads cleaner at the branch site. +func IsSubmitError(err error) bool { + var se *SubmitError + return errors.As(err, &se) +} diff --git a/internal/submit/client_test.go b/internal/submit/client_test.go new file mode 100644 index 00000000..7519c14e --- /dev/null +++ b/internal/submit/client_test.go @@ -0,0 +1,234 @@ +package submit + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// TestHTTPSubmitter_HappyPath: jobs-manager returns 201 with the +// canonical body shape; client decodes correctly + surfaces all +// three response fields. +func TestHTTPSubmitter_HappyPath(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Pin the wire format jobs-manager actually expects. + if r.Method != http.MethodPost { + t.Errorf("got method %s, want POST", r.Method) + } + if r.URL.Path != SubmitPath { + t.Errorf("got path %s, want %s", r.URL.Path, SubmitPath) + } + if got := r.Header.Get("Authorization"); got != "Bearer fake-token-deadbeef" { + t.Errorf("Authorization = %q, want Bearer fake-token-deadbeef", got) + } + if got := r.Header.Get("Content-Type"); got != "application/json" { + t.Errorf("Content-Type = %q, want application/json", got) + } + body, _ := io.ReadAll(r.Body) + // Light shape check — body_test.go pins the full JSON shape. + if !strings.Contains(string(body), `"ingest_config"`) { + t.Errorf("body missing ingest_config: %s", body) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{ + "job_name": "ingestor-abc123", + "namespace": "tracebloc", + "replay": false, + }) + })) + defer srv.Close() + + s := NewHTTPSubmitter(srv.URL, "fake-token-deadbeef") + req, _ := BuildRequest("yaml-content", "key1", "") + + resp, err := s.Submit(context.Background(), req) + if err != nil { + t.Fatalf("Submit: %v", err) + } + if resp.JobName != "ingestor-abc123" { + t.Errorf("JobName = %q, want ingestor-abc123", resp.JobName) + } + if resp.Namespace != "tracebloc" { + t.Errorf("Namespace = %q, want tracebloc", resp.Namespace) + } + if resp.Replay { + t.Errorf("Replay = true, want false") + } +} + +// TestHTTPSubmitter_ReplayResponse: replay=true is also a success +// path. The orchestrator distinguishes the two via the response +// flag, not via HTTP status — both come back 2xx. +func TestHTTPSubmitter_ReplayResponse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) // jobs-manager returns 200 for replays per source + _ = json.NewEncoder(w).Encode(map[string]any{ + "job_name": "existing-job", + "namespace": "tracebloc", + "replay": true, + }) + })) + defer srv.Close() + + s := NewHTTPSubmitter(srv.URL, "tok") + req, _ := BuildRequest("yaml", "same-key", "") + resp, err := s.Submit(context.Background(), req) + if err != nil { + t.Fatalf("Submit on replay: %v", err) + } + if !resp.Replay { + t.Errorf("Replay = false, want true") + } +} + +// TestHTTPSubmitter_4xxSurfacesBody: 4xx from jobs-manager +// surfaces the verbatim body so the customer sees jobs-manager's +// actual diagnostic (typically {"detail": "..."}), not just +// "HTTP 422". +func TestHTTPSubmitter_4xxSurfacesBody(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnprocessableEntity) + _, _ = w.Write([]byte(`{"detail":"ingest_config schema rejected: missing required field 'intent'"}`)) + })) + defer srv.Close() + + s := NewHTTPSubmitter(srv.URL, "tok") + req, _ := BuildRequest("yaml", "k", "") + _, err := s.Submit(context.Background(), req) + if err == nil { + t.Fatal("Submit returned nil on 4xx") + } + if !IsSubmitError(err) { + t.Errorf("err is not *SubmitError: %T", err) + } + for _, want := range []string{"HTTP 422", "missing required field 'intent'"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error missing %q: %v", want, err) + } + } +} + +// TestHTTPSubmitter_401IsAuthError: distinguishes the auth case +// (401/403) from generic 4xx for the orchestrator's exit-code +// mapping. Used by the CLI to return "your SA token doesn't +// work" vs "your spec was rejected." +func TestHTTPSubmitter_401IsAuthError(t *testing.T) { + for _, status := range []int{http.StatusUnauthorized, http.StatusForbidden} { + t.Run(http.StatusText(status), func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(status) + _, _ = w.Write([]byte(`{"detail":"token expired"}`)) + })) + defer srv.Close() + + s := NewHTTPSubmitter(srv.URL, "tok") + req, _ := BuildRequest("yaml", "k", "") + _, err := s.Submit(context.Background(), req) + if err == nil { + t.Fatal("Submit returned nil on auth error") + } + if !IsAuthError(err) { + t.Errorf("IsAuthError(%v) = false, want true (status=%d)", err, status) + } + }) + } +} + +// TestHTTPSubmitter_5xxNotAuthError: 5xx is server-side trouble +// (kube-apiserver flake, jobs-manager bug), NOT an auth issue. +// IsAuthError should return false so the orchestrator routes to +// the right exit-code bucket. +func TestHTTPSubmitter_5xxNotAuthError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"detail":"internal"}`)) + })) + defer srv.Close() + + s := NewHTTPSubmitter(srv.URL, "tok") + req, _ := BuildRequest("yaml", "k", "") + _, err := s.Submit(context.Background(), req) + if err == nil { + t.Fatal("Submit returned nil on 500") + } + if IsAuthError(err) { + t.Errorf("IsAuthError(500) = true, want false") + } + if !IsSubmitError(err) { + t.Errorf("err is not *SubmitError: %T", err) + } +} + +// TestHTTPSubmitter_2xxMissingJobName: a malformed 2xx response +// (server bug or version drift) without job_name is a hard error, +// not a silent success — the orchestrator wouldn't know what to +// watch. +func TestHTTPSubmitter_2xxMissingJobName(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"namespace":"tracebloc","replay":false}`)) + })) + defer srv.Close() + + s := NewHTTPSubmitter(srv.URL, "tok") + req, _ := BuildRequest("yaml", "k", "") + _, err := s.Submit(context.Background(), req) + if err == nil { + t.Fatal("Submit returned nil on missing job_name") + } + if !strings.Contains(err.Error(), "missing job_name") { + t.Errorf("error missing 'missing job_name' framing: %v", err) + } +} + +// TestHTTPSubmitter_NetworkError: unreachable server (closed +// httptest.Server) surfaces as a wrapped net error with the +// endpoint in the message so customers know what was attempted. +func TestHTTPSubmitter_NetworkError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + })) + endpoint := srv.URL + srv.Close() // kill the server + + s := NewHTTPSubmitter(endpoint, "tok") + req, _ := BuildRequest("yaml", "k", "") + _, err := s.Submit(context.Background(), req) + if err == nil { + t.Fatal("Submit returned nil on unreachable server") + } + if !strings.Contains(err.Error(), endpoint) { + t.Errorf("error missing endpoint URL %q: %v", endpoint, err) + } +} + +// TestHTTPSubmitter_RespectsContext: ctx cancellation aborts the +// in-flight POST. Critical for the SIGINT path (main.go's +// signal.NotifyContext). +func TestHTTPSubmitter_RespectsContext(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Wait longer than the test's ctx allows. + <-r.Context().Done() + w.WriteHeader(http.StatusGatewayTimeout) + })) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already cancelled + + s := NewHTTPSubmitter(srv.URL, "tok") + req, _ := BuildRequest("yaml", "k", "") + _, err := s.Submit(ctx, req) + if err == nil { + t.Fatal("Submit returned nil on cancelled ctx") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("error doesn't wrap context.Canceled: %v", err) + } +} diff --git a/internal/submit/submit.go b/internal/submit/submit.go new file mode 100644 index 00000000..7478da0a --- /dev/null +++ b/internal/submit/submit.go @@ -0,0 +1,161 @@ +package submit + +import ( + "context" + "errors" + "fmt" + "io" + + "k8s.io/client-go/kubernetes" +) + +// Options bundles every dependency Run needs. The CLI builds one +// from the resolved Phase 2/3 state (kubeconfig clientset, SA +// token, jobs-manager endpoint) + the flags from Phase 4 +// (--detach, --idempotency-key, --image-digest). +type Options struct { + // Submitter is how the POST reaches jobs-manager. Production + // uses NewHTTPSubmitter(endpoint, token); tests inject a + // fake that captures the request + returns a canned response. + Submitter Submitter + + // Client is the kubernetes.Interface for the watch loop's + // Job + Pod polls. Same clientset Phase 3 used. + Client kubernetes.Interface + + // IngestConfigYAML is the synthesized YAML body to POST. + // Phase 3 already produced this in canonical form + // (push.SpecArgs.Build → yaml.Marshal) so we don't re-marshal + // here. + IngestConfigYAML string + + // IdempotencyKey overrides the auto-generated random key. + // Empty = let BuildRequest generate a fresh one. Used by + // the --idempotency-key flag for retry-safety across + // invocations. + IdempotencyKey string + + // ImageDigest optionally pins the ingestor image. Empty = + // jobs-manager uses the cluster's configured default. + ImageDigest string + + // Detach exits immediately after the 201 — no watch, no log + // streaming, no summary. Used by CI scenarios where the + // customer just wants the Job name in stdout and the run + // proceeds asynchronously in the cluster. + Detach bool + + // Out is the customer-facing log stream. Submit writes the + // 201 announcement here, then either streams the Pod's logs + // to it (live watch) or prints the Job name (detach). The + // rendered summary panel also goes here. + Out io.Writer +} + +// Result is what Run reports back to the CLI orchestrator. +// Outcome drives the exit-code mapping (see cli/dataset.go's +// Phase 4 wiring); JobName + PodName are echoed back so the CLI +// can build "reconnect with kubectl logs -n " hints. +type Result struct { + // Submit is the 201 response from jobs-manager. Non-nil on + // any path that got past the POST (including --detach + + // failed watches). nil only if the POST itself failed. + Submit *SubmitResponse + + // Watch is the result of the watch loop. nil on --detach + // (we never started watching). nil on early POST failure. + Watch *WatchResult +} + +// Run is the Phase 4 top-level entrypoint. Steps: +// +// 1. BuildRequest from the YAML + flags +// 2. POST via opts.Submitter; surface SubmitError verbatim +// 3. Print the 201 announcement (job_name / namespace / replay flag) +// 4. If --detach, exit +// 5. WatchJob until Pod terminates or ctx cancels +// 6. Render the parsed Summary panel +// +// Returns the Result + an error. Errors come from steps 1-2-5; on +// steps 3 + 4 + 6, success means "we got far enough" and the +// outcome of the actual ingestion is in Result.Watch.Outcome. +func Run(ctx context.Context, opts Options) (*Result, error) { + if opts.Out == nil { + opts.Out = io.Discard + } + + req, err := BuildRequest(opts.IngestConfigYAML, opts.IdempotencyKey, opts.ImageDigest) + if err != nil { + return nil, fmt.Errorf("building submit request: %w", err) + } + + resp, err := opts.Submitter.Submit(ctx, req) + if err != nil { + 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. + 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) + } else { + _, _ = fmt.Fprintf(opts.Out, + "Submitted: jobs-manager spawned ingestor Job %s/%s\n", + resp.Namespace, resp.JobName) + } + + 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) + 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) + if err != nil { + return &Result{Submit: resp}, fmt.Errorf("watching ingestor Job: %w", err) + } + + // Detach-from-SIGINT path. Print the reconnect hint so the + // customer knows the cluster keeps running. + if wr.Outcome == JobOutcomeDetached { + _, _ = fmt.Fprintln(opts.Out) + _, _ = fmt.Fprintf(opts.Out, + "Detached on signal. Ingestion continues in the cluster. "+ + "Reconnect with: kubectl logs -f -n %s job/%s\n", + resp.Namespace, resp.JobName) + return &Result{Submit: resp, Watch: wr}, nil + } + + // Render the summary panel if the ingestor produced one. + // Both Succeeded and Failed paths print it — on Failed, the + // banner tells the customer what got partially through. + if wr.Summary != nil { + _, _ = fmt.Fprintln(opts.Out) + _, _ = fmt.Fprint(opts.Out, RenderPanel(wr.Summary)) + } + + return &Result{Submit: resp, Watch: wr}, nil +} + +// IsAuthError reports whether the error is the auth-flavored case +// (401/403 from jobs-manager). The orchestrator's exit-code +// mapping uses this to distinguish "your SA token doesn't work" +// from "your spec was rejected." +func IsAuthError(err error) bool { + var se *SubmitError + if !errors.As(err, &se) { + return false + } + return se.StatusCode == 401 || se.StatusCode == 403 +} diff --git a/internal/submit/submit_test.go b/internal/submit/submit_test.go new file mode 100644 index 00000000..6bb32512 --- /dev/null +++ b/internal/submit/submit_test.go @@ -0,0 +1,205 @@ +package submit + +import ( + "bytes" + "context" + "errors" + "strings" + "testing" + + "k8s.io/client-go/kubernetes/fake" +) + +// fakeSubmitter captures the request + returns a canned response. +// Used by Run() tests to exercise the orchestrator without needing +// an httptest.Server (covered by client_test.go). +type fakeSubmitter struct { + gotRequest *SubmitRequest + resp *SubmitResponse + err error +} + +func (f *fakeSubmitter) Submit(_ context.Context, req *SubmitRequest) (*SubmitResponse, error) { + f.gotRequest = req + return f.resp, f.err +} + +// TestRun_DetachPath_HappyPath: --detach exits immediately after +// the 201 with the reconnect hint. No watch loop, no log streaming. +func TestRun_DetachPath_HappyPath(t *testing.T) { + sub := &fakeSubmitter{ + resp: &SubmitResponse{ + JobName: "ingestor-abc", + Namespace: "tracebloc", + Replay: false, + }, + } + var out bytes.Buffer + + res, err := Run(context.Background(), Options{ + Submitter: sub, + Client: fake.NewClientset(), + IngestConfigYAML: "apiVersion: tracebloc.io/v1\n", + Detach: true, + Out: &out, + }) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Submit == nil || res.Submit.JobName != "ingestor-abc" { + t.Errorf("Result.Submit lost: %+v", res.Submit) + } + if res.Watch != nil { + 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)", + "kubectl logs -f -n tracebloc job/ingestor-abc", + } { + if !strings.Contains(out.String(), want) { + t.Errorf("output missing %q in:\n%s", want, out.String()) + } + } +} + +// TestRun_ReplayPath: replay=true changes the announcement +// wording — "attaching to existing Job" instead of "spawned" — +// because the cluster is already doing the work. +func TestRun_ReplayPath(t *testing.T) { + sub := &fakeSubmitter{ + resp: &SubmitResponse{ + JobName: "ingestor-existing", + Namespace: "tracebloc", + Replay: true, + }, + } + var out bytes.Buffer + + _, err := Run(context.Background(), Options{ + Submitter: sub, + Client: fake.NewClientset(), + IngestConfigYAML: "yaml", + Detach: true, // skip the watch for this test + Out: &out, + }) + 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(), "attaching to existing Job") { + t.Errorf("output missing replay-specific wording:\n%s", out.String()) + } +} + +// TestRun_SubmitErrorPropagates: a non-2xx from jobs-manager +// stops Run before any watching happens. The error surfaces with +// jobs-manager's body framing (the client.go path). +func TestRun_SubmitErrorPropagates(t *testing.T) { + sub := &fakeSubmitter{ + err: &SubmitError{ + StatusCode: 422, + Body: `{"detail":"bad spec"}`, + Endpoint: "http://jm/internal/submit-ingestion-run", + }, + } + var out bytes.Buffer + + _, err := Run(context.Background(), Options{ + Submitter: sub, + Client: fake.NewClientset(), + IngestConfigYAML: "yaml", + Out: &out, + }) + if err == nil { + t.Fatal("Run returned nil on submit error") + } + if !IsSubmitError(err) { + t.Errorf("err is not *SubmitError: %T", err) + } +} + +// TestRun_BuildRequestErrorPropagates: a crypto/rand failure in +// BuildRequest stops Run before the submitter even gets called. +// We can't easily mock crypto/rand, but we can verify the error +// path is wired by checking that any failure here doesn't reach +// the submitter. This test is more about the contract than the +// trigger. +func TestRun_PassesRequestFieldsThrough(t *testing.T) { + sub := &fakeSubmitter{ + resp: &SubmitResponse{JobName: "j", Namespace: "ns"}, + } + var out bytes.Buffer + + _, err := Run(context.Background(), Options{ + Submitter: sub, + Client: fake.NewClientset(), + IngestConfigYAML: "yaml-content-verbatim", + IdempotencyKey: "my-key-override", + ImageDigest: "sha256:abc", + Detach: true, + Out: &out, + }) + if err != nil { + t.Fatalf("Run: %v", err) + } + if sub.gotRequest == nil { + t.Fatal("submitter never called") + } + if sub.gotRequest.IngestConfig != "yaml-content-verbatim" { + t.Errorf("IngestConfig = %q, want yaml-content-verbatim", + sub.gotRequest.IngestConfig) + } + if sub.gotRequest.IdempotencyKey != "my-key-override" { + t.Errorf("IdempotencyKey = %q, want override value", + sub.gotRequest.IdempotencyKey) + } + if sub.gotRequest.ImageDigest != "sha256:abc" { + t.Errorf("ImageDigest = %q, want sha256:abc", + sub.gotRequest.ImageDigest) + } +} + +// TestRun_NilOutDefaultsToDiscard: callers passing nil Out +// shouldn't panic. The orchestrator silently discards output. +func TestRun_NilOutDefaultsToDiscard(t *testing.T) { + sub := &fakeSubmitter{ + resp: &SubmitResponse{JobName: "j", Namespace: "ns"}, + } + _, err := Run(context.Background(), Options{ + Submitter: sub, + Client: fake.NewClientset(), + IngestConfigYAML: "yaml", + Detach: true, + Out: nil, + }) + if err != nil { + t.Fatalf("Run with nil Out panicked or errored: %v", err) + } +} + +// TestIsAuthError: helper smoke test. Pin the contract used by +// the CLI's exit-code mapping. +func TestIsAuthError(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"non-submit error", errors.New("network"), false}, + {"submit 401", &SubmitError{StatusCode: 401}, true}, + {"submit 403", &SubmitError{StatusCode: 403}, true}, + {"submit 422", &SubmitError{StatusCode: 422}, false}, + {"submit 500", &SubmitError{StatusCode: 500}, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := IsAuthError(c.err); got != c.want { + t.Errorf("IsAuthError = %v, want %v", got, c.want) + } + }) + } +} diff --git a/internal/submit/summary.go b/internal/submit/summary.go new file mode 100644 index 00000000..5cf2e7df --- /dev/null +++ b/internal/submit/summary.go @@ -0,0 +1,384 @@ +package submit + +import ( + "bufio" + "bytes" + "fmt" + "io" + "regexp" + "strconv" + "strings" +) + +// Summary is the parsed contents of the ingestor's 📊 INGESTION +// SUMMARY 📊 banner. Fields mirror what +// tracebloc_ingestor/ingestors/base.py:624+ prints. Zero values +// are valid — an early failure may produce a summary with most +// counts at 0 (the ingestor still prints the banner so operators +// can see what got through). +// +// All counts are int64 to fit row counts well past int32's 2.1B +// ceiling — a customer ingesting a few-billion-row table would +// silently truncate with int. +type Summary struct { + // IngestorID is the run identifier the ingestor logs at the + // top of the banner. Useful in the customer-facing panel as + // "you can grep cluster logs for this ID." + IngestorID string + + // TotalRecords is the row count the ingestor saw in the + // source data. Includes every row regardless of outcome. + TotalRecords int64 + + // ProcessedRecords is the row count that made it through + // validation (passed FileTypeValidator, ImageResolutionValidator, + // etc.). Excludes invalid rows. + ProcessedRecords int64 + + // InsertedRecords is the row count that landed in the + // cluster-internal MySQL. The "I actually have this data + // staged" metric — this is what matters for downstream + // training jobs. + InsertedRecords int64 + + // APISentRecords is the row count that synced metadata to + // the central tracebloc backend. Only the row count + label + // is sent, not the raw data; this is the "central catalog + // knows about this dataset" metric. + APISentRecords int64 + + // SkippedRecords is the row count rejected by validators + // (wrong dimensions, missing image file, etc.). Non-fatal + // for the run but worth surfacing — a customer with 50% + // skipped wants to see that. + SkippedRecords int64 + + // FileTransferFailures is the count of files (NOT rows) that + // failed to transfer to the requests-proxy. Distinct from + // FailedRecords because file transfer is a separate stage + // from DB insertion. Non-zero here is the dominant "your + // network is flaky" signal. + FileTransferFailures int64 + + // FailedRecords is the row count that errored at the + // DB-insert stage (constraint violation, type mismatch, + // connection drop). The catch-all "something went wrong + // at the storage layer" bucket. + FailedRecords int64 +} + +// HasFailures returns true if any failure-class counter is non-zero. +// Used by the orchestrator to decide which exit code to return +// (success: 0, ingest-with-failures: non-zero) and how to color +// the rendered panel. +func (s *Summary) HasFailures() bool { + if s == nil { + return false + } + return s.FileTransferFailures > 0 || s.FailedRecords > 0 +} + +// SuccessRate returns a 0-100 percentage for the panel header. +// Defined as ProcessedRecords / TotalRecords; returns 0 when +// TotalRecords is 0 to avoid divide-by-zero in early-failure +// banners. +func (s *Summary) SuccessRate() float64 { + if s == nil || s.TotalRecords == 0 { + return 0 + } + return float64(s.ProcessedRecords) / float64(s.TotalRecords) * 100 +} + +// ansiCodeRE matches the ANSI SGR (Select Graphic Rendition) +// escape sequences the ingestor uses for its color output — +// `\x1b[1m` (bold), `\x1b[36m` (cyan), `\x1b[0m` (reset), etc. +// The ingestor prints these inline in the summary text; we strip +// them before parsing so a future palette change doesn't break +// the parser. +var ansiCodeRE = regexp.MustCompile(`\x1b\[[0-9;]*m`) + +// stripANSI removes SGR codes from a single line, returning the +// printable text. Implementation note: this is a hot path for the +// log stream so we avoid an allocation when no codes are present. +func stripANSI(line string) string { + if !strings.Contains(line, "\x1b[") { + return line + } + return ansiCodeRE.ReplaceAllString(line, "") +} + +// bannerStartMarker is the ingestor's literal banner-header line +// (with the BOLD + CYAN ANSI prefix stripped by stripANSI before +// matching). When we see this, we flip to "inside-banner" mode. +const bannerStartMarker = "📊 INGESTION SUMMARY 📊" + +// bannerEndMarker is the equals-rule the ingestor prints AFTER +// the last metric line (a second `═`x60 line). We use it as the +// terminator so a parser at end-of-stream emits a complete +// Summary even if the Pod was killed mid-log-flush. +const bannerEndMarker = "════════════════════════════════════════════════════════════" + +// fieldPatterns are the regex patterns for each Summary field, +// keyed by the human-readable label the ingestor prints. The +// values are populated by NewSummaryParser via Compile-once-at- +// init. Each pattern matches a `Label: ` shape with +// optional spacing and an optional `,`-separated digit format +// (e.g. "1,234,567"). +// +// Maintained as a parallel slice of {label, pointer-target} +// rather than a map so the order of parsing matches the order +// the ingestor prints them — useful if a future ingestor adds a +// new line, the parser doesn't have to re-scan the existing ones. +var fieldPatterns = []struct { + prefix string + apply func(s *Summary, n int64) +}{ + {"📈 Total Records Found:", func(s *Summary, n int64) { s.TotalRecords = n }}, + {"✅ Successfully Processed:", func(s *Summary, n int64) { s.ProcessedRecords = n }}, + {"💾 Inserted to Database:", func(s *Summary, n int64) { s.InsertedRecords = n }}, + {"🚀 Sent to API:", func(s *Summary, n int64) { s.APISentRecords = n }}, + {"⏭️ Skipped Records:", func(s *Summary, n int64) { s.SkippedRecords = n }}, + {"📁 File Transfer Failures:", func(s *Summary, n int64) { s.FileTransferFailures = n }}, + {"❌ Failed DB Insertion:", func(s *Summary, n int64) { s.FailedRecords = n }}, +} + +// numberRE captures the trailing digit-group on a metric line. +// Allows optional thousands-separator commas the ingestor's +// `f"{count:,}"` formatting emits. +var numberRE = regexp.MustCompile(`([0-9][0-9,]*)\s*$`) + +// ingestorIDRE matches the ingestor-ID line specifically; the +// value is a UUID-ish string, not a number, so it gets its own +// pattern. +var ingestorIDRE = regexp.MustCompile(`Ingestor ID:\s*(.+?)\s*$`) + +// SummaryParser is a streaming parser for the 📊 banner. Feed it +// log lines as they arrive (any chunk size, any line splitting); +// Result() returns the accumulated Summary at any point. The +// banner-end marker latches the result so post-banner log lines +// don't perturb it. +// +// The parser is stateful but not thread-safe — the watch loop +// uses it from a single goroutine (the log-streaming TeeReader), +// so no synchronization needed. +type SummaryParser struct { + // buf accumulates partial-line input across Feed calls. The + // log stream from the API server arrives in TCP-sized chunks + // that may split lines; we buffer until we see a '\n' to + // finalize each line. + buf bytes.Buffer + + // summary is the accumulator. nil until we see the banner + // header — Result returns nil if the run never produced one. + summary *Summary + + // finalized latches when we see the banner-end marker. After + // that, additional Feed calls don't modify summary (the + // ingestor may keep logging after the banner, e.g. shutdown + // messages; those shouldn't perturb the result). + finalized bool + + // insideBanner is true between bannerStartMarker and + // bannerEndMarker. Outside this window, lines are ignored + // (so e.g. a stray emoji in earlier log output doesn't + // trigger spurious parsing). + insideBanner bool +} + +// NewSummaryParser returns an initialized parser. Caller's +// goroutine owns it for the duration of the watch loop. +func NewSummaryParser() *SummaryParser { + return &SummaryParser{} +} + +// Feed accepts arbitrary log bytes; the parser buffers and splits +// internally. Safe to call with partial lines, multiple lines, or +// empty input. +func (p *SummaryParser) Feed(b []byte) { + if p.finalized { + return + } + _, _ = p.buf.Write(b) + for { + idx := bytes.IndexByte(p.buf.Bytes(), '\n') + if idx < 0 { + // No complete line yet — wait for more input. + return + } + line := p.buf.Next(idx + 1) // consume up to and including '\n' + p.feedLine(string(bytes.TrimRight(line, "\n"))) + } +} + +// FlushLine forces parsing of any buffered partial-line content. +// Called at end-of-stream by the watch loop in case the Pod +// terminated without a final '\n' (rare but possible if the +// container's stdout was killed mid-write). +func (p *SummaryParser) FlushLine() { + if p.buf.Len() > 0 && !p.finalized { + p.feedLine(p.buf.String()) + p.buf.Reset() + } +} + +// feedLine parses a single line, ANSI-stripped. The state machine +// has three regions: +// +// - Pre-banner: skip until we see the start marker +// - Inside banner: match each line against fieldPatterns + the +// Ingestor ID line +// - End marker: latch and stop processing +func (p *SummaryParser) feedLine(rawLine string) { + line := stripANSI(rawLine) + if strings.Contains(line, bannerStartMarker) { + p.summary = &Summary{} + p.insideBanner = true + return + } + if !p.insideBanner { + return + } + // Banner-end check: a long row of '═'. Only count when we've + // already crossed the start marker (the start banner ALSO + // has a '═' rule before the header, which we want to ignore). + if strings.Contains(line, bannerEndMarker) { + // Two ═-rules in the banner: one immediately after the + // header, one at the very end. We use a simple counter + // to distinguish — first one we see while insideBanner + // is the post-header rule (skip), second is the + // post-metrics rule (finalize). + // + // Actually the simpler approach: check whether we've + // parsed any field yet. If so, this ═ is the closing + // rule; if not, it's the opening one. The ingestor + // always prints fields between the two rules. + if p.summary != nil && p.hasParsedAnyField() { + p.finalized = true + } + return + } + + // Ingestor ID line is the first content line in the banner. + if m := ingestorIDRE.FindStringSubmatch(line); m != nil { + p.summary.IngestorID = strings.TrimSpace(m[1]) + return + } + + // Otherwise: try each field pattern. The prefix match is + // linear over a 7-element slice — microscopic overhead per + // line, and the fixed order matches the ingestor's print + // order. + for _, fp := range fieldPatterns { + if !strings.Contains(line, fp.prefix) { + continue + } + m := numberRE.FindStringSubmatch(line) + if m == nil { + return + } + // Strip thousands-separator commas. + raw := strings.ReplaceAll(m[1], ",", "") + n, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return // malformed; ignore this line + } + fp.apply(p.summary, n) + return + } +} + +// hasParsedAnyField reports whether the parser has seen and +// applied at least one fieldPatterns line. Used to disambiguate +// the two ═-rules in the banner (see feedLine). +func (p *SummaryParser) hasParsedAnyField() bool { + if p.summary == nil { + return false + } + // "Any field non-zero" is a fine proxy — the ingestor always + // prints at least one positive count even on early failure + // (TotalRecords includes everything it loaded before crashing). + return p.summary.TotalRecords > 0 || + p.summary.ProcessedRecords > 0 || + p.summary.InsertedRecords > 0 || + p.summary.APISentRecords > 0 || + p.summary.SkippedRecords > 0 || + p.summary.FileTransferFailures > 0 || + p.summary.FailedRecords > 0 +} + +// Result returns the accumulated Summary. nil if the parser never +// saw a banner header (early failure, OOM before the ingestor got +// to print its results). Safe to call at any point — the parser +// returns the in-progress accumulator if not yet finalized. +func (p *SummaryParser) Result() *Summary { + return p.summary +} + +// RenderPanel returns a multi-line, customer-facing rendering of +// the Summary for display in the orchestrator's success/failure +// message. Format: +// +// ┌─ Ingestion summary ──────────────────────────┐ +// │ Ingestor ID: │ +// │ Total records: 1,234 │ +// │ Inserted: 1,200 │ +// │ Skipped: 4 │ +// │ File transfer failures: 0 │ +// │ DB-insert failures: 30 │ +// │ Success rate: 97.2% │ +// └──────────────────────────────────────────────┘ +// +// Uses box-drawing characters for visual structure. Plain ASCII +// fallback could be added in v0.2 for terminals that don't render +// Unicode (rare on modern OS X/Linux/Windows-Terminal). +func RenderPanel(s *Summary) string { + if s == nil { + return "" + } + const labelWidth = 26 + var b strings.Builder + b.WriteString("┌─ Ingestion summary ──────────────────────────┐\n") + row := func(label, value string) { + fmt.Fprintf(&b, "│ %-*s %s\n", labelWidth, label, value) + } + if s.IngestorID != "" { + row("Ingestor ID:", s.IngestorID) + } + row("Total records:", commaSep(s.TotalRecords)) + row("Inserted:", commaSep(s.InsertedRecords)) + row("Sent to API:", commaSep(s.APISentRecords)) + row("Skipped:", commaSep(s.SkippedRecords)) + row("File transfer failures:", commaSep(s.FileTransferFailures)) + row("DB-insert failures:", commaSep(s.FailedRecords)) + row("Success rate:", fmt.Sprintf("%.1f%%", s.SuccessRate())) + b.WriteString("└──────────────────────────────────────────────┘\n") + return b.String() +} + +// commaSep formats an int64 with thousands-separator commas to +// match the ingestor's own banner format. Pure Go, no x/text. +func commaSep(n int64) string { + s := strconv.FormatInt(n, 10) + if len(s) <= 3 { + return s + } + // Insert commas every 3 digits from the right. Handle the + // optional leading '-' by carving it off first. + neg := "" + if s[0] == '-' { + neg = "-" + s = s[1:] + } + var out []byte + for i, c := range []byte(s) { + if i > 0 && (len(s)-i)%3 == 0 { + out = append(out, ',') + } + out = append(out, c) + } + return neg + string(out) +} + +// (compile-time test: bufio + io are referenced via Feed's buffer) +var _ = bufio.ScanLines +var _ io.Writer = parserWriter{} // ensures parserWriter satisfies io.Writer diff --git a/internal/submit/summary_test.go b/internal/submit/summary_test.go new file mode 100644 index 00000000..5bfe9505 --- /dev/null +++ b/internal/submit/summary_test.go @@ -0,0 +1,240 @@ +package submit + +import ( + "strings" + "testing" +) + +// realIngestorBanner mirrors what +// tracebloc_ingestor/ingestors/base.py:624+ actually prints, +// ANSI codes and all. The parser strips ANSI before matching so +// the test is bit-exact with the production logs. +// +// The leading "preamble" line + trailing "post-banner" lines +// simulate what real ingestor logs look like — the parser must +// ignore non-banner content + finalize on the closing rule. +var realIngestorBanner = "starting up...\n" + + "loaded 1234 rows from labels.csv\n" + + "\n" + + "\x1b[36m" + strings.Repeat("═", 60) + "\x1b[0m\n" + + "\x1b[1m\x1b[36m📊 INGESTION SUMMARY 📊\x1b[0m\n" + + "\x1b[36m" + strings.Repeat("═", 60) + "\x1b[0m\n" + + "\x1b[1mIngestor ID:\x1b[0m \x1b[34mrun-abc-123\x1b[0m\n" + + "\x1b[1m📈 Total Records Found:\x1b[0m \x1b[34m1,234\x1b[0m\n" + + "\x1b[1m✅ Successfully Processed:\x1b[0m \x1b[32m1,200\x1b[0m\n" + + "\x1b[1m💾 Inserted to Database:\x1b[0m \x1b[32m1,200\x1b[0m\n" + + "\x1b[1m🚀 Sent to API:\x1b[0m \x1b[32m1,150\x1b[0m\n" + + "\x1b[1m⏭️ Skipped Records:\x1b[0m \x1b[33m4\x1b[0m\n" + + "\x1b[1m📁 File Transfer Failures:\x1b[0m \x1b[32m0\x1b[0m\n" + + "\x1b[1m❌ Failed DB Insertion:\x1b[0m \x1b[31m30\x1b[0m\n" + + "\x1b[36m" + strings.Repeat("═", 60) + "\x1b[0m\n" + + "ingestor exiting cleanly\n" + +// TestSummaryParser_RealBannerEndToEnd pins the parser against +// the actual ingestor's output format. If a regression breaks +// any field's extraction, this test fails with a clear "got X +// want Y" for that specific counter. +func TestSummaryParser_RealBannerEndToEnd(t *testing.T) { + p := NewSummaryParser() + p.Feed([]byte(realIngestorBanner)) + + got := p.Result() + if got == nil { + t.Fatal("parser returned nil Result; expected populated Summary") + } + + cases := []struct { + name string + got any + want any + }{ + {"IngestorID", got.IngestorID, "run-abc-123"}, + {"TotalRecords", got.TotalRecords, int64(1234)}, + {"ProcessedRecords", got.ProcessedRecords, int64(1200)}, + {"InsertedRecords", got.InsertedRecords, int64(1200)}, + {"APISentRecords", got.APISentRecords, int64(1150)}, + {"SkippedRecords", got.SkippedRecords, int64(4)}, + {"FileTransferFailures", got.FileTransferFailures, int64(0)}, + {"FailedRecords", got.FailedRecords, int64(30)}, + } + for _, c := range cases { + if c.got != c.want { + t.Errorf("%s = %v, want %v", c.name, c.got, c.want) + } + } +} + +// TestSummaryParser_HasFailures pins the failure-detection logic +// that the orchestrator uses to choose between success exit code +// (0) and ingest-failure exit code (9). +func TestSummaryParser_HasFailures(t *testing.T) { + cases := []struct { + name string + s *Summary + want bool + }{ + {"nil", nil, false}, + {"all zero", &Summary{TotalRecords: 100, ProcessedRecords: 100}, false}, + {"file transfer failures", &Summary{FileTransferFailures: 1}, true}, + {"failed records", &Summary{FailedRecords: 1}, true}, + {"both", &Summary{FileTransferFailures: 1, FailedRecords: 1}, true}, + // Skipped records are NOT failures — they're rows that + // validators rejected. The customer wants to see the + // count but it doesn't change the exit code. + {"skipped is not failure", &Summary{SkippedRecords: 100}, false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := c.s.HasFailures(); got != c.want { + t.Errorf("HasFailures = %v, want %v", got, c.want) + } + }) + } +} + +// TestSummaryParser_SuccessRate pins the math that feeds the +// rendered panel's "Success rate: XX%" line. Divide-by-zero on +// empty banner is the critical edge case. +func TestSummaryParser_SuccessRate(t *testing.T) { + cases := []struct { + name string + s *Summary + want float64 + }{ + {"nil", nil, 0}, + {"empty banner", &Summary{}, 0}, + {"100%", &Summary{TotalRecords: 100, ProcessedRecords: 100}, 100}, + {"50%", &Summary{TotalRecords: 100, ProcessedRecords: 50}, 50}, + {"all failed", &Summary{TotalRecords: 100}, 0}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := c.s.SuccessRate(); got != c.want { + t.Errorf("SuccessRate = %v, want %v", got, c.want) + } + }) + } +} + +// TestSummaryParser_StreamedInChunks: real log streams arrive in +// arbitrary-sized chunks. Feed the banner one byte at a time to +// pin that the parser correctly buffers and finalizes despite +// partial-line input. +func TestSummaryParser_StreamedInChunks(t *testing.T) { + p := NewSummaryParser() + for i := 0; i < len(realIngestorBanner); i++ { + p.Feed([]byte{realIngestorBanner[i]}) + } + got := p.Result() + if got == nil { + t.Fatal("byte-by-byte feed produced nil Result") + } + if got.TotalRecords != 1234 || got.FailedRecords != 30 { + t.Errorf("chunked feed produced different result than monolithic: TotalRecords=%d FailedRecords=%d", + got.TotalRecords, got.FailedRecords) + } +} + +// TestSummaryParser_NoBanner: if the run died before producing a +// banner (image crashloop, OOM at startup), Result returns nil +// rather than an empty Summary. The orchestrator uses this +// nil-check to decide whether to render the panel at all. +func TestSummaryParser_NoBanner(t *testing.T) { + p := NewSummaryParser() + p.Feed([]byte("starting up...\nerror: connection refused\n")) + if got := p.Result(); got != nil { + t.Errorf("Result on no-banner log = %+v, want nil", got) + } +} + +// TestSummaryParser_PostBannerLogsIgnored: lines after the closing +// ═-rule are ignored. The ingestor may print shutdown messages +// after the summary; those shouldn't perturb the parsed counts +// (e.g. a regex-misfire interpreting "1234" in an unrelated log +// line as TotalRecords). +func TestSummaryParser_PostBannerLogsIgnored(t *testing.T) { + p := NewSummaryParser() + p.Feed([]byte(realIngestorBanner)) + pre := *p.Result() + p.Feed([]byte("📈 Total Records Found: 999999999\n")) // would alter TotalRecords if not finalized + post := *p.Result() + if pre != post { + t.Errorf("post-banner line altered Summary; pre=%+v post=%+v", pre, post) + } +} + +// TestStripANSI: the parser strips ANSI SGR codes from each line +// before matching. Validate the regex handles common shapes. +func TestStripANSI(t *testing.T) { + cases := []struct { + in, want string + }{ + {"plain text", "plain text"}, + {"\x1b[1mbold\x1b[0m", "bold"}, + {"\x1b[1;36mbold-cyan\x1b[0m", "bold-cyan"}, + {"prefix\x1b[31mred\x1b[0msuffix", "prefixredsuffix"}, + {"", ""}, + } + for _, c := range cases { + if got := stripANSI(c.in); got != c.want { + t.Errorf("stripANSI(%q) = %q, want %q", c.in, got, c.want) + } + } +} + +// TestRenderPanel_BasicShape: the panel rendering is what the +// customer sees on success; pin a few key lines so a refactor +// breaks the test rather than silently producing weird output. +func TestRenderPanel_BasicShape(t *testing.T) { + s := &Summary{ + IngestorID: "run-abc", + TotalRecords: 1234567, + InsertedRecords: 1200000, + APISentRecords: 1150000, + SkippedRecords: 4000, + FileTransferFailures: 30, + FailedRecords: 5, + } + got := RenderPanel(s) + for _, want := range []string{ + "Ingestion summary", + "run-abc", + "1,234,567", // commaSep formatting + "1,200,000", // commaSep formatting + "30", // file transfer failures + "DB-insert failures:", + } { + if !strings.Contains(got, want) { + t.Errorf("RenderPanel missing %q in:\n%s", want, got) + } + } +} + +// TestRenderPanel_Nil: nil summary returns empty string so the +// orchestrator can blind-print without a guard. +func TestRenderPanel_Nil(t *testing.T) { + if got := RenderPanel(nil); got != "" { + t.Errorf("RenderPanel(nil) = %q, want empty", got) + } +} + +// TestCommaSep: small helper test. Pin the boundary cases that +// would catch off-by-one in the comma-insertion loop. +func TestCommaSep(t *testing.T) { + cases := []struct { + in int64 + want string + }{ + {0, "0"}, + {999, "999"}, + {1000, "1,000"}, + {12345, "12,345"}, + {1234567, "1,234,567"}, + {-1234, "-1,234"}, + } + for _, c := range cases { + if got := commaSep(c.in); got != c.want { + t.Errorf("commaSep(%d) = %q, want %q", c.in, got, c.want) + } + } +} diff --git a/internal/submit/watch.go b/internal/submit/watch.go new file mode 100644 index 00000000..9a23f164 --- /dev/null +++ b/internal/submit/watch.go @@ -0,0 +1,360 @@ +package submit + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" +) + +// Watch-loop tunables. Both deliberately conservative — Phase 4's +// watching is the dominant time-spend of a typical push (the actual +// data move was done in Phase 3; what's left is the in-cluster +// ingestion which can take minutes-to-an-hour for large datasets). +const ( + // JobPollInterval is how often the watch loop re-Gets the Job. + // 2s is a sweet spot: human-perceptible enough that a + // 30-second ingestion has clean lifecycle output, light + // enough that an hour-long ingestion adds <2000 API calls + // (negligible at the apiserver's ~10k req/s ceiling). + JobPollInterval = 2 * time.Second + + // JobWatchTimeout is the absolute cap on a single + // dataset-push's watch phase. 1 hour is generous — typical + // image_classification ingestions finish in <10 min; cap + // exists to avoid infinite hangs if the cluster goes weird + // (kubelet stops reporting status, etc.). Customers running + // hours-long ingestions should use --detach. + JobWatchTimeout = 1 * time.Hour + + // PodPollInterval is how often we look for the ingestor Job's + // Pod once we have the Job name. Same 2s as the Job-level + // poll; same rationale. + PodPollInterval = 2 * time.Second + + // PodReadyTimeout caps how long the Job's Pod has to be + // schedulable + Running before we give up looking for it. + // 5 min covers image pull on a slow registry; beyond that + // the ingestion isn't going to start at all and the customer + // wants the diagnostic. + PodReadyTimeout = 5 * time.Minute +) + +// JobOutcome enumerates the terminal states the watch loop reports. +// The orchestrator (submit.go) maps these to exit codes. +type JobOutcome int + +const ( + // JobOutcomeUnknown is the zero value — never returned, used + // only as a switch-default sentinel. + JobOutcomeUnknown JobOutcome = iota + + // JobOutcomeSucceeded means the ingestor Job's Pod exited 0. + // Maps to exit code 0 in the orchestrator. + JobOutcomeSucceeded + + // JobOutcomeFailed means the ingestor Job's Pod exited + // non-zero (any cause: ingestion runtime error, OOM, image + // crashloop). Maps to the "ingest" exit code (9) in the + // orchestrator. The Pod-side summary banner may or may not + // have been printed depending on how far the run got — the + // orchestrator parses what it can. + JobOutcomeFailed + + // JobOutcomeDetached means the customer Ctrl-C'd mid-watch. + // jobs-manager already accepted the run; the cluster will + // continue without us. Maps to exit code 0 with a "reconnect + // with kubectl logs" hint in the orchestrator output. + JobOutcomeDetached +) + +func (o JobOutcome) String() string { + switch o { + case JobOutcomeSucceeded: + return "Succeeded" + case JobOutcomeFailed: + return "Failed" + case JobOutcomeDetached: + return "Detached" + default: + return "Unknown" + } +} + +// WatchResult bundles everything the orchestrator wants from the +// watch loop. Outcome drives the exit code; PodName is what the +// detach-hint prints; Summary is the structured INGESTION SUMMARY +// (nil if the run didn't produce one — early failure, OOM at +// startup, etc.). +type WatchResult struct { + Outcome JobOutcome + PodName string + + // Summary is the parsed 📊 banner, nil on early failure. + // The orchestrator decides whether to render the panel + // (success path) or include it in the failure framing + // (failed-after-summary path). + Summary *Summary +} + +// WatchJob is the top-level watch loop: poll the Job until it +// reaches a terminal phase, stream the Pod's logs while it's +// running, and return a WatchResult. +// +// SIGINT contract (Bugbot-r9 echo for the previous package): +// the caller (cli/main.go via signal.NotifyContext) is expected +// to cancel ctx on Ctrl-C. WatchJob detects ctx.Err() == Canceled +// and returns Outcome=Detached rather than treating it as a poll +// failure. The customer who Ctrl-C'd during the watch sees the +// "your ingestion is still running in the cluster; reconnect with +// kubectl logs " hint. +// +// 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 `. +func WatchJob( + ctx context.Context, + cs kubernetes.Interface, + namespace, jobName string, + out io.Writer, +) (*WatchResult, error) { + // 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(ctx, cs, namespace, jobName) + if err != nil { + if errors.Is(err, context.Canceled) { + // SIGINT before the Pod even appeared. jobs-manager + // has accepted the run, the cluster will run it, + // the CLI is just not watching anymore. + return &WatchResult{Outcome: JobOutcomeDetached}, nil + } + return nil, fmt.Errorf("waiting for ingestor Pod: %w", err) + } + + // 2. Stream Pod logs. This blocks until the Pod terminates + // or ctx is cancelled. We don't need a separate Job-status + // poll here because the Pod terminating drains the log + // stream — when GetLogs(Follow=true) returns EOF, the + // Pod has completed (success or failure). + // + // Any text the ingestor prints (including the 📊 banner + // at the end) flows verbatim through `out`. We also feed + // 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. + summary, logErr := streamPodLogsAndParse(ctx, cs, namespace, podName, out) + if logErr != nil && !errors.Is(logErr, context.Canceled) { + return nil, fmt.Errorf("streaming logs from Pod %s/%s: %w", namespace, podName, logErr) + } + + // 3. ctx cancellation = detach. Pod might still be running + // server-side, but as far as the CLI is concerned we've + // handed off. + if errors.Is(ctx.Err(), context.Canceled) { + return &WatchResult{ + Outcome: JobOutcomeDetached, + PodName: podName, + Summary: summary, // may be partial if the customer + // hit Ctrl-C right at the banner + }, nil + } + + // 4. Final status check: even though log streaming ending + // SHOULD mean Pod completion, the apiserver may not have + // posted the terminal state by the time the log stream + // drained. Poll briefly to confirm — bounded at 30s + // because beyond that something else is wrong. + outcome, err := finalJobStatus(ctx, cs, namespace, jobName) + if err != nil { + return nil, fmt.Errorf("reading final Job status for %s/%s: %w", namespace, jobName, err) + } + return &WatchResult{ + Outcome: outcome, + PodName: podName, + Summary: summary, + }, nil +} + +// waitForJobPod polls until the Job has spawned its Pod and that +// 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) { + var podName string + err := wait.PollUntilContextTimeout(ctx, PodPollInterval, PodReadyTimeout, true, + func(ctx context.Context) (bool, error) { + pods, err := cs.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ + LabelSelector: "job-name=" + jobName, + }) + if err != nil { + // Terminal errors short-circuit (echoes the + // Phase-3 r3 fix in push/pod.go). + if apierrors.IsForbidden(err) { + return false, err + } + return false, nil // transient + } + if len(pods.Items) == 0 { + return false, nil // Pod hasn't been created yet + } + // Pick the first Pod with the label; Jobs default to + // parallelism=1 so there's only one. If a future + // jobs-manager spawns parallel Pods, the first one + // to surface logs is the one we attach to. + p := pods.Items[0] + switch p.Status.Phase { + case corev1.PodRunning, corev1.PodSucceeded, corev1.PodFailed: + // Running is the happy path; Succeeded/Failed + // mean we missed the Running window (typical + // for fast-completing ingestions). In all three + // cases we can fetch logs. + podName = p.Name + return true, nil + default: + // Pending: still pulling image / scheduling. Keep + // polling. The PodReadyTimeout caps this loop. + return false, nil + } + }) + if err != nil { + return "", err + } + return podName, nil +} + +// streamPodLogsAndParse opens a streaming log read on the Pod and +// pipes it through (a) a TeeReader to `out` for verbatim display +// and (b) a Summary parser for the 📊 banner. Returns the parsed +// Summary (or nil if the Pod never produced one) + the underlying +// stream error. +// +// Streaming model: Follow=true means the API server keeps the +// connection open until the Pod terminates. When the Pod's +// container exits, the stream returns EOF and we drop out. This +// avoids the "poll twice" anti-pattern where Phase 4 would have to +// re-fetch logs after the Job is done to see the summary. +func streamPodLogsAndParse( + ctx context.Context, + cs kubernetes.Interface, + namespace, podName string, + out io.Writer, +) (*Summary, error) { + req := cs.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{ + Follow: true, + // Container omitted — Job Pods have exactly one container + // (the ingestor). If a future ingestor adds a sidecar + // (e.g. for metrics scrape), this needs to specify + // `Container: "ingestor"`. + }) + stream, err := req.Stream(ctx) + if err != nil { + return nil, err + } + defer func() { _ = stream.Close() }() + + // Wrap the stream in a TeeReader so each line flows through + // both customer-facing output AND the summary parser. The + // parser keeps a small ring buffer internally; it doesn't + // need to see the full stream in memory. + parser := NewSummaryParser() + tee := io.TeeReader(stream, parserWriter{parser: parser}) + + // Line-by-line copy so the customer sees output progressively. + // io.Copy would also work but would buffer chunks at the + // transport layer, making the output feel laggy on a fast + // ingestion. + scanner := bufio.NewScanner(tee) + // Default scanner buffer is 64 KB per line — fine for log + // lines but bump to 1 MB to handle the (rare) case where a + // single ingestion-error stacktrace has a multi-KB Python + // traceback line. + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Bytes() + // Print the line + a newline (scanner strips the trailing + // '\n'). errcheck-friendly: we discard the writer error + // because the exit code is the customer-facing contract. + _, _ = out.Write(line) + _, _ = out.Write([]byte("\n")) + } + if err := scanner.Err(); err != nil { + // EOF is normal end-of-stream; other errors (network drop + // mid-stream, ctx cancel) get propagated. + if !errors.Is(err, io.EOF) { + return parser.Result(), err + } + } + return parser.Result(), nil +} + +// parserWriter adapts a SummaryParser into an io.Writer for use +// with io.TeeReader. The TeeReader writes everything to its +// secondary sink as bytes flow through; the parser's Feed method +// accepts them line-by-line internally. +type parserWriter struct { + parser *SummaryParser +} + +func (pw parserWriter) Write(b []byte) (int, error) { + pw.parser.Feed(b) + return len(b), nil +} + +// finalJobStatus does a bounded poll on the Job's status to +// determine Succeeded vs Failed after log streaming ends. This is +// a separate step because the log-stream-end doesn't always race +// the Job-status-update; we need to wait briefly for the +// apiserver to post the terminal phase. +func finalJobStatus(ctx context.Context, cs kubernetes.Interface, namespace, jobName string) (JobOutcome, error) { + var outcome JobOutcome + err := wait.PollUntilContextTimeout(ctx, JobPollInterval, 30*time.Second, true, + func(ctx context.Context) (bool, error) { + job, err := cs.BatchV1().Jobs(namespace).Get(ctx, jobName, metav1.GetOptions{}) + if err != nil { + if apierrors.IsForbidden(err) || apierrors.IsNotFound(err) { + return false, err + } + return false, nil + } + // batch/v1 Job conditions: Complete (success) or + // Failed. Poll until one is set; in practice this + // resolves within ~1s of log stream EOF. + for _, c := range job.Status.Conditions { + if c.Status != corev1.ConditionTrue { + continue + } + switch c.Type { + case batchv1.JobComplete: + outcome = JobOutcomeSucceeded + return true, nil + case batchv1.JobFailed: + outcome = JobOutcomeFailed + return true, nil + } + } + return false, nil + }) + if err != nil { + // If the poll timed out without seeing a terminal + // condition, the apiserver is being slow. Treat as + // Unknown rather than failing the whole push — the + // orchestrator can render a useful diagnostic from the + // streamed logs. + if errors.Is(err, context.DeadlineExceeded) { + return JobOutcomeUnknown, nil + } + return JobOutcomeUnknown, err + } + return outcome, nil +} diff --git a/internal/submit/watch_test.go b/internal/submit/watch_test.go new file mode 100644 index 00000000..5a1ce974 --- /dev/null +++ b/internal/submit/watch_test.go @@ -0,0 +1,220 @@ +package submit + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +// jobPod constructs a Pod owned by `jobName` via the standard +// batch/v1 job-name label. Used to seed the fake clientset for +// waitForJobPod tests. +func jobPod(name, jobName string, phase corev1.PodPhase) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "tracebloc", + Labels: map[string]string{"job-name": jobName}, + }, + Status: corev1.PodStatus{Phase: phase}, + } +} + +// TestWaitForJobPod_RunningPodSurfaces: a Pod with job-name label +// in Phase=Running is returned. Pin the label-selector contract + +// the happy-path return. +func TestWaitForJobPod_RunningPodSurfaces(t *testing.T) { + cs := fake.NewClientset(jobPod("ingestor-abc-xyz", "ingestor-abc", corev1.PodRunning)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + name, 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) + } +} + +// TestWaitForJobPod_FastCompletionPath: ingestions that finish +// faster than the poll interval might be in Phase=Succeeded by +// the time the watch loop checks. We still want the Pod's name so +// we can fetch its (post-mortem) logs. +func TestWaitForJobPod_FastCompletionPath(t *testing.T) { + cs := fake.NewClientset(jobPod("ingestor-fast", "ingestor", corev1.PodSucceeded)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + name, 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) + } +} + +// TestWaitForJobPod_ForbiddenIsTerminal: an RBAC denial on +// `list pods` must short-circuit the wait — otherwise the +// customer sits through PodReadyTimeout for an error that +// doesn't change. +func TestWaitForJobPod_ForbiddenIsTerminal(t *testing.T) { + cs := fake.NewClientset() + cs.PrependReactor("list", "pods", + func(_ k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewForbidden( + corev1.Resource("pods"), "", + errors.New("user cannot list pods")) + }) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + start := time.Now() + _, err := waitForJobPod(ctx, cs, "tracebloc", "j") + elapsed := time.Since(start) + if err == nil { + t.Fatal("waitForJobPod returned nil on Forbidden") + } + if elapsed > 3*time.Second { + t.Errorf("waitForJobPod waited %s on Forbidden; expected immediate return", elapsed) + } +} + +// TestWaitForJobPod_NoPodEverShows: ingestor Job that doesn't +// spawn its Pod (image pull stuck, scheduling impossible) hits +// the PodReadyTimeout. Bound the test's ctx so it doesn't wait +// the full 5min. +func TestWaitForJobPod_NoPodEverShows(t *testing.T) { + cs := fake.NewClientset() // empty + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + _, err := waitForJobPod(ctx, cs, "tracebloc", "missing-job") + if err == nil { + t.Fatal("waitForJobPod returned nil when no Pod ever appeared") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("error doesn't wrap DeadlineExceeded: %v", err) + } +} + +// jobWithCondition constructs a batch/v1 Job whose status reports +// the given terminal condition. Used to seed finalJobStatus tests. +func jobWithCondition(name string, cond batchv1.JobConditionType) *batchv1.Job { + return &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "tracebloc"}, + Status: batchv1.JobStatus{ + Conditions: []batchv1.JobCondition{{ + Type: cond, + Status: corev1.ConditionTrue, + }}, + }, + } +} + +// TestFinalJobStatus_Complete: Job with Condition=Complete maps +// to JobOutcomeSucceeded. +func TestFinalJobStatus_Complete(t *testing.T) { + cs := fake.NewClientset(jobWithCondition("done", batchv1.JobComplete)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + out, err := finalJobStatus(ctx, cs, "tracebloc", "done") + if err != nil { + t.Fatalf("finalJobStatus: %v", err) + } + if out != JobOutcomeSucceeded { + t.Errorf("Outcome = %v, want Succeeded", out) + } +} + +// TestFinalJobStatus_Failed: Job with Condition=Failed maps to +// JobOutcomeFailed (drives the exit-9 "ingest failed" path). +func TestFinalJobStatus_Failed(t *testing.T) { + cs := fake.NewClientset(jobWithCondition("crashed", batchv1.JobFailed)) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + out, err := finalJobStatus(ctx, cs, "tracebloc", "crashed") + if err != nil { + t.Fatalf("finalJobStatus: %v", err) + } + if out != JobOutcomeFailed { + t.Errorf("Outcome = %v, want Failed", out) + } +} + +// TestFinalJobStatus_TimeoutIsUnknown: if no terminal condition +// posts within 30s of the log stream ending, we return Unknown +// rather than blocking the customer forever. The orchestrator +// renders a useful diagnostic from the streamed logs. +func TestFinalJobStatus_TimeoutIsUnknown(t *testing.T) { + cs := fake.NewClientset(&batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{Name: "stuck", Namespace: "tracebloc"}, + // No conditions — Job in a weird mid-state. + }) + + // Tight ctx so the test doesn't take 30s. + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + out, err := finalJobStatus(ctx, cs, "tracebloc", "stuck") + if err != nil { + t.Fatalf("finalJobStatus: %v", err) + } + if out != JobOutcomeUnknown { + t.Errorf("Outcome = %v, want Unknown", out) + } +} + +// TestJobOutcome_String: stringer pin so diagnostic output stays +// stable. +func TestJobOutcome_String(t *testing.T) { + cases := map[JobOutcome]string{ + JobOutcomeSucceeded: "Succeeded", + JobOutcomeFailed: "Failed", + JobOutcomeDetached: "Detached", + JobOutcomeUnknown: "Unknown", + } + for o, want := range cases { + if got := o.String(); got != want { + t.Errorf("%v.String() = %q, want %q", o, got, want) + } + } +} + +// TestParserWriter_FeedsParser: the io.Writer adapter that hooks +// the log-stream TeeReader to the SummaryParser. Pin that writes +// flow through to Feed correctly. +func TestParserWriter_FeedsParser(t *testing.T) { + p := NewSummaryParser() + pw := parserWriter{parser: p} + chunks := strings.Split(realIngestorBanner, "\n") + for _, line := range chunks { + _, _ = pw.Write([]byte(line + "\n")) + } + got := p.Result() + if got == nil { + t.Fatal("parserWriter didn't feed parser; Result is nil") + } + if got.TotalRecords != 1234 { + t.Errorf("TotalRecords = %d, want 1234 (via parserWriter)", got.TotalRecords) + } +} From 257c8efc45cb67b729108702552609560001e79b Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 20:34:23 +0500 Subject: [PATCH 2/8] fix(submit): enforce JobWatchTimeout + flush parser + typed watch errors + Unknown exit code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four Bugbot findings on PR #10's initial commit, all real: ## Medium — JobWatchTimeout was declared but never enforced The constant existed and documented "absolute cap on a single dataset-push's watch phase," but WatchJob never used it. Per-step timeouts (PodReadyTimeout, finalJobStatus's 30s) bounded sub-loops, but the log-stream phase ran indefinitely against ctx.Done() with no overall cap. A stuck ingestor would hang the CLI forever. Fix: context.WithTimeout(ctx, JobWatchTimeout) at the top of WatchJob. Cancel via defer. Cap fires as ctx.DeadlineExceeded through the inner loops. ## Medium — Watch failures used the submit exit code submit.Run wrapped both POST errors AND watch errors in the same return path. cli/dataset.go's switch mapped IsAuthError → 5 and "everything else" → 8 (submit-failed). But waitForJobPod / streamPodLogsAndParse / finalJobStatus failures aren't submit failures — jobs-manager already accepted the run, the cluster is doing the work, the CLI just lost the ability to follow along. Exit 8 misled customers into thinking the submit was rejected. Fix: introduce a WatchError type wrapping watch-phase errors; add IsWatchError predicate. cli/dataset.go now routes: IsAuthError → 5 (token / auth) IsWatchError → 9 (ingest-side; cluster keeps running) else → 8 (submit-side rejection) Test TestIsWatchError pins the typing including wrapped errors. ## Low — Log parser didn't flush partial last lines streamPodLogsAndParse used bufio.Scanner for line-based stream display, but the SummaryParser's own internal buffer (for chunked-write line assembly) was never flushed at end-of-stream. If the Pod terminated without a trailing '\n' on its final stdout write — rare but possible if the container's stdout was killed mid-write — the final line would be lost, potentially including the closing ═-rule that finalizes the banner. Fix: parser.FlushLine() at end-of-stream AND on non-EOF scanner errors. ## Medium — Outcome=Unknown exited zero The exit-code switch only handled Failed + Succeeded. Outcome= Unknown (returned by finalJobStatus when the 30s post-stream poll doesn't see a terminal condition) fell through to the default `return nil` → exit 0. But "we don't know if it worked" is not success. Fix: add an explicit Unknown branch that exits 9 with a diagnostic pointing the customer at `kubectl get job ...` to check the actual outcome. Local: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 23 ++++++++++++++++++++--- internal/submit/submit.go | 30 +++++++++++++++++++++++++++++- internal/submit/submit_test.go | 26 ++++++++++++++++++++++++++ internal/submit/watch.go | 18 ++++++++++++++++++ 4 files changed, 93 insertions(+), 4 deletions(-) diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 3a75dcbb..a938bf1e 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -425,10 +425,19 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush Out: out, }) if err != nil { - if submit.IsAuthError(err) { + switch { + case submit.IsAuthError(err): return &exitError{code: 5, err: err} + case submit.IsWatchError(err): + // Watch-phase failure: jobs-manager already accepted + // the run, the cluster is doing the work, the CLI + // just couldn't follow along. Exit 9 (ingest-side) + // not 8 (submit-side). Bugbot flagged the + // previously-undifferentiated mapping on PR #10. + return &exitError{code: 9, err: err} + default: + return &exitError{code: 8, err: err} } - return &exitError{code: 8, err: err} } // Detach paths (--detach flag OR SIGINT-mid-watch) are @@ -438,10 +447,18 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return nil } - // Watch outcomes. + // Watch outcomes. Both Failed and Unknown route to exit 9 + // (Unknown = finalJobStatus timed out without seeing a + // terminal condition, which we can't claim as success). + // Bugbot flagged the prior switch's missing Unknown branch + // on PR #10. switch submitRes.Watch.Outcome { case submit.JobOutcomeFailed: return &exitError{code: 9, err: errors.New("ingestion Job exited non-zero — see logs above")} + case submit.JobOutcomeUnknown: + return &exitError{code: 9, err: errors.New( + "ingestion Job's final status couldn't be determined within the watch window — " + + "check `kubectl get job -n " + submitRes.Submit.Namespace + " " + submitRes.Submit.JobName + "` for the outcome")} case submit.JobOutcomeSucceeded: if submitRes.Watch.Summary != nil && submitRes.Watch.Summary.HasFailures() { return &exitError{code: 9, err: errors.New( diff --git a/internal/submit/submit.go b/internal/submit/submit.go index 7478da0a..9869fd42 100644 --- a/internal/submit/submit.go +++ b/internal/submit/submit.go @@ -123,7 +123,11 @@ func Run(ctx context.Context, opts Options) (*Result, error) { _, _ = 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) if err != nil { - return &Result{Submit: resp}, fmt.Errorf("watching ingestor Job: %w", err) + // Tag as WatchError so the orchestrator picks the + // ingest-flavored exit code (9), not the submit-flavored + // one (8). The cluster has already accepted the run by + // this point — the CLI just failed to follow it. + return &Result{Submit: resp}, &WatchError{Err: fmt.Errorf("watching ingestor Job: %w", err)} } // Detach-from-SIGINT path. Print the reconnect hint so the @@ -159,3 +163,27 @@ func IsAuthError(err error) bool { } return se.StatusCode == 401 || se.StatusCode == 403 } + +// WatchError wraps errors that originated in the watch phase +// (waitForJobPod, log streaming, finalJobStatus). The +// orchestrator distinguishes these from submit-phase errors so +// the exit-code mapping is correct: jobs-manager accepted the +// run already, the cluster is doing the work, the CLI just +// failed to follow along. Maps to exit code 9 (ingest-side +// problem), not 8 (submit-side problem). Bugbot flagged the +// previous "everything that wasn't auth → exit 8" version on +// PR #10. +type WatchError struct { + Err error +} + +func (e *WatchError) Error() string { return e.Err.Error() } +func (e *WatchError) Unwrap() error { return e.Err } + +// IsWatchError reports whether err originated in the watch phase +// rather than the submit phase. The orchestrator's exit-code +// branch uses this directly. +func IsWatchError(err error) bool { + var we *WatchError + return errors.As(err, &we) +} diff --git a/internal/submit/submit_test.go b/internal/submit/submit_test.go index 6bb32512..cf34126d 100644 --- a/internal/submit/submit_test.go +++ b/internal/submit/submit_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "strings" "testing" @@ -180,6 +181,31 @@ func TestRun_NilOutDefaultsToDiscard(t *testing.T) { } } +// TestIsWatchError: pin the contract that the orchestrator uses +// to distinguish watch-phase failures (exit 9) from submit-phase +// failures (exit 8). Bugbot r1 found the missing distinction; +// this test guards against a regression that drops the typing. +func TestIsWatchError(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"plain error", errors.New("plain"), false}, + {"submit error", &SubmitError{StatusCode: 422}, false}, + {"watch error", &WatchError{Err: errors.New("inner")}, true}, + {"wrapped watch error", fmt.Errorf("outer: %w", &WatchError{Err: errors.New("inner")}), true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := IsWatchError(c.err); got != c.want { + t.Errorf("IsWatchError = %v, want %v", got, c.want) + } + }) + } +} + // TestIsAuthError: helper smoke test. Pin the contract used by // the CLI's exit-code mapping. func TestIsAuthError(t *testing.T) { diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 9a23f164..9cca42e7 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -127,6 +127,16 @@ func WatchJob( namespace, jobName string, out io.Writer, ) (*WatchResult, error) { + // Enforce the JobWatchTimeout absolute cap. Earlier versions + // declared the constant but only the per-step timeouts + // (PodReadyTimeout, finalJobStatus's 30s) bounded the watch — + // the log-stream phase could run indefinitely on a stuck + // ingestor. Bugbot flagged the gap on PR #10. The cap + // triggers as ctx.DeadlineExceeded, which the orchestrator + // surfaces with a clear "watch timed out" framing. + ctx, cancel := context.WithTimeout(ctx, JobWatchTimeout) + defer cancel() + // 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 @@ -292,9 +302,17 @@ func streamPodLogsAndParse( // EOF is normal end-of-stream; other errors (network drop // mid-stream, ctx cancel) get propagated. if !errors.Is(err, io.EOF) { + // Flush any buffered partial line before returning so + // the parser sees content even on mid-line failure. + parser.FlushLine() return parser.Result(), err } } + // Flush at the end of stream too. A Pod that exited without + // a trailing newline on its final stdout write would otherwise + // lose the final line — including potentially the closing + // ═-rule that finalizes the banner. Bugbot flagged on PR #10. + parser.FlushLine() return parser.Result(), nil } From 56945606dd388ce386c7054faeecdcb7bd4b9b7a Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 21:02:06 +0500 Subject: [PATCH 3/8] fix(submit): zero-metric finalize + namespace check + fresh ctx for final status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 of Bugbot fixes on PR #10. Four findings, all real — three are second-order effects of the r1 fixes: ## Medium — Zero-metric banner never finalized The hasParsedAnyField check (which gates the closing ═-rule's finalization) required ANY metric counter to be > 0. But an ingestion that ran-then-died-early can legitimately produce a banner where ALL counts are 0 — the ingestor still prints the structure, just with zeroes. The parser would never finalize on that case, leaving the parser open to perturbation from post-banner shutdown logs. Fix: track a dedicated sawAnyField bool that latches on ANY successful field-line parse, regardless of the value. The finalization check now keys off "did we see any banner field" instead of "did we see a non-zero one." ## Medium — Missing submit response namespace not rejected client.go rejected 2xx responses with no job_name but happily accepted responses with no namespace. A malformed-but-parsing response would then drive the watch loop's k8s API calls at the empty namespace, plus produce a broken `kubectl logs` reconnect hint. Fix: parallel check for empty Namespace, same error shape as the job_name check. ## Medium — Watch timeout starved finalJobStatus R1 wrapped the WHOLE WatchJob in ctx.WithTimeout(JobWatchTimeout) to enforce the 1-hour cap. Side effect: finalJobStatus (which runs AFTER log streaming) inherited the SAME ctx — so a 59-minute log stream left finalJobStatus with 1 minute of budget, and a 60-minute stream left it with 0. Successful slow ingestions misreported as Unknown → exit 9. Fix: split the ctx hierarchy. - customerCtx — original input, carries SIGINT cancellation - watchCtx = WithTimeout(customerCtx, JobWatchTimeout) — caps the pod-wait + log-stream phases - finalCtx = WithTimeout(customerCtx, 30s) — fresh budget derived from customerCtx, not from watchCtx The customerCtx-vs-watchCtx distinction also fixes the detach detection: check customerCtx.Err() not ctx.Err() so a JobWatchTimeout expiry doesn't masquerade as a SIGINT-Detach. ## Medium — SIGINT after logs yielded exit 9 After log streaming ended cleanly, Ctrl-C during the brief finalJobStatus poll returned a watch error → exit 9. But the contract is "post-201 SIGINT = graceful detach, exit 0" — the cluster already has the work, the customer is just done watching. Fix: if customerCtx is Canceled inside the finalJobStatus error branch, return Outcome=Detached with the same reconnect-hint path the log-stream-cancel case uses. Consistent detach semantics across both watch sub-phases. Local: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/submit/client.go | 9 ++++++ internal/submit/summary.go | 30 +++++++++-------- internal/submit/watch.go | 66 ++++++++++++++++++++++++++------------ 3 files changed, 72 insertions(+), 33 deletions(-) diff --git a/internal/submit/client.go b/internal/submit/client.go index 5ef433a5..972d91bc 100644 --- a/internal/submit/client.go +++ b/internal/submit/client.go @@ -153,6 +153,15 @@ func (s *HTTPSubmitter) Submit(ctx context.Context, req *SubmitRequest) (*Submit // job_name would silently break Phase 4's watch step. return nil, fmt.Errorf("submit response missing job_name (got body %q)", string(respBody)) } + if parsed.Namespace == "" { + // Same shape as the job_name check: a missing namespace + // would route subsequent k8s API calls (watch the Pod, + // stream logs) at the empty string — kubelet returns + // confusing errors, and the kubectl-logs reconnect hint + // printed in --detach output would be malformed. Bugbot + // PR #10 r2 flagged the gap. + return nil, fmt.Errorf("submit response missing namespace (got body %q)", string(respBody)) + } return &parsed, nil } diff --git a/internal/submit/summary.go b/internal/submit/summary.go index 5cf2e7df..23fd3753 100644 --- a/internal/submit/summary.go +++ b/internal/submit/summary.go @@ -183,6 +183,16 @@ type SummaryParser struct { // (so e.g. a stray emoji in earlier log output doesn't // trigger spurious parsing). insideBanner bool + + // sawAnyField latches once we successfully parse a + // fieldPatterns line (regardless of whether the count is + // zero). Used by feedLine to distinguish the opening ═-rule + // (no fields yet → ignore) from the closing one (fields + // already parsed → finalize). The earlier "any field + // non-zero" check failed on banners where every metric was + // genuinely 0 (early failure case) — Bugbot caught this on + // PR #10 round 2. + sawAnyField bool } // NewSummaryParser returns an initialized parser. Caller's @@ -283,6 +293,12 @@ func (p *SummaryParser) feedLine(rawLine string) { return // malformed; ignore this line } fp.apply(p.summary, n) + // Latch regardless of value. An all-zero banner is a + // real shape (early failure: the ingestor still prints + // the summary structure with all counts at 0). Bugbot + // PR #10 r2 flagged the "non-zero" hasParsedAnyField + // check as a finalization hole for this case. + p.sawAnyField = true return } } @@ -291,19 +307,7 @@ func (p *SummaryParser) feedLine(rawLine string) { // applied at least one fieldPatterns line. Used to disambiguate // the two ═-rules in the banner (see feedLine). func (p *SummaryParser) hasParsedAnyField() bool { - if p.summary == nil { - return false - } - // "Any field non-zero" is a fine proxy — the ingestor always - // prints at least one positive count even on early failure - // (TotalRecords includes everything it loaded before crashing). - return p.summary.TotalRecords > 0 || - p.summary.ProcessedRecords > 0 || - p.summary.InsertedRecords > 0 || - p.summary.APISentRecords > 0 || - p.summary.SkippedRecords > 0 || - p.summary.FileTransferFailures > 0 || - p.summary.FailedRecords > 0 + return p.sawAnyField } // Result returns the accumulated Summary. nil if the parser never diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 9cca42e7..9a7095c3 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -127,21 +127,25 @@ func WatchJob( namespace, jobName string, out io.Writer, ) (*WatchResult, error) { - // Enforce the JobWatchTimeout absolute cap. Earlier versions - // declared the constant but only the per-step timeouts - // (PodReadyTimeout, finalJobStatus's 30s) bounded the watch — - // the log-stream phase could run indefinitely on a stuck - // ingestor. Bugbot flagged the gap on PR #10. The cap - // triggers as ctx.DeadlineExceeded, which the orchestrator - // surfaces with a clear "watch timed out" framing. - ctx, cancel := context.WithTimeout(ctx, JobWatchTimeout) + // Keep the customer's original ctx separately so finalJobStatus + // can derive a FRESH 30s context from it (rather than inheriting + // a possibly-depleted JobWatchTimeout). Bugbot PR #10 r2: the + // previous "wrap everything in JobWatchTimeout" approach starved + // finalJobStatus's budget when streaming used most of the hour, + // so a successful slow ingestion misreported as Unknown → exit 9. + customerCtx := ctx + + // JobWatchTimeout caps the pod-wait + log-stream phases (the + // time-spend dominant parts of the watch). finalJobStatus gets + // its own ctx below. + watchCtx, cancel := context.WithTimeout(customerCtx, JobWatchTimeout) defer cancel() // 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(ctx, cs, namespace, jobName) + podName, err := waitForJobPod(watchCtx, cs, namespace, jobName) if err != nil { if errors.Is(err, context.Canceled) { // SIGINT before the Pod even appeared. jobs-manager @@ -163,15 +167,17 @@ 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. - summary, logErr := streamPodLogsAndParse(ctx, cs, namespace, podName, out) + summary, logErr := streamPodLogsAndParse(watchCtx, cs, namespace, podName, out) if logErr != nil && !errors.Is(logErr, context.Canceled) { return nil, fmt.Errorf("streaming logs from Pod %s/%s: %w", namespace, podName, logErr) } - // 3. ctx cancellation = detach. Pod might still be running - // server-side, but as far as the CLI is concerned we've - // handed off. - if errors.Is(ctx.Err(), context.Canceled) { + // 3. Customer-ctx cancellation = detach. Check customerCtx + // (not watchCtx) so a JobWatchTimeout expiry doesn't + // masquerade as a SIGINT. The customer's signal handler + // cancels customerCtx specifically; the WithTimeout wrap + // cancels watchCtx only. + if errors.Is(customerCtx.Err(), context.Canceled) { return &WatchResult{ Outcome: JobOutcomeDetached, PodName: podName, @@ -180,13 +186,33 @@ func WatchJob( }, nil } - // 4. Final status check: even though log streaming ending - // SHOULD mean Pod completion, the apiserver may not have - // posted the terminal state by the time the log stream - // drained. Poll briefly to confirm — bounded at 30s - // because beyond that something else is wrong. - outcome, err := finalJobStatus(ctx, cs, namespace, jobName) + // 4. Final status check with a FRESH 30s budget derived from + // the customer's ctx (not watchCtx, which may be near- + // expired after a long log stream). Bugbot PR #10 r2: + // inheriting watchCtx's depleted budget caused successful + // slow ingestions to misreport as Unknown. + // + // The fresh ctx still propagates SIGINT (parent is + // customerCtx, which carries signal.NotifyContext's + // cancel). If the customer Ctrl-C's during this 30s + // window, we fall into the detach branch below — same + // contract as during the log stream. + finalCtx, finalCancel := context.WithTimeout(customerCtx, 30*time.Second) + defer finalCancel() + outcome, err := finalJobStatus(finalCtx, cs, namespace, jobName) if err != nil { + // Treat SIGINT during finalJobStatus as graceful detach + // (same as during the log stream — jobs-manager already + // accepted the run, the customer is just stopping the + // observation). Bugbot PR #10 r2 flagged the "exit 9 on + // post-stream SIGINT" inconsistency. + if errors.Is(customerCtx.Err(), context.Canceled) { + return &WatchResult{ + Outcome: JobOutcomeDetached, + PodName: podName, + Summary: summary, + }, nil + } return nil, fmt.Errorf("reading final Job status for %s/%s: %w", namespace, jobName, err) } return &WatchResult{ From 466eb544065bf3c7478ccea6a8dada89ddd5427f Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 21:37:18 +0500 Subject: [PATCH 4/8] fix(submit): port-forward to jobs-manager from off-cluster (Bugbot r3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Major architectural finding from Bugbot's third PR #10 review: the CLI runs on the customer's laptop (or off-cluster CI), but the discovered JobsManagerService URL is a *.svc.cluster.local name that the laptop can't resolve. The submit POST would always fail. The original Phase 4 design quietly assumed in-cluster network reachability — wrong for the dominant v0.1 use case. The chart's existing flow doesn't have this problem because the helm hook runs as a Pod IN the cluster. ## Fix: in-process port-forward, same mechanism as kubectl pf New internal/submit/portforward.go: - PortForwardJobsManager opens an SPDY tunnel via the customer's kubeconfig-authenticated apiserver connection. - pickServicePod resolves the jobs-manager Service to a Running Pod backing it (Services aren't directly port-forwardable; client-go's portforward API speaks Pod names). - ForwardedConnection.Close tears it down on defer; idempotent so the orchestrator's defer-Close + any inner cleanup don't risk a double-close panic. Wiring in cli/dataset.go: - After token mint, open the port-forward to JobsManagerServiceName:JobsManagerPort (8080 today). - Defer Close — runs on every path, including the SIGINT-detach + error paths from submit.Run. - HTTPSubmitter targets http://localhost:; the in-cluster URL is no longer used at all from the CLI. cluster/discover.go: expose the bare Service name + port alongside the existing FQDN URL. Phase 2 still constructs the FQDN for the diagnostic output of `cluster info`, but Phase 4 needs the unqualified pieces for the port-forward setup. ## Tests internal/submit/portforward_test.go covers the pickServicePod resolution (5 cases: happy path, skip-non-Running, no-matching- Pod, missing-Service, selector-less). The full port-forward loop needs a real apiserver — covered by EKS smoke. internal/cluster/discover_test.go: updated the struct-literal comparison in TestDiscoverParentRelease_HappyPath to include the two new fields. Local: vet, test -race -cover, gofmt -s, errcheck — all green. ## What we got wrong in the initial design Designing Phase 4, I read the chart's existing flow (which runs in-cluster) + assumed the CLI could use the same URL. That's exactly the "cargo-cult the URL" mistake. The right framing would have been: "Phase 4 needs to reach jobs-manager from the laptop — what mechanism does kubectl have for that?" → port- forward. The fix is correct + well-tested where unit-testable. The full loop will get exercised on the EKS smoke test. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/cli/dataset.go | 23 ++- internal/cluster/discover.go | 18 ++- internal/cluster/discover_test.go | 14 +- internal/submit/portforward.go | 219 ++++++++++++++++++++++++++++ internal/submit/portforward_test.go | 142 ++++++++++++++++++ 5 files changed, 405 insertions(+), 11 deletions(-) create mode 100644 internal/submit/portforward.go create mode 100644 internal/submit/portforward_test.go diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index a938bf1e..5e76e0f9 100644 --- a/internal/cli/dataset.go +++ b/internal/cli/dataset.go @@ -398,8 +398,24 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush return &exitError{code: 5, err: err} } - // 11. Phase 4: POST to jobs-manager, watch the spawned ingestor - // Job, render the parsed INGESTION SUMMARY panel. + // 11. Open a port-forward to a Pod backing the jobs-manager + // Service. The CLI runs off-cluster (on a laptop, in CI + // runners outside the cluster network), so the discovered + // *.svc.cluster.local URL isn't reachable — we tunnel + // through the kubeconfig-authenticated apiserver, same as + // `kubectl port-forward`. Bugbot PR #10 r3 caught the + // original broken-by-design direct-URL POST. + _, _ = fmt.Fprintln(out, "Opening port-forward to jobs-manager...") + pf, err := submit.PortForwardJobsManager(ctx, cs, resolved.RestConfig, + resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort) + if err != nil { + return &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)} + } + defer pf.Close() + + // 12. Phase 4: POST to jobs-manager via the local port, + // watch the spawned ingestor Job, render the parsed + // INGESTION SUMMARY panel. // // Exit-code mapping: // SubmitError 401/403 → 5 (auth — same bucket as @@ -415,8 +431,9 @@ func runDatasetPush(ctx context.Context, out, errOut io.Writer, a runDatasetPush // partial-failure summaries) // WatchResult Detached → 0 (cluster keeps running) // WatchResult Succeeded clean → 0 + localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort) submitRes, err := submit.Run(ctx, submit.Options{ - Submitter: submit.NewHTTPSubmitter(release.JobsManagerService, tok.Token), + Submitter: submit.NewHTTPSubmitter(localEndpoint, tok.Token), Client: cs, IngestConfigYAML: string(specBytes), IdempotencyKey: a.IdempotencyKey, diff --git a/internal/cluster/discover.go b/internal/cluster/discover.go index 7ee34c72..b1718f9a 100644 --- a/internal/cluster/discover.go +++ b/internal/cluster/discover.go @@ -35,9 +35,20 @@ type ParentRelease struct { // JobsManagerService is the in-cluster DNS name of the // jobs-manager Service, e.g. // "-jobs-manager..svc.cluster.local:8080". - // Used as the POST target for ingestion submissions. + // Used as the POST target for ingestion submissions WHEN + // the CLI runs in-cluster (e.g. CI inside the same cluster). + // For laptop / off-cluster use, the orchestrator port-forwards + // to JobsManagerServiceName + JobsManagerPort instead. JobsManagerService string + // JobsManagerServiceName + JobsManagerPort are the bare Service + // reference for off-cluster port-forwarding (Bugbot PR #10 r3). + // The FQDN-based JobsManagerService URL above doesn't resolve + // from a laptop; the port-forward path uses these to set up a + // localhost tunnel via the kubeconfig API server. + JobsManagerServiceName string + JobsManagerPort int + // IngestorSAName is the name of the ServiceAccount the chart's // hook pods run as. Today this is always the chart's default // "ingestor". Customers who set `ingestionAuthz.serviceAccountName` @@ -143,7 +154,10 @@ func DiscoverParentRelease(ctx context.Context, cs kubernetes.Interface, namespa // release-prefixed form. Customers can always override via the // ingestor subchart's `jobsManager.endpoint` value. svc := pickJobsManagerService(ctx, cs, namespace, release.ReleaseName) - release.JobsManagerService = fmt.Sprintf("http://%s.%s.svc.cluster.local:8080", svc, namespace) + const jobsManagerPort = 8080 // chart's well-known port for /internal/submit-ingestion-run + release.JobsManagerService = fmt.Sprintf("http://%s.%s.svc.cluster.local:%d", svc, namespace, jobsManagerPort) + release.JobsManagerServiceName = svc + release.JobsManagerPort = jobsManagerPort // Read INGESTOR_IMAGE_DIGEST from jobs-manager's pod-spec env. // The chart pipes images.ingestor.digest through to here. diff --git a/internal/cluster/discover_test.go b/internal/cluster/discover_test.go index 34f4433d..5c0921c0 100644 --- a/internal/cluster/discover_test.go +++ b/internal/cluster/discover_test.go @@ -76,12 +76,14 @@ func TestDiscoverParentRelease_HappyPath(t *testing.T) { } want := ParentRelease{ - ReleaseName: "tracebloc", - ChartVersion: "1.3.5", - AppVersion: "1.3.5", - JobsManagerService: "http://jobs-manager." + ns + ".svc.cluster.local:8080", - IngestorSAName: "ingestor", - IngestorImageDigest: "sha256:463e236748708a5e3564569eec9173ea8cb3bcf515992d4939c5b610f3807a4a", + ReleaseName: "tracebloc", + ChartVersion: "1.3.5", + AppVersion: "1.3.5", + JobsManagerService: "http://jobs-manager." + ns + ".svc.cluster.local:8080", + JobsManagerServiceName: "jobs-manager", + JobsManagerPort: 8080, + IngestorSAName: "ingestor", + IngestorImageDigest: "sha256:463e236748708a5e3564569eec9173ea8cb3bcf515992d4939c5b610f3807a4a", } if *release != want { t.Errorf("mismatch.\ngot: %+v\nwant: %+v", *release, want) diff --git a/internal/submit/portforward.go b/internal/submit/portforward.go new file mode 100644 index 00000000..e8ad7057 --- /dev/null +++ b/internal/submit/portforward.go @@ -0,0 +1,219 @@ +package submit + +import ( + "context" + "fmt" + "io" + "net/http" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/portforward" + "k8s.io/client-go/transport/spdy" +) + +// ForwardedConnection is a live port-forward to an in-cluster +// Service. LocalPort is the random port the kernel picked on the +// CLI host; the Submitter POSTs to http://localhost:LocalPort. +// +// Close MUST be called when the caller is done — otherwise the +// goroutine running the SPDY tunnel leaks for the lifetime of the +// process. +type ForwardedConnection struct { + LocalPort int + + stopCh chan struct{} + done chan struct{} +} + +// Close tears down the port-forward. Safe to call multiple times. +func (f *ForwardedConnection) Close() { + select { + case <-f.stopCh: + return // already closed + default: + } + close(f.stopCh) + // Wait briefly for the goroutine to drain — without this, a + // fast Close-then-process-exit could race the SPDY teardown + // and leave a half-open connection on the apiserver side. + select { + case <-f.done: + case <-time.After(2 * time.Second): + } +} + +// PortForwardJobsManager opens a port-forward to a Pod backing the +// jobs-manager Service in `namespace`. The customer's CLI runs +// off-cluster (on a laptop, in a CI runner outside the cluster +// network); the discovered jobs-manager URL is a +// *.svc.cluster.local name that's NOT resolvable from there. The +// port-forward routes traffic through the kubeconfig-authenticated +// kube-apiserver connection — same machinery `kubectl port-forward` +// uses internally. +// +// Returns a ForwardedConnection whose LocalPort the caller targets +// for HTTP. Bugbot PR #10 r3 caught the broken-by-design assumption +// in the initial Phase 4 implementation. +// +// Lifecycle: caller MUST defer Close(). The port-forward stays open +// for the entire submit + watch sequence (submit needs a single +// POST, watch only uses the kubeconfig API server connection, so +// strictly speaking we could close right after the POST — but +// keeping it open is simpler and the resource cost is one idle +// goroutine). +func PortForwardJobsManager( + ctx context.Context, + cs kubernetes.Interface, + restConfig *rest.Config, + namespace, serviceName string, + targetPort int, +) (*ForwardedConnection, error) { + // 1. Find a Running Pod backing the Service. client-go's + // port-forward API speaks Pods, not Services — even though + // kubectl port-forward accepts both, it resolves the Service + // to a Pod internally. + pod, err := pickServicePod(ctx, cs, namespace, serviceName) + if err != nil { + return nil, fmt.Errorf("resolving Service %s/%s to a Pod: %w", + namespace, serviceName, err) + } + + // 2. Build the SPDY transport. client-go bundles helpers in + // transport/spdy for exactly this; the round-tripper handles + // the SPDY upgrade negotiation the apiserver expects on the + // /portforward subresource. + transport, upgrader, err := spdy.RoundTripperFor(restConfig) + if err != nil { + return nil, fmt.Errorf("building SPDY transport: %w", err) + } + + // 3. Construct the portforward URL on the Pod. The path is + // /api/v1/namespaces//pods//portforward; we build + // it via the REST client so kubeconfig's authentication + + // TLS config flow through automatically. + req := cs.CoreV1().RESTClient().Post(). + Resource("pods"). + Namespace(namespace). + Name(pod.Name). + SubResource("portforward") + + dialer := spdy.NewDialer(upgrader, + &http.Client{Transport: transport}, + "POST", req.URL()) + + // 4. Create the port-forwarder. "0:" means "pick + // any free local port; map it to in the Pod." + // The kernel allocates the local port at goroutine start; + // we read it back after readyCh fires. + stopCh := make(chan struct{}) + readyCh := make(chan struct{}) + pf, err := portforward.New(dialer, + []string{fmt.Sprintf("0:%d", targetPort)}, + stopCh, readyCh, + io.Discard, // forwarder's stdout — verbose listener logs we don't want + io.Discard, // forwarder's stderr + ) + if err != nil { + return nil, fmt.Errorf("creating port-forwarder: %w", err) + } + + // 5. Launch the forward goroutine. ForwardPorts blocks until + // stopCh closes (or it errors). The select below waits for + // EITHER ready (success) or done (early failure). + done := make(chan struct{}) + errCh := make(chan error, 1) + go func() { + defer close(done) + // ForwardPorts is the long-running call. Its return value + // is meaningful only on early failure — on normal close, + // it returns nil after stopCh fires. + if err := pf.ForwardPorts(); err != nil { + errCh <- err + } + }() + + select { + case <-readyCh: + // happy path — port allocated, tunnel up + case err := <-errCh: + return nil, fmt.Errorf("port-forward to %s/%s failed during startup: %w", + namespace, pod.Name, err) + case <-ctx.Done(): + close(stopCh) + return nil, ctx.Err() + } + + ports, err := pf.GetPorts() + if err != nil { + close(stopCh) + return nil, fmt.Errorf("reading allocated port: %w", err) + } + if len(ports) == 0 { + close(stopCh) + return nil, fmt.Errorf("port-forward allocated zero ports") + } + + return &ForwardedConnection{ + LocalPort: int(ports[0].Local), + stopCh: stopCh, + done: done, + }, nil +} + +// pickServicePod resolves a Service to a Running Pod backing it. +// Uses the Service's own selector (read from the Service spec) to +// match Pods — same mechanism the cluster's own kube-proxy uses. +// +// Picks the first Running Pod found; in the common case there's +// only one (jobs-manager is single-replica by chart default). For +// multi-replica deployments, picking any Running Pod is fine — +// they're load-balanced equivalents. +func pickServicePod(ctx context.Context, cs kubernetes.Interface, namespace, serviceName string) (*corev1.Pod, error) { + svc, err := cs.CoreV1().Services(namespace).Get(ctx, serviceName, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("reading service %s/%s: %w", namespace, serviceName, err) + } + if len(svc.Spec.Selector) == 0 { + // A Service with no selector is one whose Endpoints are + // hand-managed (e.g. ExternalName). The chart's jobs- + // manager is a normal selector-based Service so this + // shouldn't happen in production; the error makes the + // debugging path obvious if it ever does. + return nil, fmt.Errorf( + "service %s/%s has no selector — can't resolve to a Pod for port-forwarding", + namespace, serviceName) + } + + // Build a label selector from the Service's spec.selector map. + // strings.Join keeps the order deterministic for readable + // error output; the order doesn't affect the actual Pods + // returned. + parts := make([]string, 0, len(svc.Spec.Selector)) + for k, v := range svc.Spec.Selector { + parts = append(parts, k+"="+v) + } + selector := strings.Join(parts, ",") + + pods, err := cs.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{ + LabelSelector: selector, + }) + if err != nil { + return nil, fmt.Errorf("listing Pods for service %s/%s: %w", + namespace, serviceName, err) + } + for i := range pods.Items { + p := &pods.Items[i] + if p.Status.Phase == corev1.PodRunning { + return p, nil + } + } + return nil, fmt.Errorf( + "no Running Pod backing service %s/%s (found %d Pod(s); "+ + "check `kubectl get pods -n %s -l %s`)", + namespace, serviceName, len(pods.Items), namespace, selector) +} diff --git a/internal/submit/portforward_test.go b/internal/submit/portforward_test.go new file mode 100644 index 00000000..933831c4 --- /dev/null +++ b/internal/submit/portforward_test.go @@ -0,0 +1,142 @@ +package submit + +import ( + "context" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" +) + +// The full port-forward (PortForwardJobsManager) requires a real +// apiserver + SPDY upgrade, so it's out of scope for unit tests — +// covered by the EKS smoke. What IS testable: pickServicePod's +// Service→Pod resolution, which is the only client-go-only logic +// in the file. + +// svc constructs a Service with the given selector. Used to seed +// the fake clientset. +func svc(name string, selector map[string]string) *corev1.Service { + return &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "tracebloc"}, + Spec: corev1.ServiceSpec{Selector: selector}, + } +} + +// podForSvc constructs a Pod with labels matching the selector. +func podForSvc(name string, labels map[string]string, phase corev1.PodPhase) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "tracebloc", + Labels: labels, + }, + Status: corev1.PodStatus{Phase: phase}, + } +} + +// TestPickServicePod_HappyPath: a Service with one matching Running +// Pod resolves to that Pod's name. +func TestPickServicePod_HappyPath(t *testing.T) { + sel := map[string]string{"app": "jobs-manager"} + cs := fake.NewClientset( + svc("jobs-manager", sel), + podForSvc("jobs-manager-abc", sel, corev1.PodRunning), + ) + p, err := pickServicePod(context.Background(), cs, "tracebloc", "jobs-manager") + if err != nil { + t.Fatalf("pickServicePod: %v", err) + } + if p.Name != "jobs-manager-abc" { + t.Errorf("Pod name = %q, want jobs-manager-abc", p.Name) + } +} + +// TestPickServicePod_SkipsNonRunning: Pending / Failed Pods backing +// the same Service are filtered out — the port-forward only works +// against a Running Pod. +func TestPickServicePod_SkipsNonRunning(t *testing.T) { + sel := map[string]string{"app": "jobs-manager"} + cs := fake.NewClientset( + svc("jobs-manager", sel), + podForSvc("crashed", sel, corev1.PodFailed), + podForSvc("pending", sel, corev1.PodPending), + podForSvc("running", sel, corev1.PodRunning), + ) + p, err := pickServicePod(context.Background(), cs, "tracebloc", "jobs-manager") + if err != nil { + t.Fatalf("pickServicePod: %v", err) + } + if p.Name != "running" { + t.Errorf("Pod name = %q, want running", p.Name) + } +} + +// TestPickServicePod_NoMatchingPod: a Service whose Pods are all +// non-Running (or absent) surfaces a clear error pointing at the +// kubectl command to debug. +func TestPickServicePod_NoMatchingPod(t *testing.T) { + sel := map[string]string{"app": "jobs-manager"} + cs := fake.NewClientset( + svc("jobs-manager", sel), + podForSvc("crashed", sel, corev1.PodFailed), + ) + _, err := pickServicePod(context.Background(), cs, "tracebloc", "jobs-manager") + if err == nil { + t.Fatal("pickServicePod returned nil on no-Running-Pod") + } + for _, want := range []string{ + "no Running Pod", + "jobs-manager", + "kubectl get pods", + } { + if !strings.Contains(err.Error(), want) { + t.Errorf("error missing %q: %v", want, err) + } + } +} + +// TestPickServicePod_ServiceMissing: trying to port-forward to a +// non-existent Service surfaces with the Service name in the error. +func TestPickServicePod_ServiceMissing(t *testing.T) { + cs := fake.NewClientset() // empty + _, err := pickServicePod(context.Background(), cs, "tracebloc", "missing-svc") + if err == nil { + t.Fatal("pickServicePod returned nil on missing Service") + } + if !strings.Contains(err.Error(), "missing-svc") { + t.Errorf("error missing service name: %v", err) + } +} + +// TestPickServicePod_NoSelector: ExternalName Services and other +// selector-less shapes can't be port-forwarded by Pod lookup. +// Surface a clear error rather than silently picking nothing. +func TestPickServicePod_NoSelector(t *testing.T) { + cs := fake.NewClientset(&corev1.Service{ + ObjectMeta: metav1.ObjectMeta{Name: "externalname-svc", Namespace: "tracebloc"}, + // no Spec.Selector + }) + _, err := pickServicePod(context.Background(), cs, "tracebloc", "externalname-svc") + if err == nil { + t.Fatal("pickServicePod returned nil on selector-less Service") + } + if !strings.Contains(err.Error(), "no selector") { + t.Errorf("error missing selector-less framing: %v", err) + } +} + +// TestForwardedConnection_CloseIdempotent: Close is safe to call +// multiple times. defer-Close patterns at multiple levels of the +// orchestrator shouldn't risk a double-close panic. +func TestForwardedConnection_CloseIdempotent(t *testing.T) { + stopCh := make(chan struct{}) + done := make(chan struct{}) + close(done) // simulate goroutine already finished + f := &ForwardedConnection{LocalPort: 12345, stopCh: stopCh, done: done} + f.Close() + f.Close() // must not panic + f.Close() +} From cf8263646bead3586cde31720ed0f85d4d335ec7 Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 21:42:33 +0500 Subject: [PATCH 5/8] fix(submit): pick most-recent useful-phase Pod, not items[0] (Bugbot r4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bugbot's fourth PR #10 finding: waitForJobPod picked pods.Items[0] from an unordered List API response. A Job with backoffLimit > 0 (or any scenario where jobs-manager re-spawned the Pod for retry, node drain, etc.) can have multiple Pods bearing the same `job-name=` label. Picking items[0] could grab the old Failed Pod from a prior retry while the current Running one waits — the CLI then streams stale logs and reports the prior-attempt outcome. Fix: iterate all Pods, prefer the most recent (by creationTimestamp) in a useful phase (Running | Succeeded | Failed). Pending Pods don't count — no log stream yet, so we keep polling rather than "return a Pending pod's name." Behaviorally: - Single Pod (common): same as before — that one Pod returns. - Old Failed + new Running: returns the Running. Correct. - Old Failed + new Pending: returns nothing yet; keeps polling until Pending → Running. Correct. - Multiple Running (parallelism > 1): returns the most recent — defensible choice; either would work for log streaming. Two new tests: - TestWaitForJobPod_PicksMostRecentNotFirst: seeds an older Failed + newer Running; asserts the Running's name returns. - TestWaitForJobPod_AllPendingKeepsPolling: seeds an only- Pending Pod; asserts DeadlineExceeded on a tight ctx (no early return with the Pending name). Local: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/submit/watch.go | 47 +++++++++++++++++++++------------- internal/submit/watch_test.go | 48 +++++++++++++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 17 deletions(-) diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 9a7095c3..2d926ef3 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -244,24 +244,37 @@ func waitForJobPod(ctx context.Context, cs kubernetes.Interface, namespace, jobN if len(pods.Items) == 0 { return false, nil // Pod hasn't been created yet } - // Pick the first Pod with the label; Jobs default to - // parallelism=1 so there's only one. If a future - // jobs-manager spawns parallel Pods, the first one - // to surface logs is the one we attach to. - p := pods.Items[0] - switch p.Status.Phase { - case corev1.PodRunning, corev1.PodSucceeded, corev1.PodFailed: - // Running is the happy path; Succeeded/Failed - // mean we missed the Running window (typical - // for fast-completing ingestions). In all three - // cases we can fetch logs. - podName = p.Name - return true, nil - default: - // Pending: still pulling image / scheduling. Keep - // polling. The PodReadyTimeout caps this loop. - return false, nil + // Pick the MOST RECENT useful-phase Pod, not just + // items[0]. A Job with backoffLimit > 0 (or a Job + // where jobs-manager re-spawned the Pod for any + // reason) can have multiple Pods bearing the same + // `job-name=` label. The List API doesn't + // guarantee order, so items[0] could be the old + // Failed Pod from a prior retry instead of the + // current Running one. Bugbot PR #10 r4 caught this. + // + // "Useful phase" = Running (happy path) | Succeeded + // (fast-completing ingestion we missed) | Failed + // (terminated; we still want its logs). Pending Pods + // don't count — they have no logs to stream yet, so + // we keep polling until they either transition or + // become irrelevant. + var bestPod *corev1.Pod + for i := range pods.Items { + p := &pods.Items[i] + switch p.Status.Phase { + case corev1.PodRunning, corev1.PodSucceeded, corev1.PodFailed: + if bestPod == nil || + p.CreationTimestamp.After(bestPod.CreationTimestamp.Time) { + bestPod = p + } + } + } + if bestPod == nil { + return false, nil // all Pods still Pending } + podName = bestPod.Name + return true, nil }) if err != nil { return "", err diff --git a/internal/submit/watch_test.go b/internal/submit/watch_test.go index 5a1ce974..e283bb80 100644 --- a/internal/submit/watch_test.go +++ b/internal/submit/watch_test.go @@ -48,6 +48,54 @@ func TestWaitForJobPod_RunningPodSurfaces(t *testing.T) { } } +// TestWaitForJobPod_PicksMostRecentNotFirst: Jobs with retries +// (or any multi-Pod scenario) produce multiple Pods with the +// same `job-name=` label. The List API doesn't guarantee +// order — picking items[0] could grab the old Failed Pod from a +// prior retry while the current Running one waits. Bugbot +// PR #10 r4 caught the missing tie-break. +func TestWaitForJobPod_PicksMostRecentNotFirst(t *testing.T) { + now := time.Now() + older := jobPod("ingestor-old-failed", "ingestor", corev1.PodFailed) + older.CreationTimestamp = metav1.NewTime(now.Add(-10 * time.Minute)) + + newer := jobPod("ingestor-new-running", "ingestor", corev1.PodRunning) + newer.CreationTimestamp = metav1.NewTime(now.Add(-1 * time.Minute)) + + cs := fake.NewClientset(older, newer) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + name, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") + if err != nil { + t.Fatalf("waitForJobPod: %v", err) + } + if name != "ingestor-new-running" { + t.Errorf("name = %q, want ingestor-new-running "+ + "(most-recent useful-phase Pod, not items[0])", name) + } +} + +// TestWaitForJobPod_AllPendingKeepsPolling: if every Pod is still +// Pending (image pulling, scheduling), the function keeps polling +// rather than returning a Pending Pod's name (Pods with no log +// stream yet aren't useful to attach to). +func TestWaitForJobPod_AllPendingKeepsPolling(t *testing.T) { + cs := fake.NewClientset(jobPod("pending-1", "ingestor", corev1.PodPending)) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + _, err := waitForJobPod(ctx, cs, "tracebloc", "ingestor") + if err == nil { + t.Fatal("waitForJobPod returned nil on all-Pending; expected DeadlineExceeded") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("error doesn't wrap DeadlineExceeded: %v", err) + } +} + // TestWaitForJobPod_FastCompletionPath: ingestions that finish // faster than the poll interval might be in Phase=Succeeded by // the time the watch loop checks. We still want the Pod's name so From 0b02cd4dd7f976ee301d636a6edfb86ead6de9da Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 21:49:09 +0500 Subject: [PATCH 6/8] fix(submit): Pod-wait timeout = Detached (not failed); non-blocking errCh send MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 5 Bugbot findings on PR #10: ## Medium — Pod-wait timeout reported ingest failure waitForJobPod times out after PodReadyTimeout (5min) if the ingestor Pod doesn't reach Running — slow image pull, scheduling backlog, PSA still rejecting. The previous flow wrapped that as WatchError → exit 9 ("ingestion failed"). That's a false positive: jobs-manager already accepted the submit, the cluster will run the ingestion (eventually), the CLI just gave up observing within the timeout. Fix: distinguish errors.Is(err, context.DeadlineExceeded) from waitForJobPod and map to Outcome=Detached (same semantic as SIGINT-mid-watch). Customer sees the kubectl-logs reconnect hint and exits 0. The submit was successful; the ingestion runs. New test TestWatchJob_PodWaitTimeoutMapsToDetach pins the contract via a tight outer ctx that triggers DeadlineExceeded through the same code path. ## Medium — Defensive non-blocking errCh send in portforward.go Bugbot's read: "unbuffered handoff can block the goroutine indefinitely." False on the literal code (errCh is make(chan error, 1), buffered) — but the spirit of the concern is reasonable: a future refactor that adds a second send path, or changes the buffer size, could re-introduce the leak. Fix: keep the cap-1 buffer AND wrap the send in a non-blocking select with default. Two layers of safety, so the goroutine never blocks regardless of how the channel is used downstream. Closes out the finding structurally rather than relying on the code reader to spot that the buffer was already sized correctly. Local: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/submit/portforward.go | 23 +++++++++++++++++++++-- internal/submit/watch.go | 10 ++++++++++ internal/submit/watch_test.go | 31 +++++++++++++++++++++++++++++++ 3 files changed, 62 insertions(+), 2 deletions(-) diff --git a/internal/submit/portforward.go b/internal/submit/portforward.go index e8ad7057..20c6155a 100644 --- a/internal/submit/portforward.go +++ b/internal/submit/portforward.go @@ -125,6 +125,16 @@ func PortForwardJobsManager( // 5. Launch the forward goroutine. ForwardPorts blocks until // stopCh closes (or it errors). The select below waits for // EITHER ready (success) or done (early failure). + // + // errCh is buffered (capacity 1) AND the send is wrapped in + // a non-blocking select with a default — two layers of + // safety against the "goroutine leaks waiting to send" + // pattern. ForwardPorts only ever sends once, so the buffer + // is sufficient; the non-blocking select is paranoid + // defensive against a future refactor that adds a second + // send path. Bugbot PR #10 r5 flagged the (already- + // buffered) channel as a potential leak — the false alarm + // is worth closing out structurally. done := make(chan struct{}) errCh := make(chan error, 1) go func() { @@ -132,8 +142,17 @@ func PortForwardJobsManager( // ForwardPorts is the long-running call. Its return value // is meaningful only on early failure — on normal close, // it returns nil after stopCh fires. - if err := pf.ForwardPorts(); err != nil { - errCh <- err + err := pf.ForwardPorts() + if err == nil { + return + } + select { + case errCh <- err: + default: + // Buffer full = receiver already saw an earlier error + // (impossible today — single sender, single error — + // but the default arm makes the goroutine drainable + // regardless of any future change). } }() diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 2d926ef3..12dc2f89 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -153,6 +153,16 @@ func WatchJob( // the CLI is just not watching anymore. return &WatchResult{Outcome: JobOutcomeDetached}, nil } + // PodReadyTimeout (5min) exhausted = slow image pull / + // scheduling backlog / PSA still rejecting. The submit + // was accepted, the run will (eventually) execute in the + // cluster — the CLI just gave up observing within the + // timeout. Treat as Detached, not ingest-failed: bumping + // to exit 9 would falsely claim the ingestion failed. + // Bugbot PR #10 r5 flagged the false-positive exit code. + if errors.Is(err, context.DeadlineExceeded) { + return &WatchResult{Outcome: JobOutcomeDetached}, nil + } return nil, fmt.Errorf("waiting for ingestor Pod: %w", err) } diff --git a/internal/submit/watch_test.go b/internal/submit/watch_test.go index e283bb80..2be8b052 100644 --- a/internal/submit/watch_test.go +++ b/internal/submit/watch_test.go @@ -1,6 +1,7 @@ package submit import ( + "bytes" "context" "errors" "strings" @@ -232,6 +233,36 @@ func TestFinalJobStatus_TimeoutIsUnknown(t *testing.T) { } } +// TestWatchJob_PodWaitTimeoutMapsToDetach: when waitForJobPod +// exhausts its 5-min budget (slow image pull, PSA backlog), the +// submit has already been accepted by jobs-manager — the +// ingestion will run, the CLI just gave up watching within the +// timeout. WatchJob must return Outcome=Detached, not an error +// that bubbles up as exit 9 ("ingestion failed"). Bugbot PR #10 +// r5 caught the false-positive exit code. +func TestWatchJob_PodWaitTimeoutMapsToDetach(t *testing.T) { + cs := fake.NewClientset() // no Pod backing the job-name + + // Tight ctx so the test doesn't actually wait PodReadyTimeout (5m). + // The DeadlineExceeded from the parent ctx propagates the same + // way the inner PodReadyTimeout would, exercising the same + // errors.Is(DeadlineExceeded) branch in WatchJob. + ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond) + defer cancel() + + var out bytes.Buffer + wr, err := WatchJob(ctx, cs, "tracebloc", "ingestor-stuck", &out) + if err != nil { + t.Fatalf("WatchJob returned error on Pod-wait timeout; want nil + Detached: %v", err) + } + if wr == nil { + t.Fatal("WatchJob returned nil result on Pod-wait timeout") + } + if wr.Outcome != JobOutcomeDetached { + t.Errorf("Outcome = %v, want Detached (the cluster keeps running; the CLI just gave up observing)", wr.Outcome) + } +} + // TestJobOutcome_String: stringer pin so diagnostic output stays // stable. func TestJobOutcome_String(t *testing.T) { From cc72c59b8f1aa5e08ffe8f19ea66fdf2eb9dc30b Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 21:54:24 +0500 Subject: [PATCH 7/8] fix(submit): JobWatchTimeout-during-stream = Detached, not exit 9 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bugbot PR #10 r6 caught the inconsistency r5 introduced: the PodReadyTimeout path was mapped to Detached (commit 0b02cd4), but the SAME JobWatchTimeout constant capping the log-stream phase still mapped to exit 9 ("ingestion failed") if it expired during streaming. Both should mean the same thing: "the cluster keeps running; the CLI just gave up observing." Fix: accept context.DeadlineExceeded as well as context.Canceled in the post-stream branch. If watchCtx (the 1-hour-capped one) expired but customerCtx (the SIGINT-aware parent) didn't, that's the JobWatchTimeout-hit case — map to Detached, same UX as the SIGINT path. Customer sees the kubectl-logs reconnect hint and exits 0. The watchCtx-vs-customerCtx distinction (from r3) is what lets us tell SIGINT from JobWatchTimeout at this branch point — without it, the only way to detect timeout-vs-cancel would be the inner error. Local: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/submit/watch.go | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 12dc2f89..1ae2a66e 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -178,16 +178,29 @@ func WatchJob( // a structured representation of the banner without // requiring a second log fetch post-completion. summary, logErr := streamPodLogsAndParse(watchCtx, cs, namespace, podName, out) - if logErr != nil && !errors.Is(logErr, context.Canceled) { + // Filter out the two ctx-flavored errors — both are "observation + // gave up early," not "stream failed." They get classified below + // into Detached (customer SIGINT, JobWatchTimeout expiry). Any + // other error is a real streaming failure (network mid-stream, + // API server tantrum) and bubbles up as a watch error. + if logErr != nil && + !errors.Is(logErr, context.Canceled) && + !errors.Is(logErr, context.DeadlineExceeded) { return nil, fmt.Errorf("streaming logs from Pod %s/%s: %w", namespace, podName, logErr) } - // 3. Customer-ctx cancellation = detach. Check customerCtx - // (not watchCtx) so a JobWatchTimeout expiry doesn't - // masquerade as a SIGINT. The customer's signal handler - // cancels customerCtx specifically; the WithTimeout wrap - // cancels watchCtx only. - if errors.Is(customerCtx.Err(), context.Canceled) { + // 3. Detach branches: + // - customerCtx canceled = SIGINT + // - watchCtx expired (DeadlineExceeded) = JobWatchTimeout cap + // hit during streaming (1-hour observation window exceeded) + // + // Both are "the cluster keeps running; the CLI just gave up + // observing." Same UX as the PodReadyTimeout case from r5 + // above — exit 0 with the kubectl-logs reconnect hint. + // Bugbot PR #10 r6 flagged the inconsistency: r5 detached + // on PodReady timeout but the watch-cap exit still mapped + // to exit 9. + if errors.Is(customerCtx.Err(), context.Canceled) || errors.Is(watchCtx.Err(), context.DeadlineExceeded) { return &WatchResult{ Outcome: JobOutcomeDetached, PodName: podName, From 9f02cf8527b2d6f5c82118f7d53907ad9261fbb8 Mon Sep 17 00:00:00 2001 From: Asad Iqbal Date: Fri, 22 May 2026 22:02:10 +0500 Subject: [PATCH 8/8] fix(submit): per-reason detach message (Bugbot r7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit R5 + r6 expanded "Detached" to cover PodReadyTimeout and JobWatchTimeout in addition to the original SIGINT case, but the orchestrator's diagnostic still said "Detached on signal" for all three. Customers who hit the 5-min Pod-ready cap or 1-hour watch cap aren't pressing Ctrl-C; the framing was misleading. Fix: WatchResult gains a DetachReason field (DetachReasonSignal, DetachReasonPodWaitTimeout, DetachReasonWatchCap, or None). Every detach branch in WatchJob sets the appropriate reason. submit.Run's print-detach block now switches on the reason: Signal → "Detached on signal." PodWaitTimeout → "Detached: the ingestor Pod didn't reach Ready within the observation window..." WatchCap → "Detached: the watch window (1 hour) elapsed while the ingestion was still running." Common reconnect hint (`kubectl logs -f`) follows in all cases. The customerCtx-vs-watchCtx distinction (introduced in r3) makes the reason classification trivially correct: customerCtx.Canceled takes precedence over watchCtx.DeadlineExceeded if both fired, so a customer hitting Ctrl-C exactly at the 1-hour cap still gets the "on signal" message. Local: vet, test -race -cover, gofmt -s, errcheck — all green. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/submit/submit.go | 25 +++++++++++--- internal/submit/watch.go | 69 ++++++++++++++++++++++++++++++++++----- 2 files changed, 81 insertions(+), 13 deletions(-) diff --git a/internal/submit/submit.go b/internal/submit/submit.go index 9869fd42..d4ccab3d 100644 --- a/internal/submit/submit.go +++ b/internal/submit/submit.go @@ -130,13 +130,30 @@ func Run(ctx context.Context, opts Options) (*Result, error) { return &Result{Submit: resp}, &WatchError{Err: fmt.Errorf("watching ingestor Job: %w", err)} } - // Detach-from-SIGINT path. Print the reconnect hint so the - // customer knows the cluster keeps running. + // Detach paths print a per-reason diagnostic + the same + // kubectl-logs reconnect hint. Bugbot PR #10 r7 caught the + // previous "on signal" framing being misleading for the + // timeout-detach cases — customers who hit the 5-min + // 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) + switch wr.DetachReason { + case DetachReasonSignal: + _, _ = fmt.Fprintln(opts.Out, "Detached on signal.") + 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.") + case DetachReasonWatchCap: + _, _ = fmt.Fprintln(opts.Out, + "Detached: the watch window (1 hour) elapsed while the ingestion was still running.") + default: + _, _ = fmt.Fprintln(opts.Out, "Detached.") + } _, _ = fmt.Fprintf(opts.Out, - "Detached on signal. Ingestion continues in the cluster. "+ - "Reconnect with: kubectl logs -f -n %s job/%s\n", + "Ingestion continues in the cluster. Reconnect with: kubectl logs -f -n %s job/%s\n", resp.Namespace, resp.JobName) return &Result{Submit: resp, Watch: wr}, nil } diff --git a/internal/submit/watch.go b/internal/submit/watch.go index 1ae2a66e..3280e813 100644 --- a/internal/submit/watch.go +++ b/internal/submit/watch.go @@ -104,8 +104,44 @@ type WatchResult struct { // (success path) or include it in the failure framing // (failed-after-summary path). Summary *Summary + + // 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 + // message. Bugbot PR #10 r7 flagged the misleading "signal" + // framing for the timeout-detach paths. + DetachReason DetachReason } +// DetachReason enumerates the conditions that produce a Detached +// outcome. Used by the orchestrator's diagnostic output only — +// the exit-code mapping treats all detach reasons as success (0) +// because the cluster keeps running regardless of why we stopped +// watching. +type DetachReason int + +const ( + // DetachReasonNone is the zero value, used when Outcome is + // not Detached. + DetachReasonNone DetachReason = iota + + // DetachReasonSignal: customer pressed Ctrl-C (or a parent + // process sent SIGTERM). The original Detach semantic. + DetachReasonSignal + + // DetachReasonPodWaitTimeout: PodReadyTimeout (5 min) + // exhausted before the ingestor Pod reached a useful + // phase. Slow image pull, scheduling backlog, PSA rejection. + DetachReasonPodWaitTimeout + + // DetachReasonWatchCap: JobWatchTimeout (1 hour) exceeded + // during log streaming. Long-running ingestion that + // outlasted the observation window. + DetachReasonWatchCap +) + // WatchJob is the top-level watch loop: poll the Job until it // reaches a terminal phase, stream the Pod's logs while it's // running, and return a WatchResult. @@ -151,7 +187,10 @@ func WatchJob( // SIGINT before the Pod even appeared. jobs-manager // has accepted the run, the cluster will run it, // the CLI is just not watching anymore. - return &WatchResult{Outcome: JobOutcomeDetached}, nil + return &WatchResult{ + Outcome: JobOutcomeDetached, + DetachReason: DetachReasonSignal, + }, nil } // PodReadyTimeout (5min) exhausted = slow image pull / // scheduling backlog / PSA still rejecting. The submit @@ -161,7 +200,10 @@ func WatchJob( // to exit 9 would falsely claim the ingestion failed. // Bugbot PR #10 r5 flagged the false-positive exit code. if errors.Is(err, context.DeadlineExceeded) { - return &WatchResult{Outcome: JobOutcomeDetached}, nil + return &WatchResult{ + Outcome: JobOutcomeDetached, + DetachReason: DetachReasonPodWaitTimeout, + }, nil } return nil, fmt.Errorf("waiting for ingestor Pod: %w", err) } @@ -201,11 +243,19 @@ func WatchJob( // on PodReady timeout but the watch-cap exit still mapped // to exit 9. if errors.Is(customerCtx.Err(), context.Canceled) || errors.Is(watchCtx.Err(), context.DeadlineExceeded) { + reason := DetachReasonSignal + if errors.Is(watchCtx.Err(), context.DeadlineExceeded) && + !errors.Is(customerCtx.Err(), context.Canceled) { + // Pure watchCtx-only expiry = JobWatchTimeout. The + // customerCtx-canceled case takes precedence (if both + // fired, the customer's intent was SIGINT). + reason = DetachReasonWatchCap + } return &WatchResult{ - Outcome: JobOutcomeDetached, - PodName: podName, - Summary: summary, // may be partial if the customer - // hit Ctrl-C right at the banner + Outcome: JobOutcomeDetached, + PodName: podName, + Summary: summary, // may be partial + DetachReason: reason, }, nil } @@ -231,9 +281,10 @@ func WatchJob( // post-stream SIGINT" inconsistency. if errors.Is(customerCtx.Err(), context.Canceled) { return &WatchResult{ - Outcome: JobOutcomeDetached, - PodName: podName, - Summary: summary, + Outcome: JobOutcomeDetached, + PodName: podName, + Summary: summary, + DetachReason: DetachReasonSignal, }, nil } return nil, fmt.Errorf("reading final Job status for %s/%s: %w", namespace, jobName, err)