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

delete old non-distributed cluster state, fix storing endpoints in store

Browse Source
This commit is contained in:
Pavel Sviderski
2024-10-02 13:03:17 +10:00
parent c275273617
commit c3ba43a8fa
11 changed files with 357 additions and 933 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/machine/cluster/cluster.go
+8 -36
View File
@@ -20,16 +20,13 @@ import (
type Cluster struct {
pb.UnimplementedClusterServer
state *State
store *store.Store
// TODO: temporary channel until the state is replaced with networkDB.
newMachinesCh chan *pb.MachineInfo
}
func NewCluster(state *State, store *store.Store) *Cluster {
func NewCluster(store *store.Store) *Cluster {
return &Cluster{
state: state,
store: store,
newMachinesCh: make(chan *pb.MachineInfo, 1),
}
@@ -41,7 +38,7 @@ func (c *Cluster) Init(ctx context.Context, network netip.Prefix) error {
return err
}
if initialised {
return fmt.Errorf("cluster already initialized")
return fmt.Errorf("cluster is already initialised")
}
if err = c.store.Put(ctx, "network", network.String()); err != nil {
@@ -70,15 +67,11 @@ func (c *Cluster) checkInitialised(ctx context.Context) error {
return err
}
if !initialised {
return status.Error(codes.FailedPrecondition, "cluster is not initialized")
return status.Error(codes.FailedPrecondition, "cluster is not initialised")
}
return nil
}
func (c *Cluster) SetState(state *State) {
c.state = state
}
func (c *Cluster) Network(ctx context.Context) (netip.Prefix, error) {
if err := c.checkInitialised(ctx); err != nil {
return netip.Prefix{}, err
@@ -95,21 +88,6 @@ func (c *Cluster) Network(ctx context.Context) (netip.Prefix, error) {
return prefix, nil
}
func (c *Cluster) SetNetwork(network *pb.IPPrefix) error {
if c.state == nil {
return status.Error(codes.FailedPrecondition, "cluster is not initialized")
}
if c.state.State.Network != nil {
return fmt.Errorf("network already set and cannot be changed")
}
if network == nil {
return fmt.Errorf("network not set")
}
c.state.State.Network = network
return nil
}
// TODO: this is a temporary watcher for PoC until the state is state is replaced with networkDB.
func (c *Cluster) WatchNewMachines() <-chan *pb.MachineInfo {
return c.newMachinesCh
@@ -121,7 +99,10 @@ func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*p
return nil, err
}
if err := req.Validate(); err != nil {
if req.Network == nil {
return nil, status.Error(codes.InvalidArgument, "network not set")
}
if err := req.Network.Validate(); err != nil {
return nil, err
}
if len(req.Network.Endpoints) == 0 {
@@ -182,10 +163,10 @@ func (c *Cluster) AddMachine(ctx context.Context, req *pb.AddMachineRequest) (*p
Network: &pb.NetworkConfig{
Subnet: pb.NewIPPrefix(subnet),
ManagementIp: manageIP,
Endpoints: req.Network.Endpoints,
PublicKey: req.Network.PublicKey,
},
}
// 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 {
@@ -213,12 +194,3 @@ func (c *Cluster) ListMachines(ctx context.Context, _ *emptypb.Empty) (*pb.ListM
}
return &pb.ListMachinesResponse{Machines: machines}, nil
}
func (c *Cluster) AddUser(user *pb.User) error {
c.state.State.Users = append(c.state.State.Users, user)
return c.state.Save()
}
func (c *Cluster) ListUsers() []*pb.User {
return c.state.State.Users
}
internal/machine/cluster/state.go
-60
View File
@@ -1,60 +0,0 @@
package cluster
import (
"fmt"
"google.golang.org/protobuf/proto"
"os"
"path/filepath"
"uncloud/internal/machine/api/pb"
)
const StateFile = "cluster.pb"
type State struct {
State *pb.State
path string
}
func StatePath(dataDir string) string {
return filepath.Join(dataDir, StateFile)
}
func NewState(path string) *State {
return &State{
State: &pb.State{
Machines: make(map[string]*pb.MachineInfo),
Endpoints: make(map[string]*pb.MachineEndpoints),
},
path: path,
}
}
func (s *State) Load() error {
data, err := os.ReadFile(s.path)
if err != nil {
return fmt.Errorf("read state file %q: %w", s.path, err)
}
if err = proto.Unmarshal(data, s.State); err != nil {
return fmt.Errorf("parse state file %q: %w", s.path, err)
}
if s.State.Machines == nil {
s.State.Machines = make(map[string]*pb.MachineInfo)
}
if s.State.Endpoints == nil {
s.State.Endpoints = make(map[string]*pb.MachineEndpoints)
}
return nil
}
func (s *State) Save() error {
dir, _ := filepath.Split(s.path)
if err := os.MkdirAll(dir, 0711); err != nil {
return fmt.Errorf("create state directory %q: %w", dir, err)
}
data, err := proto.Marshal(s.State)
if err != nil {
return fmt.Errorf("marshal state: %w", err)
}
return os.WriteFile(s.path, data, 0600)
}