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

add init machine to store when initialising a cluster

Browse Source
This commit is contained in:
Pavel Sviderski
2024-10-02 11:34:54 +10:00
parent c6358474ac
commit 4ec87db8e8
3 changed files with 94 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/machine/cluster/cluster.go
+62 -75
View File
@@ -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()
internal/machine/machine.go
+1 -19
View File
@@ -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)
}
internal/machine/store/store.go
+31 -4
View File
@@ -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
}