diff --git a/cmd/uncloud/service/logs.go b/cmd/uncloud/service/logs.go index b356168a..52de1913 100644 --- a/cmd/uncloud/service/logs.go +++ b/cmd/uncloud/service/logs.go @@ -20,11 +20,12 @@ import ( ) type logsOptions struct { - follow bool - tail string - since string - until string - utc bool + follow bool + tail string + since string + until string + utc bool + machines []string } func NewLogsCommand() *cobra.Command { @@ -44,6 +45,8 @@ func NewLogsCommand() *cobra.Command { cmd.Flags().BoolVarP(&options.follow, "follow", "f", false, "Continually stream new logs.") + cmd.Flags().StringSliceVarP(&options.machines, "machine", "m", nil, + "Filter logs by machine name or ID. Can be specified multiple times or as a comma-separated list.") cmd.Flags().StringVarP(&options.tail, "tail", "n", "100", "Show the most recent logs and limit the number of lines shown per replica. Use 'all' to show all logs.") cmd.Flags().StringVar(&options.since, "since", "", @@ -82,17 +85,17 @@ func streamLogs(ctx context.Context, uncli *cli.CLI, serviceNames []string, opts defer c.Close() logsOpts := api.ServiceLogsOptions{ - Follow: opts.follow, - Tail: tail, - Since: opts.since, - Until: opts.until, + Follow: opts.follow, + Tail: tail, + Since: opts.since, + Until: opts.until, + Machines: cli.ExpandCommaSeparatedValues(opts.machines), } // Collect log streams from all services. machineIDsSet := mapset.NewSet[string]() svcStreams := make([]<-chan api.ServiceLogEntry, 0, len(serviceNames)) for _, serviceName := range serviceNames { - // TODO: set Heartbeats in the opts. svc, ch, err := c.ServiceLogs(ctx, serviceName, logsOpts) if err != nil { return fmt.Errorf("stream logs for service '%s': %w", serviceName, err) diff --git a/pkg/api/logs.go b/pkg/api/logs.go index 80660fb0..c0de7e0f 100644 --- a/pkg/api/logs.go +++ b/pkg/api/logs.go @@ -52,6 +52,9 @@ type ServiceLogsOptions struct { Tail int Since string Until 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 } // ServiceLogEntry represents a single log entry from a service container. diff --git a/pkg/client/image.go b/pkg/client/image.go index 8e4027ff..16f92d5e 100644 --- a/pkg/client/image.go +++ b/pkg/client/image.go @@ -149,18 +149,6 @@ func (cli *Client) PushImage(ctx context.Context, image string, opts PushImageOp return fmt.Errorf("list machines: %w", err) } - // Check if all specified machines were found. - if len(machineMembers) != len(opts.Machines) { - var notFound []string - for _, nameOrID := range opts.Machines { - if machineMembers.FindByNameOrID(nameOrID) == nil { - notFound = append(notFound, nameOrID) - } - } - - return fmt.Errorf("machines not found: %s", strings.Join(notFound, ", ")) - } - for _, mm := range machineMembers { machines = append(machines, mm.Machine) } diff --git a/pkg/client/logs.go b/pkg/client/logs.go index 0e5517ef..651ed46c 100644 --- a/pkg/client/logs.go +++ b/pkg/client/logs.go @@ -2,6 +2,7 @@ package client import ( "context" + "errors" "fmt" "io" @@ -27,16 +28,23 @@ func (cli *Client) ServiceLogs( return svc, nil, fmt.Errorf("no containers found for service: %s", serviceNameOrID) } - machines, err := cli.ListMachines(ctx, nil) + machines, err := cli.ListMachines(ctx, &api.MachineFilter{ + NamesOrIDs: opts.Machines, + }) if err != nil { return svc, nil, fmt.Errorf("list machines: %w", err) } ctrStreams := make([]<-chan api.ServiceLogEntry, 0, len(svc.Containers)) for _, ctr := range svc.Containers { - // Try to get machine name for ServiceLogEntry metadata and friendlier error message. - machineName := ctr.MachineID + // Skip containers not running on the specified machines. m := machines.FindByNameOrID(ctr.MachineID) + if len(opts.Machines) > 0 && m == nil { + continue + } + + // Machine name for ServiceLogEntry metadata and friendlier error message. + machineName := ctr.MachineID if m != nil { machineName = m.Machine.Name } @@ -59,6 +67,10 @@ func (cli *Client) ServiceLogs( ctrStreams = append(ctrStreams, enrichedStream) } + if len(ctrStreams) == 0 { + return svc, nil, errors.New("no service containers found on the specified machine(s)") + } + // Use the log merger to combine streams from all containers in chronological order. merger := NewLogMerger(ctrStreams, DefaultLogMergerOptions) mergedStream := merger.Stream() diff --git a/pkg/client/machine.go b/pkg/client/machine.go index 81093181..534800a6 100644 --- a/pkg/client/machine.go +++ b/pkg/client/machine.go @@ -2,7 +2,8 @@ package client import ( "context" - "slices" + "fmt" + "strings" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" @@ -32,42 +33,44 @@ func (cli *Client) ListMachines(ctx context.Context, filter *api.MachineFilter) if err != nil { return nil, err } + machines := api.MachineMembersList(resp.Machines) - machines := resp.Machines + if filter == nil { + return machines, nil + } - if filter != nil { - var matchedMachines api.MachineMembersList - for _, m := range machines { - if MachineMatchesFilter(m, filter) { - matchedMachines = append(matchedMachines, m) + // Apply the filter. + if len(filter.NamesOrIDs) > 0 { + var matched api.MachineMembersList + var notFound []string + + for _, nameOrID := range filter.NamesOrIDs { + if m := machines.FindByNameOrID(nameOrID); m != nil { + matched = append(matched, m) + } else { + notFound = append(notFound, nameOrID) } } - machines = matchedMachines + machines = matched + + if len(notFound) > 0 { + return nil, fmt.Errorf("machines not found: %s", strings.Join(notFound, ", ")) + } + } + + if filter.Available { + var available api.MachineMembersList + for _, m := range machines { + if m.State != pb.MachineMember_DOWN { + available = append(available, m) + } + } + machines = available } return machines, nil } -func MachineMatchesFilter(machine *pb.MachineMember, filter *api.MachineFilter) bool { - if filter == nil { - return true - } - - if filter.Available && machine.State == pb.MachineMember_DOWN { - return false - } - - if len(filter.NamesOrIDs) > 0 { - if !slices.ContainsFunc(filter.NamesOrIDs, func(nameOrID string) bool { - return machine.Machine.Id == nameOrID || machine.Machine.Name == nameOrID - }) { - return false - } - } - - return true -} - // UpdateMachine updates machine configuration in the cluster. func (cli *Client) UpdateMachine(ctx context.Context, req *pb.UpdateMachineRequest) (*pb.MachineInfo, error) { resp, err := cli.ClusterClient.UpdateMachine(ctx, req)