diff --git a/internal/machine/docker/server.go b/internal/machine/docker/server.go index e6e61e1c..e29e7b55 100644 --- a/internal/machine/docker/server.go +++ b/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 diff --git a/internal/machine/docker/service.go b/internal/machine/docker/service.go index 8ac14b36..e20f95a9 100644 --- a/internal/machine/docker/service.go +++ b/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. - if _, err := stdcopy.StdCopy(stdoutWriter, stderrWriter, reader); err != nil { + // Docker returns raw stdout for TTY containers and multiplexed stdout/stderr otherwise. + 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 diff --git a/internal/machine/docker/service_test.go b/internal/machine/docker/service_test.go new file mode 100644 index 00000000..414eb807 --- /dev/null +++ b/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 +} diff --git a/pkg/api/service.go b/pkg/api/service.go index ae98bfc9..a57ee47b 100644 --- a/pkg/api/service.go +++ b/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. diff --git a/pkg/client/compose/service.go b/pkg/client/compose/service.go index dc9839ab..492f435b 100644 --- a/pkg/client/compose/service.go +++ b/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), diff --git a/pkg/client/compose/service_test.go b/pkg/client/compose/service_test.go index bd6c28d0..079e9224 100644 --- a/pkg/client/compose/service_test.go +++ b/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{ diff --git a/pkg/client/compose/testdata/compose-full-spec.yaml b/pkg/client/compose/testdata/compose-full-spec.yaml index 19c643b0..528b0c2d 100644 --- a/pkg/client/compose/testdata/compose-full-spec.yaml +++ b/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 diff --git a/pkg/client/deploy/container_test.go b/pkg/client/deploy/container_test.go index e3b7e54a..664aebdd 100644 --- a/pkg/client/deploy/container_test.go +++ b/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() diff --git a/website/docs/8-compose-file-reference/1-support-matrix.md b/website/docs/8-compose-file-reference/1-support-matrix.md index 47caebaf..d6f3fa52 100644 --- a/website/docs/8-compose-file-reference/1-support-matrix.md +++ b/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 |