From 3f71aec2752a8f805bae8a06bb5ef1bd4cd83961 Mon Sep 17 00:00:00 2001 From: Pavel Sviderski Date: Thu, 5 Sep 2024 12:21:01 +1000 Subject: [PATCH] add `uncloud machine init` command to initialise a new cluster on local machine --- Makefile | 7 +- cmd/uncloud/machine/init.go | 64 ++++++++++++ cmd/uncloud/machine/root.go | 1 + cmd/uncloudd/main.go | 11 ++- internal/machine/api/server.go | 20 ---- internal/machine/daemon/daemon.go | 157 +++++++++++++++++++++++++----- internal/secret/secret.go | 15 ++- 7 files changed, 222 insertions(+), 53 deletions(-) create mode 100644 cmd/uncloud/machine/init.go delete mode 100644 internal/machine/api/server.go diff --git a/Makefile b/Makefile index 18695922..a4f7277e 100644 --- a/Makefile +++ b/Makefile @@ -1,9 +1,12 @@ .PHONY: build uncloudd-dev1: - GOOS=linux GOARCH=amd64 go build -o uncloudd-linux-amd64 cmd/uncloudd/main.go && \ + GOOS=linux GOARCH=amd64 go build -o uncloudd-linux-amd64 ./cmd/uncloudd && \ scp uncloudd-linux-amd64 spy@192.168.40.243:~/ && \ ssh spy@192.168.40.243 sudo install ./uncloudd-linux-amd64 /usr/local/bin/uncloudd + GOOS=linux GOARCH=amd64 go build -o uncloud-linux-amd64 ./cmd/uncloud && \ + scp uncloud-linux-amd64 spy@192.168.40.243:~/ && \ + ssh spy@192.168.40.243 sudo install ./uncloud-linux-amd64 /usr/local/bin/uncloud .PHONY: proto proto: - protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/machine/cluster/pb/cluster.proto + protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/machine/api/pb/cluster.proto diff --git a/cmd/uncloud/machine/init.go b/cmd/uncloud/machine/init.go new file mode 100644 index 00000000..5249614b --- /dev/null +++ b/cmd/uncloud/machine/init.go @@ -0,0 +1,64 @@ +package machine + +import ( + "fmt" + "github.com/spf13/cobra" + "net/netip" + "uncloud/internal/machine" + "uncloud/internal/machine/api/pb" + "uncloud/internal/machine/daemon" + "uncloud/internal/machine/network" + "uncloud/internal/secret" +) + +type initOptions struct { + name string + network string + userPublicKey string + dataDir string +} + +func NewInitCommand() *cobra.Command { + opts := initOptions{} + cmd := &cobra.Command{ + Use: "init", + Short: "Initialise a new cluster that consists of the local or remote machine", + RunE: func(cmd *cobra.Command, args []string) error { + netPrefix, err := netip.ParsePrefix(opts.network) + if err != nil { + return fmt.Errorf("parse network CIDR: %w", err) + } + + var users []*pb.User + if opts.userPublicKey != "" { + pubKey, uErr := secret.FromHexString(opts.userPublicKey) + if uErr != nil { + return fmt.Errorf("parse user's public key: %w", uErr) + } + user := &pb.User{ + Network: &pb.NetworkConfig{ + ManagementIp: pb.NewIP(network.ManagementIP(pubKey)), + PublicKey: pubKey, + }, + } + users = append(users, user) + } + + if err = daemon.InitCluster(opts.dataDir, opts.name, netPrefix, users); err != nil { + return fmt.Errorf("initialise cluster: %w", err) + } + return nil + }, + } + cmd.Flags().StringVarP(&opts.name, "name", "n", "", "Assign a name to the machine") + cmd.Flags().StringVar(&opts.network, "network", network.DefaultNetwork.String(), + "IPv4 network CIDR to use for machines and services") + cmd.Flags().StringVarP(&opts.userPublicKey, "user-pubkey", "u", "", + "User's public key which will be able to access the cluster (hex-encoded)") + + cmd.Flags().StringVarP(&opts.dataDir, "data-dir", "d", machine.DefaultDataDir, + "Directory for storing persistent machine state") + _ = cmd.MarkFlagDirname("data-dir") + + return cmd +} diff --git a/cmd/uncloud/machine/root.go b/cmd/uncloud/machine/root.go index bb66e23d..12313a07 100644 --- a/cmd/uncloud/machine/root.go +++ b/cmd/uncloud/machine/root.go @@ -11,6 +11,7 @@ func NewRootCommand() *cobra.Command { } cmd.AddCommand( NewAddCommand(), + NewInitCommand(), ) return cmd } diff --git a/cmd/uncloudd/main.go b/cmd/uncloudd/main.go index 806692f3..a50889f1 100644 --- a/cmd/uncloudd/main.go +++ b/cmd/uncloudd/main.go @@ -21,7 +21,14 @@ func main() { SilenceUsage: true, SilenceErrors: true, RunE: func(cmd *cobra.Command, args []string) error { - return daemon.Run(cmd.Context(), dataDir) + d, err := daemon.New(dataDir) + if err != nil { + return err + } + if err = d.Run(cmd.Context()); err == nil { + slog.Info("Daemon stopped.") + } + return err }, } cmd.PersistentFlags().StringVarP(&dataDir, "data-dir", "d", machine.DefaultDataDir, @@ -41,5 +48,5 @@ func main() { }() cobra.CheckErr(cmd.ExecuteContext(ctx)) - slog.Info("Daemon stopped.") + } diff --git a/internal/machine/api/server.go b/internal/machine/api/server.go deleted file mode 100644 index b319ea2a..00000000 --- a/internal/machine/api/server.go +++ /dev/null @@ -1,20 +0,0 @@ -package api - -import ( - "context" - pb2 "uncloud/internal/machine/api/pb" -) - -// Server is the gRPC server for the Cluster service. -type Server struct { - pb2.UnimplementedClusterServer -} - -func NewServer() *Server { - return &Server{} -} - -// AddMachine adds a machine to the cluster. -func (s *Server) AddMachine(ctx context.Context, req *pb2.AddMachineRequest) (*pb2.AddMachineResponse, error) { - return &pb2.AddMachineResponse{}, nil -} diff --git a/internal/machine/daemon/daemon.go b/internal/machine/daemon/daemon.go index 5c77999a..9b61ed76 100644 --- a/internal/machine/daemon/daemon.go +++ b/internal/machine/daemon/daemon.go @@ -2,36 +2,150 @@ package daemon import ( "context" + "errors" "fmt" "golang.org/x/sync/errgroup" - "google.golang.org/grpc" "log/slog" "net" + "net/netip" + "os" "strconv" "uncloud/internal/machine" - "uncloud/internal/machine/api" "uncloud/internal/machine/api/pb" + "uncloud/internal/machine/cluster" "uncloud/internal/machine/network" ) -const ( - MachineAPIPort = 51000 -) - -func Run(ctx context.Context, dataDir string) error { - cfg, err := machine.ParseConfig(machine.ConfigPath(dataDir)) +// InitCluster resets the local machine and initialises a new cluster with it. +// TODO: ideally, this should be an RPC call to the daemon API to correctly handle the leave request and reconfiguration. +func InitCluster(dataDir, machineName string, netPrefix netip.Prefix, users []*pb.User) error { + var err error + if machineName == "" { + machineName, err = machine.NewRandomName() + if err != nil { + return fmt.Errorf("generate machine name: %w", err) + } + } + privKey, pubKey, err := network.NewMachineKeys() if err != nil { - return fmt.Errorf("load machine config: %w", err) + return fmt.Errorf("generate machine keys: %w", err) } + state := cluster.NewState(cluster.StatePath(dataDir)) + c := cluster.NewCluster(state, "") + if err = c.SetNetwork(netPrefix); err != nil { + return fmt.Errorf("set cluster network: %w", err) + } + + // Use all routable addresses as endpoints. + addrs, err := network.ListRoutableAddresses() + if err != nil { + return fmt.Errorf("list routable addresses: %w", err) + } + endpoints := make([]*pb.IPPort, len(addrs)) + for i, addr := range addrs { + addrPort := netip.AddrPortFrom(addr, network.WireGuardPort) + endpoints[i] = pb.NewIPPort(addrPort) + } + // Register the new machine in the cluster to populate the state and get its ID and subnet. + req := &pb.AddMachineRequest{ + Name: machineName, + Network: &pb.NetworkConfig{ + Endpoints: endpoints, + PublicKey: pubKey, + }, + } + resp, err := c.AddMachine(context.Background(), req) + if err != nil { + return fmt.Errorf("add machine to cluster: %w", err) + } + + m := resp.Machine + subnet, err := m.Network.Subnet.ToPrefix() + if err != nil { + return err + } + manageIP, err := m.Network.ManagementIp.ToAddr() + if err != nil { + return err + } + mcfg := &machine.Config{ + ID: m.Id, + Name: m.Name, + Network: &network.Config{ + Subnet: subnet, + ManagementIP: manageIP, + PrivateKey: privKey, + PublicKey: pubKey, + }, + } + + // Add users to the cluster and build peers config from them. + peers := make([]network.PeerConfig, len(users)) + for i, u := range users { + if err = c.AddUser(u); err != nil { + return fmt.Errorf("add user to cluster: %w", err) + } + userManageIP, uErr := u.Network.ManagementIp.ToAddr() + if uErr != nil { + return uErr + } + peers[i] = network.PeerConfig{ + ManagementIP: userManageIP, + PublicKey: u.Network.PublicKey, + } + } + mcfg.Network.Peers = peers + + mcfg.SetPath(machine.ConfigPath(dataDir)) + if err = mcfg.Save(); err != nil { + return fmt.Errorf("save machine config: %w", err) + } + + fmt.Printf("Cluster initialised with machine %q\n", m.Name) + return nil +} + +type Daemon struct { + config *machine.Config + cluster *cluster.Cluster +} + +func New(dataDir string) (*Daemon, error) { + cfg, err := machine.ParseConfig(machine.ConfigPath(dataDir)) + if err != nil { + return nil, fmt.Errorf("load machine config: %w", err) + } + + statePath := cluster.StatePath(dataDir) + state := cluster.NewState(statePath) + if err = state.Load(); err != nil { + if !errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf("load cluster state: %w", err) + } + slog.Info("No cluster state found, creating a new one.", "path", statePath) + if err = state.Save(); err != nil { + return nil, fmt.Errorf("save cluster state: %w", err) + } + } + + apiAddr := net.JoinHostPort(cfg.Network.ManagementIP.String(), strconv.Itoa(machine.APIPort)) + c := cluster.NewCluster(state, apiAddr) + + return &Daemon{ + config: cfg, + cluster: c, + }, nil +} + +func (d *Daemon) Run(ctx context.Context) error { wgnet, err := network.NewWireGuardNetwork() if err != nil { return fmt.Errorf("create WireGuard network: %w", err) } - if err = wgnet.Configure(*cfg.Network); err != nil { + if err = wgnet.Configure(*d.config.Network); err != nil { return fmt.Errorf("configure WireGuard network: %w", err) } - //ctx, cancel := context.WithCancel(context.Background()) //go wgnet.WatchEndpoints(ctx, peerEndpointChangeNotifier) @@ -41,20 +155,12 @@ func Run(ctx context.Context, dataDir string) error { //} //fmt.Println("Addresses:", addrs) - apiAddr := net.JoinHostPort(cfg.Network.ManagementIP.String(), strconv.Itoa(MachineAPIPort)) - listener, err := net.Listen("tcp", apiAddr) - if err != nil { - return fmt.Errorf("listen API port: %w", err) - } - grpcServer := grpc.NewServer() - pb.RegisterClusterServer(grpcServer, api.NewServer()) - // Use an errgroup to coordinate error handling and graceful shutdown of multiple daemon components. errGroup, ctx := errgroup.WithContext(ctx) errGroup.Go(func() error { - slog.Info("Starting API server.", "addr", apiAddr) - if sErr := grpcServer.Serve(listener); sErr != nil { - return fmt.Errorf("API server failed: %w", sErr) + slog.Info("Starting cluster.") + if err = d.cluster.Run(); err != nil { + return fmt.Errorf("cluster failed: %w", err) } return nil }) @@ -67,10 +173,9 @@ func Run(ctx context.Context, dataDir string) error { // Shutdown goroutine. errGroup.Go(func() error { <-ctx.Done() - slog.Info("Stopping API server.") - // TODO: implement timeout for graceful shutdown. - grpcServer.GracefulStop() - slog.Info("API server stopped.") + slog.Info("Stopping cluster.") + d.cluster.Stop() + slog.Info("Cluster stopped.") return nil }) diff --git a/internal/secret/secret.go b/internal/secret/secret.go index 4b9a667c..1d8c62ac 100644 --- a/internal/secret/secret.go +++ b/internal/secret/secret.go @@ -9,6 +9,15 @@ import ( type Secret []byte +// FromHexString parses a hex-encoded string into a secret. +func FromHexString(s string) (Secret, error) { + decoded, err := hex.DecodeString(s) + if err != nil { + return nil, fmt.Errorf("invalid hex-encoded secret: %w", err) + } + return decoded, nil +} + // String returns the hex-encoded string representation of the secret. // //goland:noinspection GoMixedReceiverTypes @@ -23,11 +32,11 @@ func (s Secret) MarshalText() ([]byte, error) { //goland:noinspection GoMixedReceiverTypes func (s *Secret) UnmarshalText(text []byte) error { - decoded, err := hex.DecodeString(string(text)) + secret, err := FromHexString(string(text)) if err != nil { - return fmt.Errorf("invalid hex-encoded secret: %w", err) + return err } - *s = decoded + *s = secret return nil }