From 7bb3cb9682ccf5fa9c63000a586aa9a976e52b49 Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Wed, 17 Jun 2026 18:28:25 +1000 Subject: [PATCH] logs: stream corrosion logs with 'uc machine logs corrosion' from container instead of journal (fixes #392) --- cmd/uncloud/machine/logs.go | 53 ++++++++++--------- internal/cli/logs/formatter.go | 4 +- internal/journal/journal.go | 20 ------- internal/journal/logs.go | 5 -- internal/machine/corroservice/docker.go | 8 ++- internal/machine/docker/service.go | 4 +- internal/machine/machine.go | 28 +++++++--- pkg/api/logs.go | 12 ++++- pkg/client/logs.go | 24 ++++----- website/docs/9-cli-reference/uc_machine.md | 2 +- .../docs/9-cli-reference/uc_machine_logs.md | 14 ++--- 11 files changed, 91 insertions(+), 83 deletions(-) diff --git a/cmd/uncloud/machine/logs.go b/cmd/uncloud/machine/logs.go index 4f8b862a..33d8b9fe 100644 --- a/cmd/uncloud/machine/logs.go +++ b/cmd/uncloud/machine/logs.go @@ -3,11 +3,12 @@ package machine import ( "context" "fmt" + "slices" + "strings" "github.com/psviderski/uncloud/internal/cli" "github.com/psviderski/uncloud/internal/cli/completion" "github.com/psviderski/uncloud/internal/cli/logs" - "github.com/psviderski/uncloud/internal/journal" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client" "github.com/spf13/cobra" @@ -19,14 +20,14 @@ func NewLogsCommand() *cobra.Command { cmd := &cobra.Command{ Use: "logs [SERVICE...]", Aliases: []string{"log"}, - Short: "View systemd service logs.", - Long: `View logs from the specified systemd service(s) across all machines in the cluster. + Short: "View system service logs.", + Long: `View logs from the specified system service(s) across all machines in the cluster. Use -m to restrict to specific machines. Supported services: - uncloud the Uncloud daemon - docker the Docker daemon - uncloud-corrosion the Corrosion distributed state store + corrosion the Corrosion distributed state store + docker the Docker daemon + uncloud the Uncloud daemon If no services are specified, streams logs from 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 # 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). 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 # 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 { uncli := cmd.Context().Value("cli").(*cli.CLI) return runLogs(cmd.Context(), uncli, args, options) @@ -62,14 +63,14 @@ If no services are specified, streams logs from the uncloud service.`, return cmd } -func runLogs(ctx context.Context, uncli *cli.CLI, units []string, opts logs.Options) error { - if len(units) == 0 { - units = []string{journal.UnitUncloud} +func runLogs(ctx context.Context, uncli *cli.CLI, services []string, opts logs.Options) error { + if len(services) == 0 { + services = []string{api.SystemServiceUncloud} } - - for _, unit := range units { - if !journal.ValidUnit(unit) { - return fmt.Errorf("invalid systemd service '%s'", unit) + for _, service := range services { + if !slices.Contains(api.SystemServices, service) { + return fmt.Errorf("invalid system service '%s'; valid services: %s", + 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) } - // Collect one log stream per unit. MachineLogs merges across machines internally. - unitStreams := make([]<-chan api.ServiceLogEntry, 0, len(units)) - for _, unit := range units { - ch, err := c.MachineLogs(ctx, unit, logsOpts) + // Collect one log stream per service. MachineLogs merges across machines internally. + serviceStreams := make([]<-chan api.ServiceLogEntry, 0, len(services)) + for _, service := range services { + ch, err := c.MachineLogs(ctx, service, logsOpts) 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 - if len(unitStreams) == 1 { - stream = unitStreams[0] + if len(serviceStreams) == 1 { + stream = serviceStreams[0] } else { // Each MachineLogs stream already runs its own inner merger with stall detection, - // so the outer merger across units skips it to avoid duplicate warnings. - merger := client.NewLogMerger(unitStreams, client.LogMergerOptions{}) + // so the outer merger across services skips it to avoid duplicate warnings. + merger := client.NewLogMerger(serviceStreams, client.LogMergerOptions{}) stream = merger.Stream() } - formatter := logs.NewFormatter(machineNames, units, opts.UTC) + formatter := logs.NewFormatter(machineNames, services, opts.UTC) // Print merged logs. for entry := range stream { diff --git a/internal/cli/logs/formatter.go b/internal/cli/logs/formatter.go index 31f85de0..771eb4d6 100644 --- a/internal/cli/logs/formatter.go +++ b/internal/cli/logs/formatter.go @@ -129,7 +129,7 @@ func (f *Formatter) PrintEntry(entry api.ServiceLogEntry) { output.WriteString(f.formatMachine(entry.Metadata.MachineName)) 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(" ") @@ -160,7 +160,7 @@ func (f *Formatter) printError(entry api.ServiceLogEntry) { stringid.TruncateID(entry.Metadata.ContainerID), entry.Metadata.MachineName) } 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.MachineName) } diff --git a/internal/journal/journal.go b/internal/journal/journal.go index 075ed37f..da28c00e 100644 --- a/internal/journal/journal.go +++ b/internal/journal/journal.go @@ -10,31 +10,11 @@ import ( "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" var commandContext = exec.CommandContext // allow override for test 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 = append(args, "-n") if opts.Tail > -1 { diff --git a/internal/journal/logs.go b/internal/journal/logs.go index 020fbbd3..6749a5ad 100644 --- a/internal/journal/logs.go +++ b/internal/journal/logs.go @@ -3,7 +3,6 @@ package journal import ( "bytes" "context" - "fmt" "slices" "strconv" "time" @@ -13,10 +12,6 @@ import ( // 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) { - if !ValidUnit(unit) { - return nil, fmt.Errorf("journal logs: invalid unit: %s", unit) - } - reader, wait, err := logs(ctx, unit, opts) if err != nil { return nil, err diff --git a/internal/machine/corroservice/docker.go b/internal/machine/corroservice/docker.go index 1e23962c..afcef09f 100644 --- a/internal/machine/corroservice/docker.go +++ b/internal/machine/corroservice/docker.go @@ -17,8 +17,12 @@ import ( "github.com/psviderski/uncloud/pkg/api" ) -// Image is the Corrosion image pinned to the uncloudd version. -const Image = "ghcr.io/unlabs-dev/corrosion:2026.6.15" +const ( + // 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 { Client *client.Client diff --git a/internal/machine/docker/service.go b/internal/machine/docker/service.go index ffcf442b..076da678 100644 --- a/internal/machine/docker/service.go +++ b/internal/machine/docker/service.go @@ -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. // 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{ ShowStdout: true, ShowStderr: true, diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 44a3a76f..0f0af910 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -13,6 +13,7 @@ import ( "path/filepath" "slices" "strconv" + "strings" "sync" "time" @@ -139,7 +140,7 @@ func (c *Config) SetDefaults() (*Config, error) { cfg.CorrosionService = &corroservice.DockerService{ Client: cfg.DockerClient, Image: corroservice.Image, - Name: "uncloud-corrosion", + Name: corroservice.ContainerName, DataDir: cfg.CorrosionDataDir, RunDir: cfg.CorrosionRunDir, 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. const logsHeartbeatInterval = 200 * time.Millisecond -// MachineLogs streams logs from a systemd service. +// MachineLogs streams logs from a system service. func (m *Machine) MachineLogs( req *pb.LogsRequest, stream grpc.ServerStreamingServer[pb.LogEntry], ) error { @@ -1231,16 +1232,31 @@ func (m *Machine) MachineLogs( 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 errdefs.IsNotFound(err) { 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 systemd service logs streaming.", + log.Debug("Starting system service logs streaming.", "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 diff --git a/pkg/api/logs.go b/pkg/api/logs.go index b52badee..21697df2 100644 --- a/pkg/api/logs.go +++ b/pkg/api/logs.go @@ -16,6 +16,16 @@ const ( 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 // LogStreamTypeFromProto converts a protobuf LogEntry.StreamType to the internal LogStreamType. @@ -61,7 +71,7 @@ type ServiceLogsOptions struct { 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 { // Metadata may not be set if an error occurred (Err is not nil). Metadata ServiceLogEntryMetadata diff --git a/pkg/client/logs.go b/pkg/client/logs.go index 6ac02496..0d7874c1 100644 --- a/pkg/client/logs.go +++ b/pkg/client/logs.go @@ -156,11 +156,11 @@ func (cli *Client) ContainerLogs( 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 // machines in the cluster. func (cli *Client) MachineLogs( - ctx context.Context, unit string, opts api.ServiceLogsOptions, + ctx context.Context, service string, opts api.ServiceLogsOptions, ) (<-chan api.ServiceLogEntry, error) { machines, err := cli.ListMachines(ctx, &api.MachineFilter{NamesOrIDs: opts.Machines}) if err != nil { @@ -172,18 +172,18 @@ func (cli *Client) MachineLogs( streams := make([]<-chan api.ServiceLogEntry, 0, len(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 { // TODO: cancel already-opened streams. Currently they leak until ctx is cancelled which could // be critical when used as SDK. - return nil, fmt.Errorf("stream logs from systemd service '%s' on machine '%s': %w", - unit, m.Machine.Name, err) + return nil, fmt.Errorf("stream logs from system service '%s' on machine '%s': %w", + 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{ - ServiceID: unit, - ServiceName: unit, + ServiceID: service, + ServiceName: service, MachineID: m.Machine.Id, MachineName: m.Machine.Name, } @@ -194,14 +194,14 @@ func (cli *Client) MachineLogs( return merger.Stream(), nil } -// systemdServiceLogs streams log entries from a single systemd service on the specified machine. -func (cli *Client) systemdServiceLogs( - ctx context.Context, machineID, unit string, opts api.ServiceLogsOptions, +// systemServiceLogs streams log entries from a single system service on the specified machine. +func (cli *Client) systemServiceLogs( + ctx context.Context, machineID, service string, opts api.ServiceLogsOptions, ) (<-chan api.LogEntry, error) { proxyCtx := cli.ProxySingleMachineContext(ctx, machineID) req := &pb.LogsRequest{ - Id: unit, + Id: service, Follow: opts.Follow, Tail: int32(opts.Tail), Since: opts.Since, diff --git a/website/docs/9-cli-reference/uc_machine.md b/website/docs/9-cli-reference/uc_machine.md index 98d936b4..42d318ce 100644 --- a/website/docs/9-cli-reference/uc_machine.md +++ b/website/docs/9-cli-reference/uc_machine.md @@ -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 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 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 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. diff --git a/website/docs/9-cli-reference/uc_machine_logs.md b/website/docs/9-cli-reference/uc_machine_logs.md index 524fd08d..15ae9001 100644 --- a/website/docs/9-cli-reference/uc_machine_logs.md +++ b/website/docs/9-cli-reference/uc_machine_logs.md @@ -1,16 +1,16 @@ # uc machine logs -View systemd service logs. +View system service logs. ## 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. Supported services: - uncloud the Uncloud daemon - docker the Docker daemon - uncloud-corrosion the Corrosion distributed state store + corrosion the Corrosion distributed state store + docker the Docker daemon + uncloud the Uncloud daemon 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 # 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). uc machine logs -n 20 docker @@ -41,7 +41,7 @@ uc machine logs [SERVICE...] [flags] uc machine logs --since 3h --until 1h30m docker # View logs only from specific machines. - uc machine logs -m machine1,machine2 uncloud uncloud-corrosion + uc machine logs -m machine1,machine2 uncloud corrosion ``` ## Options