feat: support filtering service logs by machine (-m/--machine)

cmd/uncloud/service/logs.go

@@ -20,11 +20,12 @@ import (
 )
 
 type logsOptions struct {
-	follow bool
+	follow   bool
-	tail   string
+	tail     string
-	since  string
+	since    string
-	until  string
+	until    string
-	utc    bool
+	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,
+		Follow:   opts.follow,
-		Tail:   tail,
+		Tail:     tail,
-		Since:  opts.since,
+		Since:    opts.since,
-		Until:  opts.until,
+		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)

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.

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)
 		}

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.
+		// Skip containers not running on the specified machines.
-		machineName := ctr.MachineID
 		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()

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 {
+	// Apply the filter.
-		var matchedMachines api.MachineMembersList
+	if len(filter.NamesOrIDs) > 0 {
-		for _, m := range machines {
+		var matched api.MachineMembersList
-			if MachineMatchesFilter(m, filter) {
+		var notFound []string
-				matchedMachines = append(matchedMachines, m)
+
+		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)