From b852d34068bbf55dcb17ee625191acde0b46dbd9 Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Tue, 21 Apr 2026 22:00:13 +1000 Subject: [PATCH] refactor(logs): update log formatting for systemd services and handle merging in the client --- cmd/uncloud/internal/logs/formatter.go | 54 +++++++++------ cmd/uncloud/machine/logs.go | 65 ++++++++++--------- pkg/api/logs.go | 8 ++- pkg/client/logs.go | 65 +++++++++++++------ .../docs/9-cli-reference/uc_machine_logs.md | 36 +++++----- 5 files changed, 140 insertions(+), 88 deletions(-) diff --git a/cmd/uncloud/internal/logs/formatter.go b/cmd/uncloud/internal/logs/formatter.go index 656657ce..cb5756d0 100644 --- a/cmd/uncloud/internal/logs/formatter.go +++ b/cmd/uncloud/internal/logs/formatter.go @@ -10,6 +10,7 @@ import ( "charm.land/lipgloss/v2" "github.com/docker/docker/pkg/stringid" + "github.com/psviderski/uncloud/internal/cli/tui" "github.com/psviderski/uncloud/pkg/api" ) @@ -80,9 +81,9 @@ func (f *Formatter) formatMachine(name string) string { return style.Render(name) } -func (f *Formatter) formatServiceContainer(serviceName, containerID string) string { - styleService := lipgloss.NewStyle().Bold(true).PaddingRight(f.maxServiceWidth - len(serviceName)) - styleContainer := lipgloss.NewStyle().Faint(true) +func (f *Formatter) formatService(serviceName, containerID string) string { + styleService := lipgloss.NewStyle().Bold(true) + padding := f.maxServiceWidth - len(serviceName) if len(f.serviceNames) > 1 { // Service name is coloured for multi-service logs. @@ -95,10 +96,15 @@ func (f *Formatter) formatServiceContainer(serviceName, containerID string) stri styleService = styleService.Foreground(Palette[i%len(Palette)]) } - return styleService.Render(serviceName) + styleContainer.Render("["+containerID[:5]+"]") + // Journal logs are unit-scoped and have no container ID. + if containerID == "" { + return styleService.PaddingRight(padding).Render(serviceName) + } + + return styleService.Render(serviceName) + tui.Faint.PaddingRight(padding).Render("/"+containerID[:5]) } -// printEntry prints a single log entry with proper formatting. +// PrintEntry prints a single log entry with proper formatting. func (f *Formatter) PrintEntry(entry api.ServiceLogEntry) { if entry.Stream != api.LogStreamStdout && entry.Stream != api.LogStreamStderr { return @@ -114,8 +120,8 @@ func (f *Formatter) PrintEntry(entry api.ServiceLogEntry) { output.WriteString(f.formatMachine(entry.Metadata.MachineName)) output.WriteString(" ") - // Service[container_id] - output.WriteString(f.formatServiceContainer(entry.Metadata.ServiceName, entry.Metadata.ContainerID)) + // Service/container_id or service name for a systemd service. + output.WriteString(f.formatService(entry.Metadata.ServiceName, entry.Metadata.ContainerID)) output.WriteString(" ") // Message @@ -131,23 +137,31 @@ func (f *Formatter) PrintEntry(entry api.ServiceLogEntry) { // PrintError prints an error entry (e.g., stalled stream warning). func (f *Formatter) PrintError(entry api.ServiceLogEntry) { + if entry.Metadata.ServiceName == "" { + msg := fmt.Sprintf("ERROR: %v", entry.Err) + style := lipgloss.NewStyle().Bold(true).Foreground(lipgloss.BrightRed) // Bold bright red. + fmt.Fprintln(os.Stderr, style.Render(msg)) + return + } + + var msg string if entry.Metadata.ContainerID != "" { - msg := fmt.Sprintf("WARNING: log stream from %s[%s] on machine '%s'", + msg = fmt.Sprintf("WARNING: log stream from container '%s/%s' on machine '%s'", entry.Metadata.ServiceName, stringid.TruncateID(entry.Metadata.ContainerID), entry.Metadata.MachineName) - - if errors.Is(entry.Err, api.ErrLogStreamStalled) { - msg += " stopped responding" - } else { - msg += fmt.Sprintf(": %v", entry.Err) - } - - style := lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("11")) // Bold bright yellow - fmt.Fprintln(os.Stderr, style.Render(msg)) } else { - msg := fmt.Sprintf("ERROR: %v", entry.Err) - style := lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("9")) // Bold bright red - fmt.Fprintln(os.Stderr, style.Render(msg)) + msg = fmt.Sprintf("WARNING: log stream from systemd service '%s' on machine '%s'", + entry.Metadata.ServiceName, + entry.Metadata.MachineName) } + + if errors.Is(entry.Err, api.ErrLogStreamStalled) { + msg += " stopped responding" + } else { + msg += fmt.Sprintf(": %v", entry.Err) + } + + style := lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("11")) // Bold bright yellow. + fmt.Fprintln(os.Stderr, style.Render(msg)) } diff --git a/cmd/uncloud/machine/logs.go b/cmd/uncloud/machine/logs.go index 666fea3f..271b432f 100644 --- a/cmd/uncloud/machine/logs.go +++ b/cmd/uncloud/machine/logs.go @@ -16,35 +16,39 @@ func NewLogsCommand() *cobra.Command { var options logs.Options cmd := &cobra.Command{ - Use: "logs [UNIT...]", + Use: "logs [SERVICE...]", Aliases: []string{"log"}, Short: "View systemd service logs.", - Long: `View logs from all replicas of the specified units(s) (uncloud, docker or uncloud-corrosion) across all machines in the cluster. + Long: `View logs from the specified systemd service(s) across all machines in the cluster. +Use -m to restrict to specific machines. -If no units are specified, streams logs from the uncloud unit.`, - Example: ` # View recent logs for a system service. - uc logs uncloud +Supported services: + uncloud the Uncloud daemon + docker the Docker daemon + uncloud-corrosion the Corrosion distributed state store + +If no services are specified, streams logs from the uncloud service.`, + Example: ` # View recent logs for the uncloud service. + uc machine logs + uc machine logs uncloud # Stream logs in real-time (follow mode). - uc logs -f uncloud + uc machine logs -f uncloud # View logs from multiple services. - uc logs web uncloud docker + uc machine logs uncloud docker uncloud-corrosion - # View logs from uncloud - uc logs - - # Show last 20 lines per replica (default is 100). - uc logs -n 20 docker + # Show last 20 lines per machine (default is 100). + uc machine logs -n 20 docker # Show all logs without line limit. - uc logs -n all docker + uc machine logs -n all docker # View logs from a specific time range. - uc logs --since 3h --until 1h30m docker + uc machine logs --since 3h --until 1h30m docker - # View logs only from replicas running on specific machines. - uc logs -m machine1,machine2 docker corrosion`, + # View logs only from specific machines. + uc machine logs -m machine1,machine2 uncloud 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,7 +66,7 @@ func runLogs(ctx context.Context, uncli *cli.CLI, units []string, opts logs.Opti for _, unit := range units { if !journal.ValidUnit(unit) { - return fmt.Errorf("invalid unit '%s'", unit) + return fmt.Errorf("invalid systemd service '%s'", unit) } } @@ -85,8 +89,10 @@ func runLogs(ctx context.Context, uncli *cli.CLI, units []string, opts logs.Opti Machines: cli.ExpandCommaSeparatedValues(opts.Machines), } - // Fetch machine names for all machines we want the unit logs from. - machines, err := c.ListMachines(ctx, &api.MachineFilter{NamesOrIDs: opts.Machines}) + // Resolve machine records for the formatter's column width computation. + machines, err := c.ListMachines(ctx, &api.MachineFilter{ + NamesOrIDs: logsOpts.Machines, + }) if err != nil { return fmt.Errorf("list machines: %w", err) } @@ -95,23 +101,22 @@ func runLogs(ctx context.Context, uncli *cli.CLI, units []string, opts logs.Opti machineNames = append(machineNames, m.Machine.Name) } - // Collect log streams from the units on the machine machines. - unitStreams := make([]<-chan api.ServiceLogEntry, 0, len(units)+len(machines)) - - for _, machine := range machineNames { - for _, unit := range units { - ch, err := c.MachineLogs(ctx, machine, unit, logsOpts) - if err != nil { - return fmt.Errorf("stream logs for systemd service '%s': %w", unit, err) - } - unitStreams = append(unitStreams, ch) + // 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) + if err != nil { + return fmt.Errorf("stream logs for systemd service '%s': %w", unit, err) } + unitStreams = append(unitStreams, ch) } var stream <-chan api.ServiceLogEntry - if len(units) == 1 { + if len(unitStreams) == 1 { stream = unitStreams[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{}) stream = merger.Stream() } diff --git a/pkg/api/logs.go b/pkg/api/logs.go index f36d88cd..c207c9ff 100644 --- a/pkg/api/logs.go +++ b/pkg/api/logs.go @@ -57,14 +57,15 @@ type ServiceLogsOptions struct { Machines []string } -// ServiceLogEntry represents a single log entry from a service container. +// ServiceLogEntry represents a single log entry from a service container or systemd service. type ServiceLogEntry struct { // Metadata may not be set if an error occurred (Err is not nil). Metadata ServiceLogEntryMetadata LogEntry } -// ServiceLogEntryMetadata contains metadata about the source of a log entry. +// ServiceLogEntryMetadata identifies the source of a log entry. +// For systemd service logs, ServiceID and ServiceName hold the unit name and ContainerID is empty. type ServiceLogEntryMetadata struct { ServiceID string ServiceName string @@ -77,7 +78,8 @@ type ServiceLogEntryMetadata struct { type LogEntry struct { Stream LogStreamType Timestamp time.Time - Message []byte + // Message is the raw log line as bytes terminated with a trailing newline. + Message []byte // Err indicates that an error occurred while streaming logs from a container. // Other fields are not set if Err is not nil. Err error diff --git a/pkg/client/logs.go b/pkg/client/logs.go index 7fe17431..002a9e16 100644 --- a/pkg/client/logs.go +++ b/pkg/client/logs.go @@ -52,6 +52,8 @@ func (cli *Client) ServiceLogs( stream, err := cli.ContainerLogs(ctx, ctr.MachineID, ctr.Container.ID, 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 svc, nil, fmt.Errorf("stream logs from service container '%s' on machine '%s': %w", stringid.TruncateID(ctr.Container.ID), machineName, err) } @@ -140,13 +142,51 @@ func (cli *Client) ContainerLogs( return ch, nil } -// MachineLogs streams journal logs from the unit on a specified machine. +// MachineLogs streams journal logs for the given systemd 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, machineNameOrID string, unit string, opts api.ServiceLogsOptions, + ctx context.Context, unit string, opts api.ServiceLogsOptions, ) (<-chan api.ServiceLogEntry, error) { - proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineNameOrID}) + machines, err := cli.ListMachines(ctx, &api.MachineFilter{NamesOrIDs: opts.Machines}) if err != nil { - return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineNameOrID, err) + return nil, fmt.Errorf("list machines: %w", err) + } + if len(machines) == 0 { + return nil, errors.New("no machines found") + } + + streams := make([]<-chan api.ServiceLogEntry, 0, len(machines)) + for _, m := range machines { + ch, err := cli.systemdServiceLogs(ctx, m.Machine.Id, unit, 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) + } + + // Enrich journal log entries with the systemd service name and machine metadata. + metadata := api.ServiceLogEntryMetadata{ + ServiceID: unit, + ServiceName: unit, + MachineID: m.Machine.Id, + MachineName: m.Machine.Name, + } + streams = append(streams, logsStreamWithServiceMetadata(ch, metadata)) + } + + merger := NewLogMerger(streams, DefaultLogMergerOptions) + 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, +) (<-chan api.LogEntry, error) { + proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineID}) + if err != nil { + return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineID, err) } req := &pb.LogsRequest{ @@ -167,17 +207,7 @@ func (cli *Client) MachineLogs( return nil, err } - // Enrich log entries from the machine with metadata. - metadata := api.ServiceLogEntryMetadata{ - ServiceID: "", - ServiceName: "", - ContainerID: unit, - MachineID: machineNameOrID, - MachineName: machineNameOrID, - } - ch := make(chan api.LogEntry) - go func() { defer close(ch) @@ -187,9 +217,7 @@ func (cli *Client) MachineLogs( return } if err != nil { - ch <- api.LogEntry{ - Err: err, - } + ch <- api.LogEntry{Err: err} return } @@ -207,8 +235,7 @@ func (cli *Client) MachineLogs( } }() - enrichedCh := logsStreamWithServiceMetadata(ch, metadata) - return enrichedCh, nil + return ch, nil } // logsStreamWithServiceMetadata wraps a container logs stream and enriches each log entry with service metadata. diff --git a/website/docs/9-cli-reference/uc_machine_logs.md b/website/docs/9-cli-reference/uc_machine_logs.md index a1f37328..524fd08d 100644 --- a/website/docs/9-cli-reference/uc_machine_logs.md +++ b/website/docs/9-cli-reference/uc_machine_logs.md @@ -4,40 +4,44 @@ View systemd service logs. ## Synopsis -View logs from all replicas of the specified units(s) (uncloud, docker or uncloud-corrosion) across all machines in the cluster. +View logs from the specified systemd service(s) across all machines in the cluster. +Use -m to restrict to specific machines. -If no units are specified, streams logs from the uncloud unit. +Supported services: + uncloud the Uncloud daemon + docker the Docker daemon + uncloud-corrosion the Corrosion distributed state store + +If no services are specified, streams logs from the uncloud service. ``` -uc machine logs [UNIT...] [flags] +uc machine logs [SERVICE...] [flags] ``` ## Examples ``` - # View recent logs for a system service. - uc logs uncloud + # View recent logs for the uncloud service. + uc machine logs + uc machine logs uncloud # Stream logs in real-time (follow mode). - uc logs -f uncloud + uc machine logs -f uncloud # View logs from multiple services. - uc logs web uncloud docker + uc machine logs uncloud docker uncloud-corrosion - # View logs from uncloud - uc logs - - # Show last 20 lines per replica (default is 100). - uc logs -n 20 docker + # Show last 20 lines per machine (default is 100). + uc machine logs -n 20 docker # Show all logs without line limit. - uc logs -n all docker + uc machine logs -n all docker # View logs from a specific time range. - uc logs --since 3h --until 1h30m docker + uc machine logs --since 3h --until 1h30m docker - # View logs only from replicas running on specific machines. - uc logs -m machine1,machine2 docker corrosion + # View logs only from specific machines. + uc machine logs -m machine1,machine2 uncloud uncloud-corrosion ``` ## Options