diff --git a/cmd/uncloud/caddy/config.go b/cmd/uncloud/caddy/config.go index c159df72..6fa08260 100644 --- a/cmd/uncloud/caddy/config.go +++ b/cmd/uncloud/caddy/config.go @@ -48,10 +48,7 @@ func runConfig(ctx context.Context, uncli *cli.CLI, opts configOptions) error { if opts.machine != "" { // If a specific machine is requested, use it to get the Caddy configuration. - ctx, _, err = clusterClient.ProxyMachinesContext(ctx, []string{opts.machine}) - if err != nil { - return err - } + ctx = clusterClient.ProxySingleMachineContext(ctx, opts.machine) } config, err := clusterClient.Caddy.GetConfig(ctx, nil) diff --git a/cmd/uncloud/image/ls.go b/cmd/uncloud/image/ls.go index 324a7e08..a0847bde 100644 --- a/cmd/uncloud/image/ls.go +++ b/cmd/uncloud/image/ls.go @@ -88,19 +88,6 @@ func list(ctx context.Context, uncli *cli.CLI, opts listOptions) error { } defer clusterClient.Close() - // Get all machines to create ID to name mapping. - allMachines, err := clusterClient.ListMachines(ctx, nil) - if err != nil { - return fmt.Errorf("list machines: %w", err) - } - - machineIDToName := make(map[string]string) - for _, machineMember := range allMachines { - if machineMember.Machine != nil && machineMember.Machine.Id != "" && machineMember.Machine.Name != "" { - machineIDToName[machineMember.Machine.Id] = machineMember.Machine.Name - } - } - machines := cli.ExpandCommaSeparatedValues(opts.machines) clusterImages, err := clusterClient.ListImages(ctx, api.ImageFilter{ @@ -116,10 +103,7 @@ func list(ctx context.Context, uncli *cli.CLI, opts listOptions) error { for _, machineImages := range clusterImages { // Get machine name for better readability. - machineName := machineImages.Metadata.Machine - if m := allMachines.FindByNameOrID(machineName); m != nil { - machineName = m.Machine.Name - } + machineName := machineImages.Metadata.MachineName store := "docker" if machineImages.ContainerdStore { diff --git a/cmd/uncloud/machine/rm.go b/cmd/uncloud/machine/rm.go index 537079e0..3e749a85 100644 --- a/cmd/uncloud/machine/rm.go +++ b/cmd/uncloud/machine/rm.go @@ -63,15 +63,16 @@ func remove(ctx context.Context, uncli *cli.CLI, nameOrID string, opts removeOpt } defer client.Close() - // Verify the machine exists and list all service containers on it including stopped ones. - mctx, machines, err := client.ProxyMachinesContext(ctx, []string{nameOrID}) + // Verify the machine exists in the cluster. + member, err := client.InspectMachine(ctx, nameOrID) if err != nil { - return err + return fmt.Errorf("inspect machine '%s': %w", nameOrID, err) } - if len(machines) == 0 { - return fmt.Errorf("machine '%s' not found in the cluster", nameOrID) - } - m := machines[0].Machine + m := member.Machine + + // Create a proxy context for the machine being removed. + // This is used for calls that need to run directly on that machine. + rmCtx := client.ProxySingleMachineContext(ctx, m.Id) // Verify if the machine being removed is the proxy machine we're connected to. proxyMachine, err := client.MachineClient.Inspect(ctx, nil) @@ -103,7 +104,7 @@ func remove(ctx context.Context, uncli *cli.CLI, nameOrID string, opts removeOpt if reset { // Check if the machine is up and has service containers. listOpts := container.ListOptions{All: true} - machineContainers, err := client.Docker.ListServiceContainers(mctx, "", listOpts) + machineContainers, err := client.Docker.ListServiceContainers(rmCtx, "", listOpts) if err == nil { reachable = true containers = machineContainers[0].Containers @@ -156,7 +157,7 @@ func remove(ctx context.Context, uncli *cli.CLI, nameOrID string, opts removeOpt fmt.Printf("Machine '%s' removed from the cluster.\n", m.Name) if reset && reachable { - _, err = client.MachineClient.Reset(mctx, &pb.ResetRequest{}) + _, err = client.MachineClient.Reset(rmCtx, &pb.ResetRequest{}) if err != nil { fmt.Printf("WARNING: Failed to reset machine: %v\n", err) } else { diff --git a/cmd/uncloud/machine/rtt.go b/cmd/uncloud/machine/rtt.go index 2a0d51eb..387ba212 100644 --- a/cmd/uncloud/machine/rtt.go +++ b/cmd/uncloud/machine/rtt.go @@ -38,10 +38,7 @@ func rtt(ctx context.Context, uncli *cli.CLI) error { defer client.Close() // Setup context to proxy request to all machines. - ctx, _, err = client.ProxyMachinesContext(ctx, nil) - if err != nil { - return fmt.Errorf("setup proxy context: %w", err) - } + ctx = client.ProxyMachinesContext(ctx, nil) resp, err := client.MachineClient.InspectMachine(ctx, &emptypb.Empty{}) if err != nil { @@ -51,6 +48,15 @@ func rtt(ctx context.Context, uncli *cli.CLI) error { // Map machine IDs to names for display from the response. machineNames := make(map[string]string) for _, m := range resp.Machines { + // NOTE: Metadata should never be nil in practice. This is legacy fallback that will be removed. + if m.Metadata == nil { + tui.PrintWarning("metadata is missing in response from unknown server") + continue + } + if m.Metadata.Error != "" { + tui.PrintWarning(fmt.Sprintf("failed to inspect machine '%s': %s", m.Metadata.MachineName, m.Metadata.Error)) + continue + } if m.Machine == nil { continue } diff --git a/cmd/uncloud/ps.go b/cmd/uncloud/ps.go index ecde60f9..e5196bde 100644 --- a/cmd/uncloud/ps.go +++ b/cmd/uncloud/ps.go @@ -198,18 +198,7 @@ func printContainers(containers []containerInfo) error { } func collectContainers(ctx context.Context, cli *client.Client) ([]containerInfo, error) { - listCtx, machines, err := cli.ProxyMachinesContext(ctx, nil) - if err != nil { - return nil, fmt.Errorf("proxy machines context: %w", err) - } - - // Create a map of IP to machine name for resolving response metadata - machinesNamesByIP := make(map[string]string) - for _, m := range machines { - if addr, err := m.Machine.Network.ManagementIp.ToAddr(); err == nil { - machinesNamesByIP[addr.String()] = m.Machine.Name - } - } + listCtx := cli.ProxyMachinesContext(ctx, nil) // List all service containers across all machines in the cluster. machineContainers, err := cli.Docker.ListServiceContainers( @@ -221,29 +210,17 @@ func collectContainers(ctx context.Context, cli *client.Client) ([]containerInfo var containers []containerInfo for _, msc := range machineContainers { - // Metadata can be nil if the request was broadcasted to only one machine. - if msc.Metadata == nil && len(machineContainers) > 1 { - return nil, fmt.Errorf("something went wrong with gRPC proxy: metadata is missing for a machine response") + // NOTE: Metadata should never be nil in practice. This is legacy fallback that will be removed. + if msc.Metadata == nil { + tui.PrintWarning("metadata is missing in response from unknown server") + continue } - machineName := "unknown" - if msc.Metadata != nil { - var ok bool - machineName, ok = machinesNamesByIP[msc.Metadata.Machine] - if !ok { - // Fallback to machine's IP as name. - machineName = msc.Metadata.Machine - } - } else { - // Fallback to the first available machine name. - if len(machines) > 0 { - machineName = machines[0].Machine.Name - } - } + machineName := msc.Metadata.MachineName - if msc.Metadata != nil && msc.Metadata.Error != "" { - tui.PrintWarning(fmt.Sprintf("failed to list containers on machine %s: %s", machineName, - msc.Metadata.Error)) + if msc.Metadata.Error != "" { + tui.PrintWarning(fmt.Sprintf("failed to list service containers on machine %s: %s", + machineName, msc.Metadata.Error)) continue } diff --git a/cmd/uncloud/ps_test.go b/cmd/uncloud/ps_test.go index 8dd7c404..63dad60b 100644 --- a/cmd/uncloud/ps_test.go +++ b/cmd/uncloud/ps_test.go @@ -3,7 +3,6 @@ package main import ( "context" "encoding/json" - "net/netip" "testing" "github.com/psviderski/uncloud/internal/machine/api/pb" @@ -12,204 +11,20 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" - "google.golang.org/protobuf/types/known/emptypb" ) -// mockDockerClient implements pb.DockerClient type mockDockerClient struct { - pb.DockerClient // Embed to avoid implementing all methods - listResp *pb.ListServiceContainersResponse - listErr error + pb.DockerClient + listResp *pb.ListServiceContainersResponse + listErr error } func (m *mockDockerClient) ListServiceContainers(ctx context.Context, in *pb.ListServiceContainersRequest, opts ...grpc.CallOption) (*pb.ListServiceContainersResponse, error) { return m.listResp, m.listErr } -// mockClusterClient implements pb.ClusterClient -type mockClusterClient struct { - pb.ClusterClient // Embed to avoid implementing all methods - machinesResp *pb.ListMachinesResponse - machinesErr error -} - -func (m *mockClusterClient) ListMachines(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*pb.ListMachinesResponse, error) { - return m.machinesResp, m.machinesErr -} - -func TestCollectContainers_NilMetadata(t *testing.T) { - // Setup container data - containerData := map[string]any{ - "Id": "container1", - "Name": "test-container", - "Config": map[string]any{ - "Image": "test-image", - }, - "State": map[string]any{ - "Status": "running", - "StartedAt": "2023-01-01T12:00:00Z", - "FinishedAt": "0001-01-01T00:00:00Z", - }, - "NetworkSettings": map[string]any{ - "Networks": map[string]any{}, - }, - } - containerJSON, err := json.Marshal(containerData) - require.NoError(t, err) - - serviceSpecJSON, err := json.Marshal(map[string]any{}) - require.NoError(t, err) - - // Setup mocks - mockDocker := &mockDockerClient{ - listResp: &pb.ListServiceContainersResponse{ - Messages: []*pb.MachineServiceContainers{ - { - Metadata: nil, // Simulating the issue: nil metadata - Containers: []*pb.ServiceContainer{ - { - Container: containerJSON, - ServiceSpec: serviceSpecJSON, - }, - }, - }, - }, - }, - } - - machineIP := "10.0.0.1" - mockCluster := &mockClusterClient{ - machinesResp: &pb.ListMachinesResponse{ - Machines: []*pb.MachineMember{ - { - Machine: &pb.MachineInfo{ - Name: "machine-1", - Network: &pb.NetworkConfig{ - ManagementIp: pb.NewIP(netip.MustParseAddr(machineIP)), - }, - }, - State: pb.MachineMember_UP, - }, - }, - }, - } - - // Construct client with mocks - cli := &client.Client{ - Docker: &docker.Client{ - GRPCClient: mockDocker, - }, - ClusterClient: mockCluster, - } - - // Execute - containers, err := collectContainers(context.Background(), cli) - require.NoError(t, err) - - // Verify - assert.Len(t, containers, 1) - if len(containers) > 0 { - c := containers[0] - assert.Equal(t, "container1", c.id) - assert.Equal(t, "machine-1", c.machineName, "Should fall back to the single machine name when metadata is nil") - } -} - -func TestCollectContainers_NilMetadata_MultipleMachines_Error(t *testing.T) { - // If we have multiple machines but receive nil metadata, it should return an error as it is ambiguous - - // Setup container data - containerData1 := map[string]any{ - "Id": "container1", - } - containerJSON1, _ := json.Marshal(containerData1) - - containerData2 := map[string]any{ - "Id": "container2", - "Config": map[string]any{ - "Image": "test-image", - }, - "State": map[string]any{ - "Status": "running", - "StartedAt": "2023-01-01T12:00:00Z", - "FinishedAt": "0001-01-01T00:00:00Z", - }, - "NetworkSettings": map[string]any{ - "Networks": map[string]any{}, - }, - } - containerJSON2, _ := json.Marshal(containerData2) - - serviceSpecJSON, _ := json.Marshal(map[string]any{}) - - // Setup mocks - mockDocker := &mockDockerClient{ - listResp: &pb.ListServiceContainersResponse{ - Messages: []*pb.MachineServiceContainers{ - { - Metadata: nil, // Nil metadata - Containers: []*pb.ServiceContainer{ - { - Container: containerJSON1, - ServiceSpec: serviceSpecJSON, - }, - }, - }, - { - Metadata: &pb.Metadata{Machine: "10.0.0.2"}, - Containers: []*pb.ServiceContainer{ - { - Container: containerJSON2, - ServiceSpec: serviceSpecJSON, - }, - }, - }, - }, - }, - } - - mockCluster := &mockClusterClient{ - machinesResp: &pb.ListMachinesResponse{ - Machines: []*pb.MachineMember{ - { - Machine: &pb.MachineInfo{ - Name: "machine-1", - Network: &pb.NetworkConfig{ - ManagementIp: pb.NewIP(netip.MustParseAddr("10.0.0.1")), - }, - }, - }, - { - Machine: &pb.MachineInfo{ - Name: "machine-2", - Network: &pb.NetworkConfig{ - ManagementIp: pb.NewIP(netip.MustParseAddr("10.0.0.2")), - }, - }, - }, - }, - }, - } - - // Construct client with mocks - cli := &client.Client{ - Docker: &docker.Client{ - GRPCClient: mockDocker, - }, - ClusterClient: mockCluster, - } - - // Execute - _, err := collectContainers(context.Background(), cli) - require.Error(t, err) - assert.Contains(t, err.Error(), "metadata is missing for a machine response") -} - -func TestCollectContainers_MetadataPresent_MultipleMachines(t *testing.T) { - // Verify correct mapping of containers to machines when metadata is present - - // Setup container data - containerData1 := map[string]any{ +func TestCollectContainers(t *testing.T) { + containerData1 := map[string]interface{}{ "Id": "container1", "Name": "container-1", "Config": map[string]any{ @@ -245,12 +60,11 @@ func TestCollectContainers_MetadataPresent_MultipleMachines(t *testing.T) { serviceSpecJSON, _ := json.Marshal(map[string]any{}) - // Setup mocks mockDocker := &mockDockerClient{ listResp: &pb.ListServiceContainersResponse{ Messages: []*pb.MachineServiceContainers{ { - Metadata: &pb.Metadata{Machine: "10.0.0.1"}, + Metadata: &pb.Metadata{MachineAddr: "10.0.0.1", MachineName: "machine-1"}, Containers: []*pb.ServiceContainer{ { Container: containerJSON1, @@ -259,7 +73,7 @@ func TestCollectContainers_MetadataPresent_MultipleMachines(t *testing.T) { }, }, { - Metadata: &pb.Metadata{Machine: "10.0.0.2"}, + Metadata: &pb.Metadata{MachineAddr: "10.0.0.2", MachineName: "machine-2"}, Containers: []*pb.ServiceContainer{ { Container: containerJSON2, @@ -271,41 +85,14 @@ func TestCollectContainers_MetadataPresent_MultipleMachines(t *testing.T) { }, } - mockCluster := &mockClusterClient{ - machinesResp: &pb.ListMachinesResponse{ - Machines: []*pb.MachineMember{ - { - Machine: &pb.MachineInfo{ - Name: "machine-1", - Network: &pb.NetworkConfig{ - ManagementIp: pb.NewIP(netip.MustParseAddr("10.0.0.1")), - }, - }, - }, - { - Machine: &pb.MachineInfo{ - Name: "machine-2", - Network: &pb.NetworkConfig{ - ManagementIp: pb.NewIP(netip.MustParseAddr("10.0.0.2")), - }, - }, - }, - }, - }, - } - cli := &client.Client{ - Docker: &docker.Client{GRPCClient: mockDocker}, - ClusterClient: mockCluster, + Docker: &docker.Client{GRPCClient: mockDocker}, } containers, err := collectContainers(context.Background(), cli) require.NoError(t, err) assert.Len(t, containers, 2) - // Order is not guaranteed by the map iteration in logic or parallel fetch (though here it's mocked sequential), - // but collectContainers just appends. - // We'll find them by ID. for _, c := range containers { if c.id == "container1" { assert.Equal(t, "machine-1", c.machineName) @@ -316,126 +103,3 @@ func TestCollectContainers_MetadataPresent_MultipleMachines(t *testing.T) { } } } - -func TestCollectContainers_NilMetadata_NoMachines(t *testing.T) { - // Case: 1 msc with nil metadata but no machines at all - - containerData := map[string]any{ - "Id": "container1", - "Config": map[string]any{ - "Image": "test-image", - }, - "State": map[string]any{ - "Status": "running", - "StartedAt": "2023-01-01T12:00:00Z", - "FinishedAt": "0001-01-01T00:00:00Z", - }, - "NetworkSettings": map[string]any{ - "Networks": map[string]any{}, - }, - } - containerJSON, _ := json.Marshal(containerData) - serviceSpecJSON, _ := json.Marshal(map[string]any{}) - - mockDocker := &mockDockerClient{ - listResp: &pb.ListServiceContainersResponse{ - Messages: []*pb.MachineServiceContainers{ - { - Metadata: nil, - Containers: []*pb.ServiceContainer{ - { - Container: containerJSON, - ServiceSpec: serviceSpecJSON, - }, - }, - }, - }, - }, - } - - // No machines in cluster response - mockCluster := &mockClusterClient{ - machinesResp: &pb.ListMachinesResponse{ - Machines: []*pb.MachineMember{}, - }, - } - - cli := &client.Client{ - Docker: &docker.Client{GRPCClient: mockDocker}, - ClusterClient: mockCluster, - } - - containers, err := collectContainers(context.Background(), cli) - require.NoError(t, err) - - assert.Len(t, containers, 1) - if len(containers) > 0 { - assert.Equal(t, "unknown", containers[0].machineName) - } -} - -func TestCollectContainers_MetadataPresent_NotInMapping(t *testing.T) { - // Case: msc with metadata that is not in the IP-to-name mapping - - containerData := map[string]any{ - "Id": "container1", - "Config": map[string]any{ - "Image": "test-image", - }, - "State": map[string]any{ - "Status": "running", - "StartedAt": "2023-01-01T12:00:00Z", - "FinishedAt": "0001-01-01T00:00:00Z", - }, - "NetworkSettings": map[string]any{ - "Networks": map[string]any{}, - }, - } - containerJSON, _ := json.Marshal(containerData) - serviceSpecJSON, _ := json.Marshal(map[string]any{}) - - mockDocker := &mockDockerClient{ - listResp: &pb.ListServiceContainersResponse{ - Messages: []*pb.MachineServiceContainers{ - { - Metadata: &pb.Metadata{Machine: "10.0.0.99"}, // Unknown IP - Containers: []*pb.ServiceContainer{ - { - Container: containerJSON, - ServiceSpec: serviceSpecJSON, - }, - }, - }, - }, - }, - } - - mockCluster := &mockClusterClient{ - machinesResp: &pb.ListMachinesResponse{ - Machines: []*pb.MachineMember{ - { - Machine: &pb.MachineInfo{ - Name: "machine-1", - Network: &pb.NetworkConfig{ - ManagementIp: pb.NewIP(netip.MustParseAddr("10.0.0.1")), - }, - }, - }, - }, - }, - } - - cli := &client.Client{ - Docker: &docker.Client{GRPCClient: mockDocker}, - ClusterClient: mockCluster, - } - - containers, err := collectContainers(context.Background(), cli) - require.NoError(t, err) - - assert.Len(t, containers, 1) - if len(containers) > 0 { - // Should fallback to the IP/string in metadata - assert.Equal(t, "10.0.0.99", containers[0].machineName) - } -} diff --git a/cmd/uncloud/service/inspect.go b/cmd/uncloud/service/inspect.go index 09ab7156..3364c563 100644 --- a/cmd/uncloud/service/inspect.go +++ b/cmd/uncloud/service/inspect.go @@ -54,15 +54,6 @@ func inspect(ctx context.Context, uncli *cli.CLI, opts inspectOptions) error { return fmt.Errorf("inspect service: %w", err) } - machines, err := client.ListMachines(ctx, nil) - if err != nil { - return fmt.Errorf("list machines: %w", err) - } - machinesNamesByID := make(map[string]string) - for _, m := range machines { - machinesNamesByID[m.Machine.Id] = m.Machine.Name - } - fmt.Printf("Service ID: %s\n", svc.ID) fmt.Printf("Name: %s\n", svc.Name) fmt.Printf("Mode: %s\n", svc.Mode) @@ -97,7 +88,7 @@ func inspect(ctx context.Context, uncli *cli.CLI, opts inspectOptions) error { for _, ctr := range allContainers { created := units.HumanDuration(now.Sub(createdTimes[ctr.Container.ID])) + " ago" - machine := machinesNamesByID[ctr.MachineID] + machine := ctr.MachineName if machine == "" { machine = ctr.MachineID } diff --git a/cmd/uncloud/service/logs.go b/cmd/uncloud/service/logs.go index 218ee424..44acbf3e 100644 --- a/cmd/uncloud/service/logs.go +++ b/cmd/uncloud/service/logs.go @@ -74,8 +74,6 @@ If no services are specified, streams logs from all services defined in the Comp cmd.Flags().AddFlagSet(logs.Flags(&options)) completion.MachinesFlag(cmd) - completion.MachinesFlag(cmd) - return cmd } diff --git a/cmd/uncloud/wg/wg.go b/cmd/uncloud/wg/wg.go index 5a3ea680..df052e4c 100644 --- a/cmd/uncloud/wg/wg.go +++ b/cmd/uncloud/wg/wg.go @@ -60,10 +60,7 @@ func runShow(ctx context.Context, uncli *cli.CLI, opts showOptions) error { if opts.machine != "" { // Proxy requests to the specified machine. - ctx, _, err = client.ProxyMachinesContext(ctx, []string{opts.machine}) - if err != nil { - return err - } + ctx = client.ProxySingleMachineContext(ctx, opts.machine) } resp, err := client.MachineClient.InspectWireGuardNetwork(ctx, nil) diff --git a/internal/grpcversion/interceptor.go b/internal/grpcversion/interceptor.go index 0bf19f00..8205f8cd 100644 --- a/internal/grpcversion/interceptor.go +++ b/internal/grpcversion/interceptor.go @@ -31,8 +31,8 @@ const ( // // The two minimums are independent: a client might require a newer daemon for new // features, while that same daemon could still handle requests from older clients. - MinClientVersion = "0.0.0" - MinServerVersion = "0.0.0" + MinClientVersion = "0.20.0" + MinServerVersion = "0.20.0" ReleaseURL = "https://github.com/psviderski/uncloud/releases/latest" ) diff --git a/internal/machine/api/pb/common.pb.go b/internal/machine/api/pb/common.pb.go index c657b503..700ffb1e 100644 --- a/internal/machine/api/pb/common.pb.go +++ b/internal/machine/api/pb/common.pb.go @@ -81,8 +81,12 @@ type Metadata struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields + // ID of the machine the response came from. + MachineId string `protobuf:"bytes,4,opt,name=machine_id,json=machineId,proto3" json:"machine_id,omitempty"` + // Name of the machine the response came from. + MachineName string `protobuf:"bytes,5,opt,name=machine_name,json=machineName,proto3" json:"machine_name,omitempty"` // Address of the machine the response came from. - Machine string `protobuf:"bytes,1,opt,name=machine,proto3" json:"machine,omitempty"` + MachineAddr string `protobuf:"bytes,1,opt,name=machine_addr,json=machineAddr,proto3" json:"machine_addr,omitempty"` // error is set if the request to upstream failed. The rest of the response is undefined. Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` // error as a gRPC Status message. @@ -121,9 +125,23 @@ func (*Metadata) Descriptor() ([]byte, []int) { return file_internal_machine_api_pb_common_proto_rawDescGZIP(), []int{0} } -func (x *Metadata) GetMachine() string { +func (x *Metadata) GetMachineId() string { if x != nil { - return x.Machine + return x.MachineId + } + return "" +} + +func (x *Metadata) GetMachineName() string { + if x != nil { + return x.MachineName + } + return "" +} + +func (x *Metadata) GetMachineAddr() string { + if x != nil { + return x.MachineAddr } return "" } @@ -549,53 +567,58 @@ var file_internal_machine_api_pb_common_proto_rawDesc = []byte{ 0x67, 0x6c, 0x65, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, - 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x66, 0x0a, 0x08, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, - 0x61, 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x18, 0x01, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x07, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, - 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, - 0x72, 0x12, 0x2a, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, - 0x0b, 0x32, 0x12, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x72, 0x70, 0x63, 0x2e, 0x53, - 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x32, 0x0a, - 0x05, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x29, 0x0a, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, - 0x74, 0x61, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4d, - 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, - 0x61, 0x22, 0x37, 0x0a, 0x0d, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x12, 0x26, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x18, 0x01, - 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, - 0x52, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x14, 0x0a, 0x02, 0x49, 0x50, - 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x02, 0x69, 0x70, - 0x22, 0x35, 0x0a, 0x06, 0x49, 0x50, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52, - 0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, - 0x0d, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x22, 0x37, 0x0a, 0x08, 0x49, 0x50, 0x50, 0x72, 0x65, - 0x66, 0x69, 0x78, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, - 0x07, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52, 0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, - 0x62, 0x69, 0x74, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x62, 0x69, 0x74, 0x73, - 0x22, 0x75, 0x0a, 0x0b, 0x4c, 0x6f, 0x67, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, - 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, - 0x16, 0x0a, 0x06, 0x66, 0x6f, 0x6c, 0x6c, 0x6f, 0x77, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, - 0x06, 0x66, 0x6f, 0x6c, 0x6c, 0x6f, 0x77, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x61, 0x69, 0x6c, 0x18, - 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x04, 0x74, 0x61, 0x69, 0x6c, 0x12, 0x14, 0x0a, 0x05, 0x73, - 0x69, 0x6e, 0x63, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x73, 0x69, 0x6e, 0x63, - 0x65, 0x12, 0x14, 0x0a, 0x05, 0x75, 0x6e, 0x74, 0x69, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x05, 0x75, 0x6e, 0x74, 0x69, 0x6c, 0x22, 0xd2, 0x01, 0x0a, 0x08, 0x4c, 0x6f, 0x67, 0x45, - 0x6e, 0x74, 0x72, 0x79, 0x12, 0x30, 0x0a, 0x06, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x18, 0x01, - 0x20, 0x01, 0x28, 0x0e, 0x32, 0x18, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4c, 0x6f, 0x67, 0x45, 0x6e, - 0x74, 0x72, 0x79, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, 0x79, 0x70, 0x65, 0x52, 0x06, - 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, - 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, - 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, - 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, - 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, - 0x0c, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x40, 0x0a, 0x0a, 0x53, 0x74, - 0x72, 0x65, 0x61, 0x6d, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x4e, 0x4b, 0x4e, - 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x54, 0x44, 0x4f, 0x55, 0x54, 0x10, - 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x54, 0x44, 0x45, 0x52, 0x52, 0x10, 0x02, 0x12, 0x0d, 0x0a, - 0x09, 0x48, 0x45, 0x41, 0x52, 0x54, 0x42, 0x45, 0x41, 0x54, 0x10, 0x03, 0x42, 0x37, 0x5a, 0x35, - 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x76, 0x69, 0x64, - 0x65, 0x72, 0x73, 0x6b, 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x2f, 0x69, 0x6e, - 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x2f, 0x61, - 0x70, 0x69, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xb1, 0x01, 0x0a, 0x08, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, + 0x74, 0x61, 0x12, 0x1d, 0x0a, 0x0a, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x5f, 0x69, 0x64, + 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x49, + 0x64, 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x5f, 0x6e, 0x61, 0x6d, + 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, + 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x5f, + 0x61, 0x64, 0x64, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6d, 0x61, 0x63, 0x68, + 0x69, 0x6e, 0x65, 0x41, 0x64, 0x64, 0x72, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x12, 0x2a, 0x0a, + 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x12, 0x2e, + 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x72, 0x70, 0x63, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, + 0x73, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x22, 0x32, 0x0a, 0x05, 0x45, 0x6d, 0x70, + 0x74, 0x79, 0x12, 0x29, 0x0a, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x18, 0x01, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4d, 0x65, 0x74, 0x61, 0x64, + 0x61, 0x74, 0x61, 0x52, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x22, 0x37, 0x0a, + 0x0d, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x26, + 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, + 0x32, 0x0a, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x52, 0x08, 0x6d, 0x65, + 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x14, 0x0a, 0x02, 0x49, 0x50, 0x12, 0x0e, 0x0a, 0x02, + 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x02, 0x69, 0x70, 0x22, 0x35, 0x0a, 0x06, + 0x49, 0x50, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x49, 0x50, 0x52, 0x02, 0x69, 0x70, 0x12, + 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x70, + 0x6f, 0x72, 0x74, 0x22, 0x37, 0x0a, 0x08, 0x49, 0x50, 0x50, 0x72, 0x65, 0x66, 0x69, 0x78, 0x12, + 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x61, 0x70, + 0x69, 0x2e, 0x49, 0x50, 0x52, 0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x62, 0x69, 0x74, 0x73, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x62, 0x69, 0x74, 0x73, 0x22, 0x75, 0x0a, 0x0b, + 0x4c, 0x6f, 0x67, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, + 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x66, + 0x6f, 0x6c, 0x6c, 0x6f, 0x77, 0x18, 0x02, 0x20, 0x01, 0x28, 0x08, 0x52, 0x06, 0x66, 0x6f, 0x6c, + 0x6c, 0x6f, 0x77, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x61, 0x69, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, + 0x05, 0x52, 0x04, 0x74, 0x61, 0x69, 0x6c, 0x12, 0x14, 0x0a, 0x05, 0x73, 0x69, 0x6e, 0x63, 0x65, + 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x73, 0x69, 0x6e, 0x63, 0x65, 0x12, 0x14, 0x0a, + 0x05, 0x75, 0x6e, 0x74, 0x69, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x75, 0x6e, + 0x74, 0x69, 0x6c, 0x22, 0xd2, 0x01, 0x0a, 0x08, 0x4c, 0x6f, 0x67, 0x45, 0x6e, 0x74, 0x72, 0x79, + 0x12, 0x30, 0x0a, 0x06, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, + 0x32, 0x18, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x4c, 0x6f, 0x67, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x2e, + 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, 0x79, 0x70, 0x65, 0x52, 0x06, 0x73, 0x74, 0x72, 0x65, + 0x61, 0x6d, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, + 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, + 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, + 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x18, 0x0a, 0x07, + 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x6d, + 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x40, 0x0a, 0x0a, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x54, 0x79, 0x70, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, + 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x54, 0x44, 0x4f, 0x55, 0x54, 0x10, 0x01, 0x12, 0x0a, 0x0a, + 0x06, 0x53, 0x54, 0x44, 0x45, 0x52, 0x52, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x48, 0x45, 0x41, + 0x52, 0x54, 0x42, 0x45, 0x41, 0x54, 0x10, 0x03, 0x42, 0x37, 0x5a, 0x35, 0x67, 0x69, 0x74, 0x68, + 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x76, 0x69, 0x64, 0x65, 0x72, 0x73, 0x6b, + 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, + 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, + 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/internal/machine/api/pb/common.proto b/internal/machine/api/pb/common.proto index b1876158..c1af74f4 100644 --- a/internal/machine/api/pb/common.proto +++ b/internal/machine/api/pb/common.proto @@ -11,8 +11,13 @@ import "google/protobuf/timestamp.proto"; // Common metadata message nested in all reply message types, injected by the gRPC proxy to provide information // about the machine that responded to the request. message Metadata { + // ID of the machine the response came from. + string machine_id = 4; + // Name of the machine the response came from. + string machine_name = 5; // Address of the machine the response came from. - string machine = 1; + string machine_addr = 1; + // error is set if the request to upstream failed. The rest of the response is undefined. string error = 2; // error as a gRPC Status message. diff --git a/internal/machine/api/proxy/backend.go b/internal/machine/api/proxy/backend.go index abfa139b..343840f2 100644 --- a/internal/machine/api/proxy/backend.go +++ b/internal/machine/api/proxy/backend.go @@ -4,15 +4,18 @@ import ( "fmt" "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/siderolabs/grpc-proxy/proxy" "google.golang.org/grpc/status" "google.golang.org/protobuf/encoding/protowire" "google.golang.org/protobuf/proto" ) -// One2ManyResponder converts upstream responses into messages from upstreams, so that multiple -// successful and failure responses might be returned in One2Many mode. -type One2ManyResponder struct { - machine string +// MetadataBackend wraps a proxy.Backend and injects machine metadata into responses in One2Many mode. +type MetadataBackend struct { + proxy.Backend + MachineID string + MachineName string + MachineAddr string } // AppendInfo is called to enhance response from the backend with additional data. @@ -59,10 +62,12 @@ type One2ManyResponder struct { // cuts field header, rest is representation of some reply. Marshal 'Empty' as protobuf, // which builds 'common.Metadata' field, append it to original response message, build new header // for new length of some response, and add back new field header. -func (b *One2ManyResponder) AppendInfo(streaming bool, resp []byte) ([]byte, error) { +func (b *MetadataBackend) AppendInfo(streaming bool, resp []byte) ([]byte, error) { payload, err := proto.Marshal(&pb.Empty{ Metadata: &pb.Metadata{ - Machine: b.machine, + MachineAddr: b.MachineAddr, + MachineId: b.MachineID, + MachineName: b.MachineName, }, }) @@ -124,12 +129,14 @@ func (b *One2ManyResponder) AppendInfo(streaming bool, resp []byte) ([]byte, err // // Streaming responses are not wrapped into Empty, so we simply marshall EmptyResponse // message. -func (b *One2ManyResponder) BuildError(streaming bool, err error) ([]byte, error) { +func (b *MetadataBackend) BuildError(streaming bool, err error) ([]byte, error) { var resp proto.Message = &pb.Empty{ Metadata: &pb.Metadata{ - Machine: b.machine, - Error: err.Error(), - Status: status.Convert(err).Proto(), + MachineAddr: b.MachineAddr, + MachineId: b.MachineID, + MachineName: b.MachineName, + Error: err.Error(), + Status: status.Convert(err).Proto(), }, } diff --git a/internal/machine/api/proxy/director.go b/internal/machine/api/proxy/director.go index 5d6085d0..6032676a 100644 --- a/internal/machine/api/proxy/director.go +++ b/internal/machine/api/proxy/director.go @@ -2,7 +2,10 @@ package proxy import ( "context" + "errors" + "fmt" "sync" + "sync/atomic" "github.com/siderolabs/grpc-proxy/proxy" "google.golang.org/grpc/codes" @@ -15,27 +18,22 @@ type Director struct { localBackend *LocalBackend remotePort uint16 remoteBackends sync.Map - // mu synchronizes access to localAddress. - mu sync.RWMutex - localAddress string + localAddress atomic.Value + mapper MachineMapper } -func NewDirector(localSockPath string, remotePort uint16) *Director { +func NewDirector(localSockPath string, remotePort uint16, mapper MachineMapper) *Director { return &Director{ - localBackend: NewLocalBackend(localSockPath, ""), + localBackend: NewLocalBackend(localSockPath), remotePort: remotePort, + mapper: mapper, } } // UpdateLocalAddress updates the local machine address used to identify which requests should be proxied -// to the local gRPC server. +// to the local gRPC server. It is called once during machine startup before the proxy server accepts requests. func (d *Director) UpdateLocalAddress(addr string) { - d.mu.Lock() - defer d.mu.Unlock() - - d.localAddress = addr - // Replace the local backend with the one that has local address set. - d.localBackend = NewLocalBackend(d.localBackend.sockPath, addr) + d.localAddress.Store(addr) } // Director implements proxy.StreamDirector for grpc-proxy, routing requests to local or remote backends based @@ -51,39 +49,90 @@ func (d *Director) Director(ctx context.Context, fullMethodName string) (proxy.M return proxy.One2One, []proxy.Backend{d.localBackend}, nil } // If the request metadata doesn't contain machines to proxy to, send it to the local backend. - machines, ok := md["machines"] - if !ok { + machines, hasMachines := md["machines"] + machine, hasMachine := md["machine"] + if !hasMachines && !hasMachine { return proxy.One2One, []proxy.Backend{d.localBackend}, nil } - if len(machines) == 0 { - return proxy.One2One, nil, status.Error(codes.InvalidArgument, "no machines specified") - } - d.mu.RLock() - localAddress := d.localAddress - localBackend := d.localBackend - d.mu.RUnlock() - - backends := make([]proxy.Backend, len(machines)) - for i, addr := range machines { - if addr == localAddress { - backends[i] = localBackend - continue + // Handle singular "machine" case (One2One, no metadata injection) + if hasMachine { + if len(machine) != 1 { + return proxy.One2One, nil, status.Error(codes.InvalidArgument, + "proxy metadata 'machine' must have exactly one value") + } + if hasMachines { + return proxy.One2One, nil, status.Error(codes.InvalidArgument, + "both 'machine' and 'machines' proxy metadata are set") + } + targets, err := d.mapper.MapMachines(ctx, machine) + if err != nil { + return proxy.One2One, nil, mapErrorToStatus(err) } - backend, err := d.remoteBackend(addr) + backend, err := d.getBackend(targets[0].Addr) if err != nil { return proxy.One2One, nil, status.Error(codes.Internal, err.Error()) } - backends[i] = backend + + // For One2One, we don't wrap in MetadataBackend as we don't inject metadata. + return proxy.One2One, []proxy.Backend{backend}, nil } - if len(backends) == 1 { - return proxy.One2One, backends, nil + // Handle plural "machines" case (One2Many, always metadata injection) + if len(machines) == 0 { + return proxy.One2One, nil, status.Error(codes.InvalidArgument, "proxy metadata 'machines' is empty") } + + targets, err := d.mapper.MapMachines(ctx, machines) + if err != nil { + return proxy.One2One, nil, mapErrorToStatus(err) + } + + backends := make([]proxy.Backend, len(targets)) + for i, t := range targets { + backend, err := d.getBackend(t.Addr) + if err != nil { + return proxy.One2One, nil, status.Error(codes.Internal, err.Error()) + } + + // Wrap with metadata injector + backends[i] = &MetadataBackend{ + Backend: backend, + MachineID: t.ID, + MachineName: t.Name, + MachineAddr: t.Addr, + } + } + + // TODO: should we periodically close and delete outdated remote backends (the ones left after removing machines)? + // IIRC the proxy will try to reconnect to them indefinitely. This can be stopped by restarting the daemon. + // But we can clean them up, e.g. when a client requests 'machines: *' so we know all the current targets + // or run a background goroutine that periodically lists them and closes old remoteBackends. + return proxy.One2Many, backends, nil } +// mapErrorToStatus converts mapper errors to appropriate gRPC status errors. +func mapErrorToStatus(err error) error { + if notFound, ok := errors.AsType[*MachinesNotFoundError](err); ok { + return status.Error(codes.InvalidArgument, notFound.Error()) + } + // Check if already a gRPC status error. + if _, ok := status.FromError(err); ok { + return err + } + return status.Error(codes.Internal, fmt.Sprintf("failed to resolve machines: %v", err)) +} + +// getBackend returns a backend for the given address, utilizing local backend if matching local address. +func (d *Director) getBackend(addr string) (proxy.Backend, error) { + if localAddr, _ := d.localAddress.Load().(string); localAddr != "" && addr == localAddr { + return d.localBackend, nil + } + return d.remoteBackend(addr) +} + // remoteBackend returns a RemoteBackend for the given address from the cache or creates a new one. func (d *Director) remoteBackend(addr string) (*RemoteBackend, error) { b, ok := d.remoteBackends.Load(addr) diff --git a/internal/machine/api/proxy/director_test.go b/internal/machine/api/proxy/director_test.go new file mode 100644 index 00000000..55faef8c --- /dev/null +++ b/internal/machine/api/proxy/director_test.go @@ -0,0 +1,255 @@ +package proxy + +import ( + "context" + "errors" + "testing" + + "github.com/siderolabs/grpc-proxy/proxy" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +type mockMapper struct { + targets []MachineTarget + err error +} + +func (m *mockMapper) MapMachines(_ context.Context, _ []string) ([]MachineTarget, error) { + if m.err != nil { + return nil, m.err + } + return m.targets, nil +} + +func TestDirector_Director(t *testing.T) { + d := NewDirector("/tmp/test.sock", 8080, nil) + t.Cleanup(d.Close) + + // Use a valid IPv6 address for remote targets. + remoteTarget := MachineTarget{ID: "id-2", Name: "machine-b", Addr: "fd00::2"} + localTarget := MachineTarget{ID: "id-1", Name: "machine-a", Addr: "fd00::1"} + + t.Run("no metadata routes to local", func(t *testing.T) { + ctx := context.Background() + mode, backends, err := d.Director(ctx, "/Test/Method") + + require.NoError(t, err) + assert.Equal(t, proxy.One2One, mode) + assert.Len(t, backends, 1) + assert.IsType(t, (*LocalBackend)(nil), backends[0]) + }) + + t.Run("proxy-authority routes to local", func(t *testing.T) { + md := metadata.Pairs("proxy-authority", "test", "machines", remoteTarget.Name) + ctx := metadata.NewIncomingContext(context.Background(), md) + + mode, backends, err := d.Director(ctx, "/Test/Method") + + require.NoError(t, err) + assert.Equal(t, proxy.One2One, mode) + assert.Len(t, backends, 1) + assert.IsType(t, (*LocalBackend)(nil), backends[0]) + }) + + t.Run("machine singular local", func(t *testing.T) { + d.localAddress.Store(localTarget.Addr) + d.mapper = &mockMapper{targets: []MachineTarget{localTarget}} + + md := metadata.New(map[string]string{"machine": localTarget.Name}) + ctx := metadata.NewIncomingContext(context.Background(), md) + + mode, backends, err := d.Director(ctx, "/Test/Method") + + require.NoError(t, err) + assert.Equal(t, proxy.One2One, mode) + assert.Len(t, backends, 1) + assert.IsType(t, (*LocalBackend)(nil), backends[0]) + }) + + t.Run("machine singular remote", func(t *testing.T) { + d.localAddress.Store(localTarget.Addr) + d.mapper = &mockMapper{targets: []MachineTarget{remoteTarget}} + + md := metadata.New(map[string]string{"machine": remoteTarget.Name}) + ctx := metadata.NewIncomingContext(context.Background(), md) + + mode, backends, err := d.Director(ctx, "/Test/Method") + + require.NoError(t, err) + assert.Equal(t, proxy.One2One, mode) + assert.Len(t, backends, 1) + assert.IsType(t, (*RemoteBackend)(nil), backends[0]) + assert.Equal(t, "[fd00::2]:8080", backends[0].(*RemoteBackend).target) + }) + + t.Run("machine not found", func(t *testing.T) { + d.mapper = &mockMapper{err: &MachinesNotFoundError{NotFound: []string{"missing"}}} + + md := metadata.New(map[string]string{"machine": "missing"}) + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + assert.Contains(t, st.Message(), "machine not found: missing") + }) + + t.Run("machines plural single local", func(t *testing.T) { + d.localAddress.Store(localTarget.Addr) + d.mapper = &mockMapper{targets: []MachineTarget{localTarget}} + + md := metadata.Pairs("machines", localTarget.Name) + ctx := metadata.NewIncomingContext(context.Background(), md) + + mode, backends, err := d.Director(ctx, "/Test/Method") + + require.NoError(t, err) + assert.Equal(t, proxy.One2Many, mode) + assert.Len(t, backends, 1) + + mb := backends[0].(*MetadataBackend) + assert.Equal(t, localTarget.ID, mb.MachineID) + assert.Equal(t, localTarget.Name, mb.MachineName) + assert.Equal(t, localTarget.Addr, mb.MachineAddr) + assert.IsType(t, (*LocalBackend)(nil), mb.Backend) + }) + + t.Run("machines plural multiple", func(t *testing.T) { + d.localAddress.Store(localTarget.Addr) + d.mapper = &mockMapper{targets: []MachineTarget{localTarget, remoteTarget}} + + md := metadata.Pairs("machines", localTarget.Name, "machines", remoteTarget.Name) + ctx := metadata.NewIncomingContext(context.Background(), md) + + mode, backends, err := d.Director(ctx, "/Test/Method") + + require.NoError(t, err) + assert.Equal(t, proxy.One2Many, mode) + assert.Len(t, backends, 2) + + // First backend should be local. + mb0 := backends[0].(*MetadataBackend) + assert.Equal(t, localTarget.ID, mb0.MachineID) + assert.IsType(t, (*LocalBackend)(nil), mb0.Backend) + + // Second backend should be remote. + mb1 := backends[1].(*MetadataBackend) + assert.Equal(t, remoteTarget.ID, mb1.MachineID) + assert.IsType(t, (*RemoteBackend)(nil), mb1.Backend) + assert.Equal(t, "[fd00::2]:8080", mb1.Backend.(*RemoteBackend).target) + }) + + t.Run("machines empty string", func(t *testing.T) { + d.mapper = &mockMapper{err: &MachinesNotFoundError{NotFound: []string{""}}} + + md := metadata.Pairs("machines", "") + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + assert.Contains(t, st.Message(), "machine not found") + }) + + t.Run("machine empty slice", func(t *testing.T) { + md := metadata.MD{"machine": []string{}} + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + assert.Contains(t, st.Message(), "proxy metadata 'machine' must have exactly one value") + }) + + t.Run("machines empty slice", func(t *testing.T) { + md := metadata.MD{"machines": []string{}} + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + assert.Contains(t, st.Message(), "proxy metadata 'machines' is empty") + }) + + t.Run("both machine and machines set", func(t *testing.T) { + md := metadata.Pairs("machine", "m1", "machines", "m1", "machines", "m2") + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + assert.Contains(t, st.Message(), "both 'machine' and 'machines' proxy metadata are set") + }) + + t.Run("machines not found", func(t *testing.T) { + d.mapper = &mockMapper{err: &MachinesNotFoundError{NotFound: []string{"missing"}}} + + md := metadata.Pairs("machines", "missing") + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + }) + + t.Run("machines mapper generic error", func(t *testing.T) { + d.mapper = &mockMapper{err: errors.New("boom")} + + md := metadata.Pairs("machines", "any") + ctx := metadata.NewIncomingContext(context.Background(), md) + + _, _, err := d.Director(ctx, "/Test/Method") + + require.Error(t, err) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.Internal, st.Code()) + }) +} + +func TestMapErrorToStatus(t *testing.T) { + t.Run("machines not found", func(t *testing.T) { + err := mapErrorToStatus(&MachinesNotFoundError{NotFound: []string{"a", "b"}}) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.InvalidArgument, st.Code()) + }) + + t.Run("already grpc status", func(t *testing.T) { + original := status.Error(codes.DeadlineExceeded, "timeout") + err := mapErrorToStatus(original) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.DeadlineExceeded, st.Code()) + }) + + t.Run("generic error", func(t *testing.T) { + err := mapErrorToStatus(errors.New("something broke")) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.Internal, st.Code()) + assert.Contains(t, st.Message(), "something broke") + }) +} diff --git a/internal/machine/api/proxy/local.go b/internal/machine/api/proxy/local.go index 28745578..e5a5af4a 100644 --- a/internal/machine/api/proxy/local.go +++ b/internal/machine/api/proxy/local.go @@ -10,9 +10,8 @@ import ( "google.golang.org/grpc/metadata" ) -// LocalBackend is a proxy.One2ManyResponder implementation that proxies to a local gRPC server listening on a Unix socket. +// LocalBackend is a proxy.Backend implementation that proxies to a local gRPC server listening on a Unix socket. type LocalBackend struct { - One2ManyResponder sockPath string mu sync.RWMutex @@ -21,20 +20,15 @@ type LocalBackend struct { var _ proxy.Backend = (*LocalBackend)(nil) -// NewLocalBackend returns a new LocalBackend for the given Unix socket path. The addr parameter is the local address -// of the current machine which could be empty if it's not known. The address is used to populate response metadata -// in one2many mode. -func NewLocalBackend(sockPath, addr string) *LocalBackend { +// NewLocalBackend returns a new LocalBackend for the given Unix socket path. +func NewLocalBackend(sockPath string) *LocalBackend { return &LocalBackend{ - One2ManyResponder: One2ManyResponder{ - machine: addr, - }, sockPath: sockPath, } } func (b *LocalBackend) String() string { - return b.machine + return "unix://" + b.sockPath } // GetConnection returns a gRPC connection to the local server listening on the Unix socket. @@ -64,6 +58,16 @@ func (b *LocalBackend) GetConnection(ctx context.Context, _ string) (context.Con return outCtx, b.conn, err } +// AppendInfo is a no-op for LocalBackend as it does not inject metadata. +func (b *LocalBackend) AppendInfo(streaming bool, resp []byte) ([]byte, error) { + return resp, nil +} + +// BuildError is a no-op for LocalBackend. +func (b *LocalBackend) BuildError(streaming bool, err error) ([]byte, error) { + return nil, err +} + // Close closes the upstream gRPC connection. func (b *LocalBackend) Close() { b.mu.Lock() diff --git a/internal/machine/api/proxy/mapper.go b/internal/machine/api/proxy/mapper.go new file mode 100644 index 00000000..0b9e8352 --- /dev/null +++ b/internal/machine/api/proxy/mapper.go @@ -0,0 +1,108 @@ +package proxy + +import ( + "context" + "fmt" + "slices" + "strings" + + "github.com/psviderski/uncloud/internal/machine/api/pb" +) + +// MachineTarget represents a resolved machine target. +type MachineTarget struct { + ID, Name, Addr string +} + +// MachinesNotFoundError indicates that one or more requested machines were not found. +type MachinesNotFoundError struct { + NotFound []string +} + +func (e *MachinesNotFoundError) Error() string { + if len(e.NotFound) == 1 { + return fmt.Sprintf("machine not found: %s", e.NotFound[0]) + } + return fmt.Sprintf("machines not found: %s", strings.Join(e.NotFound, ", ")) +} + +// MachineMapper provides access to machine information in the cluster. +type MachineMapper interface { + // MapMachines resolves a list of machine names/IDs (or "*") to a list of machine targets. + // Returns MachinesNotFoundError if any requested machine is not found (except when "*" is used). + MapMachines(ctx context.Context, namesOrIDs []string) ([]MachineTarget, error) +} + +// Store is the interface required by MachineMapper to access the cluster store. +type Store interface { + ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) +} + +// CorrosionMapper implements MachineMapper using the corrosion store. +type CorrosionMapper struct { + store Store +} + +func NewCorrosionMapper(store Store) *CorrosionMapper { + return &CorrosionMapper{store: store} +} + +func (m *CorrosionMapper) MapMachines(ctx context.Context, namesOrIDs []string) ([]MachineTarget, error) { + if len(namesOrIDs) == 0 { + return nil, fmt.Errorf("no machines specified") + } + + machines, err := m.store.ListMachines(ctx) + if err != nil { + return nil, fmt.Errorf("list machines: %w", err) + } + + allTargets := make([]MachineTarget, 0, len(machines)) + for _, machine := range machines { + ip, err := machine.Network.ManagementIp.ToAddr() + if err != nil { + return nil, fmt.Errorf("invalid management IP for machine '%s' in store: %w", machine.Name, err) + } + allTargets = append(allTargets, MachineTarget{ + ID: machine.Id, + Name: machine.Name, + Addr: ip.String(), + }) + } + + if slices.Contains(namesOrIDs, "*") { + if len(allTargets) == 0 { + return nil, fmt.Errorf("no machines in cluster") + } + return allTargets, nil + } + + // Build a map for lookup (keyed by both ID and name) + targetByLookup := make(map[string]MachineTarget, len(allTargets)*2) + for _, t := range allTargets { + targetByLookup[t.ID] = t + targetByLookup[t.Name] = t + } + + // Resolve each requested machine. + targets := make([]MachineTarget, 0, len(namesOrIDs)) + var notFound []string + seenTarget := make(map[string]struct{}, len(namesOrIDs)) + + for _, nameOrID := range namesOrIDs { + if t, ok := targetByLookup[nameOrID]; ok { + if _, seen := seenTarget[t.ID]; !seen { + targets = append(targets, t) + seenTarget[t.ID] = struct{}{} + } + } else { + notFound = append(notFound, nameOrID) + } + } + + if len(notFound) > 0 { + return nil, &MachinesNotFoundError{NotFound: notFound} + } + + return targets, nil +} diff --git a/internal/machine/api/proxy/mapper_test.go b/internal/machine/api/proxy/mapper_test.go new file mode 100644 index 00000000..a5d0fb33 --- /dev/null +++ b/internal/machine/api/proxy/mapper_test.go @@ -0,0 +1,168 @@ +package proxy + +import ( + "context" + "errors" + "net/netip" + "testing" + + "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type mockStore struct { + machines []*pb.MachineInfo + err error +} + +func (s *mockStore) ListMachines(_ context.Context) ([]*pb.MachineInfo, error) { + if s.err != nil { + return nil, s.err + } + return s.machines, nil +} + +func machineInfo(id, name, ip string) *pb.MachineInfo { + return &pb.MachineInfo{ + Id: id, + Name: name, + Network: &pb.NetworkConfig{ + ManagementIp: pb.NewIP(netip.MustParseAddr(ip)), + }, + } +} + +func TestCorrosionMapper_MapMachines(t *testing.T) { + ctx := context.Background() + + machines := []*pb.MachineInfo{ + machineInfo("id-1", "machine-a", "fd00::1"), + machineInfo("id-2", "machine-b", "fd00::2"), + } + + tests := []struct { + name string + store *mockStore + input []string + want []MachineTarget + wantErr bool + errMsg string + }{ + { + name: "wildcard returns all machines", + store: &mockStore{machines: machines}, + input: []string{"*"}, + want: []MachineTarget{ + {ID: "id-1", Name: "machine-a", Addr: "fd00::1"}, + {ID: "id-2", Name: "machine-b", Addr: "fd00::2"}, + }, + }, + { + name: "single name", + store: &mockStore{machines: machines}, + input: []string{"machine-a"}, + want: []MachineTarget{ + {ID: "id-1", Name: "machine-a", Addr: "fd00::1"}, + }, + }, + { + name: "single id", + store: &mockStore{machines: machines}, + input: []string{"id-2"}, + want: []MachineTarget{ + {ID: "id-2", Name: "machine-b", Addr: "fd00::2"}, + }, + }, + { + name: "multiple mixed", + store: &mockStore{machines: machines}, + input: []string{"machine-a", "id-2"}, + want: []MachineTarget{ + {ID: "id-1", Name: "machine-a", Addr: "fd00::1"}, + {ID: "id-2", Name: "machine-b", Addr: "fd00::2"}, + }, + }, + { + name: "deduplicates repeated inputs", + store: &mockStore{machines: machines}, + input: []string{"machine-a", "machine-a"}, + want: []MachineTarget{ + {ID: "id-1", Name: "machine-a", Addr: "fd00::1"}, + }, + }, + { + name: "deduplicates name and id for same machine", + store: &mockStore{machines: machines}, + input: []string{"machine-a", "id-1"}, + want: []MachineTarget{ + {ID: "id-1", Name: "machine-a", Addr: "fd00::1"}, + }, + }, + { + name: "not found single", + store: &mockStore{machines: machines}, + input: []string{"missing"}, + wantErr: true, + errMsg: "machine not found: missing", + }, + { + name: "not found multiple", + store: &mockStore{machines: machines}, + input: []string{"missing", "also-missing"}, + wantErr: true, + errMsg: "machines not found: missing, also-missing", + }, + { + name: "partial not found", + store: &mockStore{machines: machines}, + input: []string{"machine-a", "missing"}, + wantErr: true, + errMsg: "machine not found: missing", + }, + { + name: "wildcard with no machines", + store: &mockStore{machines: []*pb.MachineInfo{}}, + input: []string{"*"}, + wantErr: true, + errMsg: "no machines in cluster", + }, + { + name: "store error", + store: &mockStore{err: errors.New("store down")}, + input: []string{"*"}, + wantErr: true, + errMsg: "list machines: store down", + }, + { + name: "invalid management ip", + store: &mockStore{machines: []*pb.MachineInfo{{Id: "bad", Name: "bad-ip", Network: &pb.NetworkConfig{ManagementIp: &pb.IP{}}}}}, + input: []string{"*"}, + wantErr: true, + errMsg: "invalid management IP for machine 'bad-ip' in store", + }, + { + name: "empty input returns error", + store: &mockStore{machines: machines}, + input: []string{}, + wantErr: true, + errMsg: "no machines specified", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mapper := NewCorrosionMapper(tt.store) + got, err := mapper.MapMachines(ctx, tt.input) + + if tt.wantErr { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.errMsg) + return + } + + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/internal/machine/api/proxy/remote.go b/internal/machine/api/proxy/remote.go index 2ebeef49..39bce08a 100644 --- a/internal/machine/api/proxy/remote.go +++ b/internal/machine/api/proxy/remote.go @@ -14,13 +14,11 @@ import ( "google.golang.org/grpc/metadata" ) -// RemoteBackend is a proxy.One2ManyResponder implementation that proxies to a remote gRPC server, injecting machine metadata -// into the response. +// RemoteBackend is a proxy.Backend implementation that proxies to a remote gRPC server. // // Based on the Talos apid implementation: // https://github.com/siderolabs/talos/blob/59a78da42cdea8fbccc35d0851f9b0eef928261b/internal/app/apid/pkg/backend/apid.go type RemoteBackend struct { - One2ManyResponder target string mu sync.RWMutex @@ -37,15 +35,12 @@ func NewRemoteBackend(addr string, port uint16) (*RemoteBackend, error) { } return &RemoteBackend{ - One2ManyResponder: One2ManyResponder{ - machine: addr, - }, target: netip.AddrPortFrom(ip, port).String(), }, nil } func (b *RemoteBackend) String() string { - return b.machine + return b.target } // GetConnection returns a gRPC connection to the remote server. @@ -58,6 +53,7 @@ func (b *RemoteBackend) GetConnection(ctx context.Context, _ string) (context.Co } delete(md, ":authority") delete(md, "machines") + delete(md, "machine") outCtx := metadata.NewOutgoingContext(ctx, md) @@ -100,6 +96,16 @@ func (b *RemoteBackend) GetConnection(ctx context.Context, _ string) (context.Co return outCtx, b.conn, err } +// AppendInfo is a no-op for RemoteBackend as it does not inject metadata. +func (b *RemoteBackend) AppendInfo(streaming bool, resp []byte) ([]byte, error) { + return resp, nil +} + +// BuildError is a no-op for RemoteBackend. +func (b *RemoteBackend) BuildError(streaming bool, err error) ([]byte, error) { + return nil, err +} + // Close closes the upstream gRPC connection. func (b *RemoteBackend) Close() { b.mu.Lock() diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 987e9d18..cb3b9931 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -250,7 +250,8 @@ func NewMachine(config *Config) (*Machine, error) { dockerService := machinedocker.NewService(config.DockerClient, db) // Init a local gRPC proxy server that proxies requests to the local or remote machine API servers. - proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort) + mapper := apiproxy.NewCorrosionMapper(corroStore) + proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort, mapper) localProxyServer := grpc.NewServer( grpc.ForceServerCodecV2(proxy.Codec()), grpc.UnaryInterceptor(grpcversion.ServerUnaryInterceptor), diff --git a/pkg/api/machine.go b/pkg/api/machine.go index a81f2872..8c16c0db 100644 --- a/pkg/api/machine.go +++ b/pkg/api/machine.go @@ -12,21 +12,6 @@ type MachineFilter struct { type MachineMembersList []*pb.MachineMember -func (m MachineMembersList) FindByManagementIP(ip string) *pb.MachineMember { - for _, machine := range m { - addr, err := machine.Machine.Network.ManagementIp.ToAddr() - if err != nil { - continue - } - - if addr.String() == ip { - return machine - } - } - - return nil -} - func (m MachineMembersList) FindByNameOrID(nameOrID string) *pb.MachineMember { for _, machine := range m { if machine.Machine.Id == nameOrID || machine.Machine.Name == nameOrID { diff --git a/pkg/api/service.go b/pkg/api/service.go index 5b2497b6..ae98bfc9 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -515,8 +515,9 @@ type Service struct { } type MachineServiceContainer struct { - MachineID string - Container ServiceContainer + MachineID string + MachineName string + Container ServiceContainer } // FindContainer returns the service container by exact name, ID, or unique ID prefix. diff --git a/pkg/client/client.go b/pkg/client/client.go index 10bd756e..aa1ea273 100644 --- a/pkg/client/client.go +++ b/pkg/client/client.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "os" - "strings" "github.com/docker/cli/cli/streams" "github.com/psviderski/uncloud/internal/machine/api/pb" @@ -71,47 +70,24 @@ func (cli *Client) progressOut() *streams.Out { return streams.NewOut(os.Stdout) } -// proxyToMachine returns a new context that proxies gRPC requests to the specified machine. -func proxyToMachine(ctx context.Context, machine *pb.MachineInfo) context.Context { - machineIP, _ := machine.Network.ManagementIp.ToAddr() - md := metadata.Pairs("machines", machineIP.String()) +// ProxyMachinesContext returns a new context that proxies gRPC requests to the specified machines. +// If namesOrIDs is nil or empty, all machines are included. +// This triggers One2Many proxying, which always injects metadata into the response. +func (cli *Client) ProxyMachinesContext(ctx context.Context, namesOrIDs []string) context.Context { + md := metadata.New(nil) + if len(namesOrIDs) == 0 { + md.Append("machines", "*") + } else { + md.Append("machines", namesOrIDs...) + } + return metadata.NewOutgoingContext(ctx, md) } -// ProxyMachinesContext returns a new context that proxies gRPC requests to the specified machines. -// If namesOrIDs is nil, all machines are included. -func (cli *Client) ProxyMachinesContext( - ctx context.Context, namesOrIDs []string, -) (context.Context, api.MachineMembersList, error) { - // TODO: move the machine IP resolution to the proxy router to allow setting machine names and IDs in the metadata. - machines, err := cli.ListMachines(ctx, nil) - if err != nil { - return nil, nil, fmt.Errorf("list machines: %w", err) - } - - var proxiedMachines api.MachineMembersList - var notFound []string - for _, nameOrID := range namesOrIDs { - if m := machines.FindByNameOrID(nameOrID); m != nil { - proxiedMachines = append(proxiedMachines, m) - } else { - notFound = append(notFound, nameOrID) - } - } - - if len(notFound) > 0 { - return nil, nil, fmt.Errorf("machines not found: %s", strings.Join(notFound, ", ")) - } - - if len(namesOrIDs) == 0 { - proxiedMachines = machines - } - - md := metadata.New(nil) - for _, m := range proxiedMachines { - machineIP, _ := m.Machine.Network.ManagementIp.ToAddr() - md.Append("machines", machineIP.String()) - } - - return metadata.NewOutgoingContext(ctx, md), proxiedMachines, nil +// ProxySingleMachineContext returns a new context that proxies gRPC requests to a single specified machine. +// This triggers One2One proxying, which does NOT inject metadata into the response. +// Use this for requests that expect a single response message without metadata wrapper. +func (cli *Client) ProxySingleMachineContext(ctx context.Context, nameOrID string) context.Context { + md := metadata.Pairs("machine", nameOrID) + return metadata.NewOutgoingContext(ctx, md) } diff --git a/pkg/client/container.go b/pkg/client/container.go index a2efb1ea..ae582b8c 100644 --- a/pkg/client/container.go +++ b/pkg/client/container.go @@ -74,7 +74,7 @@ func (cli *Client) createServiceContainerWithPull( resp.Name = containerName // Proxy Docker gRPC requests to the selected machine. - ctx = proxyToMachine(ctx, machine.Machine) + ctx = cli.ProxySingleMachineContext(ctx, machine.Machine.Id) pw := progress.ContextWriter(ctx) eventID := cliprogress.NewContainerEventID(ctx, containerName, machine.Machine.Name) @@ -259,27 +259,43 @@ func (cli *Client) InspectContainer( return svc.FindContainer(containerNameOrID) } -// StartContainer starts the specified container within the service. -func (cli *Client) StartContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) error { +// containerOperationContext holds the context needed to perform an operation on a container. +type containerOperationContext struct { + ctx context.Context + containerID string + eventID string +} + +// resolveContainerOperation resolves a container by name/ID and prepares the context for an operation. +func (cli *Client) resolveContainerOperation( + ctx context.Context, serviceNameOrID, containerNameOrID string, +) (containerOperationContext, error) { ctr, err := cli.InspectContainer(ctx, serviceNameOrID, containerNameOrID) if err != nil { - return err + return containerOperationContext{}, err } - machine, err := cli.InspectMachine(ctx, ctr.MachineID) + eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, ctr.MachineName) + return containerOperationContext{ + ctx: cli.ProxySingleMachineContext(ctx, ctr.MachineID), + containerID: ctr.Container.ID, + eventID: eventID, + }, nil +} + +// StartContainer starts the specified container within the service. +func (cli *Client) StartContainer(ctx context.Context, serviceNameOrID, containerNameOrID string) error { + op, err := cli.resolveContainerOperation(ctx, serviceNameOrID, containerNameOrID) if err != nil { - return fmt.Errorf("inspect machine '%s': %w", ctr.MachineID, err) - } - ctx = proxyToMachine(ctx, machine.Machine) - - pw := progress.ContextWriter(ctx) - eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, machine.Machine.Name) - - pw.Event(progress.StartingEvent(eventID)) - if err = cli.Docker.StartContainer(ctx, ctr.Container.ID, container.StartOptions{}); err != nil { return err } - pw.Event(progress.StartedEvent(eventID)) + + pw := progress.ContextWriter(op.ctx) + pw.Event(progress.StartingEvent(op.eventID)) + if err = cli.Docker.StartContainer(op.ctx, op.containerID, container.StartOptions{}); err != nil { + return err + } + pw.Event(progress.StartedEvent(op.eventID)) return nil } @@ -290,25 +306,17 @@ func (cli *Client) StartContainer(ctx context.Context, serviceNameOrID, containe func (cli *Client) StopContainer( ctx context.Context, serviceNameOrID, containerNameOrID string, opts container.StopOptions, ) error { - ctr, err := cli.InspectContainer(ctx, serviceNameOrID, containerNameOrID) + op, err := cli.resolveContainerOperation(ctx, serviceNameOrID, containerNameOrID) if err != nil { return err } - machine, err := cli.InspectMachine(ctx, ctr.MachineID) - if err != nil { - return fmt.Errorf("inspect machine '%s': %w", ctr.MachineID, err) - } - ctx = proxyToMachine(ctx, machine.Machine) - - pw := progress.ContextWriter(ctx) - eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, machine.Machine.Name) - - pw.Event(progress.StoppingEvent(eventID)) - if err = cli.Docker.StopContainer(ctx, ctr.Container.ID, opts); err != nil { + pw := progress.ContextWriter(op.ctx) + pw.Event(progress.StoppingEvent(op.eventID)) + if err = cli.Docker.StopContainer(op.ctx, op.containerID, opts); err != nil { return err } - pw.Event(progress.StoppedEvent(eventID)) + pw.Event(progress.StoppedEvent(op.eventID)) return nil } @@ -319,25 +327,18 @@ func (cli *Client) StopContainer( func (cli *Client) RemoveContainer( ctx context.Context, serviceNameOrID, containerNameOrID string, opts container.RemoveOptions, ) error { - ctr, err := cli.InspectContainer(ctx, serviceNameOrID, containerNameOrID) + op, err := cli.resolveContainerOperation(ctx, serviceNameOrID, containerNameOrID) if err != nil { return err } - machine, err := cli.InspectMachine(ctx, ctr.MachineID) - if err != nil { - return fmt.Errorf("inspect machine '%s': %w", ctr.MachineID, err) - } - ctx = proxyToMachine(ctx, machine.Machine) + pw := progress.ContextWriter(op.ctx) - pw := progress.ContextWriter(ctx) - eventID := cliprogress.ContainerEventID(ctx, ctr.Container.ServiceSpec.Name, ctr.Container.ID, machine.Machine.Name) - - pw.Event(progress.RemovingEvent(eventID)) - if err = cli.Docker.RemoveServiceContainer(ctx, ctr.Container.ID, opts); err != nil { + pw.Event(progress.RemovingEvent(op.eventID)) + if err = cli.Docker.RemoveServiceContainer(op.ctx, op.containerID, opts); err != nil { return err } - pw.Event(progress.RemovedEvent(eventID)) + pw.Event(progress.RemovedEvent(op.eventID)) return nil } @@ -374,7 +375,7 @@ func (cli *Client) ExecContainer( } // Proxy Docker gRPC requests to the machine hosting the container - ctx = proxyToMachine(ctx, machine.Machine) + ctx = cli.ProxySingleMachineContext(ctx, machine.Machine.Id) // Execute the command in the container exitCode, err := cli.Docker.ExecContainer(ctx, machinedocker.ExecConfig{ @@ -451,7 +452,7 @@ func (cli *Client) WaitContainerHealthy( } // For containers with a health check, wait until Docker reports healthy or unhealthy. - mctx := proxyToMachine(ctx, machine.Machine) + mctx := cli.ProxySingleMachineContext(ctx, machine.Machine.Id) mctx, cancel := context.WithTimeout(mctx, healthcheckTimeout(mc.Container.Config.Healthcheck)) defer cancel() ticker := time.NewTicker(1 * time.Second) diff --git a/pkg/client/image.go b/pkg/client/image.go index e9f8e6ab..b652cac8 100644 --- a/pkg/client/image.go +++ b/pkg/client/image.go @@ -25,6 +25,7 @@ import ( "github.com/docker/docker/pkg/jsonmessage" "github.com/docker/go-connections/nat" ocispec "github.com/opencontainers/image-spec/specs-go/v1" + "github.com/psviderski/uncloud/internal/cli/tui" "github.com/psviderski/uncloud/internal/docker" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/internal/machine/constants" @@ -58,10 +59,7 @@ func (cli *Client) InspectRemoteImage(ctx context.Context, id string) ([]api.Mac // it lists images on all machines. func (cli *Client) ListImages(ctx context.Context, filter api.ImageFilter) ([]api.MachineImages, error) { // Broadcast the image list request to the specified machines or all machines if none specified. - listCtx, machines, err := cli.ProxyMachinesContext(ctx, filter.Machines) - if err != nil { - return nil, fmt.Errorf("create request context to broadcast to machines: %w", err) - } + listCtx := cli.ProxyMachinesContext(ctx, filter.Machines) opts := image.ListOptions{Manifests: true} if filter.Name != "" { @@ -80,32 +78,36 @@ func (cli *Client) ListImages(ctx context.Context, filter api.ImageFilter) ([]ap return nil, err } - machineImages := make([]api.MachineImages, len(resp.Messages)) - for i, msg := range resp.Messages { - machineImages[i].Metadata = msg.Metadata - // TODO: handle this in the grpc-proxy router and always provide Metadata if possible. + machineImages := make([]api.MachineImages, 0, len(resp.Messages)) + + for _, msg := range resp.Messages { + // NOTE: Metadata should never be nil in practice. This is legacy fallback that will be removed. if msg.Metadata == nil { - // Metadata can be nil if the request was broadcasted to only one machine. - machineImages[i].Metadata = &pb.Metadata{ - Machine: machines[0].Machine.Id, - } - } else { - // Replace management IP with machine ID for friendlier error messages. - // TODO: migrate Metadata.Machine to use machine ID instead of IP in the grpc-proxy router. - if m := machines.FindByManagementIP(msg.Metadata.Machine); m != nil { - machineImages[i].Metadata.Machine = m.Machine.Id - } - if msg.Metadata.Error != "" { - continue - } + tui.PrintWarning("metadata is missing in response from unknown server") + continue + } + + if msg.Metadata.Error != "" { + // Continue processing other messages even if some machines return an error to avoid a partial failure + // of the entire command. + tui.PrintWarning(fmt.Sprintf( + "failed to list images on machine %s: %s", msg.Metadata.MachineName, msg.Metadata.Error, + )) + continue + } + + mi := api.MachineImages{ + Metadata: msg.Metadata, + ContainerdStore: msg.ContainerdStore, } if len(msg.Images) > 0 { - if err = json.Unmarshal(msg.Images, &machineImages[i].Images); err != nil { + if err = json.Unmarshal(msg.Images, &mi.Images); err != nil { return nil, fmt.Errorf("unmarshal images: %w", err) } } - machineImages[i].ContainerdStore = msg.ContainerdStore + + machineImages = append(machineImages, mi) } return machineImages, nil @@ -350,7 +352,8 @@ func (cli *Client) pushImageToMachine( pw.Event(progress.NewEvent(proxyEventID, progress.Error, err.Error())) return fmt.Errorf("run socat container with unix socket to proxy unregistry: %w", err) } - slog.Debug("Started unix socket socat proxy container.", "id", proxyCtrID, "hostPort", proxyPort, "socket", socketPath) + slog.Debug("Started unix socket socat proxy container.", + "id", proxyCtrID, "hostPort", proxyPort, "socket", socketPath) } pw.Event(progress.Event{ diff --git a/pkg/client/logs.go b/pkg/client/logs.go index bcaecab5..6ac02496 100644 --- a/pkg/client/logs.go +++ b/pkg/client/logs.go @@ -102,10 +102,7 @@ func (cli *Client) ServiceLogs( func (cli *Client) ContainerLogs( ctx context.Context, machineNameOrID string, containerID string, opts api.ServiceLogsOptions, ) (<-chan api.LogEntry, error) { - proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineNameOrID}) - if err != nil { - return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineNameOrID, err) - } + proxyCtx := cli.ProxySingleMachineContext(ctx, machineNameOrID) req := &pb.LogsRequest{ Id: containerID, @@ -201,10 +198,7 @@ func (cli *Client) MachineLogs( func (cli *Client) systemdServiceLogs( ctx context.Context, machineID, unit string, opts api.ServiceLogsOptions, ) (<-chan api.LogEntry, error) { - proxyCtx, _, err := cli.ProxyMachinesContext(ctx, []string{machineID}) - if err != nil { - return nil, fmt.Errorf("create request context to proxy to machine '%s': %w", machineID, err) - } + proxyCtx := cli.ProxySingleMachineContext(ctx, machineID) req := &pb.LogsRequest{ Id: unit, diff --git a/pkg/client/service.go b/pkg/client/service.go index 83b2f378..c2e48b44 100644 --- a/pkg/client/service.go +++ b/pkg/client/service.go @@ -4,12 +4,12 @@ import ( "context" "errors" "fmt" - "os" "slices" "sync" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/volume" + "github.com/psviderski/uncloud/internal/cli/tui" "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" "github.com/psviderski/uncloud/pkg/client/deploy/scheduler" @@ -96,16 +96,14 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser } // Broadcast the container list request to all available machines. - machineIDByManagementIP := make(map[string]string) md := metadata.New(nil) for _, m := range machines { if m.State == pb.MachineMember_UP || m.State == pb.MachineMember_SUSPECT { - machineIP, _ := m.Machine.Network.ManagementIp.ToAddr() - md.Append("machines", machineIP.String()) - - machineIDByManagementIP[machineIP.String()] = m.Machine.Id + 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())) } - // TODO: warning about machines that are DOWN. } listCtx := metadata.NewOutgoingContext(ctx, md) @@ -120,37 +118,25 @@ func (cli *Client) InspectService(ctx context.Context, nameOrID string) (api.Ser foundByID := false var containers []api.MachineServiceContainer for _, mc := range machineContainers { - // Metadata can be nil if the request was broadcasted to only one machine. - if mc.Metadata == nil && len(machineContainers) > 1 { - return svc, errors.New("something went wrong with gRPC proxy: metadata is missing for a machine response") - } - if mc.Metadata != nil && mc.Metadata.Error != "" { - // TODO: return failed machines in the response. - fmt.Printf("WARNING: failed to list containers on machine '%s': %s\n", - mc.Metadata.Machine, mc.Metadata.Error) + // 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 } - machineID := "" - if mc.Metadata == nil { - // ListServiceContainers was proxied to only one machine. - for _, v := range machineIDByManagementIP { - machineID = v - break - } - } else { - var ok bool - machineID, ok = machineIDByManagementIP[mc.Metadata.Machine] - if !ok { - return svc, fmt.Errorf("machine name not found for management IP: %s", mc.Metadata.Machine) - } + if mc.Metadata.Error != "" { + // TODO: return failed machines in the response. + tui.PrintWarning(fmt.Sprintf("failed to list containers on machine '%s': %s", + mc.Metadata.MachineName, mc.Metadata.Error)) + continue } // Collect both regular and hook containers for the service. for _, ctr := range append(mc.Containers, mc.HookContainers...) { containers = append(containers, api.MachineServiceContainer{ - MachineID: machineID, - Container: ctr, + MachineID: mc.Metadata.MachineId, + MachineName: mc.Metadata.MachineName, + Container: ctr, }) if ctr.ServiceID() == nameOrID { @@ -232,16 +218,6 @@ func (cli *Client) RemoveService(ctx context.Context, id string) error { return err } - machines, err := cli.ListMachines(ctx, nil) - if err != nil { - return fmt.Errorf("list machines: %w", err) - } - machineManagementIPByID := make(map[string]string) - for _, m := range machines { - machineIP, _ := m.Machine.Network.ManagementIp.ToAddr() - machineManagementIPByID[m.Machine.Id] = machineIP.String() - } - wg := sync.WaitGroup{} errCh := make(chan error) @@ -353,10 +329,11 @@ func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) { md := metadata.New(nil) for _, m := range machines { if m.State == pb.MachineMember_UP || m.State == pb.MachineMember_SUSPECT { - machineIP, _ := m.Machine.Network.ManagementIp.ToAddr() - md.Append("machines", machineIP.String()) + 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())) } - // TODO: warning about machines that are DOWN. } listCtx := metadata.NewOutgoingContext(ctx, md) @@ -371,10 +348,16 @@ func (cli *Client) ListServices(ctx context.Context) ([]api.Service, error) { // Most of the code can be reused in both InspectService and ListServices. servicesByID := make(map[string]api.Service) for _, mc := range machineContainers { - if mc.Metadata != nil && mc.Metadata.Error != "" { + // 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 != "" { // TODO: return failed machines in the response. - fmt.Fprintf(os.Stderr, "WARNING: failed to list containers on machine '%s': %s\n", - mc.Metadata.Machine, mc.Metadata.Error) + tui.PrintWarning(fmt.Sprintf("failed to list containers on machine '%s': %s", + mc.Metadata.MachineName, mc.Metadata.Error)) continue } diff --git a/pkg/client/volume.go b/pkg/client/volume.go index cbce4514..abe0e90f 100644 --- a/pkg/client/volume.go +++ b/pkg/client/volume.go @@ -9,7 +9,6 @@ import ( "github.com/docker/docker/api/types/volume" cliprogress "github.com/psviderski/uncloud/internal/cli/progress" "github.com/psviderski/uncloud/internal/cli/tui" - "github.com/psviderski/uncloud/internal/machine/api/pb" "github.com/psviderski/uncloud/pkg/api" ) @@ -28,7 +27,7 @@ func (cli *Client) CreateVolume( return resp, fmt.Errorf("inspect machine '%s': %w", machineNameOrID, err) } // Proxy Docker gRPC requests to the selected machine. - ctx = proxyToMachine(ctx, machine.Machine) + ctx = cli.ProxySingleMachineContext(ctx, machine.Machine.Id) pw := progress.ContextWriter(ctx) eventID := cliprogress.VolumeEventID(opts.Name, machine.Machine.Name) @@ -57,11 +56,7 @@ func (cli *Client) ListVolumes(ctx context.Context, filter *api.VolumeFilter) ([ proxyMachines = filter.Machines } - listCtx, machines, err := cli.ProxyMachinesContext(ctx, proxyMachines) - if err != nil { - return nil, fmt.Errorf("create request context to broadcast to all machines: %w", err) - } - + listCtx := cli.ProxyMachinesContext(ctx, proxyMachines) machineVolumes, err := cli.Docker.ListVolumes(listCtx, volume.ListOptions{}) if err != nil { return nil, err @@ -70,28 +65,22 @@ func (cli *Client) ListVolumes(ctx context.Context, filter *api.VolumeFilter) ([ var volumes []api.MachineVolume // Process responses from all machines. for _, mv := range machineVolumes { - if mv.Metadata != nil && mv.Metadata.Error != "" { - // TODO: return failed machines in the response. - tui.PrintWarning(fmt.Sprintf("failed to list volumes on machine '%s': %s", - mv.Metadata.Machine, mv.Metadata.Error)) + if mv.Metadata == nil { + tui.PrintWarning("metadata is missing in response from unknown server") continue } - var m *pb.MachineMember - if mv.Metadata == nil { - // ListVolumes was proxied to only one machine. - m = machines[0] - } else { - m = machines.FindByManagementIP(mv.Metadata.Machine) - if m == nil { - return nil, fmt.Errorf("machine not found by management IP: %s", mv.Metadata.Machine) - } + if mv.Metadata.Error != "" { + // TODO: return failed machines in the response. + tui.PrintWarning(fmt.Sprintf("failed to list volumes on machine '%s': %s", mv.Metadata.MachineName, + mv.Metadata.Error)) + continue } for _, vol := range mv.Response.Volumes { volumes = append(volumes, api.MachineVolume{ - MachineID: m.Machine.Id, - MachineName: m.Machine.Name, + MachineID: mv.Metadata.MachineId, + MachineName: mv.Metadata.MachineName, Volume: *vol, }) } @@ -118,7 +107,7 @@ func (cli *Client) RemoveVolume(ctx context.Context, machineNameOrID, volumeName return fmt.Errorf("inspect machine '%s': %w", machineNameOrID, err) } // Proxy Docker gRPC requests to the selected machine. - ctx = proxyToMachine(ctx, machine.Machine) + ctx = cli.ProxySingleMachineContext(ctx, machine.Machine.Id) pw := progress.ContextWriter(ctx) eventID := cliprogress.VolumeEventID(volumeName, machine.Machine.Name) diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 6cdeb717..685411eb 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -1300,7 +1300,7 @@ myapp.example.com { for _, mi := range machineImages { // Checking only DockerImages because the machines in ucind cluster don't use the containerd image store. - if !machinesWithContainers.Contains(mi.Metadata.Machine) { + if !machinesWithContainers.Contains(mi.Metadata.MachineId) { // This is the machine without service containers, it should not have the unique image. for _, img := range mi.Images { assert.NotContains(t, img.RepoTags, uniqueImage) @@ -1317,7 +1317,7 @@ myapp.example.com { } } assert.True(t, hasImage, "Machine %s with container should have image %s", - mi.Metadata.Machine, uniqueImage) + mi.Metadata.MachineId, uniqueImage) } })