diff --git a/cmd/lbagent/main.go b/cmd/lbagent/main.go index f2842002e2..e96313c96f 100644 --- a/cmd/lbagent/main.go +++ b/cmd/lbagent/main.go @@ -28,7 +28,14 @@ func main() { var haproxyHelper *lbagent.HaproxyHelper var apiHelper *lbagent.ApiHelper + var haStateWatcher *lbagent.HaStateWatcher var err error + { + haStateWatcher, err = lbagent.NewHaStateWatcher(opts) + if err != nil { + log.Fatalf("init ha state watcher failed: %s", err) + } + } { haproxyHelper, err = lbagent.NewHaproxyHelper(opts) if err != nil { @@ -40,6 +47,7 @@ func main() { if err != nil { log.Fatalf("init api helper failed: %s", err) } + apiHelper.SetHaStateProvider(haStateWatcher) } { @@ -48,7 +56,8 @@ func main() { ctx, cancelFunc := context.WithCancel(context.Background()) ctx = context.WithValue(ctx, "wg", wg) ctx = context.WithValue(ctx, "cmdChan", cmdChan) - wg.Add(2) + wg.Add(3) + go haStateWatcher.Run(ctx) go haproxyHelper.Run(ctx) go apiHelper.Run(ctx) diff --git a/pkg/compute/consts/loadbalancer_const.go b/pkg/compute/consts/loadbalancer_const.go index 2fc734aa83..56ec8848b5 100644 --- a/pkg/compute/consts/loadbalancer_const.go +++ b/pkg/compute/consts/loadbalancer_const.go @@ -256,6 +256,22 @@ const ( LB_CHARGE_TYPE_BY_HOUR = "hour" ) +const ( + LB_HA_STATE_MASTER = "MASTER" + LB_HA_STATE_BACKUP = "BACKUP" + LB_HA_STATE_FAULT = "FAULT" + LB_HA_STATE_STOP = "STOP" + LB_HA_STATE_UNKNOWN = "UNKNOWN" +) + +var LB_HA_STATES = choices.NewChoices( + LB_HA_STATE_MASTER, + LB_HA_STATE_BACKUP, + LB_HA_STATE_FAULT, + LB_HA_STATE_STOP, + LB_HA_STATE_UNKNOWN, +) + const ( LBAGENT_QUERY_ORIG_KEY = "_orig" LBAGENT_QUERY_ORIG_VAL = "lbagent" diff --git a/pkg/compute/models/loadbalanceragents.go b/pkg/compute/models/loadbalanceragents.go index 8c376e155c..35ece5fcb4 100644 --- a/pkg/compute/models/loadbalanceragents.go +++ b/pkg/compute/models/loadbalanceragents.go @@ -14,6 +14,7 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/cloudcommon/validators" + "yunion.io/x/onecloud/pkg/compute/consts" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" ) @@ -47,6 +48,9 @@ func init() { type SLoadbalancerAgent struct { db.SStandaloneResourceBase + Version string `width:"64" nullable:"true" list:"admin" update:"admin"` + IP string `width:"32" nullable:"true" list:"admin" update:"admin"` + HaState string `width:"32" nullable:"true" list:"admin" update:"admin" default:"UNKNOWN"` // LB_HA_STATE_UNKNOWN HbLastSeen time.Time `nullable:"true" list:"admin" update:"admin"` HbTimeout int `nullable:"true" list:"admin" update:"admin" create:"optional" default:"3600"` Params *SLoadbalancerAgentParams `create:"optional" get:"admin"` @@ -421,13 +425,43 @@ func (lbagent *SLoadbalancerAgent) AllowPerformHb(ctx context.Context, userCred } func (lbagent *SLoadbalancerAgent) PerformHb(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { - _, err := lbagent.GetModelManager().TableSpec().Update(lbagent, func() error { + ipV := validators.NewIPv4AddrValidator("ip") + haStateV := validators.NewStringChoicesValidator("ha_state", consts.LB_HA_STATES) + { + keyV := map[string]validators.IValidator{ + "ip": ipV, + "ha_state": haStateV, + } + for _, v := range keyV { + v.Optional(true) + if err := v.Validate(data); err != nil { + return nil, err + } + } + } + diff, err := lbagent.GetModelManager().TableSpec().Update(lbagent, func() error { lbagent.HbLastSeen = time.Now() + if jVer, err := data.Get("version"); err == nil { + if jVerStr, ok := jVer.(*jsonutils.JSONString); ok { + lbagent.Version = jVerStr.Value() + } + } + if ipV.IP != nil { + lbagent.IP = ipV.IP.String() + } + if haStateV.Value != "" { + lbagent.HaState = haStateV.Value + } return nil }) if err != nil { return nil, err } + if len(diff) > 1 { + // other things changed besides hb_last_seen + log.Infof("lbagent %s(%s) state changed: %s", lbagent.Name, lbagent.Id, diff) + db.OpsLog.LogEvent(lbagent, db.ACT_UPDATE, diff, userCred) + } return nil, nil } @@ -482,6 +516,7 @@ vrrp_instance YunionLB { auth_type PASS auth_pass {{ .vrrp.pass }} } + {{ if .vrrp.notify_script -}} notify {{ .vrrp.notify_script }} {{- end }} priority {{ .vrrp.priority }} advert_int {{ .vrrp.advert_int }} garp_master_refresh {{ .vrrp.garp_master_refresh }} diff --git a/pkg/lbagent/api.go b/pkg/lbagent/api.go index 7d59eff52a..efe3da4551 100644 --- a/pkg/lbagent/api.go +++ b/pkg/lbagent/api.go @@ -8,6 +8,7 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" + "yunion.io/x/pkg/util/version" "yunion.io/x/onecloud/pkg/compute/consts" agentmodels "yunion.io/x/onecloud/pkg/lbagent/models" @@ -16,6 +17,8 @@ import ( "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/mcclient/models" "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/mcclient/options" + "yunion.io/x/onecloud/pkg/util/netutils2" ) type ApiHelper struct { @@ -24,12 +27,16 @@ type ApiHelper struct { dataDirMan *agentutils.ConfigDirManager corpus *agentmodels.LoadbalancerCorpus agentParams *agentmodels.AgentParams + + haState string + haStateProvider HaStateProvider } func NewApiHelper(opts *Options) (*ApiHelper, error) { helper := &ApiHelper{ opts: opts, dataDirMan: agentutils.NewConfigDirManager(opts.apiDataStoreDir), + haState: consts.LB_HA_STATE_UNKNOWN, } return helper, nil } @@ -58,12 +65,27 @@ func (h *ApiHelper) Run(ctx context.Context) { if apiDataChanged || agentParamsChanged { h.doUseCorpus(ctx) } + case state := <-h.haStateProvider.StateChannel(): + switch state { + case consts.LB_HA_STATE_BACKUP: + h.doStopDaemons(ctx) + default: + if state != h.haState { + // try your best to make things up + h.doUseCorpus(ctx) + } + } + h.haState = state case <-ctx.Done(): return } } } +func (h *ApiHelper) SetHaStateProvider(hsp HaStateProvider) { + h.haStateProvider = hsp +} + func (h *ApiHelper) adminClientSession(ctx context.Context) *mcclient.ClientSession { region := h.opts.CommonOptions.Region apiVersion := "v2" @@ -131,6 +153,7 @@ func (h *ApiHelper) agentPeek(ctx context.Context) *agentPeekResult { } func (h *ApiHelper) runInit(ctx context.Context) { + h.haState = <-h.haStateProvider.StateChannel() r := h.agentPeek(ctx) if r == nil { return @@ -180,10 +203,33 @@ func (h *ApiHelper) agentUpdateSeen(ctx context.Context) *models.LoadbalancerAge return agent } +func (h *ApiHelper) newAgentHbParams(ctx context.Context) (*jsonutils.JSONDict, error) { + ip, err := netutils2.MyIP() + if err != nil { + return nil, err + } + state := h.haState + version := version.Get().GitVersion + opts := &options.LoadbalancerAgentActionHbOptions{ + IP: ip, + HaState: state, + Version: version, + } + params, err := options.StructToParams(opts) + if err != nil { + return nil, err + } + return params, nil +} + func (h *ApiHelper) doHb(ctx context.Context) (*models.LoadbalancerAgent, error) { // TODO check if things changed recently s := h.adminClientSession(ctx) - data, err := modules.LoadbalancerAgents.PerformAction(s, h.opts.ApiLbagentId, "hb", nil) + params, err := h.newAgentHbParams(ctx) + if err != nil { + return nil, fmt.Errorf("heartbeat: making params: %s", err) + } + data, err := modules.LoadbalancerAgents.PerformAction(s, h.opts.ApiLbagentId, "hb", params) if err != nil { err := fmt.Errorf("heartbeat api error: %s", err) return nil, err @@ -260,6 +306,7 @@ func (h *ApiHelper) doSyncAgentParams(ctx context.Context) bool { log.Errorf("agent params prepare failure: %s", err) return false } + agentParams.SetVrrpParams("notify_script", h.haStateProvider.StateScript()) if !agentParams.Equal(h.agentParams) { h.agentParams = agentParams return true @@ -294,3 +341,14 @@ func (h *ApiHelper) doUseCorpus(ctx context.Context) { return } } + +func (h *ApiHelper) doStopDaemons(ctx context.Context) { + cmd := &LbagentCmd{ + Type: LbagentCmdStopDaemons, + } + cmdChan := ctx.Value("cmdChan").(chan *LbagentCmd) + select { + case cmdChan <- cmd: + case <-ctx.Done(): + } +} diff --git a/pkg/lbagent/cmd.go b/pkg/lbagent/cmd.go index c04af764eb..ab8025c87f 100644 --- a/pkg/lbagent/cmd.go +++ b/pkg/lbagent/cmd.go @@ -10,6 +10,7 @@ type LbagentCmdType uintptr const ( LbagentCmdUseCorpus LbagentCmdType = iota + LbagentCmdStopDaemons ) type LbagentCmdUseCorpusData struct { diff --git a/pkg/lbagent/haproxy.go b/pkg/lbagent/haproxy.go index 0e8b1625d3..799d54a676 100644 --- a/pkg/lbagent/haproxy.go +++ b/pkg/lbagent/haproxy.go @@ -70,11 +70,44 @@ func (h *HaproxyHelper) handleCmd(ctx context.Context, cmd *LbagentCmd) { cmdData := cmd.Data.(*LbagentCmdUseCorpusData) defer cmdData.Wg.Done() h.handleUseCorpusCmd(ctx, cmd) + case LbagentCmdStopDaemons: + h.handleStopDaemonsCmd(ctx) default: log.Warningf("command type ignored: %v", cmd.Type) } } +func (h *HaproxyHelper) handleStopDaemonsCmd(ctx context.Context) { + files := map[string]string{ + "gobetween": h.gobetweenPidFile(), + "haproxy": h.haproxyPidFile(), + "telegraf": h.telegrafPidFile(), + } + wg := &sync.WaitGroup{} + wg.Add(len(files)) + + for name, f := range files { + go func(name, f string) { + defer wg.Done() + proc := agentutils.ReadPidFile(f) + if proc != nil { + log.Infof("stopping %s(%d)", name, proc.Pid) + proc.Signal(syscall.SIGTERM) + for etime := time.Now().Add(5 * time.Second); etime.Before(time.Now()); { + if err := proc.Signal(syscall.Signal(0)); err == nil { + return + } + time.Sleep(500 * time.Millisecond) + } + proc.Kill() + // TODO check whether proc.Ppid == os.Getpid() + proc.Wait() + } + }(name, f) + } + wg.Wait() +} + func (h *HaproxyHelper) handleUseCorpusCmd(ctx context.Context, cmd *LbagentCmd) { // haproxy config dir dir, err := h.configDirMan.NewDir(func(dir string) error { @@ -308,17 +341,18 @@ func (h *HaproxyHelper) gobetweenPidFile() string { func (h *HaproxyHelper) reloadGobetween(ctx context.Context) error { pidFile := h.gobetweenPidFile() - args := []string{ - h.opts.GobetweenBin, - "--config", h.gobetweenConf(), - "--format", "json", - } proc := agentutils.ReadPidFile(pidFile) if proc != nil { log.Infof("stopping gobetween(%d)", proc.Pid) proc.Kill() proc.Wait() } + + args := []string{ + h.opts.GobetweenBin, + "--config", h.gobetweenConf(), + "--format", "json", + } log.Infof("starting gobetween") cmd, err := h.startCmd(args) if err != nil { diff --git a/pkg/lbagent/hastate.go b/pkg/lbagent/hastate.go new file mode 100644 index 0000000000..b282e2d05c --- /dev/null +++ b/pkg/lbagent/hastate.go @@ -0,0 +1,195 @@ +package lbagent + +import ( + "context" + "fmt" + "io/ioutil" + "os" + "path" + "strings" + "sync" + "syscall" + "time" + "unicode" + + "github.com/fsnotify/fsnotify" + + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/compute/consts" + agentutils "yunion.io/x/onecloud/pkg/lbagent/utils" +) + +const ( + HA_STATE_SCRIPT_NAME = "ha_state.sh" + HA_STATE_SCRIPT_CONTENT = ` +#!/bin/bash +echo "$@" >%s +` + HA_STATE_FILENAME = "ha_state" +) + +type HaStateProvider interface { + StateChannel() <-chan string + StateScript() string +} + +type HaStateWatcher struct { + HaStateScriptPath string + HaStatePath string + CurrentState string // TODO hide it + + opts *Options + w *fsnotify.Watcher + C chan string +} + +func (hsw *HaStateWatcher) StateChannel() <-chan string { + return hsw.C +} + +func (hsw *HaStateWatcher) StateScript() string { + return hsw.HaStateScriptPath +} + +func (hsw *HaStateWatcher) Run(ctx context.Context) { + defer func() { + log.Infof("ha state watcher bye") + wg := ctx.Value("wg").(*sync.WaitGroup) + wg.Done() + }() + + hsw.loadHaState() + hsw.C <- hsw.CurrentState + + statePending := false + tick := time.NewTicker(3 * time.Second) + defer tick.Stop() + for { + select { + case ev := <-hsw.w.Events: + switch { + case ev.Name == hsw.opts.haproxyRunDir: + switch ev.Op { + case fsnotify.Remove, fsnotify.Rename: + log.Errorf("run dir %s", ev.Op.String()) + hsw.sayBye() + default: + log.Debugf("ignored %s", ev) + } + case ev.Name == hsw.HaStatePath: + log.Infof("hastate file: %s", ev) + switch ev.Op { + case fsnotify.Create, fsnotify.Write: + err := hsw.loadHaState() + if err != nil { + log.Errorf("load state: %s", err) + hsw.sayBye() + } + select { + case hsw.C <- hsw.CurrentState: + statePending = false + default: + statePending = true + } + } + } + case err := <-hsw.w.Errors: + log.Errorf("watcher err: %s", err) + hsw.sayBye() + case <-tick.C: + if statePending { + select { + case hsw.C <- hsw.CurrentState: + statePending = false + default: + } + } + case <-ctx.Done(): + return + } + } +} + +func (hsw *HaStateWatcher) loadHaState() (err error) { + defer func() { + if err != nil { + hsw.CurrentState = consts.LB_HA_STATE_UNKNOWN + } + }() + data, err := ioutil.ReadFile(hsw.HaStatePath) + if err != nil { + return err + } + log.Infof("got state: %s", data) + // Sample content + // + // INSTANCE YunionLB BACKUP 110 + // + fields := strings.Fields(string(data)) + if len(fields) >= 3 { + hsw.CurrentState = fields[2] + return + } + err = fmt.Errorf("ha state file contains too little info") + return +} + +func (hsw *HaStateWatcher) sayBye() { + syscall.Kill(os.Getpid(), syscall.SIGTERM) +} + +func NewHaStateWatcher(opts *Options) (hsw *HaStateWatcher, err error) { + var ( + w *fsnotify.Watcher + ) + defer func() { + if err != nil { + if w != nil { + w.Close() + } + } + }() + + haStateScriptPath := path.Join(opts.haproxyShareDir, HA_STATE_SCRIPT_NAME) + haStatePath := path.Join(opts.haproxyRunDir, HA_STATE_FILENAME) + { + content := fmt.Sprintf(HA_STATE_SCRIPT_CONTENT, haStatePath) + content = strings.TrimLeftFunc(content, unicode.IsSpace) + mode := agentutils.FileModeFileExec + err = ioutil.WriteFile(haStateScriptPath, []byte(content), mode) + if err != nil { + return + } + var fi os.FileInfo + fi, err = os.Stat(haStateScriptPath) + if err != nil { + return + } + if fi.Mode() != mode { + err = os.Chmod(haStateScriptPath, mode) + if err != nil { + return + } + } + } + + w, err = fsnotify.NewWatcher() + if err != nil { + return + } + err = w.Add(opts.haproxyRunDir) + if err != nil { + return + } + + hsw = &HaStateWatcher{ + HaStateScriptPath: haStateScriptPath, + HaStatePath: haStatePath, + CurrentState: consts.LB_HA_STATE_UNKNOWN, + + opts: opts, + w: w, + C: make(chan string), + } + return +} diff --git a/pkg/lbagent/models/agentparams.go b/pkg/lbagent/models/agentparams.go index f6dab80eda..c846fc7040 100644 --- a/pkg/lbagent/models/agentparams.go +++ b/pkg/lbagent/models/agentparams.go @@ -63,6 +63,7 @@ func NewAgentParams(agent *models.LoadbalancerAgent) (*AgentParams, error) { dataAgent := map[string]interface{}{ "id": agent.Id, "name": agent.Name, + "ip": agent.IP, } data := map[string]map[string]interface{}{ "agent": dataAgent, diff --git a/pkg/lbagent/options.go b/pkg/lbagent/options.go index 8ac80d5f8e..39cfccaf37 100644 --- a/pkg/lbagent/options.go +++ b/pkg/lbagent/options.go @@ -23,6 +23,8 @@ type LbagentOptions struct { apiDataStoreDir string haproxyConfigDir string haproxyRunDir string + haproxyShareDir string + haStateChan chan string KeepalivedBin string `default:"keepalived"` HaproxyBin string `default:"haproxy"` @@ -41,17 +43,23 @@ func (opts *Options) ValidateThenInit() error { return fmt.Errorf("negative api batch list size: %d", opts.ApiListBatchSize) } - return opts.initDirs() + if err := opts.initDirs(); err != nil { + return err + } + + return nil } func (opts *Options) initDirs() error { opts.apiDataStoreDir = filepath.Join(opts.BaseDataDir, "data") opts.haproxyConfigDir = filepath.Join(opts.BaseDataDir, "configs") opts.haproxyRunDir = filepath.Join(opts.BaseDataDir, "run") + opts.haproxyShareDir = filepath.Join(opts.BaseDataDir, "share") dirs := []string{ opts.apiDataStoreDir, opts.haproxyConfigDir, opts.haproxyRunDir, + opts.haproxyShareDir, } for _, dir := range dirs { err := os.MkdirAll(dir, agentutils.FileModeDir) @@ -60,5 +68,6 @@ func (opts *Options) initDirs() error { dir, err) } } + return nil } diff --git a/pkg/lbagent/utils/filemode.go b/pkg/lbagent/utils/filemode.go index 0bb572ebb9..d73e0f1a61 100644 --- a/pkg/lbagent/utils/filemode.go +++ b/pkg/lbagent/utils/filemode.go @@ -7,6 +7,7 @@ import ( const ( FileModeDir = os.FileMode(0755) FileModeFile = os.FileMode(0644) + FileModeFileExec = os.FileMode(0755) FileModeDirSensitive = os.FileMode(0700) FileModeFileSensitive = os.FileMode(0600) ) diff --git a/pkg/mcclient/models/loadbalancers.go b/pkg/mcclient/models/loadbalancers.go index 603f0799c8..571a7a86dd 100644 --- a/pkg/mcclient/models/loadbalancers.go +++ b/pkg/mcclient/models/loadbalancers.go @@ -163,6 +163,9 @@ type LoadbalancerCertificate struct { type LoadbalancerAgent struct { StandaloneResource + Version string + IP string + HaState string HbLastSeen time.Time HbTimeout int diff --git a/pkg/mcclient/options/loadbalanceragents.go b/pkg/mcclient/options/loadbalanceragents.go index a35b48651a..a00b624fba 100644 --- a/pkg/mcclient/options/loadbalanceragents.go +++ b/pkg/mcclient/options/loadbalanceragents.go @@ -108,6 +108,10 @@ type LoadbalancerAgentDeleteOptions struct { type LoadbalancerAgentActionHbOptions struct { ID string `json:-` + + Version string + IP string + HaState string } type LoadbalancerAgentActionPatchParamsOptions struct {