diff --git a/internal/cli/dataset.go b/internal/cli/dataset.go index 6034e03c..5e76e0f9 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,102 @@ 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. 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 + // token-mint, shared + // "your SA can't do this" + // diagnostic class) + // SubmitError other 4xx/5xx → 8 (submit failed) + // WatchResult Failed → 9 (ingest failed) + // WatchResult Succeeded + + // summary.HasFailures() → 9 (some rows failed + // even though Job exited 0; + // the ingestor surfaces + // partial-failure summaries) + // WatchResult Detached → 0 (cluster keeps running) + // WatchResult Succeeded clean → 0 + localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort) + submitRes, err := submit.Run(ctx, submit.Options{ + Submitter: submit.NewHTTPSubmitter(localEndpoint, tok.Token), + Client: cs, + IngestConfigYAML: string(specBytes), + IdempotencyKey: a.IdempotencyKey, + ImageDigest: a.ImageDigest, + Detach: a.Detach, + Out: out, + }) + if err != nil { + 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} + } + } + + // 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. 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( + "ingestion Job completed but the summary reports failures — see panel above")} + } + } return nil } 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/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..972d91bc --- /dev/null +++ b/internal/submit/client.go @@ -0,0 +1,194 @@ +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)) + } + 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 +} + +// 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/portforward.go b/internal/submit/portforward.go new file mode 100644 index 00000000..20c6155a --- /dev/null +++ b/internal/submit/portforward.go @@ -0,0 +1,238 @@ +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). + // + // 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() { + 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. + 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). + } + }() + + 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() +} diff --git a/internal/submit/submit.go b/internal/submit/submit.go new file mode 100644 index 00000000..d4ccab3d --- /dev/null +++ b/internal/submit/submit.go @@ -0,0 +1,206 @@ +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 { + // 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 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, + "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 +} + +// 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 new file mode 100644 index 00000000..cf34126d --- /dev/null +++ b/internal/submit/submit_test.go @@ -0,0 +1,231 @@ +package submit + +import ( + "bytes" + "context" + "errors" + "fmt" + "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) + } +} + +// 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) { + 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..23fd3753 --- /dev/null +++ b/internal/submit/summary.go @@ -0,0 +1,388 @@ +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 + + // 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 +// 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) + // 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 + } +} + +// 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 { + return p.sawAnyField +} + +// 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..3280e813 --- /dev/null +++ b/internal/submit/watch.go @@ -0,0 +1,491 @@ +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 + + // 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. +// +// 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) { + // 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(watchCtx, 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, + DetachReason: DetachReasonSignal, + }, 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, + DetachReason: DetachReasonPodWaitTimeout, + }, 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(watchCtx, cs, namespace, podName, out) + // 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. 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) { + 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 + DetachReason: reason, + }, nil + } + + // 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, + DetachReason: DetachReasonSignal, + }, 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 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 + } + 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) { + // 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 +} + +// 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..2be8b052 --- /dev/null +++ b/internal/submit/watch_test.go @@ -0,0 +1,299 @@ +package submit + +import ( + "bytes" + "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_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 +// 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) + } +} + +// 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) { + 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) + } +}