diff --git a/internal/machine/cluster/cluster.go b/internal/machine/cluster/cluster.go index 059e312d..8178d6db 100644 --- a/internal/machine/cluster/cluster.go +++ b/internal/machine/cluster/cluster.go @@ -12,6 +12,7 @@ import ( "net/netip" "uncloud/internal/machine/api/pb" "uncloud/internal/machine/network" + "uncloud/internal/machine/store" "uncloud/internal/secret" ) @@ -19,14 +20,16 @@ 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) *Cluster { +func NewCluster(state *State, store *store.Store) *Cluster { return &Cluster{ state: state, + store: store, newMachinesCh: make(chan *pb.MachineInfo, 1), } } diff --git a/internal/machine/corrosion/config.go b/internal/machine/corroservice/config.go similarity index 99% rename from internal/machine/corrosion/config.go rename to internal/machine/corroservice/config.go index 08282e12..3869cd9b 100644 --- a/internal/machine/corrosion/config.go +++ b/internal/machine/corroservice/config.go @@ -1,4 +1,4 @@ -package corrosion +package corroservice import ( "bytes" diff --git a/internal/machine/corrosion/service.go b/internal/machine/corroservice/service.go similarity index 67% rename from internal/machine/corrosion/service.go rename to internal/machine/corroservice/service.go index 786ac9bb..cf24bfee 100644 --- a/internal/machine/corrosion/service.go +++ b/internal/machine/corroservice/service.go @@ -1,4 +1,4 @@ -package corrosion +package corroservice type Service interface { Start() error diff --git a/internal/machine/corrosion/systemd.go b/internal/machine/corroservice/systemd.go similarity index 96% rename from internal/machine/corrosion/systemd.go rename to internal/machine/corroservice/systemd.go index edf4f6d1..c520df06 100644 --- a/internal/machine/corrosion/systemd.go +++ b/internal/machine/corroservice/systemd.go @@ -1,4 +1,4 @@ -package corrosion +package corroservice import ( "fmt" diff --git a/internal/machine/machine.go b/internal/machine/machine.go index d54931f1..1db4f38b 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -17,9 +17,10 @@ import ( "os/user" "path/filepath" "strconv" + "uncloud/internal/corrosion" "uncloud/internal/machine/api/pb" "uncloud/internal/machine/cluster" - "uncloud/internal/machine/corrosion" + "uncloud/internal/machine/corroservice" "uncloud/internal/machine/network" "uncloud/internal/machine/store" ) @@ -34,8 +35,10 @@ type Config struct { DataDir string APISockPath string - CorrosionDir string - CorrosionService corrosion.Service + CorrosionDir string + CorrosionAPIListenAddr netip.AddrPort + CorrosionAPIAddr netip.AddrPort + CorrosionService corroservice.Service } // SetDefaults returns a new Config with default values set where not provided. @@ -50,10 +53,18 @@ func (c *Config) SetDefaults() *Config { cfg.APISockPath = DefaultAPISockPath } if cfg.CorrosionDir == "" { - cfg.CorrosionDir = filepath.Join(cfg.DataDir, "corrosion") + cfg.CorrosionDir = filepath.Join(cfg.DataDir, "corroservice") + } + if !cfg.CorrosionAPIListenAddr.IsValid() { + cfg.CorrosionAPIListenAddr = netip.AddrPortFrom( + netip.AddrFrom4([4]byte{127, 0, 0, 1}), corroservice.DefaultAPIPort) + } + if !cfg.CorrosionAPIAddr.IsValid() { + cfg.CorrosionAPIAddr = netip.AddrPortFrom( + netip.AddrFrom4([4]byte{127, 0, 0, 1}), corroservice.DefaultAPIPort) } if cfg.CorrosionService == nil { - cfg.CorrosionService = corrosion.DefaultSystemdService(cfg.CorrosionDir) + cfg.CorrosionService = corroservice.DefaultSystemdService(cfg.CorrosionDir) } return &cfg } @@ -103,18 +114,24 @@ func NewMachine(config *Config) (*Machine, error) { } } + corro, err := corrosion.NewAPIClient(config.CorrosionAPIAddr) + if err != nil { + return nil, fmt.Errorf("create corrosion API client: %w", err) + } + corroStore := store.New(corro) + var c *cluster.Cluster clusterState := cluster.NewState(cluster.StatePath(config.DataDir)) if err = clusterState.Load(); err != nil { if errors.Is(err, os.ErrNotExist) { // Cluster state file does not exist, initialise the cluster without a state to fail cluster requests. - c = cluster.NewCluster(nil) + c = cluster.NewCluster(nil, corroStore) } else { return nil, fmt.Errorf("load cluster state: %w", err) } } else { // Cluster state is successfully loaded, initialise the cluster with it. - c = cluster.NewCluster(clusterState) + c = cluster.NewCluster(clusterState, corroStore) } m := &Machine{ @@ -198,7 +215,7 @@ func (m *Machine) Run(ctx context.Context) error { slog.Info("Starting network controller.") networkServer := newGRPCServer(m, m.cluster) - if err = m.configureCorrosion(m.config.CorrosionDir); err != nil { + if err = m.configureCorrosion(); err != nil { return fmt.Errorf("configure corrosion service: %w", err) } @@ -272,12 +289,12 @@ func listenUnixSocket(path string) (net.Listener, error) { return sockets.NewUnixSocket(path, gid) } -func (m *Machine) configureCorrosion(dataDir string) error { - if err := corrosion.MkDataDir(dataDir, corrosion.DefaultUser); err != nil { +func (m *Machine) configureCorrosion() error { + if err := corroservice.MkDataDir(m.config.CorrosionDir, corroservice.DefaultUser); err != nil { return fmt.Errorf("create corrosion data directory: %w", err) } - configPath := filepath.Join(dataDir, "config.toml") - schemaPath := filepath.Join(dataDir, "schema.sql") + configPath := filepath.Join(m.config.CorrosionDir, "config.toml") + schemaPath := filepath.Join(m.config.CorrosionDir, "schema.sql") // TODO: use a partial list of machine peers for bootstrapping if the cluster is large. var bootstrap []string @@ -286,33 +303,33 @@ func (m *Machine) configureCorrosion(dataDir string) error { // Skip non-machine peers. continue } - bootstrap = append(bootstrap, netip.AddrPortFrom(peer.ManagementIP, corrosion.DefaultGossipPort).String()) + bootstrap = append(bootstrap, netip.AddrPortFrom(peer.ManagementIP, corroservice.DefaultGossipPort).String()) } - cfg := corrosion.Config{ - DB: corrosion.DBConfig{ - Path: filepath.Join(dataDir, "store.db"), + cfg := corroservice.Config{ + DB: corroservice.DBConfig{ + Path: filepath.Join(m.config.CorrosionDir, "store.db"), SchemaPaths: []string{schemaPath}, }, - Gossip: corrosion.GossipConfig{ - Addr: netip.AddrPortFrom(m.state.Network.ManagementIP, corrosion.DefaultGossipPort), + Gossip: corroservice.GossipConfig{ + Addr: netip.AddrPortFrom(m.state.Network.ManagementIP, corroservice.DefaultGossipPort), Bootstrap: bootstrap, Plaintext: true, }, - API: corrosion.APIConfig{ - Addr: netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), corrosion.DefaultAPIPort), + API: corroservice.APIConfig{ + Addr: netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), corroservice.DefaultAPIPort), }, - Admin: corrosion.AdminConfig{ - Path: filepath.Join(dataDir, "admin.sock"), + Admin: corroservice.AdminConfig{ + Path: filepath.Join(m.config.CorrosionDir, "admin.sock"), }, } - if err := cfg.Write(configPath, corrosion.DefaultUser); err != nil { + if err := cfg.Write(configPath, corroservice.DefaultUser); err != nil { return fmt.Errorf("write corrosion config: %w", err) } if err := os.WriteFile(schemaPath, []byte(store.Schema), 0644); err != nil { return fmt.Errorf("write corrosion schema: %w", err) } - if err := corrosion.Chown(schemaPath, corrosion.DefaultUser); err != nil { + if err := corroservice.Chown(schemaPath, corroservice.DefaultUser); err != nil { return fmt.Errorf("chown corrosion schema: %w", err) } diff --git a/internal/machine/network.go b/internal/machine/network.go index 917101e0..e19d2653 100644 --- a/internal/machine/network.go +++ b/internal/machine/network.go @@ -10,7 +10,7 @@ import ( "net/netip" "strconv" "uncloud/internal/machine/api/pb" - "uncloud/internal/machine/corrosion" + "uncloud/internal/machine/corroservice" "uncloud/internal/machine/network" ) @@ -24,14 +24,14 @@ type networkController struct { state *State wgnet *network.WireGuardNetwork server *grpc.Server - corroService corrosion.Service + corroService corroservice.Service newMachinesCh <-chan *pb.MachineInfo // TODO: DNS server/resolver listening on the machine IP, e.g. 10.210.0.1:53. It can't listen on 127.0.X.X // like resolved does because it needs to be reachable from both the host and the containers. } func newNetworkController( - state *State, server *grpc.Server, corroService corrosion.Service, newMachCh <-chan *pb.MachineInfo, + state *State, server *grpc.Server, corroService corroservice.Service, newMachCh <-chan *pb.MachineInfo, ) ( *networkController, error, ) {