diff --git a/cmd/zk/delete.go b/cmd/zk/delete.go index 9574657..e67903c 100644 --- a/cmd/zk/delete.go +++ b/cmd/zk/delete.go @@ -1,12 +1,17 @@ package zk -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/zk" + "github.com/spf13/cobra" +) var deleteCmd = &cobra.Command{ Use: "delete ", Short: "Delete a ZooKeeper ensemble", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: delete 구현 + if err := zk.Delete(args[0]); err != nil { + panic(err) + } }, } diff --git a/cmd/zk/list.go b/cmd/zk/list.go index c8de6d4..07e6d6c 100644 --- a/cmd/zk/list.go +++ b/cmd/zk/list.go @@ -1,12 +1,17 @@ package zk -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/zk" + "github.com/spf13/cobra" +) var listCmd = &cobra.Command{ Use: "list", Short: "List ZooKeeper ensembles", Args: cobra.NoArgs, Run: func(cmd *cobra.Command, args []string) { - // TODO: list 구현 + if err := zk.List(); err != nil { + panic(err) + } }, } diff --git a/cmd/zk/start.go b/cmd/zk/start.go index dbd9482..e4acc0e 100644 --- a/cmd/zk/start.go +++ b/cmd/zk/start.go @@ -1,13 +1,21 @@ package zk -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/zk" + "github.com/spf13/cobra" +) var startCmd = &cobra.Command{ Use: "start [--node ]", Short: "Start a ZooKeeper ensemble", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: start 구현 + ensembleName := args[0] + myID, _ := cmd.Flags().GetInt("node") + + if err := zk.Start(ensembleName, myID); err != nil { + panic(err) + } }, } diff --git a/cmd/zk/status.go b/cmd/zk/status.go index b1567fe..fb2ae10 100644 --- a/cmd/zk/status.go +++ b/cmd/zk/status.go @@ -1,12 +1,17 @@ package zk -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/zk" + "github.com/spf13/cobra" +) var statusCmd = &cobra.Command{ Use: "status ", Short: "Show status of a ZooKeeper ensemble", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: status 구현 + if err := zk.Status(args[0]); err != nil { + panic(err) + } }, } diff --git a/cmd/zk/stop.go b/cmd/zk/stop.go index 5453a86..558db13 100644 --- a/cmd/zk/stop.go +++ b/cmd/zk/stop.go @@ -1,13 +1,21 @@ package zk -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/zk" + "github.com/spf13/cobra" +) var stopCmd = &cobra.Command{ Use: "stop [--node ]", Short: "Stop a ZooKeeper ensemble", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: stop 구현 + ensembleName := args[0] + myID, _ := cmd.Flags().GetInt("node") + + if err := zk.Stop(ensembleName, myID); err != nil { + panic(err) + } }, } diff --git a/internal/store/store.go b/internal/store/store.go index aab5e06..7507dcf 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -6,6 +6,7 @@ import ( "time" "github.com/jam2in/arcusctl/internal" + "github.com/jam2in/arcusctl/internal/topology" "go.yaml.in/yaml/v3" ) @@ -63,11 +64,29 @@ func LoadZKMeta(ensembleName string) (*ZKMeta, error) { return &meta, nil } +func LoadZKTopology(ensembleName string) (*topology.ZKTopology, error) { + data, err := os.ReadFile(filepath.Join(zkDir(ensembleName), topologyYML)) + if err != nil { + return nil, err + } + + var topo topology.ZKTopology + if err := yaml.Unmarshal(data, &topo); err != nil { + return nil, err + } + + return &topo, nil +} + func ZKExists(ensembleName string) bool { _, err := os.Stat(filepath.Join(zkDir(ensembleName), metaYML)) return err == nil } +func DeleteZK(ensembleName string) error { + return os.RemoveAll(zkDir(ensembleName)) +} + func ListZK() ([]string, error) { dir := zkBaseDir() entries, err := os.ReadDir(dir) diff --git a/internal/util.go b/internal/util.go index 6eb5ea1..8b7d446 100644 --- a/internal/util.go +++ b/internal/util.go @@ -1,8 +1,10 @@ package internal import ( + "bufio" "fmt" "log" + "os" "strings" "syscall" "time" @@ -90,3 +92,16 @@ func ReadStdin(msg string, isPassword bool) string { return input } } + +// FIXME: ReadStdin와 통합 가능 여부 검토 필요 +func Confirm(prompt string) bool { + fmt.Print(prompt) + reader := bufio.NewReader(os.Stdin) + input, err := reader.ReadString('\n') + if err != nil { + return false + } + + input = strings.TrimSpace(strings.ToLower(input)) + return input == "y" || input == "yes" +} diff --git a/internal/zk/delete.go b/internal/zk/delete.go new file mode 100644 index 0000000..be22d50 --- /dev/null +++ b/internal/zk/delete.go @@ -0,0 +1,87 @@ +package zk + +import ( + "fmt" + "strings" + + "github.com/jam2in/arcusctl/internal" + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/store" + "github.com/jam2in/arcusctl/internal/topology" +) + +func Delete(ensembleName string) error { + _, topo, err := loadEnsemble(ensembleName) + if err != nil { + return err + } + + if err := verifyTopology(topo); err != nil { + return err + } + + if err := verifyAllStopped(topo); err != nil { + return err + } + + fmt.Printf("This will remove ZooKeeper ensemble %q from all servers.\n", ensembleName) + if !internal.Confirm("Are you sure you want to proceed? (y/N): ") { + fmt.Println("Aborted.") + return nil + } + + hostsMap := groupServersByHost(topo.Servers) + for host, servers := range hostsMap { + fmt.Printf("Removing files on %s...\n", host) + if err := removeHostFiles(host, servers, topo); err != nil { + return fmt.Errorf("remove files on %s: %w", host, err) + } + } + + if err := store.DeleteZK(ensembleName); err != nil { + return fmt.Errorf("delete metadata: %w", err) + } + + fmt.Printf("ZooKeeper ensemble %q deleted.\n", ensembleName) + return nil +} + +func verifyTopology(topo *topology.ZKTopology) error { + for _, server := range topo.Servers { + confDir := fmt.Sprintf("%s/conf_myid_%d", topo.Path, server.MyID) + if err := ssh.Run(server.Host(), fmt.Sprintf("test -d %s", confDir)); err != nil { + return fmt.Errorf("topology mismatch: %s not found on %s", confDir, server.Host()) + } + } + return nil +} + +func verifyAllStopped(topo *topology.ZKTopology) error { + for _, server := range topo.Servers { + confPath := zkConfigPath(topo.Path, server.MyID) + cmd := fmt.Sprintf("pgrep -f 'QuorumPeerMain.*%s' > /dev/null 2>&1", confPath) + if err := ssh.Run(server.Host(), cmd); err == nil { + return fmt.Errorf("server %s (myid=%d) is still running. stop the ensemble before delete", + server.Host(), server.MyID) + } + } + return nil +} + +func groupServersByHost(servers []topology.ZKServer) map[string][]topology.ZKServer { + hosts := map[string][]topology.ZKServer{} + for _, server := range servers { + hosts[server.Host()] = append(hosts[server.Host()], server) + } + return hosts +} + +func removeHostFiles(host string, servers []topology.ZKServer, topo *topology.ZKTopology) error { + paths := []string{topo.Path} + for _, server := range servers { + paths = append(paths, fmt.Sprintf("%s/zk%d", server.Config.DataDir, server.MyID)) + } + + cmd := "rm -rf " + strings.Join(paths, " ") + return ssh.Run(host, cmd) +} diff --git a/internal/zk/deploy.go b/internal/zk/deploy.go index 1ed42f2..9acdfb0 100644 --- a/internal/zk/deploy.go +++ b/internal/zk/deploy.go @@ -1,12 +1,11 @@ package zk import ( - "bufio" "fmt" "os" - "strings" "text/tabwriter" + "github.com/jam2in/arcusctl/internal" "github.com/jam2in/arcusctl/internal/store" "github.com/jam2in/arcusctl/internal/topology" ) @@ -18,7 +17,7 @@ func Deploy(version string, topologyPath string) error { } printPlan(topo, version) - if !confirm() { + if !internal.Confirm("Proceed with deployment? (y/N): ") { fmt.Println("Aborted.") return nil } @@ -49,10 +48,7 @@ func prepareTopology(topologyPath string) (*topology.ZKTopology, []byte, error) return nil, nil, err } - for i := range topo.Servers { - merged := mergeConfig(topo.GlobalConfig, topo.Servers[i].Config) - topo.Servers[i].Config = &merged - } + mergeServerConfigs(topo) if err := topo.Validate(); err != nil { return nil, nil, err @@ -97,16 +93,3 @@ func printRecoveryGuide(deployed []topology.ZKServer, topo *topology.ZKTopology) fmt.Println("\nTo clean up, manually run on each server:") fmt.Printf(" rm -rf %s/conf_myid_\n", topo.Path) } - -func confirm() bool { - // FIXME: internal.ReadStdin()으로 변경 필요 - fmt.Print("\nProceed? [y/N]: ") - reader := bufio.NewReader(os.Stdin) - input, err := reader.ReadString('\n') - if err != nil { - return false - } - - input = strings.TrimSpace(strings.ToLower(input)) - return input == "y" || input == "yes" -} diff --git a/internal/zk/ensemble.go b/internal/zk/ensemble.go new file mode 100644 index 0000000..d64c519 --- /dev/null +++ b/internal/zk/ensemble.go @@ -0,0 +1,39 @@ +package zk + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal/store" + "github.com/jam2in/arcusctl/internal/topology" +) + +func loadEnsemble(name string) (*store.ZKMeta, *topology.ZKTopology, error) { + meta, err := store.LoadZKMeta(name) + if err != nil { + return nil, nil, fmt.Errorf("load ZooKeeper metadata %q: %w", name, err) + } + + topo, err := store.LoadZKTopology(name) + if err != nil { + return nil, nil, fmt.Errorf("load ZooKeeper topology %q: %w", name, err) + } + + if topo.Name != name { + return nil, nil, fmt.Errorf("ZooKeeper topology name mismatch: got %q, want %q", topo.Name, name) + } + + mergeServerConfigs(topo) + + if err := topo.Validate(); err != nil { + return nil, nil, fmt.Errorf("invalid ZooKeeper topology %q: %w", name, err) + } + + return meta, topo, nil +} + +func mergeServerConfigs(topo *topology.ZKTopology) { + for i := range topo.Servers { + merged := mergeConfig(topo.GlobalConfig, topo.Servers[i].Config) + topo.Servers[i].Config = &merged + } +} diff --git a/internal/zk/list.go b/internal/zk/list.go new file mode 100644 index 0000000..efead4e --- /dev/null +++ b/internal/zk/list.go @@ -0,0 +1,33 @@ +package zk + +import ( + "fmt" + "os" + "text/tabwriter" + + "github.com/jam2in/arcusctl/internal/store" +) + +func List() error { + names, err := store.ListZK() + if err != nil { + return fmt.Errorf("list ZooKeeper ensembles: %w", err) + } + + if len(names) == 0 { + fmt.Println("No ZooKeeper ensemble found.") + return nil + } + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "NAME\tVERSION\tDEPLOYED AT") + for _, name := range names { + meta, err := store.LoadZKMeta(name) + if err != nil { + fmt.Fprintf(w, "%s\t\t%v\n", name, err) + continue + } + fmt.Fprintf(w, "%s\t%s\t%s\n", meta.Name, meta.Version, meta.DeployedAt.Format("2006-01-02 15:04:05")) + } + return w.Flush() +} diff --git a/internal/zk/server.go b/internal/zk/server.go new file mode 100644 index 0000000..80969e5 --- /dev/null +++ b/internal/zk/server.go @@ -0,0 +1,27 @@ +package zk + +import ( + "fmt" + "path" + + "github.com/jam2in/arcusctl/internal/topology" +) + +func pickServer(topo *topology.ZKTopology, myID int) (*topology.ZKServer, error) { + for _, server := range topo.Servers { + if server.MyID == myID { + return &server, nil + } + } + + return nil, fmt.Errorf("ZooKeeper server myid=%d not found", myID) +} + +func zkServerScript(topo *topology.ZKTopology) string { + return path.Join(topo.Path, "bin", "zkServer.sh") +} + +func zkConfigPath(topologyPath string, myID int) string { + configDir := fmt.Sprintf("conf_myid_%d", myID) + return path.Join(topologyPath, configDir, "zoo.cfg") +} diff --git a/internal/zk/start.go b/internal/zk/start.go new file mode 100644 index 0000000..f33e270 --- /dev/null +++ b/internal/zk/start.go @@ -0,0 +1,47 @@ +package zk + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +func Start(ensembleName string, myID int) error { + _, topo, err := loadEnsemble(ensembleName) + if err != nil { + return err + } + + // Start a specific server if myID is provided + if myID != 0 { + server, err := pickServer(topo, myID) + if err != nil { + return err + } + if err := startServer(*server, topo); err != nil { + return err + } + fmt.Printf("ZooKeeper node %s (myid=%d) started successfully.\n", server.Host(), myID) + return nil + } + + for _, server := range topo.Servers { + if err := startServer(server, topo); err != nil { + return err + } + } + + fmt.Printf("ZooKeeper ensemble %q started successfully.\n", ensembleName) + return nil +} + +func startServer(server topology.ZKServer, topo *topology.ZKTopology) error { + fmt.Printf("Starting %s (myid=%d)...\n", server.Host(), server.MyID) + cmd := fmt.Sprintf("%s start %s", zkServerScript(topo), zkConfigPath(topo.Path, server.MyID)) + if err := ssh.Run(server.Host(), cmd); err != nil { + return fmt.Errorf("start %s: %w", server.Host(), err) + } + + return nil +} diff --git a/internal/zk/status.go b/internal/zk/status.go new file mode 100644 index 0000000..940c5fe --- /dev/null +++ b/internal/zk/status.go @@ -0,0 +1,32 @@ +package zk + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +func Status(ensembleName string) error { + meta, topo, err := loadEnsemble(ensembleName) + if err != nil { + return err + } + + fmt.Printf("Ensemble: %s (version: %s)\n\n", meta.Name, meta.Version) + + for _, server := range topo.Servers { + fmt.Printf("=== %s (myid=%d) ===\n", server.Host(), server.MyID) + if err := statusServer(server, topo); err != nil { + fmt.Printf(" error: %v\n", err) + } + fmt.Println() + } + + return nil +} + +func statusServer(server topology.ZKServer, topo *topology.ZKTopology) error { + cmd := fmt.Sprintf("%s status %s", zkServerScript(topo), zkConfigPath(topo.Path, server.MyID)) + return ssh.Run(server.Host(), cmd) +} diff --git a/internal/zk/stop.go b/internal/zk/stop.go new file mode 100644 index 0000000..2359d73 --- /dev/null +++ b/internal/zk/stop.go @@ -0,0 +1,49 @@ +package zk + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +func Stop(ensembleName string, myID int) error { + _, topo, err := loadEnsemble(ensembleName) + if err != nil { + return err + } + + // Stop a specific server if myID is provided + if myID != 0 { + server, err := pickServer(topo, myID) + if err != nil { + return err + } + + if err := stopServer(*server, topo); err != nil { + return err + } + + fmt.Printf("ZooKeeper server %s (myid=%d) stopped.\n", server.Host(), server.MyID) + return nil + } + + for _, server := range topo.Servers { + if err := stopServer(server, topo); err != nil { + return err + } + } + + fmt.Printf("ZooKeeper ensemble %q stopped.\n", ensembleName) + return nil +} + +func stopServer(server topology.ZKServer, topo *topology.ZKTopology) error { + fmt.Printf("Stopping %s (myid=%d)...\n", server.Host(), server.MyID) + cmd := fmt.Sprintf("%s stop %s", zkServerScript(topo), zkConfigPath(topo.Path, server.MyID)) + if err := ssh.Run(server.Host(), cmd); err != nil { + return fmt.Errorf("stop %s: %w", server.Host(), err) + } + + return nil +}