diff --git a/cluster-sample-topology.yml b/cluster-sample-topology.yml index 0df4387..228760b 100644 --- a/cluster-sample-topology.yml +++ b/cluster-sample-topology.yml @@ -10,7 +10,7 @@ servers: - address: cache2:11211 - address: cache3:11211 config: # optional - per-node override - options: "-l 192.168.1.3" # optional - override global options + options: "-l 192.168.1.3" # optional - override global options (-p, -P, -z, -d are not allowed) global_config: options: "-t 4 -c 1024 -b 1024 -B auto -m 64" # memcached command-line arguments diff --git a/cmd/cluster/delete.go b/cmd/cluster/delete.go index 0a11d2f..766e01b 100644 --- a/cmd/cluster/delete.go +++ b/cmd/cluster/delete.go @@ -1,12 +1,23 @@ package cluster -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/cluster" + "github.com/spf13/cobra" +) var deleteCmd = &cobra.Command{ Use: "delete ", Short: "Delete an Arcus cluster", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: delete 구현 + serviceCode := args[0] + purge, _ := cmd.Flags().GetBool("purge") + if err := cluster.Delete(serviceCode, purge); err != nil { + panic(err) + } }, } + +func init() { + deleteCmd.Flags().Bool("purge", false, "also remove the installation directory on each server") +} diff --git a/cmd/cluster/list.go b/cmd/cluster/list.go index e69e035..36a973d 100644 --- a/cmd/cluster/list.go +++ b/cmd/cluster/list.go @@ -1,12 +1,17 @@ package cluster -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/cluster" + "github.com/spf13/cobra" +) var listCmd = &cobra.Command{ Use: "list", Short: "List managed Arcus clusters", Args: cobra.NoArgs, Run: func(cmd *cobra.Command, args []string) { - // TODO: list 구현 + if err := cluster.List(); err != nil { + panic(err) + } }, } diff --git a/cmd/cluster/start.go b/cmd/cluster/start.go index fa3f5e8..4f43cdb 100644 --- a/cmd/cluster/start.go +++ b/cmd/cluster/start.go @@ -1,16 +1,26 @@ package cluster -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/cluster" + "github.com/spf13/cobra" +) var startCmd = &cobra.Command{ - Use: "start [--node
]", + Use: "start [--node
] [--group ]", Short: "Start an Arcus cluster", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: start 구현 + serviceCode := args[0] + nodeAddress, _ := cmd.Flags().GetString("node") + groupName, _ := cmd.Flags().GetString("group") + if err := cluster.Start(serviceCode, nodeAddress, groupName); err != nil { + panic(err) + } }, } func init() { - startCmd.Flags().String("node", "", "address of the specific node to start") + startCmd.Flags().String("node", "", "address of the specific node to start (community edition only)") + startCmd.Flags().String("group", "", "name of the specific group to start (enterprise edition only)") + startCmd.MarkFlagsMutuallyExclusive("node", "group") } diff --git a/cmd/cluster/status.go b/cmd/cluster/status.go index 61a1cc6..98fbd6a 100644 --- a/cmd/cluster/status.go +++ b/cmd/cluster/status.go @@ -1,12 +1,18 @@ package cluster -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/cluster" + "github.com/spf13/cobra" +) var statusCmd = &cobra.Command{ Use: "status ", Short: "Show status of an Arcus cluster", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: status 구현 + serviceCode := args[0] + if err := cluster.Status(serviceCode); err != nil { + panic(err) + } }, } diff --git a/cmd/cluster/stop.go b/cmd/cluster/stop.go index 372ac12..7df0eb8 100644 --- a/cmd/cluster/stop.go +++ b/cmd/cluster/stop.go @@ -1,16 +1,26 @@ package cluster -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/cluster" + "github.com/spf13/cobra" +) var stopCmd = &cobra.Command{ - Use: "stop [--node
]", + Use: "stop [--node
] [--group ]", Short: "Stop an Arcus cluster", Args: cobra.ExactArgs(1), Run: func(cmd *cobra.Command, args []string) { - // TODO: stop 구현 + serviceCode := args[0] + nodeAddress, _ := cmd.Flags().GetString("node") + groupName, _ := cmd.Flags().GetString("group") + if err := cluster.Stop(serviceCode, nodeAddress, groupName); err != nil { + panic(err) + } }, } func init() { - stopCmd.Flags().String("node", "", "address of the specific node to stop") + stopCmd.Flags().String("node", "", "address of the specific node to stop (community edition only)") + stopCmd.Flags().String("group", "", "name of the specific group to stop (enterprise edition only)") + stopCmd.MarkFlagsMutuallyExclusive("node", "group") } diff --git a/internal/cluster/cluster.go b/internal/cluster/cluster.go new file mode 100644 index 0000000..0e93624 --- /dev/null +++ b/internal/cluster/cluster.go @@ -0,0 +1,80 @@ +package cluster + +import ( + "github.com/jam2in/arcusctl/internal/store" + "github.com/jam2in/arcusctl/internal/topology" +) + +func loadCluster(serviceCode string) ( + *store.ClusterMeta, + *topology.ClusterTopology, + topology.ClusterEdition, + error, +) { + meta, err := store.LoadClusterMeta(serviceCode) + if err != nil { + return nil, nil, "", err + } + + topo, err := store.LoadClusterTopology(serviceCode) + if err != nil { + return nil, nil, "", err + } + + edition, err := topo.Edition() + if err != nil { + return nil, nil, "", err + } + + return meta, topo, edition, nil +} + +// sharingCluster return, for each host, the service code of another cluster +// installed there from same path and version. +func sharingCluster( + serviceCode string, + topoPath string, + version string, + host string, +) (string, error) { + registered, err := store.ListCluster() + if err != nil { + return "", err + } + + for _, other := range registered { + if other == serviceCode { + continue + } + + meta, err := store.LoadClusterMeta(other) + if err != nil { + continue + } + + topo, err := store.LoadClusterTopology(other) + if err != nil { + continue + } + + if topo.Path == topoPath && + meta.Version == version && + hasServerOn(topo.Servers, host) { + return other, nil + } + } + + return "", nil +} + +func hasServerOn( + topoServers []topology.CacheServer, + host string, +) bool { + for _, server := range topoServers { + if server.Host() == host { + return true + } + } + return false +} diff --git a/internal/cluster/delete.go b/internal/cluster/delete.go new file mode 100644 index 0000000..cfa4906 --- /dev/null +++ b/internal/cluster/delete.go @@ -0,0 +1,96 @@ +package cluster + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal" + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/store" + "github.com/jam2in/arcusctl/internal/topology" +) + +const removeCommandTemplate = "rm -rf %s" + +func Delete(serviceCode string, purge bool) error { + meta, topo, edition, err := loadCluster(serviceCode) + if err != nil { + return err + } + + if err := verifyAllStopped(topo, meta.Version); err != nil { + return err + } + + fmt.Printf("This will remove Arcus cluster %q from all servers.\n", serviceCode) + if !internal.Confirm("Are you sure you want to proceed? (y/N): ") { + fmt.Println("Aborted.") + return nil + } + + if err := unregisterZNodes(topo, edition); err != nil { + return err + } + + // If user specified --purge, remove the installation directories on each server. + // If the installation directory is shared with another cluster, it will not be removed. + if purge { + if err := removeInstallationDirs(serviceCode, topo, meta.Version); err != nil { + return err + } + } + + if err := store.DeleteCluster(serviceCode); err != nil { + return fmt.Errorf("delete cluster metadata: %w", err) + } + + fmt.Printf("Arcus cluster %q deleted.\n", serviceCode) + return nil +} + +func removeInstallationDirs( + serviceCode string, + topo *topology.ClusterTopology, + version string, +) error { + installPath := memcachedInstallPath(topo.Path, version) + + for _, host := range distinctHosts(topo.Servers) { + other, err := sharingCluster(serviceCode, topo.Path, version, host) + if err != nil { + return err + } + + if other != "" { + fmt.Printf( + "Skip removing directory on %s: install path is shared with cluster %q\n", + host, other, + ) + continue + } + + fmt.Printf("Removing files on %s...\n", host) + if err := ssh.Run(host, fmt.Sprintf(removeCommandTemplate, installPath)); err != nil { + return fmt.Errorf("remove files on %s: %w", host, err) + } + } + + return nil +} + +func verifyAllStopped( + topo *topology.ClusterTopology, + version string, +) error { + for _, server := range topo.Servers { + pidFile := pidFilePath(server.Address, topo.Path, version) + cmd := fmt.Sprintf("pgrep -f %q > /dev/null 2>&1", pidFile) + if err := ssh.Run(server.Host(), cmd); err == nil { + return fmt.Errorf( + "cache server %q is still running; stop the cluster before delete", + server.Address, + ) + } + } + + return nil +} diff --git a/internal/cluster/install.go b/internal/cluster/install.go index c4d8d93..85382b0 100644 --- a/internal/cluster/install.go +++ b/internal/cluster/install.go @@ -138,7 +138,3 @@ func buildCommand( strings.Join(options, " "), ) } - -func memcachedInstallPath(basePath string, version string) string { - return path.Join(basePath, version) -} diff --git a/internal/cluster/list.go b/internal/cluster/list.go new file mode 100644 index 0000000..80a444d --- /dev/null +++ b/internal/cluster/list.go @@ -0,0 +1,48 @@ +package cluster + +import ( + "fmt" + "os" + "text/tabwriter" + + "github.com/jam2in/arcusctl/internal/store" +) + +func List() error { + serviceCodes, err := store.ListCluster() + if err != nil { + return fmt.Errorf("list Arcus clusters: %w", err) + } + + if len(serviceCodes) == 0 { + fmt.Println("No Arcus cluster found.") + return nil + } + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "SERVICECODE\tVERSION\tEDITION\tNODES\tDEPLOYED_AT") + + for _, serviceCode := range serviceCodes { + meta, err := store.LoadClusterMeta(serviceCode) + if err != nil { + fmt.Fprintf(w, "%s\t\t\t\t%v\n", serviceCode, err) + continue + } + + edition := "" + servers := 0 + if topo, err := store.LoadClusterTopology(serviceCode); err == nil { + servers = len(topo.Servers) + if e, err := topo.Edition(); err == nil { + edition = string(e) + } + } + + fmt.Fprintf(w, "%s\t%s\t%s\t%d\t%s\n", + serviceCode, meta.Version, edition, servers, + meta.DeployedAt.Format("2006-01-02 15:04:05"), + ) + } + + return w.Flush() +} diff --git a/internal/cluster/server.go b/internal/cluster/server.go new file mode 100644 index 0000000..0c60307 --- /dev/null +++ b/internal/cluster/server.go @@ -0,0 +1,126 @@ +package cluster + +import ( + "fmt" + "path" + "strings" + + "github.com/jam2in/arcusctl/internal/topology" +) + +func memcachedInstallPath(basePath string, version string) string { + return path.Join(basePath, version) +} + +func pidFilePath( + serverAddress string, + topoPath string, + version string, +) string { + installPath := memcachedInstallPath(topoPath, version) + return path.Join(installPath, fmt.Sprintf("memcached-%s.pid", listenPort(serverAddress))) +} + +func listenPort(address string) string { + parts := strings.SplitN(address, ":", 2) + if len(parts) < 2 { + return "" + } + return parts[1] +} + +func pickCacheServer(servers []topology.CacheServer, address string) (*topology.CacheServer, error) { + for i := range servers { + if servers[i].Address == address { + return &servers[i], nil + } + } + return nil, fmt.Errorf("cache server %q not found in cluster", address) +} + +func serversInGroup(servers []topology.CacheServer, groupName string) []topology.CacheServer { + var result []topology.CacheServer + for _, s := range servers { + if s.Group != nil && s.Group.Name == groupName { + result = append(result, s) + } + } + return result +} + +func masterFirst(servers []topology.CacheServer) []topology.CacheServer { + return orderByRole(servers, true) +} + +func slaveFirst(servers []topology.CacheServer) []topology.CacheServer { + return orderByRole(servers, false) +} + +func orderByRole(servers []topology.CacheServer, masterFirst bool) []topology.CacheServer { + var master, slave []topology.CacheServer + for _, server := range servers { + if server.IsMaster() { + master = append(master, server) + } else { + slave = append(slave, server) + } + } + + if masterFirst { + return append(master, slave...) + } + return append(slave, master...) +} + +func selectTargets( + topo *topology.ClusterTopology, + edition topology.ClusterEdition, + nodeAddress string, + groupName string, + order func([]topology.CacheServer) []topology.CacheServer, +) ([]topology.CacheServer, error) { + if edition == topology.CommunityEdition { + if groupName != "" { + return nil, fmt.Errorf("--group is not allowed for community cluster") + } + if nodeAddress == "" { + return topo.Servers, nil + } + + server, err := pickCacheServer(topo.Servers, nodeAddress) + if err != nil { + return nil, err + } + return []topology.CacheServer{*server}, nil + } + + // Enterprise edition + if nodeAddress != "" { + return nil, fmt.Errorf("--node is not allowed for enterprise cluster") + } + + servers := topo.Servers + if groupName != "" { + servers = serversInGroup(topo.Servers, groupName) + if len(servers) == 0 { + return nil, fmt.Errorf("group %q not found in cluster", groupName) + } + } + return order(servers), nil +} + +func distinctHosts( + servers []topology.CacheServer, +) []string { + seen := map[string]bool{} + var hosts []string + for _, server := range servers { + host := server.Host() + if _, ok := seen[host]; !ok { + seen[host] = true + hosts = append(hosts, host) + } + } + + return hosts +} diff --git a/internal/cluster/start.go b/internal/cluster/start.go new file mode 100644 index 0000000..b5c114e --- /dev/null +++ b/internal/cluster/start.go @@ -0,0 +1,79 @@ +package cluster + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +const startCommandTemplate = "%s/bin/memcached" + + " -E %s/lib/default_engine.so" + + " -X %s/lib/ascii_scrub.so" + + " -X %s/lib/syslog_logger.so" + + " -p %s -P %s -z %s -d %s" + +func Start(serviceCode string, nodeAddress string, groupName string) error { + meta, topo, edition, err := loadCluster(serviceCode) + if err != nil { + return err + } + + targets, err := selectTargets(topo, edition, nodeAddress, groupName, masterFirst) + if err != nil { + return err + } + + total := len(targets) + for i, server := range targets { + fmt.Printf("[%d/%d] %s: starting...\n", i+1, total, server.Address) + if err := startServer(server, topo, meta.Version); err != nil { + return err + } + } + + fmt.Printf("Arcus cluster %q started.\n", serviceCode) + return nil +} + +func startServer( + server topology.CacheServer, + topo *topology.ClusterTopology, + version string, +) error { + command := startCommand(server, topo, version) + if err := ssh.Run(server.Host(), command); err != nil { + return fmt.Errorf("start %s: %w", server.Address, err) + } + return nil +} + +func startCommand( + server topology.CacheServer, + topo *topology.ClusterTopology, + version string, +) string { + installPath := memcachedInstallPath(topo.Path, version) + port := listenPort(server.Address) + pidFile := pidFilePath(server.Address, topo.Path, version) + options := mergedOptions(topo.GlobalConfig, server.Config) + + return fmt.Sprintf( + startCommandTemplate, + installPath, // bin/memcached + installPath, // -E default_engine.so + installPath, // -X syslog_logger.so + installPath, // -X ascii_scrub.so + port, + pidFile, + topo.ZooKeeper, + options, + ) +} + +func mergedOptions(global topology.CacheConfig, override *topology.CacheConfig) string { + if override != nil && override.Options != "" { + return global.Options + " " + override.Options + } + return global.Options +} diff --git a/internal/cluster/status.go b/internal/cluster/status.go new file mode 100644 index 0000000..b911b7e --- /dev/null +++ b/internal/cluster/status.go @@ -0,0 +1,72 @@ +package cluster + +import ( + "fmt" + "os" + "text/tabwriter" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +func Status(serviceCode string) error { + meta, topo, edition, err := loadCluster(serviceCode) + if err != nil { + return err + } + + registered, err := registeredAddresses(topo, edition) + if err != nil { + return fmt.Errorf("read cache_server_mapping from ZooKeeper: %w", err) + } + + fmt.Printf( + "Arcus cluster %q (edition: %s, version: %s)\n\n", + meta.ServiceCode, edition, meta.Version, + ) + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + + if edition == topology.EnterpriseEdition { + fmt.Fprintln(w, "GROUP\tROLE\tADDRESS\tPROCESS_STATUS\tZK_REGISTERED") + for _, s := range topo.Servers { + fmt.Fprintf( + w, "%s\t%s\t%s\t%s\t%s\n", + s.Group.Name, s.Group.Role, s.Address, + processStatus(s, topo.Path, meta.Version), + yesNo(registered[s.Address]), + ) + } + } else { + fmt.Fprintln(w, "ADDRESS\tPROCESS_STATUS\tZK_REGISTERED") + for _, s := range topo.Servers { + fmt.Fprintf( + w, "%s\t%s\t%s\n", + s.Address, processStatus(s, topo.Path, meta.Version), + yesNo(registered[s.Address]), + ) + } + } + + return w.Flush() +} + +func processStatus( + server topology.CacheServer, + topoPath string, + version string, +) string { + pidFile := pidFilePath(server.Address, topoPath, version) + cmd := fmt.Sprintf("pgrep -f %q > /dev/null 2>&1", pidFile) + if err := ssh.Run(server.Host(), cmd); err == nil { + return "running" + } + return "stopped" +} + +func yesNo(b bool) string { + if b { + return "yes" + } + return "no" +} diff --git a/internal/cluster/stop.go b/internal/cluster/stop.go new file mode 100644 index 0000000..8c1407d --- /dev/null +++ b/internal/cluster/stop.go @@ -0,0 +1,53 @@ +package cluster + +import ( + "fmt" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +const stopCommandTemplate = "[ -f %s ] && kill $(cat %s) 2>/dev/null || true" + +func Stop(serviceCode string, nodeAddress string, groupName string) error { + meta, topo, edition, err := loadCluster(serviceCode) + if err != nil { + return err + } + + targets, err := selectTargets(topo, edition, nodeAddress, groupName, slaveFirst) + if err != nil { + return err + } + + total := len(targets) + for i, server := range targets { + fmt.Printf("[%d/%d] %s: stopping...\n", i+1, total, server.Address) + if err := stopServer(server, topo.Path, meta.Version); err != nil { + return err + } + } + + fmt.Printf("Arcus cluster %q stopped.\n", serviceCode) + return nil +} + +func stopServer( + server topology.CacheServer, + topoPath string, + version string, +) error { + if err := ssh.Run(server.Host(), stopCommand(server.Address, topoPath, version)); err != nil { + return fmt.Errorf("stop %s: %w", server.Address, err) + } + return nil +} + +func stopCommand( + serverAddress string, + topoPath string, + version string, +) string { + pidPath := pidFilePath(serverAddress, topoPath, version) + return fmt.Sprintf(stopCommandTemplate, pidPath, pidPath) +} diff --git a/internal/cluster/znode.go b/internal/cluster/znode.go index ad59ae6..89df50b 100644 --- a/internal/cluster/znode.go +++ b/internal/cluster/znode.go @@ -3,7 +3,9 @@ package cluster import ( "fmt" "path" + "strings" + "github.com/go-zookeeper/zk" "github.com/jam2in/arcusctl/internal" "github.com/jam2in/arcusctl/internal/topology" ) @@ -59,7 +61,7 @@ func registerZNodes( // cache server mapping path for _, server := range topo.Servers { - leaf := mappingNode(topo, server, edition) + leaf := mappingNodeName(topo, server, edition) zpath := path.Join(root, internal.ZPATH_CACHE_SERVER_MAPPING, server.Address, leaf) if err := internal.EnsureZNode(conn, zpath); err != nil { return fmt.Errorf("create znode %s: %w", zpath, err) @@ -79,7 +81,48 @@ func registerZNodes( return nil } -func mappingNode( +func unregisterZNodes( + topo *topology.ClusterTopology, + edition topology.ClusterEdition, +) error { + conn, err := internal.ConnectZooKeeper(topo.ZooKeeper) + if err != nil { + return err + } + defer conn.Close() + + root := rootForEdition(edition) + + // cache_server_mapping + for _, server := range topo.Servers { + zpath := path.Join(root, internal.ZPATH_CACHE_SERVER_MAPPING, server.Address) + if err := internal.DeleteZNode(conn, zpath); err != nil { + return fmt.Errorf("delete znode %s: %w", zpath, err) + } + } + + // cache_list, client_list + for _, base := range []string{internal.ZPATH_CACHE_LIST, internal.ZPATH_CLIENT_LIST} { + zpath := path.Join(root, base, topo.ServiceCode) + if err := internal.DeleteZNode(conn, zpath); err != nil { + return fmt.Errorf("delete znode %s: %w", zpath, err) + } + } + + // enterprise: group_list + if edition == topology.EnterpriseEdition { + for _, server := range topo.Servers { + zpath := path.Join(root, internal.ZPATH_GROUP_LIST, topo.ServiceCode, server.Group.Name) + if err := internal.DeleteZNode(conn, zpath); err != nil { + return fmt.Errorf("delete znode %s: %w", zpath, err) + } + } + } + + return nil +} + +func mappingNodeName( topo *topology.ClusterTopology, server topology.CacheServer, edition topology.ClusterEdition, @@ -90,3 +133,52 @@ func mappingNode( return fmt.Sprintf("%s^%s^%s:%d", topo.ServiceCode, server.Group.Name, server.Host(), server.Group.Port) } + +func registeredAddresses( + topo *topology.ClusterTopology, + edition topology.ClusterEdition, +) (map[string]bool, error) { + conn, err := internal.ConnectZooKeeper(topo.ZooKeeper) + if err != nil { + return nil, err + } + defer conn.Close() + + root := rootForEdition(edition) + + registered := map[string]bool{} + for _, server := range topo.Servers { + zpath := path.Join(root, internal.ZPATH_CACHE_SERVER_MAPPING, server.Address) + + children, _, err := conn.Children(zpath) + if err == zk.ErrNoNode { + continue + } + if err != nil { + return nil, err + } + + for _, child := range children { + if mappingBelongsTo(child, topo.ServiceCode, edition) { + registered[server.Address] = true + break + } + } + } + + return registered, nil +} + +func mappingBelongsTo( + leaf string, + serviceCode string, + edition topology.ClusterEdition, +) bool { + if edition == topology.CommunityEdition { + return leaf == serviceCode + } + + // enterprise format: ^^: + prefix, _, found := strings.Cut(leaf, "^") + return found && prefix == serviceCode +} diff --git a/internal/store/store.go b/internal/store/store.go index 7507dcf..6772f37 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -131,6 +131,10 @@ func SaveCluster(serviceCode string, version string, topologyData []byte) error return os.WriteFile(filepath.Join(dir, topologyYML), topologyData, 0644) } +func DeleteCluster(serviceCode string) error { + return os.RemoveAll(clusterDir(serviceCode)) +} + func LoadClusterMeta(serviceCode string) (*ClusterMeta, error) { data, err := os.ReadFile(filepath.Join(clusterDir(serviceCode), metaYML)) if err != nil { @@ -145,6 +149,20 @@ func LoadClusterMeta(serviceCode string) (*ClusterMeta, error) { return &meta, nil } +func LoadClusterTopology(serviceCode string) (*topology.ClusterTopology, error) { + data, err := os.ReadFile(filepath.Join(clusterDir(serviceCode), topologyYML)) + if err != nil { + return nil, err + } + + var topo topology.ClusterTopology + if err := yaml.Unmarshal(data, &topo); err != nil { + return nil, err + } + + return &topo, nil +} + func ClusterExists(serviceCode string) bool { _, err := os.Stat(filepath.Join(clusterDir(serviceCode), metaYML)) return err == nil diff --git a/internal/topology/cluster.go b/internal/topology/cluster.go index 0562f95..ec38544 100644 --- a/internal/topology/cluster.go +++ b/internal/topology/cluster.go @@ -63,10 +63,6 @@ func (topo *ClusterTopology) Edition() (ClusterEdition, error) { } } -func (s *CacheServer) Host() string { - return strings.SplitN(s.Address, ":", 2)[0] -} - func (topo *ClusterTopology) Validate() error { if strings.TrimSpace(topo.ServiceCode) == "" { return fmt.Errorf("cluster servicecode is required") @@ -148,3 +144,11 @@ func (topo *ClusterTopology) validateGroups() error { return nil } + +func (s *CacheServer) IsMaster() bool { + return s.Group != nil && s.Group.Role == "master" +} + +func (s *CacheServer) Host() string { + return strings.SplitN(s.Address, ":", 2)[0] +} diff --git a/internal/util.go b/internal/util.go index 8a51ac1..8b42275 100644 --- a/internal/util.go +++ b/internal/util.go @@ -133,3 +133,25 @@ func Confirm(prompt string) bool { input = strings.TrimSpace(strings.ToLower(input)) return input == "y" || input == "yes" } + +func DeleteZNode(conn *zk.Conn, zpath string) error { + children, _, err := conn.Children(zpath) + if err == zk.ErrNoNode { + return nil + } + if err != nil { + return err + } + + for _, child := range children { + if err := DeleteZNode(conn, zpath+"/"+child); err != nil { + return err + } + } + + err = conn.Delete(zpath, -1) + if err == zk.ErrNoNode { + return nil + } + return err +}