lbagent: report ip, ha_state, version

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)
 

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"

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 }}

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():
+	}
+}

pkg/lbagent/cmd.go

@@ -10,6 +10,7 @@ type LbagentCmdType uintptr
 
 const (
 	LbagentCmdUseCorpus LbagentCmdType = iota
+	LbagentCmdStopDaemons
 )
 
 type LbagentCmdUseCorpusData struct {

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 {

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
+}

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,

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
 }

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)
 )

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
 

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 {