services/boot/consul.go

   1 package main
   2
   3 import (
   4         "context"
   5         "fmt"
   6         "os"
   7         "os/exec"
   8         "path"
   9         "strings"
  10         "sync"
  11
  12         "github.com/hashicorp/consul/api"
  13 )
  14
  15 var consul = &consulBooter{}
  16
  17 type consulBooter struct {
  18         sync.Mutex
  19 }
  20
  21 func (cb *consulBooter) Boot(ctx context.Context) error {
  22         cb.Lock()
  23         defer cb.Unlock()
  24
  25         if cb.check(ctx) == nil {
  26                 return nil
  27         }
  28         cfg := cfg(ctx)
  29         bin := cfg.UsrDir + "/bin/consul"
  30         err := (&download{
  31                 URL:  "https://releases.hashicorp.com/consul/0.7.2/consul_0.7.2_linux_amd64.zip",
  32                 Dest: bin,
  33                 Size: 29079005,
  34                 Mode: 0755,
  35         }).Boot(ctx)
  36         if err != nil {
  37                 return err
  38         }
  39         dataDir := cfg.DataDir + "/consul"
  40         if err := os.MkdirAll(dataDir, 0700); err != nil {
  41                 return err
  42         }
  43         args := []string{"agent"}
  44         {
  45                 cf := path.Join(cfg.DataDir, "consul-encrypt.json")
  46                 _, err := os.Stat(cf)
  47                 if os.IsNotExist(err) {
  48                         key, err := exec.Command(bin, "keygen").CombinedOutput()
  49                         if err != nil {
  50                                 return err
  51                         }
  52                         err = atomicWriteJSON(cf, map[string]interface{}{
  53                                 "encrypt": strings.TrimSpace(string(key)),
  54                         }, 0400)
  55                 }
  56                 if err != nil {
  57                         return err
  58                 }
  59                 args = append(args, "-config-file="+cf)
  60         }
  61         {
  62                 cf := path.Join(cfg.DataDir, "consul-ports.json")
  63                 err = atomicWriteJSON(cf, map[string]interface{}{
  64                         "client_addr":      "0.0.0.0",
  65                         "bootstrap_expect": len(cfg.ControlHosts),
  66                         "data_dir":         dataDir,
  67                         "datacenter":       cfg.SiteID,
  68                         "server":           true,
  69                         "ui":               true,
  70                         "ports": map[string]int{
  71                                 "dns":      cfg.Ports.ConsulDNS,
  72                                 "http":     cfg.Ports.ConsulHTTP,
  73                                 "https":    cfg.Ports.ConsulHTTPS,
  74                                 "rpc":      cfg.Ports.ConsulRPC,
  75                                 "serf_lan": cfg.Ports.ConsulSerfLAN,
  76                                 "serf_wan": cfg.Ports.ConsulSerfWAN,
  77                                 "server":   cfg.Ports.ConsulServer,
  78                         },
  79                 }, 0644)
  80                 if err != nil {
  81                         return err
  82                 }
  83                 args = append(args, "-config-file="+cf)
  84         }
  85         supervisor := newSupervisor(ctx, "arvados-consul", bin, args...)
  86         running, err := supervisor.Running(ctx)
  87         if err != nil {
  88                 return err
  89         }
  90         if !running {
  91                 defer feedbackf(ctx, "starting consul service")()
  92                 err = supervisor.Start(ctx)
  93                 if err != nil {
  94                         return fmt.Errorf("starting consul: %s", err)
  95                 }
  96                 if len(cfg.ControlHosts) > 1 {
  97                         cmd := exec.Command(bin, append([]string{"join"}, cfg.ControlHosts...)...)
  98                         cmd.Stdout = os.Stderr
  99                         cmd.Stderr = os.Stderr
 100                         err := cmd.Run()
 101                         if err != nil {
 102                                 return fmt.Errorf("consul join: %s", err)
 103                         }
 104                 }
 105         }
 106         return cb.check(ctx)
 107 }
 108
 109 var consulCfg = api.DefaultConfig()
 110
 111 func (cb *consulBooter) check(ctx context.Context) error {
 112         cfg := cfg(ctx)
 113         consulCfg.Address = fmt.Sprintf("127.0.0.1:%d", cfg.Ports.ConsulHTTP)
 114         consulCfg.Datacenter = cfg.SiteID
 115         consul, err := api.NewClient(consulCfg)
 116         if err != nil {
 117                 return err
 118         }
 119         _, err = consul.Catalog().Datacenters()
 120         if err != nil {
 121                 return err
 122         }
 123         return nil
 124 }
 125
 126 // OnlyNode returns true if this is the only consul node.
 127 func (cb *consulBooter) OnlyNode() (bool, error) {
 128         c, err := api.NewClient(consulCfg)
 129         if err != nil {
 130                 return false, err
 131         }
 132         nodes, _, err := c.Catalog().Nodes(nil)
 133         return len(nodes) == 1, err
 134 }