diff --git a/internal/machine/cluster/cluster.go b/internal/machine/cluster/cluster.go index 8178d6db..019bf941 100644 --- a/internal/machine/cluster/cluster.go +++ b/internal/machine/cluster/cluster.go @@ -3,6 +3,7 @@ package cluster import ( "bytes" "context" + "errors" "fmt" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -10,6 +11,7 @@ import ( "google.golang.org/protobuf/types/known/emptypb" "log/slog" "net/netip" + "time" "uncloud/internal/machine/api/pb" "uncloud/internal/machine/network" "uncloud/internal/machine/store" @@ -34,6 +36,35 @@ func NewCluster(state *State, store *store.Store) *Cluster { } } +func (c *Cluster) Init(ctx context.Context, network netip.Prefix) error { + initialised, err := c.Initialised(ctx) + if err != nil { + return err + } + if initialised { + return fmt.Errorf("cluster already initialized") + } + + if err = c.store.Put(ctx, "network", network.String()); err != nil { + return fmt.Errorf("put network to store: %w", err) + } + if err = c.store.Put(ctx, "created_at", time.Now().UTC().Format(time.RFC3339)); err != nil { + return fmt.Errorf("put created_at to store: %w", err) + } + return nil +} + +func (c *Cluster) Initialised(ctx context.Context) (bool, error) { + var createdAt string + if err := c.store.Get(ctx, "created_at", &createdAt); err != nil { + if errors.Is(err, store.ErrKeyNotFound) { + return false, nil + } + return false, fmt.Errorf("get created_at from store: %w", err) + } + return true, nil +} + func (c *Cluster) SetState(state *State) { c.state = state } diff --git a/internal/machine/machine.go b/internal/machine/machine.go index be642ec2..935ece5a 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -144,7 +144,7 @@ func NewMachine(config *Config) (*Machine, error) { } m.localServer = newGRPCServer(m, c) - if m.IsInitialised() { + if m.Initialised() { m.initialised <- struct{}{} } @@ -163,9 +163,9 @@ func (m *Machine) Started() <-chan struct{} { return m.started } -// IsInitialised returns true if the machine has been configured as a member of a cluster, +// Initialised returns true if the machine has been configured as a member of a cluster, // either by initialising a new cluster on it or joining an existing one. -func (m *Machine) IsInitialised() bool { +func (m *Machine) Initialised() bool { m.state.mu.RLock() defer m.state.mu.RUnlock() @@ -173,6 +173,19 @@ func (m *Machine) IsInitialised() bool { } func (m *Machine) Run(ctx context.Context) error { + // Configure and start the corrosion service on the loopback if the machine is not initialised as a cluster + // member. This provides the store required for the machine to initialise a new cluster on it. + if !m.Initialised() { + if err := m.configureCorrosion(); err != nil { + return fmt.Errorf("configure corrosion service: %w", err) + } + slog.Info("Configured corrosion service.", "dir", m.config.CorrosionDir) + + if err := m.config.CorrosionService.Start(); err != nil { + return fmt.Errorf("start corrosion service: %w", err) + } + } + // Use an errgroup to coordinate error handling and graceful shutdown of multiple machine components. errGroup, ctx := errgroup.WithContext(ctx) @@ -192,28 +205,10 @@ func (m *Machine) Run(ctx context.Context) error { ) close(m.started) - // Configure and start the corrosion service on the loopback if the machine is not initialised as a cluster - // member. This provides the store required for the machine to initialise a new cluster on it. - if !m.IsInitialised() { - // Needs to run in a goroutine because the corrosion systemd service depends on the readiness of the daemon - // indicated by closing the started channel. - errGroup.Go(func() error { - if err := m.configureCorrosion(); err != nil { - return fmt.Errorf("configure corrosion service: %w", err) - } - slog.Info("Configured corrosion service.", "dir", m.config.CorrosionDir) - - if err := m.config.CorrosionService.Start(); err != nil { - return fmt.Errorf("start corrosion service: %w", err) - } - return nil - }) - } - // Control loop for managing the network controller. errGroup.Go( func() error { - if !m.IsInitialised() { + if !m.Initialised() { slog.Info( "Waiting for the machine to be initialised as a member of a cluster " + "to start the network controller.", @@ -386,6 +381,15 @@ func (m *Machine) InitCluster(ctx context.Context, req *pb.InitClusterRequest) ( return nil, status.Errorf(codes.InvalidArgument, "set cluster network: %v", err) } + clusterNetwork, err := req.Network.ToPrefix() + if err != nil { + return nil, status.Errorf(codes.InvalidArgument, "invalid network: %v", err) + } + 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()) + // Use the public and all routable IPs as endpoints. ips, err := network.ListRoutableIPs() if err != nil { diff --git a/internal/machine/store/schema.sql b/internal/machine/store/schema.sql index 3f8cfe74..01be933d 100644 --- a/internal/machine/store/schema.sql +++ b/internal/machine/store/schema.sql @@ -1,3 +1,10 @@ +-- cluster table stores the key-value pairs of the cluster configuration. +CREATE TABLE cluster +( + key TEXT NOT NULL PRIMARY KEY, + value ANY +); + CREATE TABLE machines ( id TEXT NOT NULL PRIMARY KEY, diff --git a/internal/machine/store/store.go b/internal/machine/store/store.go index 111121d8..ee6c2008 100644 --- a/internal/machine/store/store.go +++ b/internal/machine/store/store.go @@ -1,14 +1,20 @@ package store import ( + "context" _ "embed" + "errors" "fmt" "uncloud/internal/corrosion" "uncloud/internal/machine/api/pb" ) -//go:embed schema.sql -var Schema string +var ( + //go:embed schema.sql + Schema string + + ErrKeyNotFound = errors.New("key not found") +) // Store is a cluster store backed by a distributed Corrosion database. type Store struct { @@ -19,6 +25,28 @@ func New(corro *corrosion.APIClient) *Store { return &Store{corro: corro} } +func (s *Store) Get(ctx context.Context, key string, value any) error { + rows, err := s.corro.QueryContext(ctx, "SELECT value FROM cluster WHERE key = ?", key) + if err != nil { + return err + } + if !rows.Next() { + if rows.Err() != nil { + return rows.Err() + } + return ErrKeyNotFound + } + if err = rows.Scan(value); err != nil { + return err + } + return nil +} + +func (s *Store) Put(ctx context.Context, key string, value any) error { + _, err := s.corro.ExecContext(ctx, "INSERT OR REPLACE INTO cluster (key, value) VALUES (?, ?)", key, value) + return err +} + func (s *Store) CreateMachine(machine *pb.MachineInfo) error { return fmt.Errorf("not implemented") } diff --git a/scripts/install.sh b/scripts/install.sh index 40eafa90..5e587b6d 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -177,8 +177,7 @@ install_corrosion_systemd() { cat > "${corrosion_service_path}" << EOF [Unit] Description=Uncloud gossip-based distributed store -After=uncloud.service -Requires=uncloud.service +PartOf=uncloud.service [Service] Type=simple