logs: stream corrosion logs with 'uc machine logs corrosion' from container instead of journal (fixes #392)

This commit is contained in:
Pasha Sviderski
2026-06-17 18:28:25 +10:00
parent 247154eeb3
commit 7bb3cb9682
11 changed files with 91 additions and 83 deletions
+26 -25
View File
@@ -3,11 +3,12 @@ package machine
import ( import (
"context" "context"
"fmt" "fmt"
"slices"
"strings"
"github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli"
"github.com/psviderski/uncloud/internal/cli/completion" "github.com/psviderski/uncloud/internal/cli/completion"
"github.com/psviderski/uncloud/internal/cli/logs" "github.com/psviderski/uncloud/internal/cli/logs"
"github.com/psviderski/uncloud/internal/journal"
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
"github.com/psviderski/uncloud/pkg/client" "github.com/psviderski/uncloud/pkg/client"
"github.com/spf13/cobra" "github.com/spf13/cobra"
@@ -19,14 +20,14 @@ func NewLogsCommand() *cobra.Command {
cmd := &cobra.Command{ cmd := &cobra.Command{
Use: "logs [SERVICE...]", Use: "logs [SERVICE...]",
Aliases: []string{"log"}, Aliases: []string{"log"},
Short: "View systemd service logs.", Short: "View system service logs.",
Long: `View logs from the specified systemd service(s) across all machines in the cluster. Long: `View logs from the specified system service(s) across all machines in the cluster.
Use -m to restrict to specific machines. Use -m to restrict to specific machines.
Supported services: Supported services:
uncloud the Uncloud daemon corrosion the Corrosion distributed state store
docker the Docker daemon docker the Docker daemon
uncloud-corrosion the Corrosion distributed state store uncloud the Uncloud daemon
If no services are specified, streams logs from the uncloud service.`, If no services are specified, streams logs from the uncloud service.`,
Example: ` # View recent logs for the uncloud service. Example: ` # View recent logs for the uncloud service.
@@ -37,7 +38,7 @@ If no services are specified, streams logs from the uncloud service.`,
uc machine logs -f uncloud uc machine logs -f uncloud
# View logs from multiple services. # View logs from multiple services.
uc machine logs uncloud docker uncloud-corrosion uc machine logs uncloud docker corrosion
# Show last 20 lines per machine (default is 100). # Show last 20 lines per machine (default is 100).
uc machine logs -n 20 docker uc machine logs -n 20 docker
@@ -49,7 +50,7 @@ If no services are specified, streams logs from the uncloud service.`,
uc machine logs --since 3h --until 1h30m docker uc machine logs --since 3h --until 1h30m docker
# View logs only from specific machines. # View logs only from specific machines.
uc machine logs -m machine1,machine2 uncloud uncloud-corrosion`, uc machine logs -m machine1,machine2 uncloud corrosion`,
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
uncli := cmd.Context().Value("cli").(*cli.CLI) uncli := cmd.Context().Value("cli").(*cli.CLI)
return runLogs(cmd.Context(), uncli, args, options) return runLogs(cmd.Context(), uncli, args, options)
@@ -62,14 +63,14 @@ If no services are specified, streams logs from the uncloud service.`,
return cmd return cmd
} }
func runLogs(ctx context.Context, uncli *cli.CLI, units []string, opts logs.Options) error { func runLogs(ctx context.Context, uncli *cli.CLI, services []string, opts logs.Options) error {
if len(units) == 0 { if len(services) == 0 {
units = []string{journal.UnitUncloud} services = []string{api.SystemServiceUncloud}
} }
for _, service := range services {
for _, unit := range units { if !slices.Contains(api.SystemServices, service) {
if !journal.ValidUnit(unit) { return fmt.Errorf("invalid system service '%s'; valid services: %s",
return fmt.Errorf("invalid systemd service '%s'", unit) service, strings.Join(api.SystemServices, ", "))
} }
} }
@@ -104,27 +105,27 @@ func runLogs(ctx context.Context, uncli *cli.CLI, units []string, opts logs.Opti
machineNames = append(machineNames, m.Machine.Name) machineNames = append(machineNames, m.Machine.Name)
} }
// Collect one log stream per unit. MachineLogs merges across machines internally. // Collect one log stream per service. MachineLogs merges across machines internally.
unitStreams := make([]<-chan api.ServiceLogEntry, 0, len(units)) serviceStreams := make([]<-chan api.ServiceLogEntry, 0, len(services))
for _, unit := range units { for _, service := range services {
ch, err := c.MachineLogs(ctx, unit, logsOpts) ch, err := c.MachineLogs(ctx, service, logsOpts)
if err != nil { if err != nil {
return fmt.Errorf("stream logs for systemd service '%s': %w", unit, err) return fmt.Errorf("stream logs for system service '%s': %w", service, err)
} }
unitStreams = append(unitStreams, ch) serviceStreams = append(serviceStreams, ch)
} }
var stream <-chan api.ServiceLogEntry var stream <-chan api.ServiceLogEntry
if len(unitStreams) == 1 { if len(serviceStreams) == 1 {
stream = unitStreams[0] stream = serviceStreams[0]
} else { } else {
// Each MachineLogs stream already runs its own inner merger with stall detection, // Each MachineLogs stream already runs its own inner merger with stall detection,
// so the outer merger across units skips it to avoid duplicate warnings. // so the outer merger across services skips it to avoid duplicate warnings.
merger := client.NewLogMerger(unitStreams, client.LogMergerOptions{}) merger := client.NewLogMerger(serviceStreams, client.LogMergerOptions{})
stream = merger.Stream() stream = merger.Stream()
} }
formatter := logs.NewFormatter(machineNames, units, opts.UTC) formatter := logs.NewFormatter(machineNames, services, opts.UTC)
// Print merged logs. // Print merged logs.
for entry := range stream { for entry := range stream {
+2 -2
View File
@@ -129,7 +129,7 @@ func (f *Formatter) PrintEntry(entry api.ServiceLogEntry) {
output.WriteString(f.formatMachine(entry.Metadata.MachineName)) output.WriteString(f.formatMachine(entry.Metadata.MachineName))
output.WriteString(" ") output.WriteString(" ")
// Service/container_id or service name for a systemd service. // Service/container_id or service name for a system service.
output.WriteString(f.formatService(entry.Metadata.ServiceName, entry.Metadata.ContainerID, entry.Metadata.Hook)) output.WriteString(f.formatService(entry.Metadata.ServiceName, entry.Metadata.ContainerID, entry.Metadata.Hook))
output.WriteString(" ") output.WriteString(" ")
@@ -160,7 +160,7 @@ func (f *Formatter) printError(entry api.ServiceLogEntry) {
stringid.TruncateID(entry.Metadata.ContainerID), stringid.TruncateID(entry.Metadata.ContainerID),
entry.Metadata.MachineName) entry.Metadata.MachineName)
} else { } else {
msg = fmt.Sprintf("WARNING: log stream from systemd service '%s' on machine '%s'", msg = fmt.Sprintf("WARNING: log stream from system service '%s' on machine '%s'",
entry.Metadata.ServiceName, entry.Metadata.ServiceName,
entry.Metadata.MachineName) entry.Metadata.MachineName)
} }
-20
View File
@@ -10,31 +10,11 @@ import (
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
) )
const (
UnitUncloud = "uncloud"
UnitDocker = "docker"
UnitCorrosion = "uncloud-corrosion"
)
func ValidUnit(unit string) bool {
switch unit {
case UnitUncloud:
case UnitDocker:
case UnitCorrosion:
default:
return false
}
return true
}
const journalctl = "journalctl" const journalctl = "journalctl"
var commandContext = exec.CommandContext // allow override for test var commandContext = exec.CommandContext // allow override for test
func logs(ctx context.Context, unit string, opts api.ServiceLogsOptions) (io.ReadCloser, func() error, error) { func logs(ctx context.Context, unit string, opts api.ServiceLogsOptions) (io.ReadCloser, func() error, error) {
if !ValidUnit(unit) {
return nil, nil, fmt.Errorf("journal logs: invalid unit: %s", unit)
}
args := []string{"-u", unit, "--no-hostname"} args := []string{"-u", unit, "--no-hostname"}
args = append(args, "-n") args = append(args, "-n")
if opts.Tail > -1 { if opts.Tail > -1 {
-5
View File
@@ -3,7 +3,6 @@ package journal
import ( import (
"bytes" "bytes"
"context" "context"
"fmt"
"slices" "slices"
"strconv" "strconv"
"time" "time"
@@ -13,10 +12,6 @@ import (
// Logs streams logs from a service and returns entries via a channel. // Logs streams logs from a service and returns entries via a channel.
func Logs(ctx context.Context, unit string, opts api.ServiceLogsOptions) (<-chan api.LogEntry, error) { func Logs(ctx context.Context, unit string, opts api.ServiceLogsOptions) (<-chan api.LogEntry, error) {
if !ValidUnit(unit) {
return nil, fmt.Errorf("journal logs: invalid unit: %s", unit)
}
reader, wait, err := logs(ctx, unit, opts) reader, wait, err := logs(ctx, unit, opts)
if err != nil { if err != nil {
return nil, err return nil, err
+6 -2
View File
@@ -17,8 +17,12 @@ import (
"github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/api"
) )
// Image is the Corrosion image pinned to the uncloudd version. const (
const Image = "ghcr.io/unlabs-dev/corrosion:2026.6.15" // Image is the Corrosion image pinned to the uncloudd version.
Image = "ghcr.io/unlabs-dev/corrosion:2026.6.15"
// ContainerName is the name of the managed Corrosion container.
ContainerName = "uncloud-corrosion"
)
type DockerService struct { type DockerService struct {
Client *client.Client Client *client.Client
+3 -1
View File
@@ -170,7 +170,9 @@ func (s *Service) ListImages(ctx context.Context, opts image.ListOptions) (Image
// ContainerLogs streams logs from a container and returns demultiplexed entries via a channel. // ContainerLogs streams logs from a container and returns demultiplexed 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(ctx context.Context, containerID string, opts api.ServiceLogsOptions) (<-chan api.LogEntry, error) { func (s *Service) ContainerLogs(
ctx context.Context, containerID string, opts api.ServiceLogsOptions,
) (<-chan api.LogEntry, error) {
dockerOpts := container.LogsOptions{ dockerOpts := container.LogsOptions{
ShowStdout: true, ShowStdout: true,
ShowStderr: true, ShowStderr: true,
+22 -6
View File
@@ -13,6 +13,7 @@ import (
"path/filepath" "path/filepath"
"slices" "slices"
"strconv" "strconv"
"strings"
"sync" "sync"
"time" "time"
@@ -139,7 +140,7 @@ func (c *Config) SetDefaults() (*Config, error) {
cfg.CorrosionService = &corroservice.DockerService{ cfg.CorrosionService = &corroservice.DockerService{
Client: cfg.DockerClient, Client: cfg.DockerClient,
Image: corroservice.Image, Image: corroservice.Image,
Name: "uncloud-corrosion", Name: corroservice.ContainerName,
DataDir: cfg.CorrosionDataDir, DataDir: cfg.CorrosionDataDir,
RunDir: cfg.CorrosionRunDir, RunDir: cfg.CorrosionRunDir,
User: fmt.Sprintf("%d:%d", uid, gid), User: fmt.Sprintf("%d:%d", uid, gid),
@@ -1217,7 +1218,7 @@ func (m *Machine) InspectService(
// logsHeartbeatInterval is the interval at which heartbeat entries are sent when there are no logs to stream. // logsHeartbeatInterval is the interval at which heartbeat entries are sent when there are no logs to stream.
const logsHeartbeatInterval = 200 * time.Millisecond const logsHeartbeatInterval = 200 * time.Millisecond
// MachineLogs streams logs from a systemd service. // MachineLogs streams logs from a system service.
func (m *Machine) MachineLogs( func (m *Machine) MachineLogs(
req *pb.LogsRequest, stream grpc.ServerStreamingServer[pb.LogEntry], req *pb.LogsRequest, stream grpc.ServerStreamingServer[pb.LogEntry],
) error { ) error {
@@ -1231,16 +1232,31 @@ func (m *Machine) MachineLogs(
Until: req.Until, Until: req.Until,
} }
logsCh, err := journal.Logs(ctx, req.Id, opts) var logsCh <-chan api.LogEntry
var err error
log := slog.With("stream_id", fmt.Sprintf("%p", stream)[2:])
switch req.Id {
case api.SystemServiceUncloud, api.SystemServiceDocker:
// These run as systemd units whose names match the service name.
logsCh, err = journal.Logs(ctx, req.Id, opts)
log = log.With("unit", req.Id)
case api.SystemServiceCorrosion:
// Corrosion runs as a daemon-managed container, not a systemd unit, so read its logs
// from the container, the same way `uc logs` does for service containers.
logsCh, err = m.dockerService.ContainerLogs(ctx, corroservice.ContainerName, opts)
log = log.With("container", corroservice.ContainerName)
default:
return status.Errorf(codes.InvalidArgument, "unsupported system service %q; supported services: %s",
req.Id, strings.Join(api.SystemServices, ", "))
}
if err != nil { if err != nil {
if errdefs.IsNotFound(err) { if errdefs.IsNotFound(err) {
return status.Error(codes.NotFound, err.Error()) return status.Error(codes.NotFound, err.Error())
} }
return status.Errorf(codes.Internal, "get journal logs: %v", err) return status.Errorf(codes.Internal, "get logs: %v", err)
} }
log := slog.With("unit", req.Id, "stream_id", fmt.Sprintf("%p", stream)[2:]) log.Debug("Starting system service logs streaming.",
log.Debug("Starting systemd service logs streaming.",
"follow", req.Follow, "tail", req.Tail, "since", req.Since, "until", req.Until) "follow", req.Follow, "tail", req.Tail, "since", req.Since, "until", req.Until)
// Heartbeats are needed only when following logs to let the client know when there are no new log entries // Heartbeats are needed only when following logs to let the client know when there are no new log entries
+11 -1
View File
@@ -16,6 +16,16 @@ const (
LogStreamHeartbeat LogStreamHeartbeat
) )
// System service names whose logs can be streamed via client.MachineLogs.
const (
SystemServiceCorrosion = "corrosion"
SystemServiceDocker = "docker"
SystemServiceUncloud = "uncloud"
)
// SystemServices lists all system services that support log streaming.
var SystemServices = []string{SystemServiceCorrosion, SystemServiceDocker, SystemServiceUncloud}
type LogStreamType int type LogStreamType int
// LogStreamTypeFromProto converts a protobuf LogEntry.StreamType to the internal LogStreamType. // LogStreamTypeFromProto converts a protobuf LogEntry.StreamType to the internal LogStreamType.
@@ -61,7 +71,7 @@ type ServiceLogsOptions struct {
Machines []string Machines []string
} }
// ServiceLogEntry represents a single log entry from a service container or systemd service. // ServiceLogEntry represents a single log entry from a service container or system service.
type ServiceLogEntry struct { type ServiceLogEntry struct {
// Metadata may not be set if an error occurred (Err is not nil). // Metadata may not be set if an error occurred (Err is not nil).
Metadata ServiceLogEntryMetadata Metadata ServiceLogEntryMetadata
+12 -12
View File
@@ -156,11 +156,11 @@ func (cli *Client) ContainerLogs(
return ch, nil return ch, nil
} }
// MachineLogs streams journal logs for the given systemd service across one or more machines in // MachineLogs streams logs for the given system service across one or more machines in
// chronological order based on timestamps. If opts.Machines is empty, logs are streamed from all // chronological order based on timestamps. If opts.Machines is empty, logs are streamed from all
// machines in the cluster. // machines in the cluster.
func (cli *Client) MachineLogs( func (cli *Client) MachineLogs(
ctx context.Context, unit string, opts api.ServiceLogsOptions, ctx context.Context, service string, opts api.ServiceLogsOptions,
) (<-chan api.ServiceLogEntry, error) { ) (<-chan api.ServiceLogEntry, error) {
machines, err := cli.ListMachines(ctx, &api.MachineFilter{NamesOrIDs: opts.Machines}) machines, err := cli.ListMachines(ctx, &api.MachineFilter{NamesOrIDs: opts.Machines})
if err != nil { if err != nil {
@@ -172,18 +172,18 @@ func (cli *Client) MachineLogs(
streams := make([]<-chan api.ServiceLogEntry, 0, len(machines)) streams := make([]<-chan api.ServiceLogEntry, 0, len(machines))
for _, m := range machines { for _, m := range machines {
ch, err := cli.systemdServiceLogs(ctx, m.Machine.Id, unit, opts) ch, err := cli.systemServiceLogs(ctx, m.Machine.Id, service, opts)
if err != nil { if err != nil {
// TODO: cancel already-opened streams. Currently they leak until ctx is cancelled which could // TODO: cancel already-opened streams. Currently they leak until ctx is cancelled which could
// be critical when used as SDK. // be critical when used as SDK.
return nil, fmt.Errorf("stream logs from systemd service '%s' on machine '%s': %w", return nil, fmt.Errorf("stream logs from system service '%s' on machine '%s': %w",
unit, m.Machine.Name, err) service, m.Machine.Name, err)
} }
// Enrich journal log entries with the systemd service name and machine metadata. // Enrich log entries with the system service name and machine metadata.
metadata := api.ServiceLogEntryMetadata{ metadata := api.ServiceLogEntryMetadata{
ServiceID: unit, ServiceID: service,
ServiceName: unit, ServiceName: service,
MachineID: m.Machine.Id, MachineID: m.Machine.Id,
MachineName: m.Machine.Name, MachineName: m.Machine.Name,
} }
@@ -194,14 +194,14 @@ func (cli *Client) MachineLogs(
return merger.Stream(), nil return merger.Stream(), nil
} }
// systemdServiceLogs streams log entries from a single systemd service on the specified machine. // systemServiceLogs streams log entries from a single system service on the specified machine.
func (cli *Client) systemdServiceLogs( func (cli *Client) systemServiceLogs(
ctx context.Context, machineID, unit string, opts api.ServiceLogsOptions, ctx context.Context, machineID, service string, opts api.ServiceLogsOptions,
) (<-chan api.LogEntry, error) { ) (<-chan api.LogEntry, error) {
proxyCtx := cli.ProxySingleMachineContext(ctx, machineID) proxyCtx := cli.ProxySingleMachineContext(ctx, machineID)
req := &pb.LogsRequest{ req := &pb.LogsRequest{
Id: unit, Id: service,
Follow: opts.Follow, Follow: opts.Follow,
Tail: int32(opts.Tail), Tail: int32(opts.Tail),
Since: opts.Since, Since: opts.Since,
+1 -1
View File
@@ -22,7 +22,7 @@ Manage machines in the cluster.
* [uc](uc.md) - A CLI tool for managing Uncloud resources such as machines, services, and volumes. * [uc](uc.md) - A CLI tool for managing Uncloud resources such as machines, services, and volumes.
* [uc machine add](uc_machine_add.md) - Add a remote machine to a cluster. * [uc machine add](uc_machine_add.md) - Add a remote machine to a cluster.
* [uc machine init](uc_machine_init.md) - Initialise a new cluster with a remote machine as the first member. * [uc machine init](uc_machine_init.md) - Initialise a new cluster with a remote machine as the first member.
* [uc machine logs](uc_machine_logs.md) - View systemd service logs. * [uc machine logs](uc_machine_logs.md) - View system service logs.
* [uc machine ls](uc_machine_ls.md) - List machines in a cluster. * [uc machine ls](uc_machine_ls.md) - List machines in a cluster.
* [uc machine rename](uc_machine_rename.md) - Rename a machine in the cluster. * [uc machine rename](uc_machine_rename.md) - Rename a machine in the cluster.
* [uc machine rm](uc_machine_rm.md) - Remove a machine from a cluster and reset it. * [uc machine rm](uc_machine_rm.md) - Remove a machine from a cluster and reset it.
@@ -1,16 +1,16 @@
# uc machine logs # uc machine logs
View systemd service logs. View system service logs.
## Synopsis ## Synopsis
View logs from the specified systemd service(s) across all machines in the cluster. View logs from the specified system service(s) across all machines in the cluster.
Use -m to restrict to specific machines. Use -m to restrict to specific machines.
Supported services: Supported services:
uncloud the Uncloud daemon corrosion the Corrosion distributed state store
docker the Docker daemon docker the Docker daemon
uncloud-corrosion the Corrosion distributed state store uncloud the Uncloud daemon
If no services are specified, streams logs from the uncloud service. If no services are specified, streams logs from the uncloud service.
@@ -29,7 +29,7 @@ uc machine logs [SERVICE...] [flags]
uc machine logs -f uncloud uc machine logs -f uncloud
# View logs from multiple services. # View logs from multiple services.
uc machine logs uncloud docker uncloud-corrosion uc machine logs uncloud docker corrosion
# Show last 20 lines per machine (default is 100). # Show last 20 lines per machine (default is 100).
uc machine logs -n 20 docker uc machine logs -n 20 docker
@@ -41,7 +41,7 @@ uc machine logs [SERVICE...] [flags]
uc machine logs --since 3h --until 1h30m docker uc machine logs --since 3h --until 1h30m docker
# View logs only from specific machines. # View logs only from specific machines.
uc machine logs -m machine1,machine2 uncloud uncloud-corrosion uc machine logs -m machine1,machine2 uncloud corrosion
``` ```
## Options ## Options