diff --git a/cluster-sample-topology.yml b/cluster-sample-topology.yml new file mode 100644 index 0000000..0df4387 --- /dev/null +++ b/cluster-sample-topology.yml @@ -0,0 +1,26 @@ +# For more configuration options, see: +# https://github.com/naver/arcus-memcached/blob/master/docs/administration/commandline_args.md + +servicecode: my-cluster # required +path: /home/arcus/arcus-memcached # required +zookeeper: zk1:2181,zk2:2181,zk3:2181 # required (comma-separated ensemble) + +servers: + - address: cache1:11211 # required (host:port) + - address: cache2:11211 + - address: cache3:11211 + config: # optional - per-node override + options: "-l 192.168.1.3" # optional - override global options + +global_config: + options: "-t 4 -c 1024 -b 1024 -B auto -m 64" # memcached command-line arguments + +# Enterprise edition: every server must have a `group` block. +# Each group must have exactly one master and at most one slave. +# Community and enterprise servers cannot be mixed in the same cluster. +# +# servers: +# - address: cache1:11211 +# group: { name: g1, role: master, port: 33533 } +# - address: cache2:11211 +# group: { name: g1, role: slave, port: 33533 } \ No newline at end of file diff --git a/cmd/cluster/deploy.go b/cmd/cluster/deploy.go index b13c721..1db5bf3 100644 --- a/cmd/cluster/deploy.go +++ b/cmd/cluster/deploy.go @@ -1,12 +1,20 @@ package cluster -import "github.com/spf13/cobra" +import ( + "github.com/jam2in/arcusctl/internal/cluster" + "github.com/spf13/cobra" +) var deployCmd = &cobra.Command{ Use: "deploy ", Short: "Deploy a new Arcus cluster", Args: cobra.ExactArgs(2), Run: func(cmd *cobra.Command, args []string) { - // TODO: deploy 구현 + version := args[0] + topologyPath := args[1] + + if err := cluster.Deploy(version, topologyPath); err != nil { + panic(err) + } }, } diff --git a/internal/cluster/deploy.go b/internal/cluster/deploy.go new file mode 100644 index 0000000..48e979b --- /dev/null +++ b/internal/cluster/deploy.go @@ -0,0 +1,150 @@ +package cluster + +import ( + "fmt" + "os" + "text/tabwriter" + + "github.com/jam2in/arcusctl/internal" + "github.com/jam2in/arcusctl/internal/store" + "github.com/jam2in/arcusctl/internal/topology" +) + +func Deploy(version string, topoPath string) error { + topo, topologyBytes, edition, err := prepareTopology(topoPath) + if err != nil { + return err + } + + printPlan(topo, version, edition) + if !internal.Confirm("Proceed with deployment? (y/N): ") { + fmt.Println("Aborted.") + return nil + } + + localTarPath, err := ensureDownloaded(version, edition) + if err != nil { + return err + } + + installed, err := installServers(topo, version, localTarPath, edition) + if err != nil { + printRecoveryGuide(topo, installed, version, edition) + return err + } + + if err := registerZNodes(topo, edition); err != nil { + printRecoveryGuide(topo, installed, version, edition) + return fmt.Errorf("register ZNodes in ZooKeeper: %w", err) + } + + if err := store.SaveCluster(topo.ServiceCode, version, topologyBytes); err != nil { + printRecoveryGuide(topo, installed, version, edition) + return fmt.Errorf("save metadata: %w", err) + } + + fmt.Printf( + "Arcus cluster %q deployed. Run 'cluster start %s' to launch.\n", + topo.ServiceCode, topo.ServiceCode, + ) + return nil +} + +func prepareTopology(topoPath string) (*topology.ClusterTopology, []byte, topology.ClusterEdition, error) { + topo, rawData, err := topology.LoadCluster(topoPath) + if err != nil { + return nil, nil, "", err + } + + if err := topo.Validate(); err != nil { + return nil, nil, "", err + } + + if store.ClusterExists(topo.ServiceCode) { + return nil, nil, "", fmt.Errorf( + "arcus cluster %q already exists", + topo.ServiceCode, + ) + } + + edition, err := topo.Edition() + if err != nil { + return nil, nil, "", err + } + + exists, err := clusterExistsInZK(topo.ZooKeeper, topo.ServiceCode, edition) + if err != nil { + return nil, nil, "", err + } + + if exists { + return nil, nil, "", fmt.Errorf( + "arcus cluster %q already exists in ZooKeeper", + topo.ServiceCode, + ) + } + + return topo, rawData, edition, nil +} + +func printPlan( + topo *topology.ClusterTopology, + version string, + edition topology.ClusterEdition, +) { + fmt.Printf( + "Arcus cluster %q will be deployed (edition: %s, version: %s)\n\n", + topo.ServiceCode, edition, version, + ) + + installPath := memcachedInstallPath(topo.Path, version) + + writer := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) + + if edition == topology.EnterpriseEdition { + fmt.Fprintln(writer, "GROUP\tROLE\tADDRESS\tDIRECTORY") + fmt.Fprintln(writer, "\t\t\t") + for _, server := range topo.Servers { + fmt.Fprintf(writer, "%s\t%s\t%s\t%s\n", + server.Group.Name, server.Group.Role, server.Address, installPath) + } + } else { + fmt.Fprintln(writer, "ADDRESS\tDIRECTORY") + fmt.Fprintln(writer, "\t") + for _, server := range topo.Servers { + fmt.Fprintf(writer, "%s\t%s\n", + server.Address, installPath) + } + } + + writer.Flush() + + fmt.Println() + fmt.Println("Attention:") + fmt.Println(" 1. If the topology is not what you expected, check your yaml file.") + fmt.Println(" 2. Please confirm there is no port/directory conflicts in same host.") +} + +func printRecoveryGuide( + topo *topology.ClusterTopology, + installed []topology.CacheServer, + version string, + edition topology.ClusterEdition, +) { + fmt.Println("\nDeployment failed. Manual recovery required.") + + if len(installed) > 0 { + fmt.Println("The following servers have been partially installed.") + for _, s := range installed { + fmt.Printf(" - %s\n", s.Host()) + } + cleanupPath := memcachedInstallPath(topo.Path, version) + fmt.Println("\nTo clean up, manually run on each server:") + fmt.Printf(" $ rm -rf %s\n", cleanupPath) + } + + root := rootForEdition(edition) + fmt.Println("\nZooKeeper nodes may have been created under:") + fmt.Printf(" %s/{cache_list, client_list, cache_server_mapping}/%s ...\n", root, topo.ServiceCode) + fmt.Println("They are left in place. To retry from a clean state, remove them.") +} diff --git a/internal/cluster/download.go b/internal/cluster/download.go new file mode 100644 index 0000000..33f3a67 --- /dev/null +++ b/internal/cluster/download.go @@ -0,0 +1,55 @@ +package cluster + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" + + "github.com/jam2in/arcusctl/internal" + "github.com/jam2in/arcusctl/internal/topology" +) + +const arcusDownloadURLTemplate = "https://github.com/naver/arcus-memcached/releases/download/%s/arcus-memcached-%s.tar.gz" + +func ensureDownloaded(version string, edition topology.ClusterEdition) (string, error) { + dir := imageDir(edition) + + if err := os.MkdirAll(dir, 0755); err != nil { + return "", fmt.Errorf("create image dir: %w", err) + } + + filename := fmt.Sprintf("arcus-memcached-%s.tar.gz", version) + localTarPath := filepath.Join(dir, filename) + + // check if the file already exists + // enterprise edition may not be downloadable, so we check for existence first + if _, err := os.Stat(localTarPath); err == nil { + fmt.Printf("Using existing file: %s\n", localTarPath) + return localTarPath, nil + } + + if edition == topology.EnterpriseEdition { + return "", fmt.Errorf("enterprise archive not found: place %s in %s", filename, dir) + } + + url := fmt.Sprintf(arcusDownloadURLTemplate, version, version) + fmt.Printf("Downloading %s...\n", url) + + cmd := exec.Command("wget", "-q", "-O", localTarPath, url) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Run(); err != nil { + _ = os.Remove(localTarPath) + return "", fmt.Errorf("download %q failed: %w", version, err) + } + + return localTarPath, nil +} + +func imageDir(edition topology.ClusterEdition) string { + if edition == topology.EnterpriseEdition { + return filepath.Join(internal.Config.Home, "images", "arcus-enterprise") + } + return filepath.Join(internal.Config.Home, "images", "arcus-community") +} diff --git a/internal/cluster/install.go b/internal/cluster/install.go new file mode 100644 index 0000000..c4d8d93 --- /dev/null +++ b/internal/cluster/install.go @@ -0,0 +1,144 @@ +package cluster + +import ( + "fmt" + "path" + "strings" + + "github.com/jam2in/arcusctl/internal/ssh" + "github.com/jam2in/arcusctl/internal/topology" +) + +func installServers( + topo *topology.ClusterTopology, + version string, + localTarPath string, + edition topology.ClusterEdition, +) ([]topology.CacheServer, error) { + var installed []topology.CacheServer + handledHosts := make(map[string]struct{}, len(topo.Servers)) + + installPath := memcachedInstallPath(topo.Path, version) + binaryPath := path.Join(installPath, "bin", "memcached") + + for i, server := range topo.Servers { + host := server.Host() + + if _, ok := handledHosts[host]; ok { + fmt.Printf("[%d/%d] %s: installation already handled, skipping\n", + i+1, len(topo.Servers), host) + continue + } + + exists, err := ssh.FileExists(host, binaryPath) + if err != nil { + return installed, fmt.Errorf("check installation on %s: %w", host, err) + } + + if exists { + fmt.Printf("[%d/%d] %s: already installed, skipping\n", + i+1, len(topo.Servers), host) + handledHosts[host] = struct{}{} + continue + } + + fmt.Printf("[%d/%d] %s: installing...\n", i+1, len(topo.Servers), host) + installed = append(installed, server) + + if err := installArchive(host, topo.Path, version, localTarPath, edition); err != nil { + return installed, fmt.Errorf("install %s: %w", host, err) + } + + handledHosts[host] = struct{}{} + } + + return installed, nil +} + +func installArchive( + host string, + basePath string, + version string, + localTarPath string, + edition topology.ClusterEdition, +) error { + installPath := memcachedInstallPath(basePath, version) + sourcePath := path.Join(installPath, "src") + + mkdirCmd := fmt.Sprintf("mkdir -p %s %s", installPath, sourcePath) + if err := ssh.Run(host, mkdirCmd); err != nil { + return fmt.Errorf("mkdir installation paths on %s: %w", host, err) + } + + remoteTarPath := path.Join( + installPath, + fmt.Sprintf("arcus-memcached-%s.tar.gz", version), + ) + if err := ssh.Copy(localTarPath, host, remoteTarPath); err != nil { + return fmt.Errorf("copy file to %s: %w", host, err) + } + + extractCmd := fmt.Sprintf( + "tar -xzf %s -C %s --strip-components=1", + remoteTarPath, + sourcePath, + ) + if err := ssh.Run(host, extractCmd); err != nil { + return fmt.Errorf("extract file on %s: %w", host, err) + } + + depsScriptPath := path.Join(sourcePath, "deps", "install.sh") + + // arcus-memcached's deps/install.sh builds cyrus-sasl, which on macOS defaults + // to building the macOS SASL2 Framework and fails. On Darwin targets only, + // inject --disable-macos-framework into the cyrus-sasl configure line. + patchCmd := fmt.Sprintf( + `if [ "$(uname -s)" = "Darwin" ]; then `+ + `sed -i '' 's|$cyrussasl $prefix|$cyrussasl "$prefix --disable-macos-framework"|' %s; `+ + `fi`, + depsScriptPath, + ) + if err := ssh.Run(host, patchCmd); err != nil { + return fmt.Errorf("patch deps script on %s: %w", host, err) + } + + depsCmd := fmt.Sprintf("%s %s", depsScriptPath, installPath) + if err := ssh.Run(host, depsCmd); err != nil { + return fmt.Errorf("install dependencies on %s: %w", host, err) + } + + buildCmd := buildCommand(sourcePath, installPath, edition) + if err := ssh.Run(host, buildCmd); err != nil { + return fmt.Errorf("build on %s: %w", host, err) + } + + return nil +} + +func buildCommand( + sourcePath string, + installPath string, + edition topology.ClusterEdition, +) string { + options := []string{ + fmt.Sprintf("--prefix=%s", installPath), + "--enable-zk-integration", + "--with-zk-reconfig", + fmt.Sprintf("--with-libevent=%s", installPath), + fmt.Sprintf("--with-zookeeper=%s", installPath), + } + + if edition == topology.EnterpriseEdition { + options = append(options, "--enable-replication") + } + + return fmt.Sprintf( + "cd %s && ./configure %s && make && make install", + sourcePath, + strings.Join(options, " "), + ) +} + +func memcachedInstallPath(basePath string, version string) string { + return path.Join(basePath, version) +} diff --git a/internal/cluster/znode.go b/internal/cluster/znode.go new file mode 100644 index 0000000..ad59ae6 --- /dev/null +++ b/internal/cluster/znode.go @@ -0,0 +1,92 @@ +package cluster + +import ( + "fmt" + "path" + + "github.com/jam2in/arcusctl/internal" + "github.com/jam2in/arcusctl/internal/topology" +) + +func clusterExistsInZK(zkAddress string, serviceCode string, edition topology.ClusterEdition) (bool, error) { + conn, err := internal.ConnectZooKeeper(zkAddress) + if err != nil { + return false, fmt.Errorf("connect to ZooKeeper: %w", err) + } + defer conn.Close() + + serviceCodePath := path.Join(rootForEdition(edition), internal.ZPATH_CACHE_LIST, serviceCode) + exists, _, err := conn.Exists(serviceCodePath) + if err != nil { + return false, fmt.Errorf("check service code %q in ZooKeeper: %w", serviceCode, err) + } + + return exists, nil +} + +func rootForEdition(edition topology.ClusterEdition) string { + switch edition { + case topology.CommunityEdition: + return internal.ZPATH_ROOT + + case topology.EnterpriseEdition: + return internal.ZPATH_REPL_ROOT + + default: + panic(fmt.Sprintf("unsupported cluster edition: %q", edition)) + } +} + +func registerZNodes( + topo *topology.ClusterTopology, + edition topology.ClusterEdition, +) error { + conn, err := internal.ConnectZooKeeper(topo.ZooKeeper) + if err != nil { + return err + } + defer conn.Close() + + root := rootForEdition(edition) + + // service code based path for cache_list and client_list + for _, base := range []string{internal.ZPATH_CACHE_LIST, internal.ZPATH_CLIENT_LIST} { + zpath := path.Join(root, base, topo.ServiceCode) + if err := internal.EnsureZNode(conn, zpath); err != nil { + return fmt.Errorf("create znode %s: %w", zpath, err) + } + } + + // cache server mapping path + for _, server := range topo.Servers { + leaf := mappingNode(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) + } + } + + // enterprise: group list path + 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.EnsureZNode(conn, zpath); err != nil { + return fmt.Errorf("create znode %s: %w", zpath, err) + } + } + } + + return nil +} + +func mappingNode( + topo *topology.ClusterTopology, + server topology.CacheServer, + edition topology.ClusterEdition, +) string { + if edition == topology.CommunityEdition { + return topo.ServiceCode + } + return fmt.Sprintf("%s^%s^%s:%d", + topo.ServiceCode, server.Group.Name, server.Host(), server.Group.Port) +} diff --git a/internal/const.go b/internal/const.go index dc7a825..e732ea7 100644 --- a/internal/const.go +++ b/internal/const.go @@ -1,9 +1,12 @@ package internal const ( - ZPATH_ACL_ROOT = "/arcus_acl" - ZPATH_ACL_MAPPING_ROOT = "/arcus_acl_mapping" - ZPATH_ROOT = "/arcus" - ZPATH_REPL_ROOT = "/arcus_repl" - ZPATH_CLIENT_LIST = "/client_list" + ZPATH_ACL_ROOT = "/arcus_acl" + ZPATH_ACL_MAPPING_ROOT = "/arcus_acl_mapping" + ZPATH_ROOT = "/arcus" + ZPATH_REPL_ROOT = "/arcus_repl" + ZPATH_CACHE_LIST = "/cache_list" + ZPATH_CLIENT_LIST = "/client_list" + ZPATH_CACHE_SERVER_MAPPING = "/cache_server_mapping" + ZPATH_GROUP_LIST = "/group_list" ) diff --git a/internal/ssh/ssh.go b/internal/ssh/ssh.go index 8e1051d..4e638cf 100644 --- a/internal/ssh/ssh.go +++ b/internal/ssh/ssh.go @@ -1,6 +1,7 @@ package ssh import ( + "errors" "fmt" "os" "os/exec" @@ -91,3 +92,29 @@ func Copy(localPath string, host string, remotePath string) error { cmd.Stderr = os.Stderr return cmd.Run() } + +func FileExists(host string, remotePath string) (bool, error) { + client, err := newClient(host) + if err != nil { + return false, err + } + defer client.Close() + + session, err := client.NewSession() + if err != nil { + return false, err + } + defer session.Close() + + err = session.Run(fmt.Sprintf("test -f %s", remotePath)) + if err == nil { + return true, nil + } + + var exitErr *gossh.ExitError + if errors.As(err, &exitErr) && exitErr.ExitStatus() == 1 { + return false, nil + } + + return false, err +} diff --git a/internal/topology/cluster.go b/internal/topology/cluster.go index 1186829..0562f95 100644 --- a/internal/topology/cluster.go +++ b/internal/topology/cluster.go @@ -1,6 +1,17 @@ package topology -import "strings" +import ( + "errors" + "fmt" + "strings" +) + +type ClusterEdition string + +const ( + CommunityEdition ClusterEdition = "community" + EnterpriseEdition ClusterEdition = "enterprise" +) type ClusterTopology struct { ServiceCode string `yaml:"servicecode"` @@ -19,29 +30,121 @@ type CacheServer struct { type GroupInfo struct { Name string `yaml:"name"` Role string `yaml:"role"` + Port int `yaml:"port"` } type CacheConfig struct { - Threads int `yaml:"threads,omitempty"` - MaxConnections int `yaml:"max_connections,omitempty"` - Listen string `yaml:"listen,omitempty"` - Engine *EngineConfig `yaml:"engine,omitempty"` + Options string `yaml:"options,omitempty"` } -type EngineConfig struct { - MemoryLimit int `yaml:"memory_limit,omitempty"` - Eviction bool `yaml:"eviction,omitempty"` -} +func (topo *ClusterTopology) Edition() (ClusterEdition, error) { + groupCount := 0 -func (t *ClusterTopology) IsEnterprise() bool { - for _, s := range t.Servers { - if s.Group != nil { - return true + for _, server := range topo.Servers { + if server.Group != nil { + groupCount++ } } - return false + + switch { + case groupCount == 0: + return CommunityEdition, nil + + case groupCount == len(topo.Servers): + return EnterpriseEdition, nil + + default: + return "", fmt.Errorf( + "%w: grouped=%d, ungrouped=%d", + errors.New("community and enterprise servers cannot be mixed"), + groupCount, + len(topo.Servers)-groupCount, + ) + } } 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") + } + + if strings.TrimSpace(topo.Path) == "" { + return fmt.Errorf("cluster installation path is required") + } + + if strings.TrimSpace(topo.ZooKeeper) == "" { + return fmt.Errorf("ZooKeeper address is required") + } + + if len(topo.Servers) == 0 { + return fmt.Errorf("no servers defined in topology") + } + + seenAddress := make(map[string]struct{}, len(topo.Servers)) + for _, s := range topo.Servers { + if strings.TrimSpace(s.Address) == "" { + return fmt.Errorf("server address is required") + } + + if _, ok := seenAddress[s.Address]; ok { + return fmt.Errorf("duplicate address: %s", s.Address) + } + seenAddress[s.Address] = struct{}{} + } + + edition, err := topo.Edition() + if err != nil { + return err + } + + if edition == EnterpriseEdition { + if err := topo.validateGroups(); err != nil { + return err + } + } + + return nil +} + +func (topo *ClusterTopology) validateGroups() error { + type groupCount struct{ masters, slaves int } + groups := map[string]*groupCount{} + + for _, server := range topo.Servers { + if server.Group.Port <= 0 || server.Group.Port > 65535 { + return fmt.Errorf("server %s: invalid group port %d (must be between 1 and 65535)", + server.Address, server.Group.Port) + } + + group := groups[server.Group.Name] + if group == nil { + group = &groupCount{} + groups[server.Group.Name] = group + } + + switch server.Group.Role { + case "master": + group.masters++ + case "slave": + group.slaves++ + default: + return fmt.Errorf("server %s: invalid group role %q (must be master or slave)", + server.Address, server.Group.Role) + } + } + + for name, count := range groups { + if count.masters != 1 { + return fmt.Errorf("group %q: must have exactly 1 master, got %d", name, count.masters) + } + if count.slaves > 1 { + return fmt.Errorf("group %q: must have at most 1 slave, got %d", name, count.slaves) + } + } + + return nil +} diff --git a/internal/util.go b/internal/util.go index 8b7d446..8a51ac1 100644 --- a/internal/util.go +++ b/internal/util.go @@ -20,6 +20,34 @@ func ConnectZooKeeper(addr string) (*zk.Conn, error) { return conn, err } +// FXIME: EnsureZPath와 통합 가능 여부 검토 필요 +func EnsureZNode(conn *zk.Conn, zpath string) error { + if zpath == "" || zpath == "/" { + return nil + } + + exists, _, err := conn.Exists(zpath) + if err != nil { + return err + } + if exists { + return nil + } + + parent := zpath[:strings.LastIndex(zpath, "/")] + if parent != "" { + if err := EnsureZNode(conn, parent); err != nil { + return err + } + } + + _, err = conn.Create(zpath, []byte{}, 0, zk.WorldACL(zk.PermAll)) + if err == zk.ErrNodeExists { + return nil + } + return err +} + func EnsureZPath(conn *zk.Conn, zpath string) { if zpath == "" || zpath == "/" { return