diff --git a/internal/machine/cluster/cluster.go b/internal/machine/cluster/cluster.go index 019bf941..e24f8597 100644 --- a/internal/machine/cluster/cluster.go +++ b/internal/machine/cluster/cluster.go @@ -7,7 +7,6 @@ import ( "fmt" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" - "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/emptypb" "log/slog" "net/netip" @@ -60,24 +59,40 @@ func (c *Cluster) Initialised(ctx context.Context) (bool, error) { if errors.Is(err, store.ErrKeyNotFound) { return false, nil } - return false, fmt.Errorf("get created_at from store: %w", err) + return false, status.Errorf(codes.Internal, "get created_at from store: %v", err) } return true, nil } +func (c *Cluster) checkInitialised(ctx context.Context) error { + initialised, err := c.Initialised(ctx) + if err != nil { + return err + } + if !initialised { + return status.Error(codes.FailedPrecondition, "cluster is not initialized") + } + return nil +} + func (c *Cluster) SetState(state *State) { c.state = state } -func (c *Cluster) Network() (netip.Prefix, error) { - if c.state == nil { - return netip.Prefix{}, status.Error(codes.FailedPrecondition, "cluster is not initialized") +func (c *Cluster) Network(ctx context.Context) (netip.Prefix, error) { + if err := c.checkInitialised(ctx); err != nil { + return netip.Prefix{}, err } - if c.state.State.Network == nil { - return netip.Prefix{}, fmt.Errorf("network not set") + var net string + if err := c.store.Get(ctx, "network", &net); err != nil { + return netip.Prefix{}, status.Errorf(codes.Internal, "get network from store: %v", err) } - return c.state.State.Network.ToPrefix() + prefix, err := netip.ParsePrefix(net) + if err != nil { + return netip.Prefix{}, status.Errorf(codes.Internal, "parse network prefix: %v", err) + } + return prefix, nil } func (c *Cluster) SetNetwork(network *pb.IPPrefix) error { @@ -102,8 +117,8 @@ func (c *Cluster) WatchNewMachines() <-chan *pb.MachineInfo { // AddMachine adds a machine to the cluster. func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*pb.AddMachineResponse, error) { - if c.state == nil { - return nil, status.Error(codes.FailedPrecondition, "cluster is not initialized") + if err := c.checkInitialised(ctx); err != nil { + return nil, err } if err := req.Validate(); err != nil { @@ -113,16 +128,16 @@ func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*p return nil, status.Error(codes.InvalidArgument, "endpoints not set") } - machines := c.state.State.Machines + machines, err := c.store.ListMachines(ctx) + if err != nil { + return nil, status.Errorf(codes.Internal, "list machines: %v", err) + } allocatedSubnets := make([]netip.Prefix, len(machines)) - var err error - i := 0 - for _, m := range machines { + for i, m := range machines { if req.Name != "" && m.Name == req.Name { return nil, status.Errorf(codes.AlreadyExists, "machine with name %q already exists", req.Name) } - if m.Network.ManagementIp != nil && req.Network.ManagementIp != nil && - m.Network.ManagementIp.Equal(req.Network.ManagementIp) { + if req.Network.ManagementIp != nil && req.Network.ManagementIp.Equal(m.Network.ManagementIp) { manageIP, _ := req.Network.ManagementIp.ToAddr() return nil, status.Errorf(codes.AlreadyExists, "machine with management IP %q already exists", manageIP) } @@ -130,37 +145,25 @@ func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*p publicKey := secret.Secret(m.Network.PublicKey) return nil, status.Errorf(codes.AlreadyExists, "machine with public key %q already exists", publicKey) } - - if allocatedSubnets[i], err = m.Network.Subnet.ToPrefix(); err != nil { - return nil, err - } - i++ + allocatedSubnets[i], _ = m.Network.Subnet.ToPrefix() } mid, err := NewMachineID() if err != nil { return nil, status.Errorf(codes.Internal, "generate machine ID: %v", err) } + name := req.Name + if name == "" { + if name, err = NewRandomMachineName(); err != nil { + return nil, status.Errorf(codes.Internal, "generate machine name: %v", err) + } + } manageIP := req.Network.ManagementIp if manageIP == nil { manageIP = pb.NewIP(network.ManagementIP(req.Network.PublicKey)) } - m := &pb.MachineInfo{ - Id: mid, - Name: req.Name, - Network: &pb.NetworkConfig{ - ManagementIp: manageIP, - PublicKey: req.Network.PublicKey, - }, - } - if m.Name == "" { - m.Name, err = NewRandomMachineName() - if err != nil { - return nil, status.Errorf(codes.Internal, "generate machine name: %v", err) - } - } - - clusterNetwork, err := c.Network() + // Allocate a subnet for the machine from the cluster network. + clusterNetwork, err := c.Network(ctx) if err != nil { return nil, status.Errorf(codes.Internal, "get cluster network: %v", err) } @@ -172,25 +175,26 @@ func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*p if err != nil { return nil, status.Errorf(codes.Internal, "allocate subnet for machine: %v", err) } - m.Network.Subnet = pb.NewIPPrefix(subnet) - // TODO: announce the new machine to the cluster network using Serf and achieve consensus. - mCopy := proto.Clone(m).(*pb.MachineInfo) - c.state.State.Machines[mCopy.Id] = mCopy - // Store machine endpoints in a separate collection to allow modifying them with limited permissions. - c.state.State.Endpoints[mCopy.Id] = &pb.MachineEndpoints{ - Id: mCopy.Id, - Endpoints: req.Network.Endpoints, + m := &pb.MachineInfo{ + Id: mid, + Name: name, + Network: &pb.NetworkConfig{ + Subnet: pb.NewIPPrefix(subnet), + ManagementIp: manageIP, + PublicKey: req.Network.PublicKey, + }, } - if err = c.state.Save(); err != nil { - return nil, status.Errorf(codes.Internal, "save state: %v", err) + + // TODO: announce the new machine to the cluster members and achieve consensus. + // We should perhaps not proceed if this machine is in a minority partition. + if err = c.store.CreateMachine(ctx, m); err != nil { + return nil, status.Errorf(codes.Internal, "create machine: %v", err) } - slog.Info("Machine added to the cluster.", "id", m.Id, "name", m.Name) + slog.Info("Machine added to the cluster.", + "id", m.Id, "name", m.Name, "subnet", subnet, "public_key", secret.Secret(m.Network.PublicKey)) - // Include the machine endpoints in the response. - m.Network.Endpoints = req.Network.Endpoints - - // TODO: notify all cluster machines about the new machine so they can update their peers config. + // TODO: Subscribe all cluster members to updates about the new machine so they can update their peers config. // In PoC we just notify the local machine. c.newMachinesCh <- m @@ -199,34 +203,17 @@ func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*p } func (c *Cluster) ListMachines(ctx context.Context, _ *emptypb.Empty) (*pb.ListMachinesResponse, error) { - if c.state == nil { - return nil, status.Error(codes.FailedPrecondition, "cluster is not initialized") + if err := c.checkInitialised(ctx); err != nil { + return nil, err } - // TODO: consider creating MachineInfo type that pb.MachineInfo is mapped to always carry valid data to reduce - // error handling caused by conversions and nil pointers. - machines := make([]*pb.MachineInfo, 0, len(c.state.State.Machines)) - for _, sm := range c.state.State.Machines { - m := proto.Clone(sm).(*pb.MachineInfo) - m.Network.Endpoints = c.state.State.Endpoints[m.Id].Endpoints - machines = append(machines, m) + + machines, err := c.store.ListMachines(ctx) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) } return &pb.ListMachinesResponse{Machines: machines}, nil } -func (c *Cluster) ListMachineEndpoints( - ctx context.Context, req *pb.ListMachineEndpointsRequest, -) (*pb.ListMachineEndpointsResponse, error) { - if c.state == nil { - return nil, status.Error(codes.FailedPrecondition, "cluster is not initialized") - } - - endpoints, ok := c.state.State.Endpoints[req.Id] - if !ok { - return nil, status.Errorf(codes.NotFound, "machine %q not found", req.Id) - } - return &pb.ListMachineEndpointsResponse{Endpoints: endpoints}, nil -} - func (c *Cluster) AddUser(user *pb.User) error { c.state.State.Users = append(c.state.State.Users, user) return c.state.Save() diff --git a/internal/machine/machine.go b/internal/machine/machine.go index a4ec7398..98997c18 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -390,7 +390,7 @@ func (m *Machine) InitCluster(ctx context.Context, req *pb.InitClusterRequest) ( if err = m.cluster.Init(ctx, clusterNetwork); err != nil { return nil, status.Errorf(codes.Internal, "init cluster: %v", err) } - slog.Info("Cluster initialised.", "network", clusterNetwork.String()) + slog.Info("Cluster state initialised.", "network", clusterNetwork.String()) // Use the public and all routable IPs as endpoints. ips, err := network.ListRoutableIPs() @@ -439,24 +439,6 @@ func (m *Machine) InitCluster(ctx context.Context, req *pb.InitClusterRequest) ( PrivateKey: m.state.Network.PrivateKey, PublicKey: m.state.Network.PublicKey, } - - // Add a user to the cluster and build a peers config from it if provided. - if req.User != nil { - if err = m.cluster.AddUser(req.User); err != nil { - return nil, status.Errorf(codes.Internal, "add user to cluster: %v", err) - } - userManageIP, uErr := req.User.Network.ManagementIp.ToAddr() - if uErr != nil { - return nil, status.Error(codes.Internal, uErr.Error()) - } - - m.state.Network.Peers = make([]network.PeerConfig, 1) - m.state.Network.Peers[0] = network.PeerConfig{ - ManagementIP: userManageIP, - PublicKey: req.User.Network.PublicKey, - } - } - if err = m.state.Save(); err != nil { return nil, status.Errorf(codes.Internal, "save machine state: %v", err) } diff --git a/internal/machine/store/store.go b/internal/machine/store/store.go index ee6c2008..40530830 100644 --- a/internal/machine/store/store.go +++ b/internal/machine/store/store.go @@ -5,6 +5,7 @@ import ( _ "embed" "errors" "fmt" + "google.golang.org/protobuf/encoding/protojson" "uncloud/internal/corrosion" "uncloud/internal/machine/api/pb" ) @@ -47,10 +48,36 @@ func (s *Store) Put(ctx context.Context, key string, value any) error { return err } -func (s *Store) CreateMachine(machine *pb.MachineInfo) error { - return fmt.Errorf("not implemented") +func (s *Store) CreateMachine(ctx context.Context, m *pb.MachineInfo) error { + mJSON, err := protojson.Marshal(m) + if err != nil { + return fmt.Errorf("marshal machine info: %w", err) + } + _, err = s.corro.ExecContext(ctx, "INSERT INTO machines (id, info) VALUES (?, ?)", m.Id, string(mJSON)) + if err != nil { + return fmt.Errorf("insert query: %w", err) + } + return nil } -func (s *Store) ListMachines() ([]*pb.MachineInfo, error) { - return nil, fmt.Errorf("not implemented") +func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) { + rows, err := s.corro.QueryContext(ctx, "SELECT info FROM machines ORDER BY name") + if err != nil { + return nil, err + } + defer rows.Close() + + var machines []*pb.MachineInfo + for rows.Next() { + var mJSON string + if err = rows.Scan(&mJSON); err != nil { + return nil, err + } + var m pb.MachineInfo + if err = protojson.Unmarshal([]byte(mJSON), &m); err != nil { + return nil, fmt.Errorf("unmarshal machine info: %w", err) + } + machines = append(machines, &m) + } + return machines, nil }