philipp/uncloud
1
0
Fork 0
mirror of https://github.com/psviderski/uncloud.git synced 2026-08-26 11:03:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add RemoveMachine API endpoint to remove a machine from the cluster

Browse Source
This commit is contained in:
Pasha Sviderski
2025-07-25 19:30:59 +10:00
parent 4166474ee8
commit 2714587ec5
5 changed files with 618 additions and 473 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/machine/api/pb/cluster.pb.go
+501 -432
View File
File diff suppressed because it is too large Load Diff
internal/machine/api/pb/cluster.proto
+19 -15
View File
@@ -12,7 +12,7 @@ service Cluster {
rpc AddMachine(AddMachineRequest) returns (AddMachineResponse);
rpc ListMachines(google.protobuf.Empty) returns (ListMachinesResponse);
rpc UpdateMachine(UpdateMachineRequest) returns (UpdateMachineResponse);
// TODO: add RemoveMachine (see stashed changes)
rpc RemoveMachine(RemoveMachineRequest) returns (google.protobuf.Empty);
rpc ReserveDomain(ReserveDomainRequest) returns (Domain);
rpc GetDomain(google.protobuf.Empty) returns (Domain);
@@ -51,6 +51,24 @@ message ListMachinesResponse {
repeated MachineMember machines = 1;
}
message UpdateMachineRequest {
// Machine to update
string machine_id = 1;
// Updated machine information
optional string name = 2;
optional IP public_ip = 3;
repeated IPPort endpoints = 4;
}
message UpdateMachineResponse {
MachineInfo machine = 1;
}
message RemoveMachineRequest {
string id = 1;
}
message Domain {
string name = 1;
}
@@ -78,17 +96,3 @@ message DNSRecord {
RecordType type = 2;
repeated string values = 3;
}
message UpdateMachineRequest {
// Machine to update
string machine_id = 1;
// Updated machine information
optional string name = 2;
optional IP public_ip = 3;
repeated IPPort endpoints = 4;
}
message UpdateMachineResponse {
MachineInfo machine = 1;
}
internal/machine/api/pb/cluster_grpc.pb.go
+38
View File
@@ -23,6 +23,7 @@ const (
Cluster_AddMachine_FullMethodName = "/api.Cluster/AddMachine"
Cluster_ListMachines_FullMethodName = "/api.Cluster/ListMachines"
Cluster_UpdateMachine_FullMethodName = "/api.Cluster/UpdateMachine"
Cluster_RemoveMachine_FullMethodName = "/api.Cluster/RemoveMachine"
Cluster_ReserveDomain_FullMethodName = "/api.Cluster/ReserveDomain"
Cluster_GetDomain_FullMethodName = "/api.Cluster/GetDomain"
Cluster_ReleaseDomain_FullMethodName = "/api.Cluster/ReleaseDomain"
@@ -36,6 +37,7 @@ type ClusterClient interface {
AddMachine(ctx context.Context, in *AddMachineRequest, opts ...grpc.CallOption) (*AddMachineResponse, error)
ListMachines(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*ListMachinesResponse, error)
UpdateMachine(ctx context.Context, in *UpdateMachineRequest, opts ...grpc.CallOption) (*UpdateMachineResponse, error)
RemoveMachine(ctx context.Context, in *RemoveMachineRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
ReserveDomain(ctx context.Context, in *ReserveDomainRequest, opts ...grpc.CallOption) (*Domain, error)
GetDomain(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*Domain, error)
ReleaseDomain(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*Domain, error)
@@ -80,6 +82,16 @@ func (c *clusterClient) UpdateMachine(ctx context.Context, in *UpdateMachineRequ
return out, nil
}
func (c *clusterClient) RemoveMachine(ctx context.Context, in *RemoveMachineRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(emptypb.Empty)
err := c.cc.Invoke(ctx, Cluster_RemoveMachine_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *clusterClient) ReserveDomain(ctx context.Context, in *ReserveDomainRequest, opts ...grpc.CallOption) (*Domain, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(Domain)
@@ -127,6 +139,7 @@ type ClusterServer interface {
AddMachine(context.Context, *AddMachineRequest) (*AddMachineResponse, error)
ListMachines(context.Context, *emptypb.Empty) (*ListMachinesResponse, error)
UpdateMachine(context.Context, *UpdateMachineRequest) (*UpdateMachineResponse, error)
RemoveMachine(context.Context, *RemoveMachineRequest) (*emptypb.Empty, error)
ReserveDomain(context.Context, *ReserveDomainRequest) (*Domain, error)
GetDomain(context.Context, *emptypb.Empty) (*Domain, error)
ReleaseDomain(context.Context, *emptypb.Empty) (*Domain, error)
@@ -150,6 +163,9 @@ func (UnimplementedClusterServer) ListMachines(context.Context, *emptypb.Empty)
func (UnimplementedClusterServer) UpdateMachine(context.Context, *UpdateMachineRequest) (*UpdateMachineResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method UpdateMachine not implemented")
}
func (UnimplementedClusterServer) RemoveMachine(context.Context, *RemoveMachineRequest) (*emptypb.Empty, error) {
return nil, status.Errorf(codes.Unimplemented, "method RemoveMachine not implemented")
}
func (UnimplementedClusterServer) ReserveDomain(context.Context, *ReserveDomainRequest) (*Domain, error) {
return nil, status.Errorf(codes.Unimplemented, "method ReserveDomain not implemented")
}
@@ -237,6 +253,24 @@ func _Cluster_UpdateMachine_Handler(srv interface{}, ctx context.Context, dec fu
return interceptor(ctx, in, info, handler)
}
func _Cluster_RemoveMachine_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RemoveMachineRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ClusterServer).RemoveMachine(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: Cluster_RemoveMachine_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ClusterServer).RemoveMachine(ctx, req.(*RemoveMachineRequest))
}
return interceptor(ctx, in, info, handler)
}
func _Cluster_ReserveDomain_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ReserveDomainRequest)
if err := dec(in); err != nil {
@@ -328,6 +362,10 @@ var Cluster_ServiceDesc = grpc.ServiceDesc{
MethodName: "UpdateMachine",
Handler: _Cluster_UpdateMachine_Handler,
},
{
MethodName: "RemoveMachine",
Handler: _Cluster_RemoveMachine_Handler,
},
{
MethodName: "ReserveDomain",
Handler: _Cluster_ReserveDomain_Handler,
internal/machine/cluster/cluster.go
+21
View File
@@ -326,3 +326,24 @@ func (c *Cluster) ListMachines(ctx context.Context, _ *emptypb.Empty) (*pb.ListM
return &pb.ListMachinesResponse{Machines: members}, nil
}
// RemoveMachine removes a machine from the cluster.
func (c *Cluster) RemoveMachine(ctx context.Context, req *pb.RemoveMachineRequest) (*emptypb.Empty, error) {
if err := c.checkInitialised(ctx); err != nil {
return nil, err
}
if req.Id == "" {
return nil, status.Error(codes.InvalidArgument, "machine ID not set")
}
if err := c.store.DeleteMachine(ctx, req.Id); err != nil {
if errors.Is(err, store.ErrMachineNotFound) {
return nil, status.Errorf(codes.NotFound, "machine not found: %s", req.Id)
}
return nil, status.Errorf(codes.Internal, "delete machine from store: %v", err)
}
slog.Info("Machine removed from the cluster.", "id", req.Id)
return &emptypb.Empty{}, nil
}
internal/machine/store/store.go
+39 -26
View File
@@ -68,32 +68,6 @@ func (s *Store) CreateMachine(ctx context.Context, m *pb.MachineInfo) error {
return nil
}
func (s *Store) UpdateMachine(ctx context.Context, m *pb.MachineInfo) error {
if m == nil {
return fmt.Errorf("machine info cannot be nil")
}
if m.Id == "" {
return fmt.Errorf("machine ID cannot be empty")
}
mJSON, err := protojson.Marshal(m)
if err != nil {
return fmt.Errorf("marshal machine info: %w", err)
}
result, err := s.corro.ExecContext(ctx, "UPDATE machines SET info = ? WHERE id = ?", string(mJSON), m.Id)
if err != nil {
return fmt.Errorf("update machine: %w", err)
}
// Check if machine exists
if result.RowsAffected == 0 {
return fmt.Errorf("%w: %s", ErrMachineNotFound, m.Id)
}
return nil
}
func (s *Store) GetMachine(ctx context.Context, machineID string) (*pb.MachineInfo, error) {
if machineID == "" {
return nil, fmt.Errorf("machine ID cannot be empty")
@@ -170,6 +144,45 @@ func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) {
return machines, nil
}
func (s *Store) UpdateMachine(ctx context.Context, m *pb.MachineInfo) error {
if m == nil {
return fmt.Errorf("machine info cannot be nil")
}
if m.Id == "" {
return fmt.Errorf("machine ID cannot be empty")
}
mJSON, err := protojson.Marshal(m)
if err != nil {
return fmt.Errorf("marshal machine info: %w", err)
}
result, err := s.corro.ExecContext(ctx, "UPDATE machines SET info = ? WHERE id = ?", string(mJSON), m.Id)
if err != nil {
return fmt.Errorf("update machine: %w", err)
}
// Check if machine exists
if result.RowsAffected == 0 {
return fmt.Errorf("%w: %s", ErrMachineNotFound, m.Id)
}
return nil
}
func (s *Store) DeleteMachine(ctx context.Context, id string) error {
result, err := s.corro.ExecContext(ctx, "DELETE FROM machines WHERE id = ?", id)
if err != nil {
return fmt.Errorf("delete machine: %w", err)
}
// Check if machine was deleted.
if result.RowsAffected == 0 {
return fmt.Errorf("%w: %s", ErrMachineNotFound, id)
}
return nil
}
// SubscribeMachines returns a list of machines and a channel that signals changes to the list. The channel doesn't
// receive any values, it just signals when a machine has been added, updated, or deleted in the database.
func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-chan struct{}, error) {