diff --git a/cmd/zk/delete.go b/cmd/zk/delete.go index e67903c..54e1057 100644 --- a/cmd/zk/delete.go +++ b/cmd/zk/delete.go @@ -10,8 +10,13 @@ var deleteCmd = &cobra.Command{ Short: "Delete a ZooKeeper ensemble", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - if err := zk.Delete(args[0]); err != nil { + purge, _ := cmd.Flags().GetBool("purge") + if err := zk.Delete(args[0], purge); err != nil { panic(err) } }, } + +func init() { + deleteCmd.Flags().Bool("purge", false, "also remove the installation directory on each server") +} diff --git a/internal/cluster/cluster.go b/internal/cluster/cluster.go index 0e93624..8e55953 100644 --- a/internal/cluster/cluster.go +++ b/internal/cluster/cluster.go @@ -33,8 +33,7 @@ func loadCluster(serviceCode string) ( // installed there from same path and version. func sharingCluster( serviceCode string, - topoPath string, - version string, + installPath string, host string, ) (string, error) { registered, err := store.ListCluster() @@ -57,8 +56,8 @@ func sharingCluster( continue } - if topo.Path == topoPath && - meta.Version == version && + otherInstallPath := memcachedInstallPath(topo.Path, meta.Version) + if installPath == otherInstallPath && hasServerOn(topo.Servers, host) { return other, nil } @@ -68,10 +67,10 @@ func sharingCluster( } func hasServerOn( - topoServers []topology.CacheServer, + cacheServers []topology.CacheServer, host string, ) bool { - for _, server := range topoServers { + for _, server := range cacheServers { if server.Host() == host { return true } diff --git a/internal/cluster/delete.go b/internal/cluster/delete.go index 7e7c813..525c0bf 100644 --- a/internal/cluster/delete.go +++ b/internal/cluster/delete.go @@ -55,7 +55,7 @@ func removeInstallationDirs( installPath := memcachedInstallPath(topo.Path, version) for _, host := range distinctHosts(topo.Servers) { - other, err := sharingCluster(serviceCode, topo.Path, version, host) + other, err := sharingCluster(serviceCode, installPath, host) if err != nil { return err } diff --git a/internal/store/store.go b/internal/store/store.go index 6772f37..194d541 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -190,7 +190,7 @@ func ListCluster() ([]string, error) { } func zkBaseDir() string { - return filepath.Join(internal.Config.Home, "clusters", "zk") + return filepath.Join(internal.Config.Home, "clusters", "zookeeper") } func zkDir(ensembleName string) string { diff --git a/internal/topology/zk.go b/internal/topology/zk.go index 2bbc334..53b2668 100644 --- a/internal/topology/zk.go +++ b/internal/topology/zk.go @@ -63,10 +63,6 @@ func (topo *ZKTopology) Validate() error { return fmt.Errorf("duplicate address: %s", s.Address) } seenAddress[s.Address] = true - - if s.Config.DataDir == "" { - return fmt.Errorf("server myid=%d: data_dir is required", s.MyID) - } } return nil diff --git a/internal/zk/config.go b/internal/zk/config.go index e1dccf5..1e3d597 100644 --- a/internal/zk/config.go +++ b/internal/zk/config.go @@ -17,13 +17,15 @@ func buildConfig(server topology.ZKServer, topo *topology.ZKTopology) string { var sb strings.Builder cfg := server.Config - dynamicConfigPath := fmt.Sprintf("%s/conf_myid_%d/zoo.cfg.dynamic", topo.Path, server.MyID) + dataDir := zkNodeDataPath(cfg.DataDir, server.MyID) + dataLogDir := zkNodeDataPath(cfg.DataLogDir, server.MyID) + dynamicConfigPath := zkDynamicConfigPath(topo.Path, topo.Name, server.MyID) fmt.Fprintf(&sb, "tickTime=%d\n", cfg.TickTime) fmt.Fprintf(&sb, "initLimit=%d\n", cfg.InitLimit) fmt.Fprintf(&sb, "syncLimit=%d\n", cfg.SyncLimit) - fmt.Fprintf(&sb, "dataDir=%s/zk%d\n", cfg.DataDir, server.MyID) - fmt.Fprintf(&sb, "dataLogDir=%s/zk%d\n", cfg.DataLogDir, server.MyID) + fmt.Fprintf(&sb, "dataDir=%s\n", dataDir) + fmt.Fprintf(&sb, "dataLogDir=%s\n", dataLogDir) fmt.Fprintf(&sb, "dynamicConfigFile=%s\n", dynamicConfigPath) sb.WriteString("standaloneEnabled=false\n") diff --git a/internal/zk/delete.go b/internal/zk/delete.go index e11f942..b00e406 100644 --- a/internal/zk/delete.go +++ b/internal/zk/delete.go @@ -11,17 +11,19 @@ import ( "github.com/jam2in/arcusctl/internal/topology" ) -func Delete(ensembleName string) error { - _, topo, err := loadEnsemble(ensembleName) +const removeCommandTemplate = "rm -rf %s" + +func Delete(ensembleName string, purge bool) error { + meta, topo, err := loadEnsemble(ensembleName) if err != nil { return err } - if err := verifyTopology(topo.Servers, topo.Path); err != nil { + if err := verifyTopology(topo.Servers, topo.Path, topo.Name); err != nil { return err } - if err := verifyAllStopped(topo.Servers, topo.Path); err != nil { + if err := verifyAllStopped(topo.Servers, topo.Path, topo.Name); err != nil { return err } @@ -34,11 +36,19 @@ func Delete(ensembleName string) error { hostsMap := groupServersByHost(topo.Servers) for host, servers := range hostsMap { fmt.Printf("Removing files on %s...\n", host) - if err := removeHostFiles(host, servers, topo.Path); err != nil { + if err := removeHostFiles(host, servers, topo.Path, topo.Name); err != nil { return fmt.Errorf("remove files on %s: %w", host, err) } } + // If user specified --purge, remove the installation directories on each server. + // If the installation directory is shared with another ensemble, it will not be removed. + if purge { + if err := removeInstallationDirs(ensembleName, topo, meta.Version); err != nil { + return err + } + } + if err := store.DeleteZK(ensembleName); err != nil { return fmt.Errorf("delete metadata: %w", err) } @@ -47,9 +57,14 @@ func Delete(ensembleName string) error { return nil } -func verifyTopology(servers []topology.ZKServer, topoPath string) error { +func verifyTopology( + servers []topology.ZKServer, + topoPath string, + ensembleName string, +) error { for _, server := range servers { - confDir := fmt.Sprintf("%s/conf_myid_%d", topoPath, server.MyID) + confDir := zkConfigDir(topoPath, ensembleName, 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()) } @@ -57,10 +72,14 @@ func verifyTopology(servers []topology.ZKServer, topoPath string) error { return nil } -func verifyAllStopped(servers []topology.ZKServer, topoPath string) error { +func verifyAllStopped( + servers []topology.ZKServer, + topoPath string, + ensembleName string, +) error { for _, server := range servers { - confPath := zkConfigPath(topoPath, server.MyID) - cmd := fmt.Sprintf("pgrep -f '[Q]uorumPeerMain.*%s' > /dev/null 2>&1", confPath) + confDir := zkConfigDir(topoPath, ensembleName, server.MyID) + cmd := fmt.Sprintf("pgrep -f '[Q]uorumPeerMain.*%s' > /dev/null 2>&1", confDir) 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) @@ -77,15 +96,56 @@ func groupServersByHost(servers []topology.ZKServer) map[string][]topology.ZKSer return hosts } -func removeHostFiles(host string, servers []topology.ZKServer, topoPath string) error { - removePaths := []string{topoPath} +func removeHostFiles( + host string, + servers []topology.ZKServer, + topoPath string, + ensembleName string, +) error { + var removePaths []string + for _, server := range servers { + // Remove conf//zk.cfg and data/log directories + removePaths = append( + removePaths, + zkConfigDir(topoPath, ensembleName, server.MyID), + ) removePaths = append(removePaths, nodeDataPaths(server)...) } return ssh.Run(host, "rm -rf "+strings.Join(removePaths, " ")) } +func removeInstallationDirs( + ensembleName string, + topo *topology.ZKTopology, + version string, +) error { + installPath := zkInstallPath(topo.Path, version) + + for host := range groupServersByHost(topo.Servers) { + other, err := sharingEnsemble(ensembleName, installPath, host) + if err != nil { + return err + } + + if other != "" { + fmt.Printf( + "Skip removing directory on %s: install path is shared with ensemble %q\n", + host, other, + ) + continue + } + + fmt.Printf("Removing installation directory on %s...\n", host) + if err := ssh.Run(host, fmt.Sprintf(removeCommandTemplate, installPath)); err != nil { + return fmt.Errorf("remove installation on %s: %w", host, err) + } + } + + return nil +} + func nodeDataPaths(server topology.ZKServer) []string { nodeDirName := fmt.Sprintf("zk%d", server.MyID) diff --git a/internal/zk/deploy.go b/internal/zk/deploy.go index 82e837c..a7043e9 100644 --- a/internal/zk/deploy.go +++ b/internal/zk/deploy.go @@ -63,14 +63,15 @@ func prepareTopology(topoPath string) (*topology.ZKTopology, []byte, error) { func printPlan(topo *topology.ZKTopology, version string) { fmt.Printf("ZooKeeper ensemble %q will be deployed (version: %s)\n\n", topo.Name, version) + installPath := zkInstallPath(topo.Path, version) w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) - fmt.Fprintln(w, "ROLE\tHOST\tPORTS\tDIRECTORIES") + fmt.Fprintln(w, "MYID\tHOST\tPORTS\tDIRECTORIES") fmt.Fprintln(w, "\t\t\t\t") for _, s := range topo.Servers { host, clientPort, quorumPort, electionPort := s.ParseAddress() ports := fmt.Sprintf("%s/%s/%s", clientPort, quorumPort, electionPort) - fmt.Fprintf(w, "zookeeper\t%s\t%s\t%s\n", host, ports, topo.Path) + fmt.Fprintf(w, "%d\t%s\t%s\t%s\n", s.MyID, host, ports, installPath) } w.Flush() @@ -91,5 +92,12 @@ func printRecoveryGuide(deployed []topology.ZKServer, topo *topology.ZKTopology) fmt.Printf(" - %s (myid=%d)\n", s.Host(), s.MyID) } fmt.Println("\nTo clean up, manually run on each server:") - fmt.Printf(" rm -rf %s/conf_myid_\n", topo.Path) + for _, server := range deployed { + confDir := zkConfigDir(topo.Path, topo.Name, server.MyID) + dataDir := zkNodeDataPath(server.Config.DataDir, server.MyID) + dataLogDir := zkNodeDataPath(server.Config.DataLogDir, server.MyID) + + fmt.Printf(" %s:\n", server.Host()) + fmt.Printf(" rm -rf %s %s %s\n", confDir, dataDir, dataLogDir) + } } diff --git a/internal/zk/ensemble.go b/internal/zk/ensemble.go index d64c519..8897a7d 100644 --- a/internal/zk/ensemble.go +++ b/internal/zk/ensemble.go @@ -2,6 +2,7 @@ package zk import ( "fmt" + "path" "github.com/jam2in/arcusctl/internal/store" "github.com/jam2in/arcusctl/internal/topology" @@ -32,8 +33,64 @@ func loadEnsemble(name string) (*store.ZKMeta, *topology.ZKTopology, error) { } func mergeServerConfigs(topo *topology.ZKTopology) { + globalConfig := topo.GlobalConfig + + if topo.GlobalConfig.DataDir == "" { + globalConfig.DataDir = path.Join(topo.Path, "data", topo.Name) + } + for i := range topo.Servers { - merged := mergeConfig(topo.GlobalConfig, topo.Servers[i].Config) + merged := mergeConfig(globalConfig, topo.Servers[i].Config) topo.Servers[i].Config = &merged } } + +// sharingEnsemble returns the name of another ensemble +// that shares the same installation path and has a server on the given host. +func sharingEnsemble( + ensembleName string, + installPath string, + host string, +) (string, error) { + registered, err := store.ListZK() + if err != nil { + return "", err + } + + for _, other := range registered { + if other == ensembleName { + continue + } + + meta, err := store.LoadZKMeta(other) + if err != nil { + continue + } + + topo, err := store.LoadZKTopology(other) + if err != nil { + continue + } + + otherInstallPath := zkInstallPath(topo.Path, meta.Version) + if otherInstallPath == installPath && + hasServerOn(topo.Servers, host) { + return other, nil + } + } + + return "", nil +} + +func hasServerOn( + zkServers []topology.ZKServer, + host string, +) bool { + for _, server := range zkServers { + if server.Host() == host { + return true + } + } + + return false +} diff --git a/internal/zk/install.go b/internal/zk/install.go index a38e3be..f757a8d 100644 --- a/internal/zk/install.go +++ b/internal/zk/install.go @@ -3,13 +3,16 @@ package zk import ( "fmt" "os" - "path/filepath" "github.com/jam2in/arcusctl/internal/ssh" "github.com/jam2in/arcusctl/internal/topology" ) -func installServers(topo *topology.ZKTopology, version string, localTarPath string) ([]topology.ZKServer, error) { +func installServers( + topo *topology.ZKTopology, + version string, + localTarPath string, +) ([]topology.ZKServer, error) { var installed []topology.ZKServer archiveInstalled := map[string]bool{} @@ -33,7 +36,23 @@ func installServers(topo *topology.ZKTopology, version string, localTarPath stri return installed, nil } -func installArchive(host string, installPath string, version string, localTarPath string) error { +func installArchive( + host string, + topoPath string, + version string, + localTarPath string, +) error { + installPath := zkInstallPath(topoPath, version) + scriptPath := zkServerScript(topoPath, version) + + exists, err := ssh.FileExists(host, scriptPath) + if err != nil { + return fmt.Errorf("check installation on %s: %w", host, err) + } + if exists { + return nil + } + if err := ssh.Run(host, fmt.Sprintf("mkdir -p %s", installPath)); err != nil { return fmt.Errorf("mkdir base path on %s: %w", host, err) } @@ -43,7 +62,11 @@ func installArchive(host string, installPath string, version string, localTarPat return fmt.Errorf("copy file to %s: %w", host, err) } - extractCmd := fmt.Sprintf("tar -xzf %s -C %s --strip-components=1", remoteTarPath, installPath) + extractCmd := fmt.Sprintf( + "tar -xzf %s -C %s --strip-components=1", + remoteTarPath, + installPath, + ) if err := ssh.Run(host, extractCmd); err != nil { return fmt.Errorf("extract file on %s: %w", host, err) } @@ -53,16 +76,17 @@ func installArchive(host string, installPath string, version string, localTarPat func configureServer(server topology.ZKServer, topo *topology.ZKTopology) error { host := server.Host() - confDir := fmt.Sprintf("%s/conf_myid_%d", topo.Path, server.MyID) - dataDirPath := fmt.Sprintf("%s/zk%d", server.Config.DataDir, server.MyID) - dataLogDirPath := fmt.Sprintf("%s/zk%d", server.Config.DataLogDir, server.MyID) + + confDir := zkConfigDir(topo.Path, topo.Name, server.MyID) + dataDirPath := zkNodeDataPath(server.Config.DataDir, server.MyID) + dataLogDirPath := zkNodeDataPath(server.Config.DataLogDir, server.MyID) + mkdirCmd := fmt.Sprintf( "mkdir -p %s %s %s", confDir, dataDirPath, dataLogDirPath, ) - if err := ssh.Run(host, mkdirCmd); err != nil { return fmt.Errorf("mkdir on %s: %w", host, err) } @@ -73,12 +97,14 @@ func configureServer(server topology.ZKServer, topo *topology.ZKTopology) error } config := buildConfig(server, topo) - if err := uploadFile(host, config, filepath.Join(confDir, "zoo.cfg")); err != nil { + configPath := zkConfigPath(topo.Path, topo.Name, server.MyID) + if err := uploadFile(host, config, configPath); err != nil { return fmt.Errorf("upload zoo.cfg to %s: %w", host, err) } dynamicConfig := buildDynamicConfig(topo) - if err := uploadFile(host, dynamicConfig, filepath.Join(confDir, "zoo.cfg.dynamic")); err != nil { + dynamicConfigPath := zkDynamicConfigPath(topo.Path, topo.Name, server.MyID) + if err := uploadFile(host, dynamicConfig, dynamicConfigPath); err != nil { return fmt.Errorf("upload zoo.cfg.dynamic to %s: %w", host, err) } diff --git a/internal/zk/server.go b/internal/zk/server.go index cfc4038..3fb375d 100644 --- a/internal/zk/server.go +++ b/internal/zk/server.go @@ -17,11 +17,26 @@ func pickServer(topoServers []topology.ZKServer, myID int) (*topology.ZKServer, return nil, fmt.Errorf("ZooKeeper server myid=%d not found", myID) } -func zkServerScript(topoPath string) string { - return path.Join(topoPath, "bin", "zkServer.sh") +func zkServerScript(topoPath string, version string) string { + return path.Join(zkInstallPath(topoPath, version), "bin", "zkServer.sh") } -func zkConfigPath(topoPath string, myID int) string { - configDir := fmt.Sprintf("conf_myid_%d", myID) - return path.Join(topoPath, configDir, "zoo.cfg") +func zkInstallPath(topoPath string, version string) string { + return path.Join(topoPath, version) +} + +func zkConfigDir(topoPath string, ensembleName string, myID int) string { + return path.Join(topoPath, "conf", ensembleName, fmt.Sprintf("zk%d", myID)) +} + +func zkConfigPath(topoPath string, ensembleName string, myID int) string { + return path.Join(zkConfigDir(topoPath, ensembleName, myID), "zoo.cfg") +} + +func zkDynamicConfigPath(topoPath string, ensembleName string, myID int) string { + return path.Join(zkConfigDir(topoPath, ensembleName, myID), "zoo.cfg.dynamic") +} + +func zkNodeDataPath(dataDir string, myID int) string { + return path.Join(dataDir, fmt.Sprintf("zk%d", myID)) } diff --git a/internal/zk/start.go b/internal/zk/start.go index f8a0102..1ad8bf7 100644 --- a/internal/zk/start.go +++ b/internal/zk/start.go @@ -8,7 +8,7 @@ import ( ) func Start(ensembleName string, myID int) error { - _, topo, err := loadEnsemble(ensembleName) + meta, topo, err := loadEnsemble(ensembleName) if err != nil { return err } @@ -19,7 +19,7 @@ func Start(ensembleName string, myID int) error { if err != nil { return err } - if err := startServer(*server, topo.Path); err != nil { + if err := startServer(*server, topo.Path, topo.Name, meta.Version); err != nil { return err } fmt.Printf("ZooKeeper node %s (myid=%d) started successfully.\n", server.Host(), myID) @@ -27,7 +27,7 @@ func Start(ensembleName string, myID int) error { } for _, server := range topo.Servers { - if err := startServer(server, topo.Path); err != nil { + if err := startServer(server, topo.Path, topo.Name, meta.Version); err != nil { return err } } @@ -36,9 +36,18 @@ func Start(ensembleName string, myID int) error { return nil } -func startServer(server topology.ZKServer, topoPath string) error { +func startServer( + server topology.ZKServer, + topoPath string, + ensembleName string, + version string, +) error { fmt.Printf("Starting %s (myid=%d)...\n", server.Host(), server.MyID) - cmd := fmt.Sprintf("%s start %s", zkServerScript(topoPath), zkConfigPath(topoPath, server.MyID)) + + scriptPath := zkServerScript(topoPath, version) + configPath := zkConfigPath(topoPath, ensembleName, server.MyID) + cmd := fmt.Sprintf("%s start %s", scriptPath, configPath) + if err := ssh.Run(server.Host(), cmd); err != nil { return fmt.Errorf("start %s: %w", server.Host(), err) } diff --git a/internal/zk/status.go b/internal/zk/status.go index a791319..8f1083d 100644 --- a/internal/zk/status.go +++ b/internal/zk/status.go @@ -17,7 +17,7 @@ func Status(ensembleName string) error { for _, server := range topo.Servers { fmt.Printf("=== %s (myid=%d) ===\n", server.Host(), server.MyID) - if err := statusServer(server, topo.Path); err != nil { + if err := statusServer(server, topo.Path, topo.Name, meta.Version); err != nil { fmt.Printf(" error: %v\n", err) } fmt.Println() @@ -26,7 +26,14 @@ func Status(ensembleName string) error { return nil } -func statusServer(server topology.ZKServer, topoPath string) error { - cmd := fmt.Sprintf("%s status %s", zkServerScript(topoPath), zkConfigPath(topoPath, server.MyID)) +func statusServer( + server topology.ZKServer, + topoPath string, + ensembleName string, + version string, +) error { + scriptPath := zkServerScript(topoPath, version) + configPath := zkConfigPath(topoPath, ensembleName, server.MyID) + cmd := fmt.Sprintf("%s status %s", scriptPath, configPath) return ssh.Run(server.Host(), cmd) } diff --git a/internal/zk/stop.go b/internal/zk/stop.go index 80976f8..1e18127 100644 --- a/internal/zk/stop.go +++ b/internal/zk/stop.go @@ -8,7 +8,7 @@ import ( ) func Stop(ensembleName string, myID int) error { - _, topo, err := loadEnsemble(ensembleName) + meta, topo, err := loadEnsemble(ensembleName) if err != nil { return err } @@ -20,7 +20,7 @@ func Stop(ensembleName string, myID int) error { return err } - if err := stopServer(*server, topo.Path); err != nil { + if err := stopServer(*server, topo.Path, topo.Name, meta.Version); err != nil { return err } @@ -29,7 +29,7 @@ func Stop(ensembleName string, myID int) error { } for _, server := range topo.Servers { - if err := stopServer(server, topo.Path); err != nil { + if err := stopServer(server, topo.Path, topo.Name, meta.Version); err != nil { return err } } @@ -38,9 +38,17 @@ func Stop(ensembleName string, myID int) error { return nil } -func stopServer(server topology.ZKServer, topoPath string) error { +func stopServer( + server topology.ZKServer, + topoPath string, + ensembleName string, + version string, +) error { fmt.Printf("Stopping %s (myid=%d)...\n", server.Host(), server.MyID) - cmd := fmt.Sprintf("%s stop %s", zkServerScript(topoPath), zkConfigPath(topoPath, server.MyID)) + + scriptPath := zkServerScript(topoPath, version) + configPath := zkConfigPath(topoPath, ensembleName, server.MyID) + cmd := fmt.Sprintf("%s stop %s", scriptPath, configPath) if err := ssh.Run(server.Host(), cmd); err != nil { return fmt.Errorf("stop %s: %w", server.Host(), err) } diff --git a/zk-sample-topology.yml b/zk-sample-topology.yml index 49c220a..e2f8754 100644 --- a/zk-sample-topology.yml +++ b/zk-sample-topology.yml @@ -26,10 +26,10 @@ global_config: # optional (default: 5) - ticks between request and ack sync_limit: 5 - # required + # optional (default: /data/) data_dir: /var/lib/zk/data - # optional (default: data_dir) - dedicated disk recommended + # optional (default: data_dir) data_log_dir: /var/lib/zk/datalog # optional - other ZK options (string values only)