mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 19:13:34 +00:00
Compare commits
2
Commits
351698c280
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac56754281 | ||
|
|
b7e224a1ef |
@@ -562,7 +562,9 @@ func (s *Server) CreateServiceContainer(
|
|||||||
api.LabelServiceMode: spec.Mode,
|
api.LabelServiceMode: spec.Mode,
|
||||||
api.LabelManaged: "",
|
api.LabelManaged: "",
|
||||||
},
|
},
|
||||||
User: spec.Container.User,
|
User: spec.Container.User,
|
||||||
|
Tty: spec.Container.Tty,
|
||||||
|
OpenStdin: spec.Container.OpenStdin,
|
||||||
}
|
}
|
||||||
if spec.Mode == "" {
|
if spec.Mode == "" {
|
||||||
config.Labels[api.LabelServiceMode] = api.ServiceModeReplicated
|
config.Labels[api.LabelServiceMode] = api.ServiceModeReplicated
|
||||||
|
|||||||
@@ -1,12 +1,14 @@
|
|||||||
package docker
|
package docker
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bufio"
|
||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -233,11 +235,17 @@ func (s *Service) ListImages(ctx context.Context, opts image.ListOptions) (Image
|
|||||||
return imagesResp, nil
|
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.
|
// The channel is closed when streaming completes or context is cancelled.
|
||||||
func (s *Service) ContainerLogs(
|
func (s *Service) ContainerLogs(
|
||||||
ctx context.Context, containerID string, opts api.ServiceLogsOptions,
|
ctx context.Context, containerID string, opts api.ServiceLogsOptions,
|
||||||
) (<-chan api.LogEntry, error) {
|
) (<-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{
|
dockerOpts := container.LogsOptions{
|
||||||
ShowStdout: true,
|
ShowStdout: true,
|
||||||
ShowStderr: true,
|
ShowStderr: true,
|
||||||
@@ -257,25 +265,31 @@ func (s *Service) ContainerLogs(
|
|||||||
stdoutWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: false}
|
stdoutWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: false}
|
||||||
stderrWriter := &logsChannelWriter{ctx: ctx, ch: outCh, isStderr: true}
|
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)
|
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() {
|
go func() {
|
||||||
defer close(outCh)
|
defer close(outCh)
|
||||||
defer cancel()
|
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.
|
// Send error as the last entry.
|
||||||
select {
|
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():
|
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() {
|
go func() {
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
reader.Close()
|
reader.Close()
|
||||||
@@ -284,7 +298,32 @@ func (s *Service) ContainerLogs(
|
|||||||
return outCh, nil
|
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 {
|
type logsChannelWriter struct {
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
ch chan<- api.LogEntry
|
ch chan<- api.LogEntry
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -265,6 +265,11 @@ type ContainerSpec struct {
|
|||||||
LogDriver *LogDriver
|
LogDriver *LogDriver
|
||||||
// PidMode sets the PID namespace mode for the container. Currently only "" or "host" is supported.
|
// PidMode sets the PID namespace mode for the container. Currently only "" or "host" is supported.
|
||||||
PidMode string
|
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 gives extended privileges to the container. This is a security risk and should be used with caution.
|
||||||
Privileged bool
|
Privileged bool
|
||||||
// PullPolicy determines when to pull the image from the registry or use the image already available in the cluster.
|
// PullPolicy determines when to pull the image from the registry or use the image already available in the cluster.
|
||||||
|
|||||||
@@ -54,6 +54,8 @@ func ServiceSpecFromCompose(project *types.Project, serviceName string) (api.Ser
|
|||||||
Image: service.Image,
|
Image: service.Image,
|
||||||
Init: service.Init,
|
Init: service.Init,
|
||||||
PidMode: service.Pid,
|
PidMode: service.Pid,
|
||||||
|
Tty: service.Tty,
|
||||||
|
OpenStdin: service.StdinOpen,
|
||||||
Privileged: service.Privileged,
|
Privileged: service.Privileged,
|
||||||
PullPolicy: pullPolicy,
|
PullPolicy: pullPolicy,
|
||||||
Resources: resourcesFromCompose(service),
|
Resources: resourcesFromCompose(service),
|
||||||
|
|||||||
@@ -128,6 +128,8 @@ func TestServiceSpecFromCompose(t *testing.T) {
|
|||||||
},
|
},
|
||||||
},
|
},
|
||||||
PidMode: "host",
|
PidMode: "host",
|
||||||
|
Tty: true,
|
||||||
|
OpenStdin: true,
|
||||||
Privileged: true,
|
Privileged: true,
|
||||||
PullPolicy: api.PullPolicyAlways,
|
PullPolicy: api.PullPolicyAlways,
|
||||||
Resources: api.ContainerResources{
|
Resources: api.ContainerResources{
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ services:
|
|||||||
command: ["nginx", "updated", "command"]
|
command: ["nginx", "updated", "command"]
|
||||||
cpus: 0.5
|
cpus: 0.5
|
||||||
pid: host
|
pid: host
|
||||||
|
tty: true
|
||||||
|
stdin_open: true
|
||||||
deploy:
|
deploy:
|
||||||
update_config:
|
update_config:
|
||||||
order: stop-first
|
order: stop-first
|
||||||
|
|||||||
@@ -66,6 +66,26 @@ func TestEvalContainerSpecChange_ContainerPidMode(t *testing.T) {
|
|||||||
assert.Equal(t, ContainerNeedsRecreate, EvalContainerSpecChange(newSpec, currentSpec))
|
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) {
|
func TestEvalContainerSpecChange_ContainerResources(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -57,6 +57,12 @@ curl -fsS https://get.uncloud.run/install.sh | VERSION=nightly sh
|
|||||||
|
|
||||||
More information about nightly builds of the CLI and daemon can be found [here](https://github.com/psviderski/uncloud/releases/tag/nightly).
|
More information about nightly builds of the CLI and daemon can be found [here](https://github.com/psviderski/uncloud/releases/tag/nightly).
|
||||||
|
|
||||||
|
## mise
|
||||||
|
You can install uncloud with [mise](https://mise.jdx.dev/):
|
||||||
|
```
|
||||||
|
mise use github:psviderski/uncloud[exe=uc]
|
||||||
|
```
|
||||||
|
|
||||||
## GitHub download (macOS, Linux)
|
## GitHub download (macOS, Linux)
|
||||||
|
|
||||||
You can manually download and use a pre-built binary from the
|
You can manually download and use a pre-built binary from the
|
||||||
|
|||||||
@@ -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 |
|
| `secrets` | ⚠️ Limited | Reference secrets in `environment`, see [Secrets](../3-concepts/8-secrets.md). File mounts not supported |
|
||||||
| `security_opt` | ❌ Not supported | |
|
| `security_opt` | ❌ Not supported | |
|
||||||
| `shm_size` | ✅ Supported | Shared memory size |
|
| `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 |
|
| `stop_grace_period` | ✅ Supported | Time to wait after SIGTERM before SIGKILL |
|
||||||
| `storage_opt` | ❌ Not supported | |
|
| `storage_opt` | ❌ Not supported | |
|
||||||
| `sysctls` | ✅ Supported | Namespaced kernel parameters |
|
| `sysctls` | ✅ Supported | Namespaced kernel parameters |
|
||||||
|
| `tty` | ✅ Supported | Allocate a pseudo-TTY and connect the container's standard streams to it |
|
||||||
| `ulimits` | ✅ Supported | Resource limits |
|
| `ulimits` | ✅ Supported | Resource limits |
|
||||||
| `user` | ✅ Supported | Set container user |
|
| `user` | ✅ Supported | Set container user |
|
||||||
| `volumes` | ✅ Supported | Named volumes, bind mounts, tmpfs |
|
| `volumes` | ✅ Supported | Named volumes, bind mounts, tmpfs |
|
||||||
|
|||||||
Reference in New Issue
Block a user