feat(compose): support stdin_open and tty (#419)

* feat: support stdin_open and tty

This can be useful to leave a container running without specifying a
command like `sleep`, and probably in other situations as well.

Signed-off-by: Miek Gieben <miek@miek.nl>

* Complete full spec test case

Signed-off-by: Miek Gieben <miek@miek.nl>

* fix container log streaming for containers with TTY

---------

Signed-off-by: Miek Gieben <miek@miek.nl>
Co-authored-by: Pasha Sviderski <me@psviderski.name>

internal/machine/docker/server.go

@@ -562,7 +562,9 @@ func (s *Server) CreateServiceContainer(
 			api.LabelServiceMode: spec.Mode,
 			api.LabelManaged:     "",
 		},
-		User: spec.Container.User,
+		User:      spec.Container.User,
+		Tty:       spec.Container.Tty,
+		OpenStdin: spec.Container.OpenStdin,
 	}
 	if spec.Mode == "" {
 		config.Labels[api.LabelServiceMode] = api.ServiceModeReplicated

internal/machine/docker/service.go

@@ -1,12 +1,14 @@
 package docker
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"strconv"
 	"strings"
@@ -233,11 +235,17 @@ func (s *Service) ListImages(ctx context.Context, opts image.ListOptions) (Image
 	return imagesResp, nil
 }
 
-// ContainerLogs streams logs from a container and returns demultiplexed entries via a channel.
+// ContainerLogs streams logs from a container and returns entries via a channel.
 // The channel is closed when streaming completes or context is cancelled.
 func (s *Service) ContainerLogs(
 	ctx context.Context, containerID string, opts api.ServiceLogsOptions,
 ) (<-chan api.LogEntry, error) {
+	ctr, err := s.Client.ContainerInspect(ctx, containerID)
+	if err != nil {
+		return nil, fmt.Errorf("inspect container '%s': %w", containerID, err)
+	}
+	isTTY := ctr.Config != nil && ctr.Config.Tty
+
 	dockerOpts := container.LogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
@@ -257,25 +265,31 @@ func (s *Service) ContainerLogs(
 	stdoutWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: false}
 	stderrWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: true}
 
-	// Wrap the context in a cancellable one to unblock the second goroutine below when StdCopy completes.
+	// Wrap the context in a cancellable one to unblock the second goroutine when log copying completes.
 	ctx, cancel := context.WithCancel(ctx)
 
-	// Run StdCopy in a goroutine to be able to handle context cancellation.
+	// Copy logs in a goroutine to be able to handle context cancellation.
 	go func() {
 		defer close(outCh)
 		defer cancel()
 
-		// StdCopy is blocking and will return when the reader is closed in another goroutine below or on error.
+		// Docker returns raw stdout for TTY containers and multiplexed stdout/stderr otherwise.
-		if _, err := stdcopy.StdCopy(stdoutWriter, stderrWriter, reader); err != nil {
+		var err error
+		if isTTY {
+			_, err = copyRawContainerLogs(stdoutWriter, reader)
+		} else {
+			_, err = stdcopy.StdCopy(stdoutWriter, stderrWriter, reader)
+		}
+		if err != nil {
 			// Send error as the last entry.
 			select {
-			case outCh <- api.LogEntry{Err: fmt.Errorf("demultiplex container logs: %w", err)}:
+			case outCh <- api.LogEntry{Err: fmt.Errorf("copy container logs: %w", err)}:
 			case <-ctx.Done():
 			}
 		}
 	}()
 
-	// Close the reader when the context is done to cancel StdCopy if it's still running.
+	// Close the reader when the context is done to cancel log copying if it's still running.
 	go func() {
 		<-ctx.Done()
 		reader.Close()
@@ -284,7 +298,32 @@ func (s *Service) ContainerLogs(
 	return outCh, nil
 }
 
-// logsChannelWriter is a writer for stdcopy.StdCopy that sends demultiplexed container logs to a channel.
+// copyRawContainerLogs copies a raw TTY log stream one line at a time so each write produces one log entry.
+func copyRawContainerLogs(dst io.Writer, src io.Reader) (written int64, _ error) {
+	reader := bufio.NewReader(src)
+	for {
+		line, readErr := reader.ReadBytes('\n')
+		if len(line) > 0 {
+			n, writeErr := dst.Write(line)
+			written += int64(n)
+			if writeErr != nil {
+				return written, writeErr
+			}
+			if n != len(line) {
+				return written, io.ErrShortWrite
+			}
+		}
+
+		if readErr != nil {
+			if errors.Is(readErr, io.EOF) {
+				return written, nil
+			}
+			return written, readErr
+		}
+	}
+}
+
+// logsChannelWriter sends container log writes to a channel.
 type logsChannelWriter struct {
 	ctx      context.Context
 	ch       chan<- api.LogEntry

internal/machine/docker/service_test.go

@@ -0,0 +1,122 @@
+package docker
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/docker/docker/api/types/container"
+	dockerclient "github.com/docker/docker/client"
+	"github.com/docker/docker/pkg/stdcopy"
+	"github.com/psviderski/uncloud/pkg/api"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestServiceContainerLogs(t *testing.T) {
+	t.Parallel()
+
+	const (
+		containerID = "container-id"
+		firstLog    = "2025-01-01T00:00:00.000000000Z first message\n"
+		secondLog   = "2025-01-01T00:00:01.000000000Z second message\n"
+	)
+
+	var multiplexedLogs bytes.Buffer
+	_, err := stdcopy.NewStdWriter(&multiplexedLogs, stdcopy.Stdout).Write([]byte(firstLog))
+	require.NoError(t, err)
+	_, err = stdcopy.NewStdWriter(&multiplexedLogs, stdcopy.Stderr).Write([]byte(secondLog))
+	require.NoError(t, err)
+
+	tests := []struct {
+		name     string
+		tty      bool
+		logs     []byte
+		streams  []api.LogStreamType
+		messages []string
+	}{
+		{
+			name:     "TTY raw stream",
+			tty:      true,
+			logs:     []byte(firstLog + secondLog),
+			streams:  []api.LogStreamType{api.LogStreamStdout, api.LogStreamStdout},
+			messages: []string{"first message\n", "second message\n"},
+		},
+		{
+			name:     "non-TTY multiplexed stream",
+			logs:     multiplexedLogs.Bytes(),
+			streams:  []api.LogStreamType{api.LogStreamStdout, api.LogStreamStderr},
+			messages: []string{"first message\n", "second message\n"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			dockerClient := newLogsTestClient(t, tt.tty, tt.logs)
+			service := NewService(dockerClient, nil)
+
+			logsCh, err := service.ContainerLogs(context.Background(), containerID, api.ServiceLogsOptions{})
+			require.NoError(t, err)
+
+			var entries []api.LogEntry
+			for entry := range logsCh {
+				require.NoError(t, entry.Err)
+				entries = append(entries, entry)
+			}
+
+			require.Len(t, entries, len(tt.messages))
+			for i := range entries {
+				assert.Equal(t, tt.streams[i], entries[i].Stream)
+				assert.Equal(t, tt.messages[i], string(entries[i].Message))
+				assert.False(t, entries[i].Timestamp.IsZero())
+			}
+		})
+	}
+}
+
+func newLogsTestClient(t *testing.T, tty bool, logs []byte) *dockerclient.Client {
+	t.Helper()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.HasSuffix(r.URL.Path, "/containers/container-id/json"):
+			w.Header().Set("Content-Type", "application/json")
+			if err := json.NewEncoder(w).Encode(container.InspectResponse{
+				ContainerJSONBase: &container.ContainerJSONBase{ID: "container-id"},
+				Config:            &container.Config{Tty: tty},
+			}); err != nil {
+				t.Errorf("encode inspect response: %v", err)
+			}
+		case strings.HasSuffix(r.URL.Path, "/containers/container-id/logs"):
+			w.Header().Set("Content-Type", "application/vnd.docker.raw-stream")
+			if _, err := w.Write(logs); err != nil {
+				t.Errorf("write logs response: %v", err)
+			}
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	t.Cleanup(server.Close)
+
+	serverURL, err := url.Parse(server.URL)
+	require.NoError(t, err)
+
+	dockerClient, err := dockerclient.NewClientWithOpts(
+		dockerclient.WithHost("tcp://"+serverURL.Host),
+		dockerclient.WithHTTPClient(server.Client()),
+		dockerclient.WithVersion("1.48"),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, dockerClient.Close())
+	})
+
+	return dockerClient
+}

pkg/api/service.go

@@ -265,6 +265,11 @@ type ContainerSpec struct {
 	LogDriver *LogDriver
 	// PidMode sets the PID namespace mode for the container. Currently only "" or "host" is supported.
 	PidMode string
+	// Tty allocates a pseudo-TTY and connects the container's standard streams to it.
+	// Standard output and standard error share one stream.
+	Tty bool
+	// OpenStdin allocates standard input and keeps it open.
+	OpenStdin bool
 	// Privileged gives extended privileges to the container. This is a security risk and should be used with caution.
 	Privileged bool
 	// PullPolicy determines when to pull the image from the registry or use the image already available in the cluster.

pkg/client/compose/service.go

@@ -54,6 +54,8 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser
 			Image:       service.Image,
 			Init:        service.Init,
 			PidMode:     service.Pid,
+			Tty:         service.Tty,
+			OpenStdin:   service.StdinOpen,
 			Privileged:  service.Privileged,
 			PullPolicy:  pullPolicy,
 			Resources:   resourcesFromCompose(service),

pkg/client/compose/service_test.go

@@ -128,6 +128,8 @@ func TestServiceSpecFromCompose(t *testing.T) {
 							},
 						},
 						PidMode:    "host",
+						Tty:        true,
+						OpenStdin:  true,
 						Privileged: true,
 						PullPolicy: api.PullPolicyAlways,
 						Resources: api.ContainerResources{

pkg/client/compose/testdata/compose-full-spec.yaml

@@ -7,6 +7,8 @@ services:
     command: ["nginx", "updated", "command"]
     cpus: 0.5
     pid: host
+    tty: true
+    stdin_open: true
     deploy:
       update_config:
         order: stop-first

pkg/client/deploy/container_test.go

@@ -66,6 +66,26 @@ func TestEvalContainerSpecChange_ContainerPidMode(t *testing.T) {
 	assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(newSpec, currentSpec))
 }
 
+func TestEvalContainerSpecChange_ContainerTty(t *testing.T) {
+	t.Parallel()
+
+	currentSpec := api.ServiceSpec{
+		Container: api.ContainerSpec{
+			Image: "nginx:latest",
+		},
+	}
+	newSpec := api.ServiceSpec{
+		Container: api.ContainerSpec{
+			Image:     "nginx:latest",
+			Tty:       true,
+			OpenStdin: true,
+		},
+	}
+
+	assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(currentSpec, newSpec))
+	assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(newSpec, currentSpec))
+}
+
 func TestEvalContainerSpecChange_ContainerResources(t *testing.T) {
 	t.Parallel()
 

website/docs/8-compose-file-reference/1-support-matrix.md

@@ -45,9 +45,11 @@ If you rely on a specific Compose feature that is not supported by Uncloud, plea
 | `secrets`                        | ⚠️ Limited         | Reference secrets in `environment`, see [Secrets](../3-concepts/8-secrets.md). File mounts not supported                                   |
 | `security_opt`                   | ❌ Not supported    |                                                                                                                                            |
 | `shm_size`                       | ✅ Supported        | Shared memory size                                                                                                                         |
+| `stdin_open`                     | ✅ Supported        | Allocate standard input and keep it open                                                                                                   |
 | `stop_grace_period`              | ✅ Supported        | Time to wait after SIGTERM before SIGKILL                                                                                                  |
 | `storage_opt`                    | ❌ Not supported    |                                                                                                                                            |
 | `sysctls`                        | ✅ Supported        | Namespaced kernel parameters                                                                                                               |
+| `tty`                            | ✅ Supported        | Allocate a pseudo-TTY and connect the container's standard streams to it                                                                   |
 | `ulimits`                        | ✅ Supported        | Resource limits                                                                                                                            |
 | `user`                           | ✅ Supported        | Set container user                                                                                                                         |
 | `volumes`                        | ✅ Supported        | Named volumes, bind mounts, tmpfs                                                                                                          |