refactor(client): speed up service list call (N -> 1 broadcast RPC)

This commit is contained in:
Pasha Sviderski committed 2026-10-01 19:52:18 +10:00
1 parent 9c21ae6b08
commit 0e53f505c0
1 file changed
+25 -41
+25 -41
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"maps"
"slices" "slices"
"sync" "sync"
@@ -320,40 +321,18 @@ func (cli *Client) StartService(ctx context.Context, id string) error {
// ListServices returns a list of all services and their containers. // ListServices returns a list of all services and their containers.
func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) { func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) {
machines, err := cli.ListMachines(ctx, nil) // Broadcast the container list request to all machines in the cluster.
if err != nil { listCtx := ProxyMachinesContext(ctx, nil)
return nil, fmt.Errorf("list machines: %w", err) // List all containers including stopped ones and deployment hooks.
}
// Broadcast the container list request to all available machines.
md := metadata.New(nil)
for _, m := range machines {
if m.State == pb.MachineMember_UP || m.State == pb.MachineMember_SUSPECT {
md.Append("machines", m.Machine.Id)
} else {
tui.PrintWarning(fmt.Sprintf("failed to list service containers on machine '%s' (state is %s). "+
"The results may be incomplete.", m.Machine.Name, m.State.String()))
}
}
listCtx := metadata.NewOutgoingContext(ctx, md)
// List all containers including stopped ones.
opts := container.ListOptions{All: true} opts := container.ListOptions{All: true}
machineContainers, err := cli.Docker.ListServiceContainers(listCtx, "", opts) machineContainers, err := cli.Docker.ListServiceContainers(listCtx, "", opts)
if err != nil { if err != nil {
return nil, fmt.Errorf("list containers: %w", err) return nil, fmt.Errorf("list containers: %w", err)
} }
// TODO: optimise by extracting services from the list of all containers instead of inspecting each service. // Group containers by service ID using the broadcast results.
// Most of the code can be reused in both InspectService and ListServices.
servicesByID := make(map[string]api.Service) servicesByID := make(map[string]api.Service)
for _, mc := range machineContainers { for _, mc := range machineContainers {
// NOTE: Metadata should never be nil in practice. This is legacy fallback that will be removed.
if mc.Metadata == nil {
tui.PrintWarning("metadata is missing in response from unknown server")
continue
}
if mc.Metadata.Error != "" { if mc.Metadata.Error != "" {
// TODO: return failed machines in the response. // TODO: return failed machines in the response.
tui.PrintWarning(fmt.Sprintf("failed to list containers on machine '%s': %s", tui.PrintWarning(fmt.Sprintf("failed to list containers on machine '%s': %s",
@@ -362,25 +341,30 @@ func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) {
} }
for _, ctr := range append(mc.Containers, mc.HookContainers...) { for _, ctr := range append(mc.Containers, mc.HookContainers...) {
if _, ok := servicesByID[ctr.ServiceID()]; ok { serviceID := ctr.ServiceID()
continue svc, ok := servicesByID[serviceID]
} if !ok {
svc = api.Service{
svc, err := cli.InspectService(ctx, ctr.ServiceID()) ID: serviceID,
if err != nil { Name: ctr.ServiceName(),
if errors.Is(err, api.ErrNotFound) { Mode: ctr.ServiceMode(),
continue
} }
return nil, fmt.Errorf("inspect service: %w", err)
} }
servicesByID[ctr.ServiceID()] = svc machineContainer := api.MachineServiceContainer{
MachineID: mc.Metadata.MachineId,
MachineName: mc.Metadata.MachineName,
Container: ctr,
}
if ctr.IsHook() {
svc.HookContainers = append(svc.HookContainers, machineContainer)
} else {
svc.Containers = append(svc.Containers, machineContainer)
}
servicesByID[serviceID] = svc
} }
} }
services := make([]api.Service, 0, len(servicesByID)) return slices.Collect(maps.Values(servicesByID)), nil
for _, svc := range servicesByID {
services = append(services, svc)
}
return services, nil
} }