diff --git a/cmd/uncloud/internal/logs/formatter.go b/cmd/uncloud/internal/logs/formatter.go index cb5756d0..3b1bd078 100644 --- a/cmd/uncloud/internal/logs/formatter.go +++ b/cmd/uncloud/internal/logs/formatter.go @@ -3,6 +3,7 @@ package logs import ( "errors" "fmt" + "image/color" "os" "slices" "strings" @@ -75,7 +76,7 @@ func (f *Formatter) formatMachine(name string) string { i = len(f.machineNames) - 1 } - style = style.Foreground(Palette[i%len(Palette)]) + style = style.Foreground(palette[i%len(palette)]) } return style.Render(name) @@ -93,7 +94,7 @@ func (f *Formatter) formatService(serviceName, containerID string) string { i = len(f.serviceNames) - 1 } - styleService = styleService.Foreground(Palette[i%len(Palette)]) + styleService = styleService.Foreground(palette[i%len(palette)]) } // Journal logs are unit-scoped and have no container ID. @@ -165,3 +166,17 @@ func (f *Formatter) PrintError(entry api.ServiceLogEntry) { style := lipgloss.NewStyle().Bold(true).Foreground(lipgloss.Color("11")) // Bold bright yellow. fmt.Fprintln(os.Stderr, style.Render(msg)) } + +// palette is available colors for machine/service differentiation. +var palette = []color.Color{ + lipgloss.BrightGreen, + lipgloss.BrightYellow, + lipgloss.BrightBlue, + lipgloss.BrightMagenta, + lipgloss.BrightCyan, + lipgloss.Green, + lipgloss.Yellow, + lipgloss.Blue, + lipgloss.Magenta, + lipgloss.Cyan, +} diff --git a/cmd/uncloud/internal/logs/logs.go b/cmd/uncloud/internal/logs/logs.go index 5ca0d75c..fe5df5be 100644 --- a/cmd/uncloud/internal/logs/logs.go +++ b/cmd/uncloud/internal/logs/logs.go @@ -2,10 +2,9 @@ package logs import ( "fmt" - "image/color" "strconv" + "strings" - "charm.land/lipgloss/v2" "github.com/spf13/pflag" ) @@ -59,16 +58,59 @@ func Tail(tail string) (int, error) { return tailInt, nil } -// Available colors for machine/service differentiation. -var Palette = []color.Color{ - lipgloss.BrightGreen, - lipgloss.BrightYellow, - lipgloss.BrightBlue, - lipgloss.BrightMagenta, - lipgloss.BrightCyan, - lipgloss.Green, - lipgloss.Yellow, - lipgloss.Blue, - lipgloss.Magenta, - lipgloss.Cyan, +// ServiceArg pairs a service name with the list of container filters parsed from `uc service logs` arguments. +// An empty Containers slice means "stream logs from all containers of this service". +type ServiceArg struct { + Service string + Containers []string +} + +// ParseServiceArgs groups raw `uc service logs` positional arguments into per-service entries, +// preserving first-seen order. Each argument is either a service name (e.g. "web") or a +// service/container reference (e.g. "web/61d57fd3428f"). +// Arguments for the same service are merged: if any argument for a service lacks a container suffix, all containers +// of that service are streamed regardless of any other service/container arguments for it. +func ParseServiceArgs(args []string) ([]ServiceArg, error) { + indexByService := make(map[string]int, len(args)) + allContainers := make(map[string]bool, len(args)) + result := make([]ServiceArg, 0, len(args)) + + for _, arg := range args { + arg = strings.TrimSpace(arg) + if arg == "" { + return nil, fmt.Errorf("empty service argument") + } + + service, container, hasSlash := strings.Cut(arg, "/") + if service == "" { + return nil, fmt.Errorf("invalid service argument '%s': service name is empty", arg) + } + if hasSlash && container == "" { + return nil, fmt.Errorf("invalid service argument '%s': container name or ID is empty", arg) + } + + idx, seen := indexByService[service] + if !seen { + entry := ServiceArg{Service: service} + if hasSlash { + entry.Containers = []string{container} + } else { + allContainers[service] = true + } + result = append(result, entry) + indexByService[service] = len(result) - 1 + continue + } + + if !hasSlash { + result[idx].Containers = nil + allContainers[service] = true + continue + } + if !allContainers[service] { + result[idx].Containers = append(result[idx].Containers, container) + } + } + + return result, nil } diff --git a/cmd/uncloud/service/logs.go b/cmd/uncloud/service/logs.go index 912ec9be..9832eff1 100644 --- a/cmd/uncloud/service/logs.go +++ b/cmd/uncloud/service/logs.go @@ -20,11 +20,14 @@ func NewLogsCommand(groupID string) *cobra.Command { var options logs.Options cmd := &cobra.Command{ - Use: "logs [SERVICE...]", + Use: "logs [SERVICE[/CONTAINER]...]", Aliases: []string{"log"}, Short: "View service logs.", Long: `View logs from all replicas of the specified service(s) across all machines in the cluster. +To view logs from specific replicas (containers) within a service, use the SERVICE/CONTAINER form, +where CONTAINER is a container name, full ID, or unique ID prefix. + If no services are specified, streams logs from all services defined in the Compose file (compose.yaml by default or the file(s) specified with --file).`, Example: ` # View recent logs for a service. @@ -48,6 +51,9 @@ If no services are specified, streams logs from all services defined in the Comp # View logs from a specific time range. uc logs --since 3h --until 1h30m web + # View logs only from specific replicas (containers). + uc logs web/61d57fd3428f api/2f60 + # View logs only from replicas running on specific machines. uc logs -m machine1,machine2 web api`, RunE: func(cmd *cobra.Command, args []string) error { @@ -64,10 +70,15 @@ If no services are specified, streams logs from all services defined in the Comp return cmd } -func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts logs.Options) error { +func runLogs(ctx context.Context, uncli *cli.CLI, args []string, opts logs.Options) error { + serviceArgs, err := logs.ParseServiceArgs(args) + if err != nil { + return err + } + // If no services specified, try to load them from the Compose file(s). fromCompose := false - if len(serviceNames) == 0 { + if len(serviceArgs) == 0 { fromCompose = true project, err := compose.LoadProject(ctx, opts.Files) if err != nil { @@ -77,10 +88,15 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo uncli.SetClusterContextIfUnset(compose.ClusterContext(project)) // View logs for all services, including disabled by inactive profiles. - serviceNames = append(project.ServiceNames(), project.DisabledServiceNames()...) - if len(serviceNames) == 0 { + composeServices := append(project.ServiceNames(), project.DisabledServiceNames()...) + if len(composeServices) == 0 { return errors.New("no services found in Compose file(s)") } + + serviceArgs = make([]logs.ServiceArg, len(composeServices)) + for i, name := range composeServices { + serviceArgs[i] = logs.ServiceArg{Service: name} + } } // Parse tail option. @@ -95,7 +111,7 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo } defer c.Close() - logsOpts := api.ServiceLogsOptions{ + baseOpts := api.ServiceLogsOptions{ Follow: opts.Follow, Tail: tail, Since: opts.Since, @@ -106,20 +122,23 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo // Collect log streams from all services. When service names come from a Compose file, // skip the ones that are not found in the cluster (they may have been removed or not deployed yet). machineIDsSet := mapset.NewSet[string]() - svcStreams := make([]<-chan api.ServiceLogEntry, 0, len(serviceNames)) + svcStreams := make([]<-chan api.ServiceLogEntry, 0, len(serviceArgs)) var foundServices, notFoundServices []string - for _, serviceName := range serviceNames { - svc, ch, err := c.ServiceLogs(ctx, serviceName, logsOpts) + for _, sa := range serviceArgs { + svcOpts := baseOpts + svcOpts.Containers = sa.Containers + + svc, ch, err := c.ServiceLogs(ctx, sa.Service, svcOpts) if err != nil { if errors.Is(err, api.ErrNotFound) && fromCompose { - notFoundServices = append(notFoundServices, serviceName) + notFoundServices = append(notFoundServices, sa.Service) continue } - return fmt.Errorf("stream logs for service '%s': %w", serviceName, err) + return fmt.Errorf("stream logs for service '%s': %w", sa.Service, err) } svcStreams = append(svcStreams, ch) - foundServices = append(foundServices, serviceName) + foundServices = append(foundServices, sa.Service) machineIDs := svc.MachineIDs() machineIDsSet.Append(machineIDs...) @@ -130,7 +149,6 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo return fmt.Errorf("stream logs for services defined in %s: no services found in the cluster", strings.Join(opts.Files, ", ")) } - serviceNames = foundServices for _, name := range notFoundServices { tui.PrintWarning(fmt.Sprintf("service '%s' not found in the cluster, skipping", name)) @@ -138,7 +156,7 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo } var stream <-chan api.ServiceLogEntry - if len(serviceNames) == 1 { + if len(svcStreams) == 1 { stream = svcStreams[0] } else { // Merge all service streams into a single sorted stream without stall detection as its handled per-service. @@ -147,6 +165,8 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo } // Fetch machine names for all machines (machineIDsSet) service containers are running on. + // Note: this is the full set per service, not narrowed by --machine or per-container filters, + // so the formatter may pad columns wider than strictly needed when filters are active. machines, err := c.ListMachines(ctx, &api.MachineFilter{NamesOrIDs: machineIDsSet.ToSlice()}) if err != nil { return fmt.Errorf("list machines: %w", err) @@ -156,7 +176,7 @@ func runLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts lo machineNames = append(machineNames, m.Machine.Name) } - formatter := logs.NewFormatter(machineNames, serviceNames, opts.UTC) + formatter := logs.NewFormatter(machineNames, foundServices, opts.UTC) // Print merged logs. for entry := range stream { diff --git a/pkg/api/logs.go b/pkg/api/logs.go index c207c9ff..2ac15db1 100644 --- a/pkg/api/logs.go +++ b/pkg/api/logs.go @@ -46,12 +46,16 @@ func LogStreamTypeToProto(s LogStreamType) pb.LogEntry_StreamType { } } -// ServiceLogsOptions specifies parameters for ServiceLogs. +// ServiceLogsOptions specifies parameters for ServiceLogs and MachineLogs. type ServiceLogsOptions struct { Follow bool Tail int Since string Until string + // Containers filters logs to only include the specified service containers (names, full IDs, + // or unique ID prefixes). If empty, logs from all containers in the service are included. + // Ignored by MachineLogs. + Containers []string // Machines filters logs to only include containers running on the specified machines (names or IDs). // If empty, logs from all machines are included. Machines []string diff --git a/pkg/api/service.go b/pkg/api/service.go index 4536b477..5b2497b6 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -519,6 +519,29 @@ type MachineServiceContainer struct { Container ServiceContainer } +// FindContainer returns the service container by exact name, ID, or unique ID prefix. +// Returns ErrNotFound if no container matches, or an error if an ID prefix matches more than one container. +func (s *Service) FindContainer(nameOrID string) (MachineServiceContainer, error) { + var prefixMatches []MachineServiceContainer + for _, c := range append(s.Containers, s.HookContainers...) { + if c.Container.ID == nameOrID || c.Container.Name == nameOrID { + return c, nil + } + if strings.HasPrefix(c.Container.ID, nameOrID) { + prefixMatches = append(prefixMatches, c) + } + } + + if len(prefixMatches) == 1 { + return prefixMatches[0], nil + } + if len(prefixMatches) > 1 { + return MachineServiceContainer{}, fmt.Errorf("multiple containers found with ID prefix '%s'", nameOrID) + } + + return MachineServiceContainer{}, ErrNotFound +} + // MachineIDs returns a list of unique machine IDs where the service containers are running. func (s *Service) MachineIDs() []string { ids := mapset.NewSet[string]() diff --git a/pkg/client/container.go b/pkg/client/container.go index 98b6baf3..a2efb1ea 100644 --- a/pkg/client/container.go +++ b/pkg/client/container.go @@ -256,26 +256,7 @@ func (cli *Client) InspectContainer( return api.MachineServiceContainer{}, fmt.Errorf("inspect service: %w", err) } - prefixMatchCandidates := []api.MachineServiceContainer{} - for _, c := range append(svc.Containers, svc.HookContainers...) { - if c.Container.ID == containerNameOrID || - c.Container.Name == containerNameOrID { - return c, nil - } - - if strings.HasPrefix(c.Container.ID, containerNameOrID) { - prefixMatchCandidates = append(prefixMatchCandidates, c) - } - } - - if len(prefixMatchCandidates) == 1 { - return prefixMatchCandidates[0], nil - } else if len(prefixMatchCandidates) > 1 { - return api.MachineServiceContainer{}, fmt.Errorf( - "multiple containers found with ID prefix '%s'", containerNameOrID) - } - - return api.MachineServiceContainer{}, api.ErrNotFound + return svc.FindContainer(containerNameOrID) } // StartContainer starts the specified container within the service. diff --git a/pkg/client/logs.go b/pkg/client/logs.go index 002a9e16..cdadb903 100644 --- a/pkg/client/logs.go +++ b/pkg/client/logs.go @@ -5,6 +5,8 @@ import ( "errors" "fmt" "io" + "maps" + "slices" "github.com/docker/docker/pkg/stringid" "github.com/psviderski/uncloud/internal/machine/api/pb" @@ -24,11 +26,25 @@ func (cli *Client) ServiceLogs( return svc, nil, fmt.Errorf("inspect service: %w", err) } - allContainers := append(svc.Containers, svc.HookContainers...) - if len(allContainers) == 0 { + containers := append(svc.Containers, svc.HookContainers...) + if len(containers) == 0 { return svc, nil, fmt.Errorf("no containers found for service: %s", serviceNameOrID) } + if len(opts.Containers) > 0 { + selected := make(map[string]api.MachineServiceContainer, len(opts.Containers)) + for _, nameOrID := range opts.Containers { + ctr, err := svc.FindContainer(nameOrID) + if err != nil { + return svc, nil, fmt.Errorf("find container '%s' in service '%s': %w", + nameOrID, serviceNameOrID, err) + } + selected[ctr.Container.ID] = ctr + } + + containers = slices.Collect(maps.Values(selected)) + } + machines, err := cli.ListMachines(ctx, &api.MachineFilter{ NamesOrIDs: opts.Machines, }) @@ -36,8 +52,8 @@ func (cli *Client) ServiceLogs( return svc, nil, fmt.Errorf("list machines: %w", err) } - ctrStreams := make([]<-chan api.ServiceLogEntry, 0, len(allContainers)) - for _, ctr := range allContainers { + ctrStreams := make([]<-chan api.ServiceLogEntry, 0, len(containers)) + for _, ctr := range containers { // Skip containers not running on the specified machines. m := machines.FindByNameOrID(ctr.MachineID) if len(opts.Machines) > 0 && m == nil { diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 7d135aee..6cdeb717 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -2026,6 +2026,52 @@ func TestServiceLifecycle(t *testing.T) { }) require.Error(t, err) assert.Contains(t, err.Error(), "machines not found") + + // Test filter logs by container. + targetCtr := svc.Containers[0].Container + expectedStdout := fmt.Sprintf("Hello from %s\n", targetCtr.Name) + expectedStderr := fmt.Sprintf("Hello stderr from %s\n", targetCtr.Name) + + // Filter by full container ID. + _, ctrStream, err := cli.ServiceLogs(ctx, name, api.ServiceLogsOptions{ + Containers: []string{targetCtr.ID}, + }) + require.NoError(t, err) + assert.Equal(t, []string{expectedStdout, expectedStderr}, collectLogs(ctrStream)) + + // Filter by container name. + _, ctrStream, err = cli.ServiceLogs(ctx, name, api.ServiceLogsOptions{ + Containers: []string{targetCtr.Name}, + }) + require.NoError(t, err) + assert.Equal(t, []string{expectedStdout, expectedStderr}, collectLogs(ctrStream)) + + // Filter by short ID prefix (12 chars is Docker's standard short ID length). + _, ctrStream, err = cli.ServiceLogs(ctx, name, api.ServiceLogsOptions{ + Containers: []string{targetCtr.ID[:12]}, + }) + require.NoError(t, err) + assert.Equal(t, []string{expectedStdout, expectedStderr}, collectLogs(ctrStream)) + + // Filter by multiple containers in the same service. + secondCtr := svc.Containers[1].Container + _, ctrStream, err = cli.ServiceLogs(ctx, name, api.ServiceLogsOptions{ + Containers: []string{targetCtr.ID, secondCtr.ID}, + }) + require.NoError(t, err) + multiLogs := collectLogs(ctrStream) + require.Len(t, multiLogs, 4, "should have 4 log entries from 2 containers") + assert.Contains(t, multiLogs, expectedStdout) + assert.Contains(t, multiLogs, expectedStderr) + assert.Contains(t, multiLogs, fmt.Sprintf("Hello from %s\n", secondCtr.Name)) + assert.Contains(t, multiLogs, fmt.Sprintf("Hello stderr from %s\n", secondCtr.Name)) + + // Test non-existent container filter returns error. + _, _, err = cli.ServiceLogs(ctx, name, api.ServiceLogsOptions{ + Containers: []string{"non-existent-container"}, + }) + require.Error(t, err) + assert.ErrorContains(t, err, "container 'non-existent-container' in service 'test-service-logs': not found") }) t.Run("internal DNS", func(t *testing.T) {