refactor(grpc-proxy): use Metadata.Err() method to get upstream error for a broadcasted request

This commit is contained in:
Pasha Sviderski
2026-10-27 10:05:35 +10:00
parent bb8fb3a380
commit fab2a20242
9 changed files with 28 additions and 28 deletions
+8
View File
@@ -4,8 +4,16 @@ import (
"bytes" "bytes"
"fmt" "fmt"
"net/netip" "net/netip"
"google.golang.org/grpc/status"
) )
// Err returns the upstream error, preserving its gRPC status code.
// It returns nil for nil metadata or an absent or successful status.
func (m *Metadata) Err() error {
return status.ErrorProto(m.GetStatus())
}
func NewIP(addr netip.Addr) *IP { func NewIP(addr netip.Addr) *IP {
// MarshalBinary always returns a nil error. // MarshalBinary always returns a nil error.
ip, _ := addr.MarshalBinary() ip, _ := addr.MarshalBinary()
+1 -1
View File
@@ -103,7 +103,7 @@ func list(ctx context.Context, uncli *cli.CLI, opts listOptions) error {
var rows []imageRow var rows []imageRow
for _, machineImages := range clusterImages { for _, machineImages := range clusterImages {
if err := machineImages.Error(); err != nil { if err := machineImages.Metadata.Err(); err != nil {
tui.PrintWarning(fmt.Sprintf("failed to list images on machine '%s': %s", tui.PrintWarning(fmt.Sprintf("failed to list images on machine '%s': %s",
machineImages.Metadata.MachineName, err)) machineImages.Metadata.MachineName, err))
continue continue
+2 -2
View File
@@ -54,8 +54,8 @@ func rtt(ctx context.Context, uncli *cli.CLI) error {
tui.PrintWarning("metadata is missing in response from unknown server") tui.PrintWarning("metadata is missing in response from unknown server")
continue continue
} }
if m.Metadata.Error != "" { if err := m.Metadata.Err(); err != nil {
tui.PrintWarning(fmt.Sprintf("failed to inspect machine '%s': %s", m.Metadata.MachineName, m.Metadata.Error)) tui.PrintWarning(fmt.Sprintf("failed to inspect machine '%s': %s", m.Metadata.MachineName, err))
continue continue
} }
if m.Machine == nil { if m.Machine == nil {
+2 -2
View File
@@ -207,9 +207,9 @@ func collectContainers(ctx context.Context, cli *client.Client) ([]containerInfo
machineName := msc.Metadata.MachineName machineName := msc.Metadata.MachineName
if msc.Metadata.Error != "" { if err := msc.Metadata.Err(); err != nil {
tui.PrintWarning(fmt.Sprintf("failed to list service containers on machine %s: %s", tui.PrintWarning(fmt.Sprintf("failed to list service containers on machine %s: %s",
machineName, msc.Metadata.Error)) machineName, err))
continue continue
} }
+6 -6
View File
@@ -167,7 +167,7 @@ func (c *Client) ListContainers(ctx context.Context, opts container.ListOptions)
machineContainers := make([]MachineContainers, len(resp.Messages)) machineContainers := make([]MachineContainers, len(resp.Messages))
for i, msg := range resp.Messages { for i, msg := range resp.Messages {
machineContainers[i].Metadata = msg.Metadata machineContainers[i].Metadata = msg.Metadata
if msg.Metadata != nil && msg.Metadata.Error != "" { if msg.Metadata.Err() != nil {
continue continue
} }
@@ -258,7 +258,7 @@ func (c *Client) InspectImage(ctx context.Context, id string) ([]api.MachineImag
notFoundCount := 0 notFoundCount := 0
for _, msg := range resp.Messages { for _, msg := range resp.Messages {
if msg.Metadata != nil && msg.Metadata.Status != nil && codes.Code(msg.Metadata.Status.Code) == codes.NotFound { if status.Code(msg.Metadata.Err()) == codes.NotFound {
notFoundCount++ notFoundCount++
} }
} }
@@ -269,7 +269,7 @@ func (c *Client) InspectImage(ctx context.Context, id string) ([]api.MachineImag
images := make([]api.MachineImage, len(resp.Messages)) images := make([]api.MachineImage, len(resp.Messages))
for i, msg := range resp.Messages { for i, msg := range resp.Messages {
images[i].Metadata = msg.Metadata images[i].Metadata = msg.Metadata
if msg.Metadata != nil && msg.Metadata.Error != "" { if msg.Metadata.Err() != nil {
continue continue
} }
@@ -312,7 +312,7 @@ func parseRemoteImageMessage(msg *pb.RemoteImage) (api.MachineRemoteImage, error
mri := api.MachineRemoteImage{ mri := api.MachineRemoteImage{
Metadata: msg.Metadata, Metadata: msg.Metadata,
} }
if msg.Metadata != nil && msg.Metadata.Error != "" { if msg.Metadata.Err() != nil {
return mri, nil return mri, nil
} }
@@ -393,7 +393,7 @@ func (c *Client) ListVolumes(ctx context.Context, opts volume.ListOptions) ([]Ma
machineVolumes := make([]MachineVolumes, len(resp.Messages)) machineVolumes := make([]MachineVolumes, len(resp.Messages))
for i, msg := range resp.Messages { for i, msg := range resp.Messages {
machineVolumes[i].Metadata = msg.Metadata machineVolumes[i].Metadata = msg.Metadata
if msg.Metadata != nil && msg.Metadata.Error != "" { if msg.Metadata.Err() != nil {
continue continue
} }
@@ -472,7 +472,7 @@ func (c *Client) ListServiceContainers(
machineContainers := make([]MachineServiceContainers, len(resp.Messages)) machineContainers := make([]MachineServiceContainers, len(resp.Messages))
for i, msg := range resp.Messages { for i, msg := range resp.Messages {
machineContainers[i].Metadata = msg.Metadata machineContainers[i].Metadata = msg.Metadata
if msg.Metadata != nil && msg.Metadata.Error != "" { if msg.Metadata.Err() != nil {
continue continue
} }
-9
View File
@@ -1,7 +1,6 @@
package api package api
import ( import (
"errors"
"slices" "slices"
"github.com/distribution/reference" "github.com/distribution/reference"
@@ -25,14 +24,6 @@ type MachineImages struct {
ContainerdStore bool ContainerdStore bool
} }
// Error returns a non-nil error if listing images on this machine failed.
func (mi MachineImages) Error() error {
if mi.Metadata != nil && mi.Metadata.Error != "" {
return errors.New(mi.Metadata.Error)
}
return nil
}
// ImageFilter defines criteria to filter images in ListImages. // ImageFilter defines criteria to filter images in ListImages.
type ImageFilter struct { type ImageFilter struct {
// Machines filters images to those present on the specified machines (names or IDs). // Machines filters images to those present on the specified machines (names or IDs).
+3 -2
View File
@@ -92,7 +92,7 @@ func (cli *Client) ListImages(ctx context.Context, filter api.ImageFilter) ([]ap
ContainerdStore: msg.ContainerdStore, ContainerdStore: msg.ContainerdStore,
} }
if msg.Metadata.Error == "" && len(msg.Images) > 0 { if msg.Metadata.Err() == nil && len(msg.Images) > 0 {
if err = json.Unmarshal(msg.Images, &mi.Images); err != nil { if err = json.Unmarshal(msg.Images, &mi.Images); err != nil {
return nil, fmt.Errorf("unmarshal images: %w", err) return nil, fmt.Errorf("unmarshal images: %w", err)
} }
@@ -104,6 +104,7 @@ func (cli *Client) ListImages(ctx context.Context, filter api.ImageFilter) ([]ap
return machineImages, nil return machineImages, nil
} }
// TODO: move PushImageOptions to api package and include PushImage in ImageClient interface.
type PushImageOptions struct { type PushImageOptions struct {
// AllMachines pushes the image to all machines in the cluster. Takes precedence over Machines field. // AllMachines pushes the image to all machines in the cluster. Takes precedence over Machines field.
AllMachines bool AllMachines bool
@@ -215,7 +216,7 @@ func (cli *Client) pushImageToMachine(
Name: "%invalid-name-to-only-check-store-type%", Name: "%invalid-name-to-only-check-store-type%",
}) })
if err == nil { if err == nil {
err = images[0].Error() err = images[0].Metadata.Err()
} }
if err != nil { if err != nil {
return fmt.Errorf("check Docker image store type on machine '%s': %w", machine.Name, err) return fmt.Errorf("check Docker image store type on machine '%s': %w", machine.Name, err)
+4 -4
View File
@@ -125,10 +125,10 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser
continue continue
} }
if mc.Metadata.Error != "" { if err := mc.Metadata.Err(); err != nil {
// 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",
mc.Metadata.MachineName, mc.Metadata.Error)) mc.Metadata.MachineName, err))
continue continue
} }
@@ -333,10 +333,10 @@ func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) {
// Group containers by service ID using the broadcast results. // Group containers by service ID using the broadcast results.
servicesByID := make(map[string]api.Service) servicesByID := make(map[string]api.Service)
for _, mc := range machineContainers { for _, mc := range machineContainers {
if mc.Metadata.Error != "" { if err := mc.Metadata.Err(); err != nil {
// 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",
mc.Metadata.MachineName, mc.Metadata.Error)) mc.Metadata.MachineName, err))
continue continue
} }
+2 -2
View File
@@ -70,10 +70,10 @@ func (cli *Client) ListVolumes(ctx context.Context, filter *api.VolumeFilter) ([
continue continue
} }
if mv.Metadata.Error != "" { if err := mv.Metadata.Err(); err != nil {
// TODO: return failed machines in the response. // TODO: return failed machines in the response.
tui.PrintWarning(fmt.Sprintf("failed to list volumes on machine '%s': %s", mv.Metadata.MachineName, tui.PrintWarning(fmt.Sprintf("failed to list volumes on machine '%s': %s", mv.Metadata.MachineName,
mv.Metadata.Error)) err))
continue continue
} }