diff --git a/README.md b/README.md index be55566..a1a0556 100644 --- a/README.md +++ b/README.md @@ -92,9 +92,10 @@ the standard `nvidia.com/gpu` resource limit, so scheduling and provisioning rea the same number. Do not set `nodeName` or a provider `nodeSelector` yourself — the placement controller owns those. -> `kubectl logs` works, `-f` and `--tail` included: the manager serves the one kubelet -> route the API server proxies for logs. `--timestamps`/`--previous`/`--since` are -> ignored, and `kubectl exec` still does not work — use the provider's shell for that. +> `kubectl logs` and `kubectl exec` both work on Modal, `-f`/`--tail` and `-it` +> included: the manager serves the two kubelet routes the API server proxies. +> `--timestamps`/`--previous`/`--since` and `-c` are ignored, and a terminal resize is +> not forwarded. On providers that do not support them yet, both answer NotFound. ## Getting started @@ -104,6 +105,8 @@ placement controller owns those. - See [docs/architecture.md](docs/architecture.md) for design details. - See [docs/status.md](docs/status.md) for how instance lifecycle becomes Pod and NodeClaim status, per provider. +- See [docs/kubelet-api.md](docs/kubelet-api.md) for how `kubectl logs` and `kubectl exec` + reach a Pod with no kubelet. - See [docs/metrics.md](docs/metrics.md) for what is instrumented and how to query it. ## License diff --git a/docs/add-a-provider.md b/docs/add-a-provider.md index 2052d2d..1dff040 100644 --- a/docs/add-a-provider.md +++ b/docs/add-a-provider.md @@ -87,7 +87,7 @@ carry weight: `kubectl logs -f` client disconnects, and on a long-polling API that is one open request per abandoned client if it does not land. -See [status.md § Logs](status.md#logs) for what the kubelet side does with the stream, +See [kubelet-api.md](kubelet-api.md) for what the kubelet side does with the stream, including which kubectl flags are ignored and why. ## 2. Add the price/availability catalog diff --git a/docs/architecture.md b/docs/architecture.md index 8e2aabd..49d9adf 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -339,7 +339,7 @@ serves one HTTPS listener in the manager pod and every virtual node advertises i `manager.Runnable`, hence leader-scoped — correct, because only the leader holds the tracked Pods a log request resolves against. A provider opts in by implementing `provider.LogStreamer`; one that does not answers `NotFound`. See -[status.md § Logs](status.md#logs) for the transport, the trust model, and which kubectl +[kubelet-api.md](kubelet-api.md) for the transport, the trust model, and which kubectl flags are honoured. Exec, attach, stats, and port-forward are not implemented — those need an agent inside diff --git a/docs/deploy.md b/docs/deploy.md index d345325..663618f 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -133,7 +133,7 @@ Manager flags worth knowing (edit `config/manager/manager.yaml` `args`): The endpoint needs `POD_IP` (projected via `fieldRef` in `config/manager/manager.yaml`) because virtual nodes advertise the leader's Pod IP, not a Service. Running the manager off-cluster leaves it unset, and logs degrade to unsupported. See -[status.md § Logs](status.md#logs). +[kubelet-api.md](kubelet-api.md). --- diff --git a/docs/kubelet-api.md b/docs/kubelet-api.md new file mode 100644 index 0000000..8c00424 --- /dev/null +++ b/docs/kubelet-api.md @@ -0,0 +1,95 @@ +# Logs and exec + +`kubectl logs` (`-f` included) and `kubectl exec` (`-it` included) both work against a +Nebula Pod. It is worth spelling out how, because none of the usual kubelet machinery is +present. + +- [The transport](#the-transport) +- [The provider seam](#the-provider-seam) +- [What logs honour, and the one heuristic](#what-logs-honour-and-the-one-heuristic) +- [Containers are not addressable](#containers-are-not-addressable) +- [Exec needs no agent in the image](#exec-needs-no-agent-in-the-image) + +--- + +## The transport + +Neither is a control-plane read: the API server proxies both to the kubelet of the node +the Pod is on, dialing the address in the Node's `status.addresses` and the port in +`status.daemonEndpoints`. A virtual node has no kubelet, so the manager serves those +routes itself (`pkg/vnode/kubelet.go`) and every virtual node advertises the manager's +Pod IP and that port. Consequences worth knowing: + +- The endpoint is **leader-scoped and dialed by Pod IP**, not through a Service. The + tracked Pods live in one process's memory, so a Service balancing across replicas + would send requests to a replica that answers `NotFound`. +- It serves TLS with a self-signed, in-memory certificate — what the API server + expects of a kubelet, which does not verify it unless + `--kubelet-certificate-authority` is set. Client certificates are **not** verified + by default, because which CA signs the API server's kubelet client cert is not + portable across distributions. Anything that can reach the port can therefore read the + logs of, and **run commands in**, any Pod on these virtual nodes, with no RBAC check: + keep it closed with a NetworkPolicy, or pass `--kubelet-client-ca` to require mTLS. +- No POD_IP (running the manager off-cluster) means no endpoint. Logs and exec degrade + to unsupported; nothing else is affected. + +## The provider seam + +Both are optional: a provider opts in by implementing `provider.LogStreamer` and +`provider.Executor`, and one that does not answers `NotFound` rather than carrying a +stub. Modal implements both; AWS implements neither. Each seam is deliberately minimal — +logs are one stream from the instance's first byte, stdout and stderr merged; exec only +STARTS the command and hands back its streams — so every kubectl option is honoured +once, for all providers, in `pkg/vnode/logs.go` and `pkg/vnode/exec.go`. + +## What logs honour, and the one heuristic + +A provider stream has no EOF while the instance lives and no marker for "you have now +caught up", which the real kubelet gets for free from a file on disk. So: + +| flag | behaviour | +|---|---| +| (none) | ends at the first silent gap (1s) or a 30s ceiling, whichever comes first | +| `--follow` | runs until the instance exits, the client disconnects, or the manager shuts down; a silent gap does NOT end it | +| `--tail=N` | the backlog is buffered into a ring of the last N lines and only those are emitted, then `--follow` continues from there | +| `--limit-bytes` | hard cap on bytes handed to the client, applied last | +| `--timestamps`, `--previous`, `--since`, `--since-time` | accepted and **ignored** | + +The idle gap is the heuristic, and it is unavoidable: the alternative for a one-shot +read is hanging until the workload exits, which for a long-running server is forever. +The ceiling covers the opposite case, a workload chatty enough that the stream never +goes idle — it truncates rather than failing, and `--follow` has no ceiling. + +`--tail` costs what buffering costs: the full backlog still crosses the provider's API, +because there is no seek. Only the delivery is trimmed, bounded by N lines. + +The ignored flags cannot be served from this seam rather than merely being unfinished. +`--timestamps` would have to invent a receive-time stamp, attributing the backlog's +whole history to the moment it was fetched. `--previous` and `--since` need a +per-container restart history and time-indexed storage that no provider exposes. They +are ignored rather than rejected so a habitual `--since=1h` prints the full stream +instead of an error. + +## Containers are not addressable + +`-c` is accepted and ignored, for logs and exec alike: a Nebula Pod maps to exactly one +external instance with one console, so there is no per-container stream to select or +shell to enter. Honouring the name would mean rejecting `kubectl logs pod` (which sends +no container) or lying about a second container's output. + +## Exec needs no agent in the image + +The provider's own worker starts the command, so `kubectl exec -it pod -- bash` works +against an unmodified user image — including the `sleep infinity` placeholder a Sandbox +runs. What the exec does need is a **running** instance: a sandbox still queued for +capacity has no container to run in, and the attempt fails rather than waiting. + +Nebula only pumps bytes: stdin is forwarded and closed at EOF (so `exec -i -- cat < f` +ends), stdout and stderr are copied back, and a non-zero exit reaches kubectl as +`command terminated with exit code N` rather than a server error. Two gaps, both from the +provider side: + +| behaviour | why | +|---|---| +| terminal **resize** is ignored | no provider exposes a window-size call, so `-it` opens at the remote default and stays there | +| under `-t`, **stderr is merged into stdout** | a PTY is one stream; kubectl forbids asking for both anyway | diff --git a/docs/status.md b/docs/status.md index ae54726..94d4c6d 100644 --- a/docs/status.md +++ b/docs/status.md @@ -26,7 +26,7 @@ enters the system. - [AWS](#aws) - [Modal](#modal) - [fake](#fake) -- [Logs](#logs) +- [Logs and exec](#logs-and-exec) - [What is not observable](#what-is-not-observable) --- @@ -212,7 +212,8 @@ only two signals and has to record a third fact itself. one stream), so the sandbox console reaches kubectl over the kubelet log route. Two Modal-specific limits: the stream always replays from the sandbox's FIRST byte (there is no seek, which is why `--tail` is served by buffering — see - [Logs](#logs)), and v2 sandboxes report stdio as unsupported, which surfaces as an + [kubelet-api.md](kubelet-api.md#what-logs-honour-and-the-one-heuristic)), and v2 + sandboxes report stdio as unsupported, which surfaces as an error rather than an empty stream. - **Exit codes are lossy.** The control plane's result carries eight statuses (`SUCCESS`, `FAILURE`, `INIT_FAILURE`, `INTERNAL_FAILURE`, `TERMINATED`, @@ -239,71 +240,12 @@ paths without a real backend, so it has no boot or readiness phase to model. --- -## Logs - -`kubectl logs` works against a Nebula Pod, `-f` included. It is worth spelling out -how, because none of the usual kubelet machinery is present. - -**The transport.** `kubectl logs` is not a control-plane read: the API server proxies -it to the kubelet of the node the Pod is on, dialing the address in the Node's -`status.addresses` and the port in `status.daemonEndpoints`. A virtual node has no -kubelet, so the manager serves that one route itself -(`pkg/vnode/kubelet.go`) and every virtual node advertises the manager's Pod IP and -that port. Consequences worth knowing: - -- The endpoint is **leader-scoped and dialed by Pod IP**, not through a Service. The - tracked Pods live in one process's memory, so a Service balancing across replicas - would send requests to a replica that answers `NotFound`. -- It serves TLS with a self-signed, in-memory certificate — what the API server - expects of a kubelet, which does not verify it unless - `--kubelet-certificate-authority` is set. Client certificates are **not** verified - by default, because which CA signs the API server's kubelet client cert is not - portable across distributions. Anything that can reach the port can therefore read - the logs of any Pod on these virtual nodes: keep it closed with a NetworkPolicy, or - pass `--kubelet-client-ca` to require mTLS. -- No POD_IP (running the manager off-cluster) means no endpoint. Logs degrade to - unsupported; nothing else is affected. - -**The provider seam.** Log support is optional: a provider opts in by implementing -`provider.LogStreamer`, and one that does not answers `NotFound` rather than carrying -a stub. The seam is deliberately option-free — one stream, from the instance's first -byte, following until the instance exits, stdout and stderr merged — so every kubectl -option is honoured once, for all providers, in `pkg/vnode/logs.go`. - -**What is honoured, and the one heuristic.** A provider stream has no EOF while the -instance lives and no marker for "you have now caught up", which the real kubelet gets -for free from a file on disk. So: - -| flag | behaviour | -|---|---| -| (none) | ends at the first silent gap (1s) or a 30s ceiling, whichever comes first | -| `--follow` | runs until the instance exits, the client disconnects, or the manager shuts down; a silent gap does NOT end it | -| `--tail=N` | the backlog is buffered into a ring of the last N lines and only those are emitted, then `--follow` continues from there | -| `--limit-bytes` | hard cap on bytes handed to the client, applied last | -| `--timestamps`, `--previous`, `--since`, `--since-time` | accepted and **ignored** | - -The idle gap is the heuristic, and it is unavoidable: the alternative for a one-shot -read is hanging until the workload exits, which for a long-running server is forever. -The ceiling covers the opposite case, a workload chatty enough that the stream never -goes idle — it truncates rather than failing, and `--follow` has no ceiling. - -`--tail` costs what buffering costs: the full backlog still crosses the provider's API, -because there is no seek. Only the delivery is trimmed, bounded by N lines. - -The ignored flags cannot be served from this seam rather than merely being unfinished. -`--timestamps` would have to invent a receive-time stamp, attributing the backlog's -whole history to the moment it was fetched. `--previous` and `--since` need a -per-container restart history and time-indexed storage that no provider exposes. They -are ignored rather than rejected so a habitual `--since=1h` prints the full stream -instead of an error. - -**Containers are not addressable.** `-c` is accepted and ignored: a Nebula Pod maps to -exactly one external instance with one console, so there is no per-container stream to -select. Honouring the name would mean rejecting `kubectl logs pod` (which sends no -container) or lying about a second container's output. - -**`kubectl exec` still does not work.** It needs an agent inside the container, which -is not part of the project today; the route answers `NotImplemented`. +## Logs and exec + +Both are read paths onto a live instance rather than status, and they share one +transport, so they live in [docs/kubelet-api.md](kubelet-api.md): how the manager serves +the kubelet routes, the trust model on that port, and which `kubectl logs` and `kubectl +exec` options are honoured. ## What is not observable diff --git a/internal/controller/sandbox_controller.go b/internal/controller/sandbox_controller.go index 8dbf715..d2b4ba0 100644 --- a/internal/controller/sandbox_controller.go +++ b/internal/controller/sandbox_controller.go @@ -204,10 +204,10 @@ func (r *SandboxReconciler) buildPod(sbx *nebulav1alpha1.Sandbox) *corev1.Pod { // and takes the instance down: the box would surface as Failed seconds after // being provisioned. // - // A PLACEHOLDER process, not a control surface. There is no agent in the - // container, so `kubectl exec`/`logs` against a sandbox do not work yet. - // Whatever restores them replaces this command; a bare `sleep` keeps any - // bootstrap from having to inject a binary into an arbitrary user image. + // A PLACEHOLDER process, not a control surface. `kubectl exec` and + // `kubectl logs` do not go through it: the provider starts the command + // itself, so a bare `sleep` is enough and no binary has to be injected + // into an arbitrary user image. // // No tension with a user-supplied command: SandboxSpec has no command field. Command: []string{"sleep", "infinity"}, diff --git a/pkg/provider/modal/client.go b/pkg/provider/modal/client.go index 2a85103..48c279b 100644 --- a/pkg/provider/modal/client.go +++ b/pkg/provider/modal/client.go @@ -18,6 +18,7 @@ package modal import ( "context" + "errors" "fmt" "io" "strconv" @@ -31,6 +32,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/intstr" + "github.com/InftyAI/Nebula/pkg/provider" "github.com/InftyAI/Nebula/pkg/provider/catalog" ) @@ -318,13 +320,135 @@ func (c *sdkClient) SandboxLogs(ctx context.Context, id string) (io.ReadCloser, return mergeStreams(ctx, sb.Stdout, sb.Stderr), nil } +// SandboxExec implements Client: it starts cmd in the sandbox's running container and +// hands back the process handle, which is what `kubectl exec` is pumped from. +// +// No agent is installed for this. Modal's own worker runs the command, so exec works on +// any image — but it goes through the container's TASK, so the sandbox has to be running: +// the SDK polls briefly for a task id and then errors, which is the honest answer for a +// sandbox still queued for a GPU. +// +// A TTY is requested when the client asked for one, and Modal then multiplexes stderr into +// stdout — the same thing a real terminal does. Its window size is Modal's fixed 24x80: the +// SDK exposes no way to set it and the command router has no resize call. +// +// ctx bounds the start only, as provider.Executor requires. That holds because the SDK runs +// the stdio streams and stdin writes on their own background contexts — the sandbox lookup, +// the task-id wait and ExecStart are all this ctx covers, and those are exactly what a +// caller wants to give up on. +// +// Setup is paid PER EXEC and dominates it (~16s cold: resolve the sandbox, poll for its +// task id, dial a fresh TLS connection to that task's router). Caching the handle would +// make later execs cheap, but the manager would hold one connection per sandbox with +// nothing reliable to close it — a sandbox that dies on its own leaves Modal's list +// (IncludeFinished: false) and is never observed again. A slow exec beats a leak here. +// +// No Timeout is set, so the command runs until it exits or the sandbox does. Any cap we +// picked would truncate somebody's job — and, since it surfaces as a stream error rather +// than an exit code, it would look like a network glitch. The cost is that a command whose +// client disconnected keeps running: Modal has no "kill this exec" call, and closing the +// connection leaves the process alive in the container. The sandbox's own lifetime is what +// finally reaps it. +func (c *sdkClient) SandboxExec( + ctx context.Context, id string, cmd []string, opts provider.ExecOptions, +) (provider.Process, error) { + sb, err := c.mc.Sandboxes.FromID(ctx, id, &modal.SandboxFromIDParams{}) + if err != nil { + if isNotFound(err) { + return nil, fmt.Errorf("modal: sandbox %s not found: %w", id, err) + } + return nil, err + } + cp, err := sb.Exec(ctx, cmd, &modal.SandboxExecParams{PTY: opts.TTY}) + if err != nil { + return nil, fmt.Errorf("modal: exec in sandbox %s: %w", id, err) + } + return &sandboxProcess{sb: sb, cp: cp, tty: opts.TTY}, nil +} + +// sandboxProcess adapts Modal's ContainerProcess to provider.Process. +// +// It holds the *modal.Sandbox as well as the process because Exec dialled a private gRPC +// connection to the container's command router through it, and Detach is the only way to +// close that — without it every exec would leak a connection for the manager's life. +type sandboxProcess struct { + sb *modal.Sandbox + cp *modal.ContainerProcess + tty bool + + closeOnce sync.Once +} + +var _ provider.Process = (*sandboxProcess)(nil) + +func (p *sandboxProcess) Stdin() io.WriteCloser { return p.cp.Stdin } +func (p *sandboxProcess) Stdout() io.Reader { return p.cp.Stdout } + +// Stderr is nil under a TTY: Modal multiplexes both streams into stdout there, leaving +// this one permanently empty, and reading it would just park a goroutine. +func (p *sandboxProcess) Stderr() io.Reader { + if p.tty { + return nil + } + return p.cp.Stderr +} + +// Wait blocks until the command exits and reports its exit code. Modal renders a signal +// death as 128+signal, the shell convention, so ^C reads as 130. +func (p *sandboxProcess) Wait(ctx context.Context) (int, error) { + return p.cp.Wait(ctx, &modal.ContainerProcessWaitParams{}) +} + +// Close releases the streams and the command-router connection. Idempotent, since the +// caller may close after a teardown already ran. Errors are dropped: the exec is over +// either way, and there is no caller left to act on them. +func (p *sandboxProcess) Close() error { + p.closeOnce.Do(func() { + _ = p.cp.Stdin.Close() + _ = p.cp.Stdout.Close() + _ = p.cp.Stderr.Close() + _ = p.sb.Detach() + }) + return nil +} + // mergeStreams fans log streams into one ReadCloser. Close tears everything down: it // closes the sources (unblocking the copiers) and the pipe (returning an in-flight // Read). ctx does the same, so a disconnected `kubectl logs -f` cannot leave // goroutines long-polling Modal forever. +// +// A stream that FAILS ends the merged stream with its error, not at EOF. Silence is the +// normal answer here — Modal serves only recent output — so a swallowed error made an +// outage look exactly like a workload that had logged nothing. func mergeStreams(ctx context.Context, streams ...io.ReadCloser) io.ReadCloser { pr, pw := io.Pipe() + // Closed before the sources are, so a read that fails because WE tore down is not + // reported as Modal breaking. + tearing := make(chan struct{}) + + var ( + mu sync.Mutex + firstErr error + ) + // io.Copy reports EOF as success, so anything arriving here is a real failure — + // except a closed pipe, which is the client having gone away. + fail := func(err error) { + if err == nil || errors.Is(err, io.ErrClosedPipe) { + return + } + select { + case <-tearing: + return + default: + } + mu.Lock() + defer mu.Unlock() + if firstErr == nil { + firstErr = err + } + } + var wg sync.WaitGroup for _, s := range streams { if s == nil { @@ -333,16 +457,20 @@ func mergeStreams(ctx context.Context, streams ...io.ReadCloser) io.ReadCloser { wg.Add(1) go func(src io.ReadCloser) { defer wg.Done() - // Copy errors are not reported: one stream ending must not truncate the other, - // and the shared failure (sandbox gone) already surfaces as EOF. - _, _ = io.Copy(pw, src) + // Held, not raised here: one stream ending must not truncate the other. + _, err := io.Copy(pw, src) + fail(err) }(s) } // Only close the pipe once BOTH sources are done, so EOF means no more output. go func() { wg.Wait() - _ = pw.Close() + mu.Lock() + err := firstErr + mu.Unlock() + // CloseWithError(nil) is a plain EOF, so this covers both endings. + _ = pw.CloseWithError(err) }() closeAll := func() { @@ -360,6 +488,7 @@ func mergeStreams(ctx context.Context, streams ...io.ReadCloser) io.ReadCloser { case <-ctx.Done(): case <-stop: } + close(tearing) closeAll() _ = pw.CloseWithError(ctx.Err()) }() diff --git a/pkg/provider/modal/modal.go b/pkg/provider/modal/modal.go index 9d5c864..715548e 100644 --- a/pkg/provider/modal/modal.go +++ b/pkg/provider/modal/modal.go @@ -66,12 +66,14 @@ import ( // different ceiling sets spec.activeDeadlineSeconds, which maps straight through. const defaultSandboxTimeout = 24 * time.Hour -// compile-time assertions that Provider satisfies the interfaces. LogStreamer is -// the optional half: it is what makes `kubectl logs` work here, and asserting it -// separately is the point — a provider is free not to serve logs. +// compile-time assertions that Provider satisfies the interfaces. LogStreamer and +// Executor are the optional halves: they are what make `kubectl logs` and `kubectl exec` +// work here, and asserting them separately is the point — a provider is free to serve +// neither. var ( _ provider.Provider = (*Provider)(nil) _ provider.LogStreamer = (*Provider)(nil) + _ provider.Executor = (*Provider)(nil) ) // Client is the narrow seam over Modal's API. It is intentionally small: only @@ -95,6 +97,9 @@ type Client interface { // SandboxLogs returns merged stdout+stderr, from the first byte, following until // the sandbox exits (see provider.LogStreamer). The caller owns Close. SandboxLogs(ctx context.Context, id string) (io.ReadCloser, error) + // SandboxExec starts cmd inside a running sandbox and returns the handle to its + // streams and exit code (see provider.Executor). The caller owns Close. + SandboxExec(ctx context.Context, id string, cmd []string, opts provider.ExecOptions) (provider.Process, error) } // SandboxSpec is the resolved, Modal-shaped request the Client turns into a @@ -356,6 +361,22 @@ func (p *Provider) Logs(ctx context.Context, instanceID string) (io.ReadCloser, return p.client.SandboxLogs(ctx, instanceID) } +// Exec implements provider.Executor, and is what `kubectl exec` runs. A pass-through: +// the streams are the caller's job (pkg/vnode/exec.go pumps them). +// +// Modal needs no agent in the image for this — its own worker runs the command — so exec +// works on any sandbox, including the `sleep infinity` placeholder a Sandbox gets. The +// sandbox must be RUNNING though: Modal routes an exec through the container's task, so +// one still queued fails here rather than waiting. +func (p *Provider) Exec( + ctx context.Context, instanceID string, cmd []string, opts provider.ExecOptions, +) (provider.Process, error) { + if instanceID == "" { + return nil, fmt.Errorf("modal: no sandbox for this pod yet") + } + return p.client.SandboxExec(ctx, instanceID, cmd, opts) +} + // Get implements provider.Provider. func (p *Provider) Get(ctx context.Context, instanceID string) (*provider.Instance, error) { sb, err := p.client.GetSandbox(ctx, instanceID) diff --git a/pkg/provider/modal/modal_test.go b/pkg/provider/modal/modal_test.go index c514140..36d8fc0 100644 --- a/pkg/provider/modal/modal_test.go +++ b/pkg/provider/modal/modal_test.go @@ -22,6 +22,7 @@ import ( "io" "slices" "strings" + "sync" "testing" "time" @@ -52,6 +53,12 @@ type fakeClient struct { logs string logsErr error logsFor string + + // The exec path, recorded the same way: what the adapter passed down. + execErr error + execFor string + execCmd []string + execOpts provider.ExecOptions } func (f *fakeClient) CreateSandbox(_ context.Context, spec SandboxSpec) (string, Credential, error) { @@ -95,6 +102,25 @@ func (f *fakeClient) SandboxLogs(_ context.Context, id string) (io.ReadCloser, e return io.NopCloser(strings.NewReader(f.logs)), nil } +func (f *fakeClient) SandboxExec( + _ context.Context, id string, cmd []string, opts provider.ExecOptions, +) (provider.Process, error) { + f.execFor, f.execCmd, f.execOpts = id, cmd, opts + if f.execErr != nil { + return nil, f.execErr + } + return fakeProcess{}, nil +} + +// fakeProcess is a command that produced nothing and exited 0. +type fakeProcess struct{} + +func (fakeProcess) Stdin() io.WriteCloser { return nil } +func (fakeProcess) Stdout() io.Reader { return strings.NewReader("") } +func (fakeProcess) Stderr() io.Reader { return nil } +func (fakeProcess) Wait(_ context.Context) (int, error) { return 0, nil } +func (fakeProcess) Close() error { return nil } + // fakeCatalog is a trivial provider.Catalog for tests. type fakeCatalog struct{ rows []provider.Offering } @@ -1115,6 +1141,161 @@ func TestLogs_ClientErrorPropagates(t *testing.T) { } } -// The handler resolves LogStreamer by type assertion, so drift would compile fine and -// silently revert `kubectl logs` to NotFound. Assert it here. -var _ provider.LogStreamer = (*Provider)(nil) +// Exec is the same shape of pass-through: the command, the TTY request and the id must all +// reach the client unchanged, since the adapter decides nothing else here. +func TestExec_PassesCommandThrough(t *testing.T) { + f := &fakeClient{} + p := newTestProvider(f) + + proc, err := p.Exec(context.Background(), "sb-1", []string{"bash", "-lc", "ls /"}, + provider.ExecOptions{TTY: true}) + if err != nil { + t.Fatalf("Exec: %v", err) + } + defer func() { _ = proc.Close() }() + + if f.execFor != "sb-1" { + t.Errorf("client asked for sandbox %q, want sb-1", f.execFor) + } + if !slices.Equal(f.execCmd, []string{"bash", "-lc", "ls /"}) { + t.Errorf("command = %v, want it passed through verbatim", f.execCmd) + } + if !f.execOpts.TTY { + t.Error("TTY was not passed through; an interactive shell needs it") + } +} + +// An empty id is a Pod still inside Provision: there is no sandbox to run in, so this +// fails here instead of reaching Modal. +func TestExec_NoSandboxIDErrors(t *testing.T) { + f := &fakeClient{} + p := newTestProvider(f) + + if _, err := p.Exec(context.Background(), "", []string{"sh"}, provider.ExecOptions{}); err == nil { + t.Fatal("Exec(\"\"): expected an error") + } + if f.execFor != "" { + t.Fatalf("client was called with %q; an empty id must not reach Modal", f.execFor) + } +} + +// A sandbox that is gone, or still queued for a GPU (no task to exec into), must surface +// as an error rather than a silent no-op. +func TestExec_ClientErrorPropagates(t *testing.T) { + f := &fakeClient{execErr: fmt.Errorf("timed out waiting for task ID")} + p := newTestProvider(f) + + if _, err := p.Exec(context.Background(), "sb-queued", []string{"sh"}, provider.ExecOptions{}); err == nil { + t.Fatal("expected the client error to propagate") + } +} + +// The handler resolves both optional halves by type assertion, so drift would compile fine +// and silently revert `kubectl logs`/`kubectl exec` to NotFound. Assert them here. +var ( + _ provider.LogStreamer = (*Provider)(nil) + _ provider.Executor = (*Provider)(nil) +) + +// --- mergeStreams ----------------------------------------------------------- + +// errStream yields s, then fails. A Modal log stream that dies mid-poll looks like this. +type errStream struct { + s string + err error + n int +} + +func (e *errStream) Read(p []byte) (int, error) { + if e.n < len(e.s) { + n := copy(p, e.s[e.n:]) + e.n += n + return n, nil + } + return 0, e.err +} + +func (e *errStream) Close() error { return nil } + +// Modal serves only recent output, so an empty log is ORDINARY. A broken stream must +// therefore say so: read as EOF, an outage is indistinguishable from a quiet workload. +func TestMergeStreams_StreamFailureSurfaces(t *testing.T) { + boom := fmt.Errorf("error getting output stream: unavailable") + rc := mergeStreams(context.Background(), &errStream{s: "some output\n", err: boom}) + defer func() { _ = rc.Close() }() + + got, err := io.ReadAll(rc) + if err == nil { + t.Fatal("ReadAll: expected the stream failure to surface") + } + if !strings.Contains(err.Error(), "unavailable") { + t.Fatalf("err = %v, want the provider's own message", err) + } + // Output already received is still delivered: the error explains the END of the log, + // it does not discard it. + if string(got) != "some output\n" { + t.Fatalf("logs = %q, want what arrived before the failure", got) + } +} + +// One failed stream must not truncate its sibling — stdout is the interesting one, and a +// stderr that breaks would otherwise cut it short. +func TestMergeStreams_OneFailureKeepsTheOther(t *testing.T) { + ok := io.NopCloser(strings.NewReader("stdout line\n")) + bad := &errStream{err: fmt.Errorf("stderr gone")} + + rc := mergeStreams(context.Background(), ok, bad) + defer func() { _ = rc.Close() }() + + got, err := io.ReadAll(rc) + if string(got) != "stdout line\n" { + t.Fatalf("logs = %q, want the healthy stream in full", got) + } + if err == nil { + t.Fatal("expected the failure of the other stream to be reported") + } +} + +// A clean end stays clean: both streams EOF, so `kubectl logs` must not report an error +// for a workload that simply stopped logging. +func TestMergeStreams_CleanEndIsEOF(t *testing.T) { + rc := mergeStreams(context.Background(), + io.NopCloser(strings.NewReader("a\n")), io.NopCloser(strings.NewReader(""))) + defer func() { _ = rc.Close() }() + + if _, err := io.ReadAll(rc); err != nil { + t.Fatalf("ReadAll: %v, want a clean EOF", err) + } +} + +// blockingStream is a following log stream: silent until closed, and then it reports a +// failure — which is what a source WE closed looks like from the copier. +type blockingStream struct { + closed chan struct{} + once sync.Once +} + +func (b *blockingStream) Read([]byte) (int, error) { + <-b.closed + return 0, fmt.Errorf("stream closed by teardown") +} + +func (b *blockingStream) Close() error { + b.once.Do(func() { close(b.closed) }) + return nil +} + +// Our own teardown is not a provider failure: a client walking away from `kubectl logs -f` +// closes the sources, and the read errors that follow must not be reported as an outage. +func TestMergeStreams_TeardownIsNotAFailure(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + rc := mergeStreams(ctx, &blockingStream{closed: make(chan struct{})}) + cancel() + + // The context error or a clean EOF, never the source's own failure. + _, err := io.ReadAll(rc) + if err != nil && strings.Contains(err.Error(), "stream closed by teardown") { + t.Fatalf("err = %v, want teardown not to be reported as a provider failure", err) + } + _ = rc.Close() +} diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index 8ffaa51..1efe911 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -166,6 +166,48 @@ type LogStreamer interface { Logs(ctx context.Context, instanceID string) (io.ReadCloser, error) } +// Executor is the other OPTIONAL half: run one command inside a live instance, which is +// what makes `kubectl exec` work. A backend with no way in (no agent, no SSH key) does +// not implement it, and the virtual node answers NotFound. +// +// Exec only STARTS the command; the caller pumps the streams and waits. That keeps the +// copy loop in pkg/vnode/exec.go, so stdin EOF, output draining and exit-code reporting +// are identical for every provider and no adapter can quietly get one wrong. +// +// ctx covers the START ONLY, and the caller bounds it — an interactive shell outlives it +// by hours. So the returned Process must not tie its streams or Wait to ctx, or a long +// exec would die the moment the start budget ran out. +type Executor interface { + Exec(ctx context.Context, instanceID string, cmd []string, opts ExecOptions) (Process, error) +} + +// ExecOptions is what the client asked for that the provider must know at start time. +// Everything else about the command is in cmd. +type ExecOptions struct { + // TTY asks for a pseudo-terminal, so the command believes it is interactive and line + // editing works (`kubectl exec -it`). A provider that cannot allocate one still runs + // the command: a shell without a TTY beats no shell. + TTY bool +} + +// Process is one command running inside an instance. The caller always calls Close. +type Process interface { + // Stdin is the command's input; closing it is how the command sees EOF. Nil when the + // provider cannot write stdin, which makes the exec output-only. + Stdin() io.WriteCloser + // Stdout is the command's output — under a TTY, everything it writes. + Stdout() io.Reader + // Stderr is separate error output, or nil when there is none to separate. Nil is the + // normal case under a TTY, where one terminal carries both. + Stderr() io.Reader + // Wait blocks until the command exits and returns its exit code. A non-zero code is + // the command's own answer, NOT a failure — err is for "we could not find out". + Wait(ctx context.Context) (int, error) + // Close releases the streams and whatever connection carried them. It must unblock a + // parked Read/Write, so a disconnected client cannot leave the transport running. + Close() error +} + // ProvisionRequest carries only the placement decisions that are NOT already on the Pod. // Everything about the workload — image, command, env, ports, cpu/memory, accelerator type // and count — is read from the Pod, the single source of truth. That leaves the tier, the diff --git a/pkg/vnode/exec.go b/pkg/vnode/exec.go new file mode 100644 index 0000000..074f637 --- /dev/null +++ b/pkg/vnode/exec.go @@ -0,0 +1,144 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package vnode + +import ( + "context" + "fmt" + "io" + "sync" + "time" + + vkapi "github.com/virtual-kubelet/virtual-kubelet/node/api" + utilexec "k8s.io/utils/exec" + + "github.com/InftyAI/Nebula/pkg/provider" +) + +const ( + // execDrainGrace is how long output may still arrive after the exit code has. The two + // travel on separate streams, so the last bytes can lose the race. Bounded, because a + // stream that never signals EOF must not hang the request forever. + execDrainGrace = 2 * time.Second + + // execStartTimeout caps how long a provider may take to START the command — dialling + // the instance, waiting for its container. Matches the stream creation timeout the + // kubelet routes use, so the whole handshake has one budget. + execStartTimeout = 30 * time.Second +) + +// runExec pumps one `kubectl exec`: the client's terminal (vkapi.AttachIO) on one side, +// the provider's command (provider.Process) on the other. The provider seam only starts +// the command, so this is the whole contract in one place and every provider behaves the +// same. +// +// The returned error is what the VK exec route turns into a status: nil for a clean run, +// a utilexec.ExitError for a non-zero exit, anything else an internal error. +func runExec(ctx context.Context, proc provider.Process, attach vkapi.AttachIO) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + // Cancel is what releases the provider's streams, so nothing keeps streaming after + // this returns — including when the copies below end because the client hung up. + go func() { + <-ctx.Done() + _ = proc.Close() + }() + + // Resize events are dropped: no provider carries a window size today. They are still + // drained, or the VK goroutine sending them parks until the exec ends. + if resize := attach.Resize(); resize != nil { + go func() { + for { + select { + case <-resize: + case <-ctx.Done(): + return + } + } + }() + } + + // Stdin is copied without being waited on: an interactive shell's stdin ends only when + // the client goes away, so the command decides when the exec is over. Closing on EOF is + // what makes `kubectl exec -i -- cat < file` terminate. + if src, dst := attach.Stdin(), proc.Stdin(); src != nil && dst != nil { + go func() { + _, _ = io.Copy(dst, src) + _ = dst.Close() + }() + } + + // A copy that FAILS means the client is gone or the transport died, and it is the only + // signal we get: the VK exec route runs on a context of its own, so a disconnect never + // reaches ctx. Without this, closing a laptop mid-shell leaves Wait blocked for as long + // as the command lives. + output := copyOutput(attach, proc, cancel) + + code, err := proc.Wait(ctx) + // Let output still in flight land before reporting the outcome, so a command's last + // line is not lost to its exit code overtaking it. + select { + case <-output: + case <-ctx.Done(): + case <-time.After(execDrainGrace): + } + + if err != nil { + return fmt.Errorf("wait for command: %w", err) + } + if code != 0 { + // A TYPED exit error: the VK route reports anything else as an internal error, so + // the user would see a 500 instead of "command terminated with exit code N". + return utilexec.CodeExitError{ + Err: fmt.Errorf("command terminated with exit code %d", code), + Code: code, + } + } + return nil +} + +// copyOutput streams the command's stdout and stderr to the client and returns a channel +// closed once BOTH have ended — i.e. the command has produced everything it will. A copy +// that fails calls broken, since neither side can be reached any more. +func copyOutput(attach vkapi.AttachIO, proc provider.Process, broken func()) <-chan struct{} { + var wg sync.WaitGroup + pump := func(dst io.Writer, src io.Reader) { + // A nil stream means the client did not ask for it, or the command has none (stderr + // under a TTY, where the terminal carries both). + if dst == nil || src == nil { + return + } + wg.Add(1) + go func() { + defer wg.Done() + // EOF is the command finishing normally and returns nil; any other error means + // this exec has nowhere left to go, so end it rather than wait out the command. + if _, err := io.Copy(dst, src); err != nil { + broken() + } + }() + } + pump(attach.Stdout(), proc.Stdout()) + pump(attach.Stderr(), proc.Stderr()) + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + return done +} diff --git a/pkg/vnode/exec_test.go b/pkg/vnode/exec_test.go new file mode 100644 index 0000000..f0d5baf --- /dev/null +++ b/pkg/vnode/exec_test.go @@ -0,0 +1,436 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package vnode + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + "sync" + "testing" + "time" + + "github.com/virtual-kubelet/virtual-kubelet/errdefs" + vkapi "github.com/virtual-kubelet/virtual-kubelet/node/api" + utilexec "k8s.io/utils/exec" + + "github.com/InftyAI/Nebula/pkg/provider" +) + +// --- fakes ------------------------------------------------------------------ + +// syncBuffer is a writable sink the test reads while copy goroutines write it. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer + closed chan struct{} + once sync.Once +} + +func newSyncBuffer() *syncBuffer { return &syncBuffer{closed: make(chan struct{})} } + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) Close() error { + b.once.Do(func() { close(b.closed) }) + return nil +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// execProcess is a fake provider.Process. wait, when set, is what the command waits on +// before exiting, so a test decides when it ends. +type execProcess struct { + stdin *syncBuffer + stdout io.Reader + stderr io.Reader + code int + waitErr error + wait <-chan struct{} + + waitBounded bool + + closed chan struct{} + once sync.Once +} + +func newExecProcess(stdout, stderr string, code int) *execProcess { + p := &execProcess{stdin: newSyncBuffer(), code: code, closed: make(chan struct{})} + if stdout != "" { + p.stdout = strings.NewReader(stdout) + } + if stderr != "" { + p.stderr = strings.NewReader(stderr) + } + return p +} + +func (p *execProcess) Stdin() io.WriteCloser { return p.stdin } +func (p *execProcess) Stdout() io.Reader { return p.stdout } +func (p *execProcess) Stderr() io.Reader { return p.stderr } + +func (p *execProcess) Wait(ctx context.Context) (int, error) { + _, p.waitBounded = ctx.Deadline() + if p.wait != nil { + select { + case <-p.wait: + case <-ctx.Done(): + return 0, ctx.Err() + } + } + return p.code, p.waitErr +} + +func (p *execProcess) Close() error { + p.once.Do(func() { close(p.closed) }) + return nil +} + +// execProvider is a fakeProvider that also serves exec, and records what it was asked. +type execProvider struct { + *fakeProvider + proc *execProcess + execErr error + askedFor string + askedCmd []string + askedTTY bool + + startDeadline time.Time + startBounded bool +} + +func newExecProvider(fp *fakeProvider, proc *execProcess) *execProvider { + return &execProvider{fakeProvider: fp, proc: proc} +} + +func (p *execProvider) Exec( + ctx context.Context, instanceID string, cmd []string, opts provider.ExecOptions, +) (provider.Process, error) { + p.askedFor, p.askedCmd, p.askedTTY = instanceID, cmd, opts.TTY + p.startDeadline, p.startBounded = ctx.Deadline() + if p.execErr != nil { + return nil, p.execErr + } + return p.proc, nil +} + +// fakeAttach is the client side of an exec: the streams kubectl would have opened. A nil +// stream means the client did not ask for it. +type fakeAttach struct { + stdin io.Reader + stdout io.WriteCloser + stderr io.WriteCloser + tty bool + resize chan vkapi.TermSize +} + +func (a *fakeAttach) Stdin() io.Reader { return a.stdin } +func (a *fakeAttach) Stdout() io.WriteCloser { return a.stdout } +func (a *fakeAttach) Stderr() io.WriteCloser { return a.stderr } +func (a *fakeAttach) TTY() bool { return a.tty } +func (a *fakeAttach) Resize() <-chan vkapi.TermSize { return a.resize } + +// --- runExec ---------------------------------------------------------------- + +// The ordinary case: both output streams reach the client and a clean exit is no error. +func TestRunExec_CopiesOutput(t *testing.T) { + proc := newExecProcess("on stdout\n", "on stderr\n", 0) + out, errOut := newSyncBuffer(), newSyncBuffer() + + if err := runExec(context.Background(), proc, &fakeAttach{stdout: out, stderr: errOut}); err != nil { + t.Fatalf("runExec: %v", err) + } + if out.String() != "on stdout\n" { + t.Errorf("stdout = %q", out.String()) + } + if errOut.String() != "on stderr\n" { + t.Errorf("stderr = %q", errOut.String()) + } + // The provider transport must be released, or a disconnected client leaks a stream. + select { + case <-proc.closed: + case <-time.After(2 * time.Second): + t.Error("process was not closed after the exec finished") + } +} + +// A non-zero exit is the command's own answer, and has to reach kubectl AS an exit code: +// only a typed exit error becomes "command terminated with exit code N" instead of a 500. +func TestRunExec_NonZeroExitIsAnExitError(t *testing.T) { + proc := newExecProcess("", "no such file\n", 2) + out := newSyncBuffer() + + err := runExec(context.Background(), proc, &fakeAttach{stdout: out, stderr: newSyncBuffer()}) + if err == nil { + t.Fatal("runExec: expected an error for a non-zero exit") + } + var exitErr utilexec.ExitError + if !errors.As(err, &exitErr) { + t.Fatalf("err = %v (%T), want a utilexec.ExitError", err, err) + } + if !exitErr.Exited() || exitErr.ExitStatus() != 2 { + t.Fatalf("exit status = %d, want 2", exitErr.ExitStatus()) + } +} + +// Not knowing how the command ended is a different failure from it ending badly, and must +// NOT be reported as an exit code the command never returned. +func TestRunExec_WaitErrorIsNotAnExitError(t *testing.T) { + proc := newExecProcess("", "", 0) + proc.waitErr = errors.New("command router unreachable") + + err := runExec(context.Background(), proc, &fakeAttach{stdout: newSyncBuffer()}) + if err == nil { + t.Fatal("runExec: expected the wait failure to surface") + } + var exitErr utilexec.ExitError + if errors.As(err, &exitErr) { + t.Fatalf("err = %v, want a plain error rather than an exit status", err) + } +} + +// Stdin is forwarded, and CLOSED at EOF — that close is what makes +// `kubectl exec -i -- cat < file` end instead of hanging forever. +func TestRunExec_ForwardsStdinAndClosesIt(t *testing.T) { + proc := newExecProcess("", "", 0) + // The command exits when its stdin closes, like `cat`. + proc.wait = proc.stdin.closed + + err := runExec(context.Background(), proc, &fakeAttach{ + stdin: strings.NewReader("hello\n"), + stdout: newSyncBuffer(), + }) + if err != nil { + t.Fatalf("runExec: %v", err) + } + if proc.stdin.String() != "hello\n" { + t.Fatalf("stdin = %q, want it forwarded verbatim", proc.stdin.String()) + } +} + +// `kubectl exec -it`: the terminal carries one stream, so the provider reports no stderr +// and the client asked for none. Neither may be treated as a missing stream to copy. +func TestRunExec_TTYWithoutStderr(t *testing.T) { + proc := newExecProcess("prompt$ ", "", 0) + out := newSyncBuffer() + + attach := &fakeAttach{stdout: out, tty: true, resize: make(chan vkapi.TermSize, 1)} + // kubectl sends the window size immediately; nothing consumes it downstream, but a + // dropped resize must not stall the exec. + attach.resize <- vkapi.TermSize{Width: 80, Height: 24} + + if err := runExec(context.Background(), proc, attach); err != nil { + t.Fatalf("runExec: %v", err) + } + if out.String() != "prompt$ " { + t.Fatalf("stdout = %q", out.String()) + } +} + +// A client that disconnects mid-command (ctx cancelled) must release the provider stream +// rather than leave it running for the rest of the manager's life. +func TestRunExec_CancelReleasesTheProcess(t *testing.T) { + proc := newExecProcess("", "", 0) + proc.wait = make(chan struct{}) // never exits on its own + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- runExec(ctx, proc, &fakeAttach{stdout: newSyncBuffer()}) }() + + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("runExec did not return after the context was cancelled") + } + select { + case <-proc.closed: + case <-time.After(2 * time.Second): + t.Fatal("process was not closed after cancellation") + } +} + +// --- RunInContainer --------------------------------------------------------- + +// The whole path: Pod → the instance it provisioned → a command run there. The instance +// id is the assertion that matters; a wrong one would shell into another tenant's box. +func TestRunInContainer_RunsInTrackedInstance(t *testing.T) { + proc := newExecProcess("root@sandbox:/#\n", "", 0) + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, proc) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + out := newSyncBuffer() + attach := &fakeAttach{stdout: out, tty: true} + err := h.RunInContainer(context.Background(), "default", "p1", "main", []string{"bash"}, attach) + if err != nil { + t.Fatalf("RunInContainer: %v", err) + } + if ep.askedFor != "inst-1" { + t.Errorf("provider asked for instance %q, want inst-1", ep.askedFor) + } + if len(ep.askedCmd) != 1 || ep.askedCmd[0] != "bash" { + t.Errorf("command = %v, want it passed through", ep.askedCmd) + } + if !ep.askedTTY { + t.Error("TTY was not passed to the provider") + } + if out.String() != "root@sandbox:/#\n" { + t.Errorf("stdout = %q", out.String()) + } +} + +// One instance, one container: `-c whatever` must not fail on a name with nothing behind +// it, exactly as for logs. +func TestRunInContainer_IgnoresContainerName(t *testing.T) { + for _, container := range []string{"", "main", "not-a-container"} { + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("ok\n", "", 0)) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + err := h.RunInContainer(context.Background(), "default", "p1", container, + []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) + if err != nil { + t.Fatalf("RunInContainer(container=%q): %v", container, err) + } + } +} + +// Every miss must read as NotFound, which kubectl reports as such rather than as a 500 +// the user is invited to retry. +func TestRunInContainer_NotFoundCases(t *testing.T) { + // No exec support at all — a legitimate configuration (no agent, no key), not a bug. + t.Run("provider does not support exec", func(t *testing.T) { + h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + err := h.RunInContainer(context.Background(), "default", "p1", "main", + []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) + assertNotFound(t, err) + }) + + // Another node's pod, or one this process never adopted. + t.Run("pod not tracked", func(t *testing.T) { + ep := newExecProvider(&fakeProvider{}, newExecProcess("", "", 0)) + h := NewHandler(ep, nil, nil) + err := h.RunInContainer(context.Background(), "default", "ghost", "main", + []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) + assertNotFound(t, err) + if ep.askedFor != "" { + t.Fatalf("provider was asked for instance %q; it must not be called at all", ep.askedFor) + } + }) + + // Tracked with no instance id: what a rejected Provision leaves behind. There is + // nothing to run in. + t.Run("tracked without an instance", func(t *testing.T) { + ep := newExecProvider(&fakeProvider{provisionErr: errors.New("no capacity")}, newExecProcess("", "", 0)) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { + t.Fatal("CreatePod: expected the provision rejection to surface") + } + err := h.RunInContainer(context.Background(), "default", "p1", "main", + []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) + assertNotFound(t, err) + }) +} + +// Starting is bounded, running is not. Without the cap, a provider that waits for a +// queued instance (Modal polls for five minutes) leaves the user at a blank terminal; with +// it applied too widely, an idle shell would be cut off after 30s. +func TestRunInContainer_OnlyTheStartIsBounded(t *testing.T) { + proc := newExecProcess("ok\n", "", 0) + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, proc) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + err := h.RunInContainer(context.Background(), "default", "p1", "main", + []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) + if err != nil { + t.Fatalf("RunInContainer: %v", err) + } + if !ep.startBounded { + t.Fatal("the provider was given an unbounded context to start the command in") + } + if left := time.Until(ep.startDeadline); left > execStartTimeout { + t.Fatalf("start budget = %v, want at most %v", left, execStartTimeout) + } + if proc.waitBounded { + t.Fatal("Wait inherited the start deadline; a long-running command would be killed") + } +} + +// A provider that cannot start the command (sandbox still queued, API down) is a real +// error: NotFound would claim the Pod cannot be exec'd into at all. +func TestRunInContainer_StartErrorIsNotNotFound(t *testing.T) { + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("", "", 0)) + ep.execErr = errors.New("timed out waiting for task id") + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + err := h.RunInContainer(context.Background(), "default", "p1", "main", + []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) + if err == nil { + t.Fatal("expected an error when the provider cannot start the command") + } + if errdefs.IsNotFound(err) { + t.Fatalf("err = %v, want a plain error rather than NotFound", err) + } + if !strings.Contains(err.Error(), "inst-1") { + t.Fatalf("err = %v, want the instance id in the message", err) + } +} + +// An empty command is a malformed request, not a missing Pod, and must never reach the +// provider — a provider that ran a default shell for it would be a surprise. +func TestRunInContainer_EmptyCommandRejected(t *testing.T) { + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("", "", 0)) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + err := h.RunInContainer(context.Background(), "default", "p1", "main", + nil, &fakeAttach{stdout: newSyncBuffer()}) + if err == nil { + t.Fatal("expected an error for an empty command") + } + if ep.askedFor != "" { + t.Fatalf("provider was asked for instance %q; an empty command must not reach it", ep.askedFor) + } +} + +// compile-time check that the fake matches the seam the handler asserts on. +var _ provider.Executor = (*execProvider)(nil) diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index 6cbef94..7b37a42 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -892,6 +892,39 @@ func blocklistTTL(pod *corev1.Pod) time.Duration { func ptrNow(t metav1.Time) *metav1.Time { return &t } +// Tracks reports whether this virtual node is the one running the Pod. The kubelet API +// has one listener for every provider and its routes carry no node name, so this is how a +// request finds its handler; see kubelet.go. +func (h *Handler) Tracks(namespace, podName string) bool { + h.mu.Lock() + defer h.mu.Unlock() + _, ok := h.tracked[key(namespace, podName)] + return ok +} + +// instanceFor returns the external instance backing a Pod, for the kubelet routes that +// need one. Both failures are NotFound, since both mean "there is nothing to read or run +// against": the Pod is not tracked here, or it is but Provision has not returned an id +// yet. +func (h *Handler) instanceFor(namespace, podName string) (string, error) { + h.mu.Lock() + tp, tracked := h.tracked[key(namespace, podName)] + var instance string + if tracked { + instance = tp.instance + } + h.mu.Unlock() + + if !tracked { + return "", errdefs.NotFoundf("pod %q is not known to virtual node %q", + key(namespace, podName), NodeName(h.prov.Name())) + } + if instance == "" { + return "", errdefs.NotFoundf("pod %q has no external instance yet", key(namespace, podName)) + } + return instance, nil +} + // GetContainerLogs serves `kubectl logs` for a Pod on this virtual node (kubelet.go // carries the endpoint). Three things must line up, each failing as NotFound: the Pod is // TRACKED here, it has an instance id (one still inside Provision has nothing to read), @@ -907,20 +940,9 @@ func (h *Handler) GetContainerLogs( return nil, errdefs.NotFoundf("provider %q does not support container logs", h.prov.Name()) } - h.mu.Lock() - tp, tracked := h.tracked[key(namespace, podName)] - var instance string - if tracked { - instance = tp.instance - } - h.mu.Unlock() - - if !tracked { - return nil, errdefs.NotFoundf("pod %q is not known to virtual node %q", - key(namespace, podName), NodeName(h.prov.Name())) - } - if instance == "" { - return nil, errdefs.NotFoundf("pod %q has no external instance yet", key(namespace, podName)) + instance, err := h.instanceFor(namespace, podName) + if err != nil { + return nil, err } logf.FromContext(ctx).WithName("vnode-logs").V(1).Info("streaming container logs", @@ -937,16 +959,55 @@ func (h *Handler) GetContainerLogs( return kubeletLogStream(ctx, src, opts), nil } +// RunInContainer serves `kubectl exec` for a Pod on this virtual node: it starts cmd in +// the external instance and hands the streams to runExec, which owns the pumping. +// +// Same three preconditions as logs, each a NotFound: the Pod is tracked here, it has an +// instance, and the provider implements provider.Executor (exec is optional — a backend +// with no way into the box simply cannot serve it). +// +// containerName is IGNORED, as for logs: a Nebula Pod is one external instance, so there +// is no second container to pick. +func (h *Handler) RunInContainer( + ctx context.Context, namespace, podName, containerName string, cmd []string, attach vkapi.AttachIO, +) error { + executor, ok := h.prov.(provider.Executor) + if !ok { + return errdefs.NotFoundf("provider %q does not support exec", h.prov.Name()) + } + if len(cmd) == 0 { + return errdefs.InvalidInput("exec: no command given") + } + + instance, err := h.instanceFor(namespace, podName) + if err != nil { + return err + } + + logf.FromContext(ctx).WithName("vnode-exec").V(1).Info("running command in instance", + "provider", h.prov.Name(), "pod", key(namespace, podName), "container", containerName, + "instanceID", instance, "command", cmd, "tty", attach.TTY()) + + // Starting is bounded, running is not: a provider may wait for the instance to be + // ready (Modal polls for a task id for five minutes), and a client staring at a blank + // terminal that long is worse than being told the box is not ready. Only the start is + // capped — the command itself runs under ctx, for as long as the client stays. + startCtx, cancelStart := context.WithTimeout(ctx, execStartTimeout) + defer cancelStart() + + proc, err := executor.Exec(startCtx, instance, cmd, provider.ExecOptions{TTY: attach.TTY()}) + if err != nil { + return fmt.Errorf("start command in instance %s: %w", instance, err) + } + return runExec(ctx, proc, attach) +} + // --- Unused nodeutil.Provider surface -------------------------------------- // -// Exec/attach/stats/port-forward are out of v1 scope: beyond logs, Nebula does not proxy +// Attach/stats/port-forward are out of scope: beyond logs and exec, Nebula does not proxy // a workload's console. These satisfy the nodeutil.Provider interface and return NotFound // so the VK core reports them cleanly rather than panicking. -func (h *Handler) RunInContainer(context.Context, string, string, string, []string, vkapi.AttachIO) error { - return errdefs.NotFound("exec is not supported by the Nebula virtual node") -} - func (h *Handler) AttachToContainer(context.Context, string, string, string, vkapi.AttachIO) error { return errdefs.NotFound("attach is not supported by the Nebula virtual node") } diff --git a/pkg/vnode/kubelet.go b/pkg/vnode/kubelet.go index a3d8c01..ebec767 100644 --- a/pkg/vnode/kubelet.go +++ b/pkg/vnode/kubelet.go @@ -55,14 +55,29 @@ const certValidity = 365 * 24 * time.Hour // starts. Short on purpose: it avoids cutting a response mid-write, it is not a drain. const kubeletShutdownGrace = 5 * time.Second -// KubeletServer serves the one kubelet API route Nebula implements — container logs — -// for every provider's virtual node. +const ( + // execIdleTimeout is how long an exec survives without a sign of life FROM THE CLIENT. + // Not a limit on silence: client-go pings every 5s and any frame resets the timer, so a + // shell parked at a prompt stays up while kubectl is attached, and only a client that + // vanished (laptop closed, network gone) expires. Short, because expiry is what frees + // our streams and the connection behind them. + // + // Expiry does NOT stop the command — no provider can kill one — so a disconnected + // exec keeps running until the instance goes away. + execIdleTimeout = 5 * time.Minute + // execCreationTimeout bounds setting the streams up, before any command runs. + execCreationTimeout = 30 * time.Second +) + +// KubeletServer serves the kubelet API routes Nebula implements — container logs and exec +// — for every provider's virtual node. // -// Why it exists: `kubectl logs` is not a control-plane read. The API server proxies it to -// the kubelet of the Pod's node, at that Node's addresses and daemonEndpoints. A virtual -// node has no kubelet, so without a listener logs fail whatever the provider can serve. +// Why it exists: `kubectl logs` and `kubectl exec` are not control-plane reads. The API +// server proxies them to the kubelet of the Pod's node, at that Node's addresses and +// daemonEndpoints. A virtual node has no kubelet, so without a listener both fail whatever +// the provider can serve. // -// One listener for all nodes: the route carries only namespace/pod/container, so a request +// One listener for all nodes: the routes carry only namespace/pod/container, so a request // is resolved by asking each registered Handler whether it tracks that Pod — at most one // can. Cheaper than a port per provider, and than reading the Pod to learn its node. // @@ -74,7 +89,8 @@ const kubeletShutdownGrace = 5 * time.Second // Client certs are verified only when ClientCAPath is set. Off by default because which CA // signs the API server's kubelet client cert is not portable (kubeadm uses the cluster CA, // EKS/GKE their own), so requiring it would break `kubectl logs` on managed control -// planes. The cost: anything that can reach this port can read these Pods' logs — set the +// planes. The cost is now larger than logs: anything that can reach this port can also RUN +// COMMANDS in these Pods' instances, without passing through the API server's RBAC. Set the // CA if you can name it, else close the port with a NetworkPolicy. type KubeletServer struct { // addr is the listen address, e.g. ":10250". @@ -160,9 +176,17 @@ func (s *KubeletServer) Start(ctx context.Context) error { } mux := http.NewServeMux() - // Only GetContainerLogs is wired; the nil funcs make VK answer NotImplemented on - // exec/attach/portForward, which is the honest answer — see handler.go. - vkapi.AttachPodRoutes(vkapi.PodHandlerConfig{GetContainerLogs: s.getContainerLogs}, mux, false) + // Logs and exec are wired; the nil funcs make VK answer NotImplemented on + // attach/portForward, which is the honest answer — see handler.go. + vkapi.AttachPodRoutes(vkapi.PodHandlerConfig{ + GetContainerLogs: s.getContainerLogs, + RunInContainer: s.runInContainer, + // A real kubelet's --streaming-connection-idle-timeout, which an interactive shell + // needs: VK's own default is 30s, so `kubectl exec -it` would be cut off half a + // minute after the user stopped typing. + StreamIdleTimeout: execIdleTimeout, + StreamCreationTimeout: execCreationTimeout, + }, mux, false) srv := &http.Server{ Handler: mux, @@ -187,7 +211,7 @@ func (s *KubeletServer) Start(ctx context.Context) error { } }() - log.Info("serving kubelet api (container logs)", + log.Info("serving kubelet api (container logs, exec)", "addr", s.addr, "advertisedIP", s.nodeIP, "clientCertRequired", s.clientCAPath != "") // The cert and key are already in TLSConfig, hence the empty paths. if err := srv.ServeTLS(ln, "", ""); err != nil && !errors.Is(err, http.ErrServerClosed) { @@ -197,29 +221,43 @@ func (s *KubeletServer) Start(ctx context.Context) error { return nil } -// getContainerLogs finds the Handler tracking the Pod and streams from it. NotFound -// means "not my Pod", so the walk continues; any other error is a real failure to read -// logs we could have served, and is returned as-is rather than hidden behind NotFound. +// handlerFor finds the virtual node running this Pod. The routes carry only +// namespace/pod/container, so the owner is found by asking each registered Handler — at +// most one tracks a given Pod. nil means no node here runs it (the Pod is elsewhere, or +// leadership moved and the Runners have not re-adopted it yet). +func (s *KubeletServer) handlerFor(namespace, podName string) *Handler { + s.mu.RLock() + defer s.mu.RUnlock() + for _, h := range s.handlers { + if h.Tracks(namespace, podName) { + return h + } + } + return nil +} + +// getContainerLogs serves the containerLogs route from the Handler that owns the Pod. Its +// error is returned as-is: a provider that cannot read logs is a real failure, not an +// unknown Pod. func (s *KubeletServer) getContainerLogs( ctx context.Context, namespace, podName, containerName string, opts vkapi.ContainerLogOpts, ) (io.ReadCloser, error) { - s.mu.RLock() - handlers := make([]*Handler, 0, len(s.handlers)) - for _, h := range s.handlers { - handlers = append(handlers, h) + h := s.handlerFor(namespace, podName) + if h == nil { + return nil, errdefs.NotFoundf("no Nebula virtual node is running pod %q", key(namespace, podName)) } - s.mu.RUnlock() + return h.GetContainerLogs(ctx, namespace, podName, containerName, opts) +} - for _, h := range handlers { - rc, err := h.GetContainerLogs(ctx, namespace, podName, containerName, opts) - if err == nil { - return rc, nil - } - if !errdefs.IsNotFound(err) { - return nil, err - } +// runInContainer serves the exec route, the same way. +func (s *KubeletServer) runInContainer( + ctx context.Context, namespace, podName, containerName string, cmd []string, attach vkapi.AttachIO, +) error { + h := s.handlerFor(namespace, podName) + if h == nil { + return errdefs.NotFoundf("no Nebula virtual node is running pod %q", key(namespace, podName)) } - return nil, errdefs.NotFoundf("no Nebula virtual node is running pod %q", key(namespace, podName)) + return h.RunInContainer(ctx, namespace, podName, containerName, cmd, attach) } // tlsConfig: a fresh self-signed keypair, plus client verification if a CA is set. diff --git a/pkg/vnode/kubelet_test.go b/pkg/vnode/kubelet_test.go index 22e23af..4f099e1 100644 --- a/pkg/vnode/kubelet_test.go +++ b/pkg/vnode/kubelet_test.go @@ -17,9 +17,11 @@ limitations under the License. package vnode import ( + "bytes" "context" "crypto/tls" "crypto/x509" + "errors" "fmt" "io" "net" @@ -30,6 +32,9 @@ import ( "time" corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/remotecommand" + utilexec "k8s.io/utils/exec" ) func TestNewKubeletServer_Validation(t *testing.T) { @@ -127,6 +132,81 @@ func TestKubeletServer_ServesLogsOverTLS(t *testing.T) { } } +// The end-to-end shape of a `kubectl exec`: the API server's own SPDY client against the +// exec route. Covers what unit tests cannot — the route is attached, the streams are +// negotiated, and the command's output and exit code both survive the wire. +func TestKubeletServer_ServesExecOverTLS(t *testing.T) { + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("hi from exec\n", "", 0)) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + _, base := startTestKubeletServer(t, map[string]*Handler{"nebula-fake": h}) + + stdout, err := execCommand(t, base, "default", "p1", []string{"echo", "hi"}) + if err != nil { + t.Fatalf("exec: %v", err) + } + if stdout != "hi from exec\n" { + t.Fatalf("stdout = %q", stdout) + } + if ep.askedFor != "inst-1" { + t.Fatalf("provider asked for %q, want inst-1", ep.askedFor) + } + + // A Pod nobody tracks fails, rather than exec'ing into someone else's instance. + if _, err := execCommand(t, base, "default", "ghost", []string{"echo", "hi"}); err == nil { + t.Fatal("unknown pod: expected an error") + } +} + +// The exit code is the one thing an exec must not lose: a failed command has to look +// failed to the client, with its own status, not like a broken kubelet. +func TestKubeletServer_ExecReportsExitCode(t *testing.T) { + ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("", "boom\n", 3)) + h := NewHandler(ep, nil, nil) + if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + _, base := startTestKubeletServer(t, map[string]*Handler{"nebula-fake": h}) + + _, err := execCommand(t, base, "default", "p1", []string{"false"}) + var exitErr utilexec.ExitError + if !errors.As(err, &exitErr) { + t.Fatalf("err = %v (%T), want an exit error the client can read a code from", err, err) + } + if exitErr.ExitStatus() != 3 { + t.Fatalf("exit status = %d, want 3", exitErr.ExitStatus()) + } +} + +// execCommand runs one command through the exec route the way the API server does, +// returning what the command wrote to stdout. +func execCommand(t *testing.T, base, namespace, pod string, cmd []string) (string, error) { + t.Helper() + u, err := url.Parse(fmt.Sprintf("%s/exec/%s/%s/main", base, url.PathEscape(namespace), url.PathEscape(pod))) + if err != nil { + t.Fatalf("parse url: %v", err) + } + q := url.Values{"command": cmd, "output": {"1"}, "error": {"1"}} + u.RawQuery = q.Encode() + + // Self-signed, like a real kubelet's; the API server does not verify it either. + cfg := &rest.Config{Host: base, TLSClientConfig: rest.TLSClientConfig{Insecure: true}} + exec, err := remotecommand.NewSPDYExecutor(cfg, http.MethodPost, u) + if err != nil { + t.Fatalf("NewSPDYExecutor: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + var stdout, stderr bytes.Buffer + err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{Stdout: &stdout, Stderr: &stderr}) + return stdout.String(), err +} + // One listener serves every provider and the route carries no node name, so a request // must find the ONE Handler tracking that Pod. func TestKubeletServer_ResolvesAcrossProviders(t *testing.T) { diff --git a/pkg/vnode/logs.go b/pkg/vnode/logs.go index 8c43d25..0a5daef 100644 --- a/pkg/vnode/logs.go +++ b/pkg/vnode/logs.go @@ -19,7 +19,9 @@ package vnode import ( "bytes" "context" + "errors" "io" + "sync" "time" vkapi "github.com/virtual-kubelet/virtual-kubelet/node/api" @@ -69,7 +71,7 @@ const ( // // Ignored, because this seam cannot serve them: Timestamps (raw bytes, no per-line time), // Previous, SinceSeconds, SinceTime (no restart history, no time index). Ignored rather -// than rejected, so a habitual --since still prints the log. See docs/status.md. +// than rejected, so a habitual --since still prints the log. See docs/kubelet-api.md. // // Close tears down src and every goroutine below, and the VK log route always calls it, // including when a --follow client disconnects. @@ -90,6 +92,14 @@ func newLogStream( done := make(chan struct{}) chunks := make(chan []byte, 8) + // A provider stream that BROKE, kept for the close below. Ending on it instead of at + // EOF is what stops an outage from printing as an empty log — the VK route turns a + // non-nil error into an HTTP error while no bytes have gone out yet. + var ( + srcMu sync.Mutex + srcErr error + ) + // The only goroutine touching src. It cannot block forever on a full channel: // teardown closes done, and src.Close unblocks the Read. go func() { @@ -107,13 +117,30 @@ func newLogStream( } } if err != nil { + // EOF is the instance's log ending. Anything else is a failure worth + // reporting, unless we caused it by closing src in teardown. + if !errors.Is(err, io.EOF) { + select { + case <-done: + default: + srcMu.Lock() + srcErr = err + srcMu.Unlock() + } + } return } } }() go func() { - defer func() { _ = pw.Close() }() + defer func() { + srcMu.Lock() + err := srcErr + srcMu.Unlock() + // CloseWithError(nil) is a plain EOF, so this covers both endings. + _ = pw.CloseWithError(err) + }() copyLogs(ctx, pw, chunks, done, opts, t) }() diff --git a/pkg/vnode/logs_test.go b/pkg/vnode/logs_test.go index 9e55237..5ba6ec2 100644 --- a/pkg/vnode/logs_test.go +++ b/pkg/vnode/logs_test.go @@ -521,3 +521,80 @@ func waitFor(t *testing.T, cond func() bool, what string) { } t.Fatalf("timed out waiting for %s", what) } + +// --- stream failures -------------------------------------------------------- + +// failingLogSource yields s, then fails — a provider stream that died mid-poll. +type failingLogSource struct { + s string + err error + n int + closed bool +} + +func (f *failingLogSource) Read(p []byte) (int, error) { + if f.n < len(f.s) { + n := copy(p, f.s[f.n:]) + f.n += n + return n, nil + } + return 0, f.err +} + +func (f *failingLogSource) Close() error { f.closed = true; return nil } + +// A stream that BROKE must not read as the end of the log. Providers serve only recent +// output, so an empty result is ordinary — swallowing the error left an outage and a quiet +// workload looking identical, which is how a working `kubectl logs` was read as broken. +func TestKubeletLogStream_SourceFailureSurfaces(t *testing.T) { + src := &failingLogSource{s: "before the failure\n", err: errors.New("provider stream unavailable")} + rc := newLogStream(context.Background(), src, vkapi.ContainerLogOpts{Follow: true}, testTiming) + defer func() { _ = rc.Close() }() + + got, err := io.ReadAll(rc) + if err == nil { + t.Fatal("ReadAll: expected the stream failure to surface") + } + if !strings.Contains(err.Error(), "provider stream unavailable") { + t.Fatalf("err = %v, want the provider's own message", err) + } + // What arrived is still delivered: the error ends the log, it does not discard it. + if string(got) != "before the failure\n" { + t.Fatalf("logs = %q, want the output that preceded the failure", got) + } +} + +// EOF is the log ENDING, not a failure: an instance that exited must not make +// `kubectl logs` report an error over its final output. +func TestKubeletLogStream_EOFIsNotAnError(t *testing.T) { + src := &failingLogSource{s: "all of it\n", err: io.EOF} + rc := newLogStream(context.Background(), src, vkapi.ContainerLogOpts{Follow: true}, testTiming) + defer func() { _ = rc.Close() }() + + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v, want a clean EOF", err) + } + if string(got) != "all of it\n" { + t.Fatalf("logs = %q", got) + } +} + +// Closing is OUR teardown — a `kubectl logs -f` client hanging up. The read errors it +// causes must not be reported as a provider failure. +func TestKubeletLogStream_CloseIsNotAFailure(t *testing.T) { + src := newFakeLogSource("first\n") + rc := newLogStream(context.Background(), src, vkapi.ContainerLogOpts{Follow: true}, testTiming) + + if got := readN(t, rc, len("first\n")); got != "first\n" { + t.Fatalf("first read = %q", got) + } + if err := rc.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + // The reader is closed either way; what matters is that nothing reports the closed + // source as an outage. + if _, err := io.ReadAll(rc); err != nil && !errors.Is(err, io.ErrClosedPipe) { + t.Fatalf("err = %v, want teardown to end the stream cleanly", err) + } +}