diff --git a/.github/workflows/pr-ci.yml b/.github/workflows/pr-ci.yml index a9439b3ca3..d3843b938e 100644 --- a/.github/workflows/pr-ci.yml +++ b/.github/workflows/pr-ci.yml @@ -723,6 +723,15 @@ jobs: test-timeout: 1200s job-timeout-minutes: 25 + - label: up-provider-microsandbox-logs-cancel + runner: ubuntu-latest + free-disk-space: false + install-kind: false + requires-secret: false + install-microsandbox: true + test-timeout: 1200s + job-timeout-minutes: 25 + # Snapshot tests - label: snapshot diff --git a/e2e/e2e_suite_test.go b/e2e/e2e_suite_test.go index 87d3355917..abe2f0af1c 100644 --- a/e2e/e2e_suite_test.go +++ b/e2e/e2e_suite_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/devsy-org/devsy/cmd" "github.com/devsy-org/devsy/e2e/framework" // Register tests. _ "github.com/devsy-org/devsy/e2e/tests/build" @@ -43,6 +44,15 @@ import ( "github.com/onsi/gomega" ) +func TestMain(m *testing.M) { + // The external runtime self-execs this test binary as the real supervisor. + if len(os.Args) > 2 && os.Args[1] == "internal" && os.Args[2] == "runtime-supervisor" { + cmd.Execute() + return + } + os.Exit(m.Run()) +} + // TestRunE2ETests checks configuration parameters (specified through flags) and then runs // E2E tests using the Ginkgo runner. // If a "report directory" is specified, one or more JUnit test reports will be diff --git a/e2e/tests/up/provider_microsandbox_logs_cancel.go b/e2e/tests/up/provider_microsandbox_logs_cancel.go new file mode 100644 index 0000000000..41cc08ca24 --- /dev/null +++ b/e2e/tests/up/provider_microsandbox_logs_cancel.go @@ -0,0 +1,591 @@ +package up + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "time" + + "al.essio.dev/pkg/shellescape" + "github.com/devsy-org/devsy/e2e/framework" + "github.com/devsy-org/devsy/pkg/agent" + pkgconfig "github.com/devsy-org/devsy/pkg/config" + "github.com/devsy-org/devsy/pkg/driver" + "github.com/devsy-org/devsy/pkg/driver/drivercreate" + "github.com/devsy-org/devsy/pkg/encoding" + "github.com/devsy-org/devsy/pkg/provider" + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" +) + +const ( + microsandboxCancelBound = 10 * time.Second + microsandboxCancelSentinel = "source-sentinel" + microsandboxCancelShell = "/bin/sh" +) + +type microsandboxLogCancelState struct { + Status string `json:"status"` +} + +// The child blocks on its own FIFO; the parent either waits for it or reads the +// caller's stdin. Neither process has a self-exit timer. The nonce is in both +// argv vectors, and readiness includes each process's kernel start time. +const microsandboxCancelScript = `set -eu +nonce=$1 +mode=$2 +start_time() { sed 's/^.*) //' "/proc/$1/stat" | awk '{print $20}'; } +if [ "$mode" = child ]; then + fifo="/root/$nonce.fifo" + mkfifo "$fifo" + exec 9<>"$fifo" + printf 'READY %s child %s %s\n' "$nonce" "$$" "$(start_time $$)" + read -r ignored <&9 + exit 19 +fi +/bin/sh -c "$3" "$nonce-child" "$nonce" child & +child=$! +printf 'READY %s parent %s %s\n' "$nonce" "$$" "$(start_time $$)" +if [ "$mode" = stdin ]; then + read -r ignored + exit 20 +fi +wait "$child" +` + +type microsandboxLogCancelFixture struct { + framework *framework.Framework + workspace string + sandbox string + createdAt string + bootID string + nonce string + uid string + label string + runtime driver.ArgvExecDriver +} + +var _ = ginkgo.Describe("microsandbox logs and cancellation parity", + ginkgo.Label("up-provider-microsandbox-logs-cancel"), func() { + for _, selection := range []struct{ name, source, runtime string }{ + {"builtin-logs-cancel", microsandboxResourceBuiltin, provider.MicrosandboxDriver}, + {"external-logs-cancel", microsandboxExternalProvider, provider.ExternalDriver}, + } { + ginkgo.Context(selection.name, func() { + var fixture *microsandboxLogCancelFixture + ginkgo.BeforeEach(func(ctx context.Context) { + fixture = newMicrosandboxLogCancelFixture(ctx, selection.name, selection.source) + fixture.loadRuntime(ctx, selection.runtime) + }) + ginkgo.It( + "returns actual stdout and stderr workload logs while running and stopped", + func(ctx context.Context) { fixture.assertLogs(ctx) }, + ginkgo.SpecTimeout(framework.TimeoutLong()), + ) + ginkgo.It("cancels a ready Exec parent and child without replacing the VM", + func(ctx context.Context) { fixture.assertCancellation(ctx, false) }, + ginkgo.SpecTimeout(framework.TimeoutLong()), + ) + ginkgo.It("cancels a ready Exec with blocked owned stdin without replacing the VM", + func(ctx context.Context) { fixture.assertCancellation(ctx, true) }, + ginkgo.SpecTimeout(framework.TimeoutLong()), + ) + }) + } + }) + +func newMicrosandboxLogCancelFixture( + ctx context.Context, name, source string, +) *microsandboxLogCancelFixture { + skipIfNoMicrosandbox(ctx) + ginkgo.GinkgoT().Setenv("DEVSY_HOME", ginkgo.GinkgoT().TempDir()) + ginkgo.GinkgoT().Setenv("DEVSY_CONFIG", "") + initialDir, err := os.Getwd() + framework.ExpectNoError(err) + f := framework.NewDefaultFramework(filepath.Join(initialDir, "bin")) + workspace, err := framework.CreateTempDir() + framework.ExpectNoError(err) + ginkgo.DeferCleanup(framework.CleanupTempDir, initialDir, workspace) + var random [16]byte + _, err = rand.Read(random[:]) + framework.ExpectNoError(err) + fixture := µsandboxLogCancelFixture{ + framework: f, workspace: workspace, + nonce: "exec-parity-" + hex.EncodeToString(random[:]), + uid: encoding.CreateNewUID("", filepath.Base(workspace)), + } + fixture.sandbox = "devsy-" + fixture.uid + fixture.label = "devsy.logs-cancel-parity=" + fixture.uid + assertMicrosandboxEphemeralAbsent(ctx, fixture.label) + out, err := microsandboxResourceCommand(ctx, "list", "--quiet") + framework.ExpectNoError(err) + gomega.Expect(strings.Fields(string(out))).NotTo(gomega.ContainElement(fixture.sandbox)) + framework.ExpectNoError(f.DevsyProviderAdd(ctx, source, "--name", name)) + fixture.registerCleanup(name) + framework.ExpectNoError(f.DevsyProviderUse(ctx, name)) + writeMicrosandboxResourceConfig(workspace, nil) + framework.ExpectNoError(os.WriteFile( + filepath.Join(workspace, microsandboxCancelSentinel), []byte(fixture.nonce), 0o600, + )) + fixture.up(ctx) + gomega.Expect(microsandboxResourceSandbox(ctx, f, workspace)).To(gomega.Equal(fixture.sandbox)) + fixture.createdAt = microsandboxCreationTime(ctx, fixture.sandbox) + fixture.bootID = microsandboxStorageBootID(ctx, f, workspace) + return fixture +} + +func (f *microsandboxLogCancelFixture) registerCleanup(name string) { + ginkgo.DeferCleanup(func() error { + cleanup, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + return f.framework.DevsyProviderDelete(cleanup, name) + }) + // Registered only after proving the fresh UID/name/label absent. A failed + // startup can remove its Devsy record before deleting the owned backend VM. + ginkgo.DeferCleanup(func() error { + cleanup, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + out, err := microsandboxResourceCommand(cleanup, "list", "--quiet") + if err != nil { + return err + } + for sandbox := range strings.FieldsSeq(string(out)) { + if sandbox != f.sandbox { + continue + } + out, err = microsandboxResourceCommand(cleanup, "remove", "--force", sandbox) + if err != nil { + return fmt.Errorf("remove exact test-owned sandbox %s: %w: %s", sandbox, err, out) + } + } + return nil + }) + ginkgo.DeferCleanup(f.framework.CleanupWorkspace, f.workspace) +} + +func (f *microsandboxLogCancelFixture) up(ctx context.Context) { + framework.ExpectNoError(f.framework.DevsyUp(ctx, f.workspace, + "--uid", f.uid, "--id-label", f.label, "--devcontainer", ".devcontainer.json")) +} + +func (f *microsandboxLogCancelFixture) loadRuntime(ctx context.Context, expected string) { + workspace, err := f.framework.FindWorkspace(ctx, f.workspace) + framework.ExpectNoError(err) + origin, err := provider.GetWorkspaceAgentDir(pkgconfig.DefaultContext, workspace.ID) + framework.ExpectNoError(err) + info, err := agent.ParseAgentWorkspaceInfo(filepath.Join(origin, provider.WorkspaceConfigFile)) + framework.ExpectNoError(err) + gomega.Expect(info.Workspace).NotTo(gomega.BeNil()) + gomega.Expect(string(info.Agent.Local)).To(gomega.Equal("true")) + gomega.Expect(info.Agent.Driver).To(gomega.Equal(expected)) + gomega.Expect(info.Origin).To(gomega.Equal(origin)) + gomega.Expect(info.Workspace.UID).To(gomega.Equal(workspace.UID)) + gomega.Expect(info.Workspace.UID).To(gomega.Equal(f.uid)) + gomega.Expect(info.Workspace.UID).NotTo(gomega.BeEmpty()) + if expected == provider.ExternalDriver { + binaries, resolveErr := agent.GetAgentBinariesDirFromWorkspaceDir(origin) + framework.ExpectNoError(resolveErr) + _, resolveErr = provider.ResolveExternalRuntimeBinary(info.Agent, binaries) + framework.ExpectNoError(resolveErr) + } + bundle, err := drivercreate.New(ctx, info) + framework.ExpectNoError(err) + var ok bool + f.runtime, ok = bundle.Runtime.(driver.ArgvExecDriver) + gomega.Expect(ok).To(gomega.BeTrue(), "selected runtime must implement literal argv Exec") + f.uid = info.Workspace.UID +} + +func (f *microsandboxLogCancelFixture) assertLogs(ctx context.Context) { + outMarker, errMarker := f.nonce+"-stdout", f.nonce+"-stderr" + // Earlier Exec traffic can leave binary log bytes without a line boundary. + command, cancel := context.WithTimeout(ctx, 30*time.Second) + err := f.runtime.CommandContainerArgv(command, f.uid, + []string{ + microsandboxCancelShell, "-c", "printf '\\n%s\\n' \"$1\"; printf '\\n%s\\n' \"$2\" >&2", + f.nonce, outMarker, errMarker, + }, + driver.Streams{Stdout: io.Discard, Stderr: io.Discard}) + cancel() + framework.ExpectNoError(err) + f.assertLogMarkers(ctx, outMarker, errMarker) + f.assertUnchanged(ctx) + framework.ExpectNoError(f.framework.DevsyStop(ctx, f.workspace)) + assertMicrosandboxEphemeralStopped(ctx, f.sandbox, f.createdAt) + f.assertLogMarkers(ctx, outMarker, errMarker) + assertMicrosandboxEphemeralStopped(ctx, f.sandbox, f.createdAt) + f.up(ctx) + gomega.Expect(microsandboxCreationTime(ctx, f.sandbox)).To(gomega.Equal(f.createdAt)) + gomega.Expect(microsandboxStorageBootID(ctx, f.framework, f.workspace)). + NotTo(gomega.Equal(f.bootID)) + f.assertSourceAndCommand(ctx) +} + +func (f *microsandboxLogCancelFixture) assertLogMarkers(ctx context.Context, markers ...string) { + logsCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + out, err := f.framework.DevsyLogs(logsCtx, f.workspace) + framework.ExpectNoError(err) + lines := strings.Split(strings.ReplaceAll(out, "\r\n", "\n"), "\n") + for _, marker := range markers { + gomega.Expect(lines).To(gomega.ContainElement(marker)) + } +} + +type microsandboxExecIdentity struct { + Role string + PID int + Start string +} + +type microsandboxExecReadiness struct { + mu sync.Mutex + pending string + nonce string + frames chan microsandboxExecIdentity + seen []microsandboxExecIdentity +} + +func (r *microsandboxExecReadiness) Write(p []byte) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.pending += string(p) + if len(r.pending) > 4096 { + return 0, fmt.Errorf("unexpectedly large Exec readiness output") + } + for { + line, rest, found := strings.Cut(r.pending, "\n") + if !found { + return len(p), nil + } + r.pending = rest + identity, err := parseMicrosandboxExecIdentity(line, r.nonce) + if err != nil { + return 0, err + } + r.seen = append(r.seen, identity) + select { + case r.frames <- identity: + default: + return 0, fmt.Errorf("too many Exec readiness frames") + } + } +} + +func parseMicrosandboxExecIdentity(line, nonce string) (microsandboxExecIdentity, error) { + fields := strings.Fields(line) + if len(fields) != 5 || fields[0] != "READY" || fields[1] != nonce { + return microsandboxExecIdentity{}, fmt.Errorf("unexpected Exec readiness frame %q", line) + } + pid, err := strconv.Atoi(fields[3]) + if err != nil || pid <= 1 { + return microsandboxExecIdentity{}, fmt.Errorf("invalid Exec PID in %q", line) + } + if _, err = strconv.ParseUint(fields[4], 10, 64); err != nil { + return microsandboxExecIdentity{}, fmt.Errorf("invalid Exec start time in %q", line) + } + return microsandboxExecIdentity{Role: fields[2], PID: pid, Start: fields[4]}, nil +} + +func (r *microsandboxExecReadiness) snapshot() []microsandboxExecIdentity { + r.mu.Lock() + defer r.mu.Unlock() + return append([]microsandboxExecIdentity(nil), r.seen...) +} + +type microsandboxOwnedInput struct { + *io.PipeReader + entered chan struct{} + closed chan struct{} + readOnce sync.Once + closeOnce sync.Once +} + +func (r *microsandboxOwnedInput) Read(p []byte) (int, error) { + r.readOnce.Do(func() { close(r.entered) }) + return r.PipeReader.Read(p) +} + +func (r *microsandboxOwnedInput) Close() error { + err := r.PipeReader.Close() + r.closeOnce.Do(func() { close(r.closed) }) + return err +} + +type microsandboxExecOperation struct { + cancel context.CancelFunc + done chan struct{} + err error + input *microsandboxOwnedInput + writer *io.PipeWriter +} + +func (f *microsandboxLogCancelFixture) startExec( + ctx context.Context, blocked bool, readiness *microsandboxExecReadiness, +) *microsandboxExecOperation { + commandCtx, cancel := context.WithCancel(ctx) + op := µsandboxExecOperation{cancel: cancel, done: make(chan struct{})} + mode := "wait" + var input io.Reader + if blocked { + reader, writer := io.Pipe() + op.input = µsandboxOwnedInput{ + PipeReader: reader, entered: make(chan struct{}), closed: make(chan struct{}), + } + op.writer, input, mode = writer, op.input, "stdin" + } + ginkgo.DeferCleanup(op.cleanup) + go func() { + defer close(op.done) + op.err = f.runtime.CommandContainerArgv(commandCtx, f.uid, + []string{ + microsandboxCancelShell, "-c", microsandboxCancelScript, f.nonce, + f.nonce, mode, microsandboxCancelScript, + }, + driver.Streams{Stdin: input, Stdout: readiness, Stderr: io.Discard}) + }() + return op +} + +func (op *microsandboxExecOperation) cleanup() { + op.cancel() + if op.input != nil { + _ = op.input.Close() + _ = op.writer.Close() + } + select { + case <-op.done: + case <-time.After(microsandboxCancelBound): + ginkgo.Fail("owned Exec goroutine failed to join after cancellation and input cleanup") + } +} + +func (f *microsandboxLogCancelFixture) assertCancellation(ctx context.Context, blocked bool) { + ready := µsandboxExecReadiness{ + nonce: f.nonce, + frames: make(chan microsandboxExecIdentity, 2), + } + op := f.startExec(ctx, blocked, ready) + ginkgo.DeferCleanup(func() { f.cleanupProcesses(ready.snapshot()) }) + identities := f.awaitReadiness(ctx, ready, op) + observer, cancelObserver := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + alive, err := f.processesAlive(observer, identities) + cancelObserver() + framework.ExpectNoError(err) + gomega.Expect(alive). + To(gomega.Equal(len(identities)), "both exact identities must be live before cancellation") + + // Both bounds start at the cancellation event, not after waiting for caller exit. + deathCtx, cancelDeath := context.WithTimeout( + context.WithoutCancel(ctx), + microsandboxCancelBound, + ) + defer cancelDeath() + op.cancel() + select { + case <-op.done: + gomega.Expect(op.err). + To(gomega.HaveOccurred(), "cancellation must not look like successful completion") + case <-deathCtx.Done(): + ginkgo.Fail("ready Runtime Exec did not return within 10s of cancellation") + } + if blocked { + select { + case <-op.input.closed: + default: + ginkgo.Fail("Runtime Exec returned without closing its blocked owned input") + } + _, err = op.writer.Write([]byte("late producer")) + gomega.Expect(err).To(gomega.MatchError(io.ErrClosedPipe)) + } + f.awaitProcessDeath(deathCtx, identities) + f.assertUnchanged(ctx) +} + +func (f *microsandboxLogCancelFixture) awaitReadiness( + ctx context.Context, ready *microsandboxExecReadiness, op *microsandboxExecOperation, +) []microsandboxExecIdentity { + readinessCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + identities := make([]microsandboxExecIdentity, 0, 2) + roles := make(map[string]bool) + for len(identities) < 2 { + select { + case identity := <-ready.frames: + gomega.Expect(identity.Role).To(gomega.BeElementOf("parent", "child")) + gomega.Expect(roles[identity.Role]).To(gomega.BeFalse(), "duplicate readiness role") + roles[identity.Role] = true + identities = append(identities, identity) + case <-op.done: + ginkgo.Fail(fmt.Sprintf("Exec exited before readiness: %v", op.err)) + case <-readinessCtx.Done(): + ginkgo.Fail("guest parent and child were not ready within 30s") + } + } + gomega.Expect(identities[0].PID).NotTo(gomega.Equal(identities[1].PID)) + op.awaitInputRead(readinessCtx) + return identities +} + +func (op *microsandboxExecOperation) awaitInputRead(ctx context.Context) { + if op.input == nil { + return + } + select { + case <-op.input.entered: + case <-op.done: + ginkgo.Fail(fmt.Sprintf("Exec exited before stdin Read entered: %v", op.err)) + case <-ctx.Done(): + ginkgo.Fail("host input Read was not entered within the readiness bound") + } +} + +// The observer is a separate SSH operation, never the canceled Exec session. +// Identity replacement or a zombie means the original process is not live; +// unexpected procfs or SSH errors fail, rather than being interpreted as death. +func (f *microsandboxLogCancelFixture) processesAlive( + ctx context.Context, identities []microsandboxExecIdentity, +) (int, error) { + out, err := f.framework.DevsySSHOnce(ctx, f.workspace, f.identityCommand(identities, false)) + if err != nil { + return 0, err + } + count, err := strconv.Atoi(strings.TrimSpace(out)) + if err != nil { + return 0, fmt.Errorf("invalid process observer response %q: %w", out, err) + } + if count < 0 || count > len(identities) { + return 0, fmt.Errorf("unexpected live process count %d", count) + } + return count, nil +} + +func (f *microsandboxLogCancelFixture) identityCommand( + identities []microsandboxExecIdentity, cleanup bool, +) string { + var command strings.Builder + command.WriteString("set -eu; alive=0; ") + for _, identity := range identities { + fmt.Fprintf( + &command, + "p=%d; expected=%s; ", + identity.PID, + shellescape.Quote(identity.Start), + ) + command.WriteString(`if [ -d "/proc/$p" ]; then ` + + `if stat=$(cat "/proc/$p/stat"); then ` + + `stat=${stat##*) }; set -- $stat; state=$1; shift 19; actual=$1; ` + + `if [ "$actual" = "$expected" ] && [ "$state" != Z ] && [ "$state" != X ]; then ` + + `if cmdline=$(tr '\000' ' ' < "/proc/$p/cmdline"); then `) + fmt.Fprintf(&command, "case \"$cmdline\" in *%s*) ", shellescape.Quote(f.nonce)) + if cleanup { + command.WriteString(`kill -KILL "$p" || { [ ! -d "/proc/$p" ] || exit 31; }; `) + } else { + command.WriteString("alive=$((alive + 1)); ") + } + if cleanup { + command.WriteString(";; *) : ;; esac; ") + } else { + command.WriteString(`;; *) +if latest=$(cat "/proc/$p/stat"); then + latest=${latest##*) }; set -- $latest; latest_state=$1; shift 19 + if [ "$1" = "$expected" ] && [ "$latest_state" != Z ] && [ "$latest_state" != X ]; then exit 34; fi +else [ ! -d "/proc/$p" ] || exit 35; fi +;; esac; `) + } + command.WriteString(`else [ ! -d "/proc/$p" ] || exit 32; fi; fi; ` + + `else [ ! -d "/proc/$p" ] || exit 33; fi; fi; `) + } + command.WriteString("printf '%s\\n' \"$alive\"") + return command.String() +} + +func (f *microsandboxLogCancelFixture) awaitProcessDeath( + ctx context.Context, identities []microsandboxExecIdentity, +) { + ticker := time.NewTicker(200 * time.Millisecond) + defer ticker.Stop() + for { + alive, err := f.processesAlive(ctx, identities) + framework.ExpectNoError(err) + if alive == 0 { + return + } + select { + case <-ticker.C: + case <-ctx.Done(): + ginkgo.Fail( + "original guest process identities remained live after the 10s cancellation bound", + ) + } + } +} + +func (f *microsandboxLogCancelFixture) cleanupProcesses(identities []microsandboxExecIdentity) { + if len(identities) == 0 { + return + } + ctx, cancel := context.WithTimeout(context.Background(), microsandboxCancelBound) + defer cancel() + _, err := f.framework.DevsySSHOnce(ctx, f.workspace, f.identityCommand(identities, true)) + framework.ExpectNoError(err) +} + +func (f *microsandboxLogCancelFixture) assertUnchanged(ctx context.Context) { + out, err := microsandboxResourceCommand( + ctx, + "inspect", + f.sandbox, + "--format", + microsandboxResourceJSON, + ) + framework.ExpectNoError(err) + var state microsandboxLogCancelState + framework.ExpectNoError(json.Unmarshal(out, &state)) + gomega.Expect(state.Status).To(gomega.Equal("Running")) + gomega.Expect(microsandboxCreationTime(ctx, f.sandbox)).To(gomega.Equal(f.createdAt)) + gomega.Expect(microsandboxStorageBootID(ctx, f.framework, f.workspace)). + To(gomega.Equal(f.bootID)) + f.assertSourceAndCommand(ctx) +} + +func (f *microsandboxLogCancelFixture) assertSourceAndCommand(ctx context.Context) { + command, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + var out strings.Builder + err := f.runtime.CommandContainerArgv( + command, + f.uid, + []string{ + microsandboxCancelShell, + "-c", + "printf '%s\\n' \"$1\"", + f.nonce, + f.nonce + "-next", + }, + driver.Streams{Stdout: &out, Stderr: io.Discard}, + ) + framework.ExpectNoError(err) + gomega.Expect(out.String()).To(gomega.Equal(f.nonce + "-next\n")) + data, err := os.ReadFile(filepath.Join(f.workspace, microsandboxCancelSentinel)) + framework.ExpectNoError(err) + gomega.Expect(string(data)).To(gomega.Equal(f.nonce)) + guestPath := filepath.Join( + "/workspaces", + filepath.Base(f.workspace), + microsandboxCancelSentinel, + ) + out.Reset() + err = f.runtime.CommandContainerArgv(command, f.uid, []string{"cat", guestPath}, + driver.Streams{Stdout: &out, Stderr: io.Discard}) + framework.ExpectNoError(err) + gomega.Expect(out.String()).To(gomega.Equal(f.nonce)) +} diff --git a/pkg/agent/delivery/exec_stream_cancellation_unix_test.go b/pkg/agent/delivery/exec_stream_cancellation_unix_test.go new file mode 100644 index 0000000000..9967166970 --- /dev/null +++ b/pkg/agent/delivery/exec_stream_cancellation_unix_test.go @@ -0,0 +1,261 @@ +//go:build linux || darwin + +package delivery + +import ( + "context" + "io" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/devsy-org/devsy/pkg/driver" + "github.com/devsy-org/devsy/pkg/driver/microsandbox" + "github.com/devsy-org/devsy/pkg/provider" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const execStreamCancellationTestBound = 3 * time.Second + +type cancellationBinary struct { + io.Reader + closer io.Closer + entered chan struct{} + once sync.Once + closes atomic.Int32 +} + +func (s *cancellationBinary) Read(p []byte) (int, error) { + s.once.Do(func() { close(s.entered) }) + return s.Reader.Read(p) +} + +func (s *cancellationBinary) Close() error { + s.closes.Add(1) + if s.closer != nil { + return s.closer.Close() + } + return nil +} + +type cancellationBinarySource struct { + mu sync.Mutex + blocked int + sources []*cancellationBinary + writers []*io.PipeWriter + acquired chan *cancellationBinary +} + +func (s *cancellationBinarySource) acquire(ctx context.Context, _ string) (io.ReadCloser, error) { + s.mu.Lock() + defer s.mu.Unlock() + // Cleanup cancels before taking this mutex; a concurrent retry must not + // allocate a new blocked reader after cleanup has closed the existing ones. + if err := ctx.Err(); err != nil { + return nil, err + } + source := &cancellationBinary{entered: make(chan struct{})} + if len(s.sources) < s.blocked { + reader, writer := io.Pipe() + source.Reader, source.closer = reader, reader + s.writers = append(s.writers, writer) + } else { + source.Reader = strings.NewReader("fresh-source-binary") + } + s.sources = append(s.sources, source) + s.acquired <- source + return source, nil +} + +func (s *cancellationBinarySource) cleanup() { + s.mu.Lock() + defer s.mu.Unlock() + for _, source := range s.sources { + if source.closer != nil { + _ = source.closer.Close() + } + } + for _, writer := range s.writers { + _ = writer.Close() + } +} + +func (s *cancellationBinarySource) assertOwnership(t *testing.T, expected int) { + t.Helper() + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.sources, expected, "every retry must acquire a fresh source") + for index, source := range s.sources { + select { + case <-source.entered: + default: + t.Errorf("source %d was acquired but never read", index) + } + assert.EqualValues(t, 1, source.closes.Load(), "source %d close count", index) + if index > 0 { + assert.NotSame(t, s.sources[index-1], source, "retry reused the previous source") + } + } +} + +func cancellationDeliveryRuntime(t *testing.T, blockCompletion bool) driver.ArgvExecDriver { + t.Helper() + dir := t.TempDir() + script := "#!/bin/sh\nexec /bin/cat\n" + if blockCompletion { + // A stage's small finite source drains into the pipe, but its process + // remains alive. CommandContext must kill it on the real completion timer. + script = "#!/bin/sh\ncase \"$*\" in *\"cat >\"*) exec /bin/sleep 60;; *) exit 0;; esac\n" + } + // #nosec G306 -- Fixed fake executable in the test's private temporary directory. + require.NoError(t, os.WriteFile(filepath.Join(dir, "msb"), []byte(script), 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + runtime, err := microsandbox.NewMicrosandboxDriver( + context.Background(), + &provider.AgentWorkspaceInfo{}, + ) + require.NoError(t, err) + argvRuntime, ok := runtime.(driver.ArgvExecDriver) + require.True(t, ok) + return argvRuntime +} + +func cancellationDeliveryExecutor( + t *testing.T, runtime driver.ArgvExecDriver, canceled chan<- error, +) func(context.Context, []string, driver.Streams) error { + return func(ctx context.Context, argv []string, streams driver.Streams) error { + if streams.Stdin != nil { + callbackDone := make(chan struct{}) + stop := context.AfterFunc(ctx, func() { + defer close(callbackDone) + select { + case canceled <- context.Cause(ctx): + default: + t.Error("unexpected extra attempt cancellation") + } + }) + defer func() { + if !stop() { + select { + case <-callbackDone: + case <-time.After(execStreamCancellationTestBound): + t.Error("attempt observer callback did not join") + } + } + }() + } + return runtime.CommandContainerArgv(ctx, "owned-delivery-test", argv, streams) + } +} + +type cancellationDeliveryCall struct { + done chan struct{} + cancel context.CancelFunc + err error +} + +func startCancellationDelivery( + t *testing.T, delivery *KubernetesDelivery, sources *cancellationBinarySource, +) (context.Context, *cancellationDeliveryCall) { + t.Helper() + outer, cancel := context.WithCancel(context.Background()) + call := &cancellationDeliveryCall{done: make(chan struct{}), cancel: cancel} + t.Cleanup(func() { + cancel() + sources.cleanup() + select { + case <-call.done: + case <-time.After(execStreamCancellationTestBound): + t.Error("delivery failed to join after explicit source cleanup") + } + }) + go func() { + defer close(call.done) + // The fake msb never interprets this stage destination or writes to it. + call.err = delivery.deliverViaExecStream( + outer, + "/tmp/owned-delivery-never-written", + PostStartOptions{ + Arch: "amd64", SkipVersionCheck: true, BinarySource: sources.acquire, + }, + ) + }() + return outer, call +} + +func awaitDeliveryRead(t *testing.T, sources *cancellationBinarySource) { + t.Helper() + select { + case source := <-sources.acquired: + select { + case <-source.entered: + case <-time.After(execStreamCancellationTestBound): + t.Fatal("actual source Read not entered") + } + case <-time.After(execStreamCancellationTestBound): + t.Fatal("source not acquired") + } +} + +func TestExecStreamAttemptCancellationOwnsSource(t *testing.T) { + for _, testCase := range []struct { + name string + blockedSources int + blockCompletion bool + cause error + result error + acquisitions int + }{ + {"retry acquires fresh source", 1, false, errExecStreamIdleTimeout, nil, 2}, + {"idle retries exhausted", 2, false, errExecStreamIdleTimeout, errExecStreamIdleTimeout, 2}, + {"completion retries exhausted", 0, true, errExecStreamCompletionTimeout, errExecStreamCompletionTimeout, 2}, + {"normal completion", 0, false, nil, nil, 1}, + } { + t.Run(testCase.name, func(t *testing.T) { + runtime := cancellationDeliveryRuntime(t, testCase.blockCompletion) + sources := &cancellationBinarySource{ + blocked: testCase.blockedSources, + acquired: make(chan *cancellationBinary, execStreamMaxAttempts), + } + canceled := make(chan error, execStreamMaxAttempts) + delivery := &KubernetesDelivery{ + execStreamIdleTimeout: 300 * time.Millisecond, + execStreamCompletionTimeout: 300 * time.Millisecond, + Exec: cancellationDeliveryExecutor(t, runtime, canceled), + } + outer, call := startCancellationDelivery(t, delivery, sources) + awaitDeliveryRead(t, sources) + if testCase.cause != nil { + select { + case cause := <-canceled: + require.ErrorIs(t, cause, testCase.cause) + case <-time.After(execStreamCancellationTestBound): + t.Fatal("actual attempt cancellation was not observed") + } + } + select { + case <-call.done: + case <-time.After(execStreamCancellationTestBound): + t.Fatal("delivery remained blocked after actual attempt cancellation") + } + if testCase.result == nil { + require.NoError(t, call.err) + } else { + require.ErrorIs( + t, + call.err, + testCase.result, + "original attempt cause must survive cleanup and retry", + ) + } + require.NoError(t, outer.Err(), "attempt timeout must not cancel the outer operation") + call.cancel() + sources.assertOwnership(t, testCase.acquisitions) + }) + } +} diff --git a/pkg/agent/delivery/kubernetes.go b/pkg/agent/delivery/kubernetes.go index 2ad79924d5..3b1ef0574e 100644 --- a/pkg/agent/delivery/kubernetes.go +++ b/pkg/agent/delivery/kubernetes.go @@ -203,9 +203,10 @@ func (d *KubernetesDelivery) deliverViaExecStream( if err != nil { return &permanentDeliveryError{fmt.Errorf("acquire binary: %w", err)} } - defer func() { _ = binary.Close() }() + owned := &execStreamSource{ReadCloser: binary} + defer func() { _ = owned.Close() }() - streamErr := d.execStreamOnce(ctx, destPath, binary, opts.SkipVersionCheck) + streamErr := d.execStreamOnce(ctx, destPath, owned, opts.SkipVersionCheck) if streamErr != nil && isTransientDeliveryError(streamErr) && attempt < execStreamMaxAttempts { log.Warnf( @@ -222,6 +223,18 @@ func (d *KubernetesDelivery) deliverViaExecStream( return err } +// execStreamSource joins cancel-time closure with the acquiring owner's cleanup. +type execStreamSource struct { + io.ReadCloser + closeOnce sync.Once + closeErr error +} + +func (s *execStreamSource) Close() error { + s.closeOnce.Do(func() { s.closeErr = s.ReadCloser.Close() }) + return s.closeErr +} + type permanentDeliveryError struct{ err error } func (e *permanentDeliveryError) Error() string { return e.err.Error() } @@ -241,6 +254,18 @@ func (d *KubernetesDelivery) execStreamOnce( attemptCtx, cancel := context.WithCancelCause(ctx) defer cancel(nil) + if source, ok := binary.(io.Closer); ok { + closed := make(chan struct{}) + stop := context.AfterFunc(attemptCtx, func() { + _ = source.Close() + close(closed) + }) + defer func() { + if !stop() { + <-closed + } + }() + } progress := newProgressReader(binary) monitorDone := make(chan struct{}) monitorStop := make(chan struct{}) diff --git a/pkg/driver/microsandbox/cliclient.go b/pkg/driver/microsandbox/cliclient.go index 4c0230b314..799f017aa2 100644 --- a/pkg/driver/microsandbox/cliclient.go +++ b/pkg/driver/microsandbox/cliclient.go @@ -153,6 +153,8 @@ func (cliClient) Remove(ctx context.Context, sandbox string) error { // Exec uses --stream (byte-faithful stdio, no PTY) because the agent binary // injection and the SSH-over-stdio tunnel require it; a PTY stalls the tunnel. +// Cancellation closes non-file ReadCloser input; Close must unblock Read. +// File input and input after normal completion remain caller-owned. func (cliClient) Exec(ctx context.Context, sandbox string, req execRequest) error { args := []string{"exec", "--stream"} if req.User != "" { @@ -169,7 +171,27 @@ func (cliClient) Exec(ctx context.Context, sandbox string, req execRequest) erro cmd.Stdin = req.Stdin cmd.Stdout = req.Stdout cmd.Stderr = req.Stderr - return cmd.Run() + input, closable := req.Stdin.(io.ReadCloser) + _, file := req.Stdin.(*os.File) + if !closable || file { + return cmd.Run() + } + if err := cmd.Start(); err != nil { + return err + } + // Non-file input is delegated for cancellation only. Keep the callback + // alive while Wait drains stdin, even if the child has already exited. + closed := make(chan struct{}) + stop := context.AfterFunc(ctx, func() { + _ = input.Close() + close(closed) + }) + defer func() { + if !stop() { + <-closed + } + }() + return cmd.Wait() } func (cliClient) Logs(ctx context.Context, sandbox string, w io.Writer) error { diff --git a/pkg/driver/microsandbox/exec_cancellation_unix_test.go b/pkg/driver/microsandbox/exec_cancellation_unix_test.go new file mode 100644 index 0000000000..9dd09a88e8 --- /dev/null +++ b/pkg/driver/microsandbox/exec_cancellation_unix_test.go @@ -0,0 +1,248 @@ +//go:build linux || darwin + +package microsandbox + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const execCancellationTestBound = 3 * time.Second + +type cancellationInput struct { + io.Reader + closer func() error + entered chan struct{} + readOnce sync.Once + closes atomic.Int32 + closeEntered chan struct{} + closeGate <-chan struct{} +} + +func (r *cancellationInput) Read(p []byte) (int, error) { + if r.entered != nil { + r.readOnce.Do(func() { close(r.entered) }) + } + return r.Reader.Read(p) +} + +func (r *cancellationInput) Close() error { + r.closes.Add(1) + if r.closeEntered != nil { + close(r.closeEntered) + } + if r.closeGate != nil { + <-r.closeGate + } + if r.closer != nil { + return r.closer() + } + return nil +} + +func cancellationMSB(t *testing.T, exitImmediately bool) string { + t.Helper() + dir := t.TempDir() + ready := filepath.Join(dir, "ready") + script := "#!/bin/sh\nprintf '%s\\n' \"$$\" > \"$DEVSY_CANCEL_READY\"\n" + if exitImmediately { + script += "exit 0\n" + } else { + script += "exec /bin/cat\n" + } + // #nosec G306 -- This executable is a fixed test-owned fake in a private temporary directory. + require.NoError(t, os.WriteFile(filepath.Join(dir, "msb"), []byte(script), 0o700)) + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("DEVSY_CANCEL_READY", ready) + return ready +} + +func awaitCancellationPID(t *testing.T, path string) int { + t.Helper() + var pid int + require.Eventually(t, func() bool { + data, err := os.ReadFile(path) // #nosec G304 -- Only the test-owned readiness file is read. + if err != nil { + return false + } + pid, err = strconv.Atoi(strings.TrimSpace(string(data))) + return err == nil && pid > 1 + }, execCancellationTestBound, time.Millisecond, "child readiness absent") + return pid +} + +func awaitCancellationSignal(t *testing.T, signal <-chan struct{}, message string) { + t.Helper() + select { + case <-signal: + case <-time.After(execCancellationTestBound): + t.Fatal(message) + } +} + +type cancellationCall struct { + cancel context.CancelFunc + done chan struct{} + err error +} + +func startCancellationCall(t *testing.T, input io.Reader, cleanup func()) *cancellationCall { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + call := &cancellationCall{cancel: cancel, done: make(chan struct{})} + t.Cleanup(func() { + cancel() + if cleanup != nil { + cleanup() + } + select { + case <-call.done: + case <-time.After(execCancellationTestBound): + t.Error("Exec failed to join after explicit input cleanup") + } + }) + go func() { + defer close(call.done) + call.err = (cliClient{}).Exec(ctx, "owned-cancellation-test", execRequest{ + Stdin: input, Stdout: io.Discard, Stderr: io.Discard, + }) + }() + return call +} + +func TestExecCancellationClosesBlockedInput(t *testing.T) { + for _, testCase := range []struct { + name string + exitImmediately bool + }{ + {name: "running child"}, + {name: "child exited while stdin copy remains blocked", exitImmediately: true}, + } { + t.Run(testCase.name, func(t *testing.T) { + ready := cancellationMSB(t, testCase.exitImmediately) + reader, writer := io.Pipe() + input := &cancellationInput{ + Reader: reader, + closer: reader.Close, + entered: make(chan struct{}), + } + call := startCancellationCall(t, input, func() { + _ = reader.Close() + _ = writer.Close() + }) + awaitCancellationSignal(t, input.entered, "source Read not entered") + pid := awaitCancellationPID(t, ready) + if testCase.exitImmediately { + require.Eventually(t, func() bool { + return errors.Is(syscall.Kill(pid, 0), syscall.ESRCH) + }, execCancellationTestBound, time.Millisecond, "child did not exit before cancellation") + } + call.cancel() + awaitCancellationSignal(t, call.done, "Exec stayed blocked after cancellation") + if !testCase.exitImmediately { + require.Error(t, call.err, "cancellation must not report successful execution") + } + assert.EqualValues(t, 1, input.closes.Load()) + }) + } +} + +func TestExecCompletionPreservesInputOwnership(t *testing.T) { + cancellationMSB(t, false) + input := &cancellationInput{Reader: strings.NewReader("complete")} + call := startCancellationCall(t, input, nil) + awaitCancellationSignal(t, call.done, "normal execution did not finish") + require.NoError(t, call.err) + call.cancel() + assert.Never( + t, + func() bool { return input.closes.Load() != 0 }, + 50*time.Millisecond, + time.Millisecond, + "normal completion must disarm the cancellation callback", + ) +} + +func TestExecCancellationPreservesCallerFile(t *testing.T) { + ready := cancellationMSB(t, false) + reader, writer, err := os.Pipe() + require.NoError(t, err) + call := startCancellationCall(t, reader, func() { + _ = reader.Close() + _ = writer.Close() + }) + awaitCancellationPID(t, ready) + call.cancel() + awaitCancellationSignal(t, call.done, "file-backed Exec cancellation did not return") + require.Error(t, call.err) + _, err = reader.Stat() + assert.NoError(t, err, "the caller retains ownership of its file") +} + +func TestExecStartFailurePreservesInput(t *testing.T) { + dir := t.TempDir() + // #nosec G306 -- Intentionally executable fixture with an invalid interpreter, in a private directory. + require.NoError( + t, + os.WriteFile(filepath.Join(dir, "msb"), []byte("#!/nonexistent-interpreter\n"), 0o700), + ) + t.Setenv("PATH", dir) + input := &cancellationInput{Reader: strings.NewReader("unused")} + call := startCancellationCall(t, input, nil) + awaitCancellationSignal(t, call.done, "failed Start did not return") + require.Error(t, call.err) + call.cancel() + assert.Zero(t, input.closes.Load(), "failed Start must not transfer input ownership") +} + +func TestExecAlreadyCanceledPreservesInput(t *testing.T) { + cancellationMSB(t, false) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + input := &cancellationInput{Reader: strings.NewReader("unused")} + err := (cliClient{}).Exec(ctx, "owned-cancellation-test", execRequest{Stdin: input}) + require.ErrorIs(t, err, context.Canceled) + assert.Zero(t, input.closes.Load(), "pre-start cancellation must not transfer input ownership") +} + +func TestExecCancellationJoinsInputClose(t *testing.T) { + ready := cancellationMSB(t, false) + reader, writer := io.Pipe() + gate := make(chan struct{}) + var release sync.Once + input := &cancellationInput{ + Reader: reader, closer: reader.Close, entered: make(chan struct{}), + closeEntered: make(chan struct{}), closeGate: gate, + } + call := startCancellationCall(t, input, func() { + release.Do(func() { close(gate) }) + _ = reader.Close() + _ = writer.Close() + }) + awaitCancellationPID(t, ready) + awaitCancellationSignal(t, input.entered, "source Read not entered") + call.cancel() + awaitCancellationSignal(t, input.closeEntered, "cancellation did not invoke Close") + select { + case <-call.done: + t.Fatal("Exec returned before its Close callback completed") + default: + } + release.Do(func() { close(gate) }) + awaitCancellationSignal(t, call.done, "Exec did not join the completed Close callback") + require.Error(t, call.err) + assert.EqualValues(t, 1, input.closes.Load()) +} diff --git a/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx b/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx index e2e470e2f7..87adb7f93f 100644 --- a/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx +++ b/sites/docs-devsy-sh/content/docs/developing-providers/runtime-protocol.mdx @@ -119,6 +119,14 @@ This suite requires Linux with KVM and public access to the base and Dockerless DEVSY_REQUIRE_MICROSANDBOX=true task cli:test:e2e:suite -- up-provider-microsandbox-dockerless ``` -These scenarios do not establish complete parity. Logs, cancellation, and runtime compatibility failures still require coverage before replacing the built-in provider. Green lifecycle, image, mount, resource, persistent-storage, ephemeral-root, TCP-egress, prebuild, and Dockerless checks alone do not authorize that cutover. Dockerless refusal checks cover fresh creation, not preservation after an invalid recreation request. Egress coverage does not establish UDP, DNS, TLS, ingress, or live-policy modification behavior. +The `up-provider-microsandbox-logs-cancel` label checks completed stdout/stderr workload logs while running and stopped, direct runtime Exec cancellation of a ready parent and child, and cancellation with blocked owned stdin against each provider. A separate SSH observer verifies process identity using PID, kernel start time, and a unique command marker. Caller return and process death must both occur within ten seconds of cancellation; the VM's creation time, boot ID, and source files must remain intact. Logs do not start a stopped VM. These checks do not establish log ordering, binary log fidelity, TTY cancellation, or cancellation of arbitrary nonclosable input. + +```sh +DEVSY_REQUIRE_MICROSANDBOX=true task cli:test:e2e:suite -- up-provider-microsandbox-logs-cancel +``` + +For built-in Exec, cancellation closes non-file `io.ReadCloser` input whose `Close` unblocks `Read`; ordinary completion and caller-owned file input retain their ownership. Agent binary delivery owns its acquired source and closes it when the individual transfer attempt is canceled, before retrying with a fresh source. Cleanup joins cancellation callbacks before returning. + +These scenarios do not establish complete parity. Runtime compatibility failures and invalid-recreate preservation still require coverage before replacing the built-in provider. Green lifecycle, image, mount, resource, persistent-storage, ephemeral-root, TCP-egress, prebuild, Dockerless, logs, and cancellation checks alone do not authorize that cutover. Dockerless refusal checks cover fresh creation, not preservation after an invalid recreation request. Egress coverage does not establish UDP, DNS, TLS, ingress, or live-policy modification behavior. Prebuild invalidation coverage changes the Dockerfile; it does not independently establish hashing of local Feature script contents.