Merge pull request #1271 in YUNIONIO/onecloud from ~ZHOUYOUSONG/onecloud:feature/yousong-lbagent-managed-270 to release/2.7.0

* commit '4f7a4177ac8838e13b03f96b5a1e8a0b099918a8':
  lbagent: try using unicast vrrp
  lbagent: report ip, ha_state, version
  lbagent: always reload on startup
  lbagent: making agent params with reflect
  netutils2: add MyIP() for finding default source ip

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,9 +48,12 @@ 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"`
+	Params     *SLoadbalancerAgentParams `create:"optional" list:"admin" get:"admin"`
 
 	Loadbalancers             time.Time `nullable:"true" list:"admin" update:"admin"`
 	LoadbalancerListeners     time.Time `nullable:"true" list:"admin" update:"admin"`
@@ -421,14 +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) {
-	diff, err := db.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
 	}
-	db.OpsLog.LogEvent(lbagent, db.ACT_UPDATE, diff, userCred)
+	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
 }
 
@@ -484,6 +517,11 @@ vrrp_instance YunionLB {
 		auth_type PASS
 		auth_pass {{ .vrrp.pass }}
 	}
+	{{ if .vrrp.notify_script -}} notify {{ .vrrp.notify_script }} {{- end }}
+	{{ if .vrrp.unicast_peer -}} unicast_peer { {{- println }}
+		{{- range .vrrp.unicast_peer }}		{{ println . }} {{- end }}
+	}
+	{{- end }}
 	priority {{ .vrrp.priority }}
 	advert_int {{ .vrrp.advert_int }}
 	garp_master_refresh {{ .vrrp.garp_master_refresh }}

pkg/lbagent/api.go

@@ -3,11 +3,13 @@ package lbagent
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
 	"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 +18,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 +28,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
 }
@@ -56,14 +64,30 @@ func (h *ApiHelper) Run(ctx context.Context) {
 			apiDataChanged := h.doSyncApiData(ctx)
 			agentParamsChanged := h.doSyncAgentParams(ctx)
 			if apiDataChanged || agentParamsChanged {
+				log.Infof("things changed: params: %v, data %v", agentParamsChanged, apiDataChanged)
 				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"
@@ -89,6 +113,31 @@ func (h *ApiHelper) agentPeekOnce(ctx context.Context) (*models.LoadbalancerAgen
 	return agent, nil
 }
 
+func (h *ApiHelper) agentPeekPeers(ctx context.Context, vri int) ([]*models.LoadbalancerAgent, error) {
+	s := h.adminClientSession(ctx)
+	params := jsonutils.NewDict()
+	params.Set(consts.LBAGENT_QUERY_ORIG_KEY, jsonutils.NewString(consts.LBAGENT_QUERY_ORIG_VAL))
+	listResult, err := modules.LoadbalancerAgents.List(s, params)
+	if err != nil {
+		err := fmt.Errorf("agent listing error: %s", err)
+		return nil, err
+	}
+	peers := []*models.LoadbalancerAgent{}
+	for _, data := range listResult.Data {
+		agent := &models.LoadbalancerAgent{}
+		err := data.Unmarshal(agent)
+		if err != nil {
+			err := fmt.Errorf("agent data unmarshal error: %s", err)
+			return nil, err
+		}
+		if agent.Params.Vrrp.VirtualRouterId != vri {
+			continue
+		}
+		peers = append(peers, agent)
+	}
+	return peers, nil
+}
+
 type agentPeekResult models.LoadbalancerAgent
 
 func (r *agentPeekResult) staleInFuture(s int) bool {
@@ -131,12 +180,13 @@ 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
 	}
 	if !r.staleInFuture(h.opts.ApiLbagentHbTimeoutRelaxation) {
-		log.Warningf("agent will stale in %d seconds, re-sync",
+		log.Warningf("agent will stale in %d seconds, ignore old corpus",
 			h.opts.ApiLbagentHbTimeoutRelaxation)
 	} else {
 		h.doHb(ctx)
@@ -147,9 +197,10 @@ func (h *ApiHelper) runInit(ctx context.Context) {
 			log.Errorf("load local api data failed: %s", err)
 		}
 	}
-	if changed := h.doSyncApiData(ctx); changed {
-		h.doUseCorpus(ctx)
-	}
+	// better reload now because agent data is not in corpus yet
+	h.doSyncApiData(ctx)
+	h.doSyncAgentParams(ctx)
+	h.doUseCorpus(ctx)
 }
 
 func (h *ApiHelper) loadLocalData(ctx context.Context) (*agentmodels.LoadbalancerCorpus, error) {
@@ -179,10 +230,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
@@ -254,12 +328,36 @@ func (h *ApiHelper) doSyncAgentParams(ctx context.Context) bool {
 		log.Errorf("agent params get failure: %s", err)
 		return false
 	}
+	peers, err := h.agentPeekPeers(ctx, agent.Params.Vrrp.VirtualRouterId)
+	if err != nil {
+		log.Errorf("agent get peers failure: %s", err)
+		return false
+	}
+	useUnicast := true
+	unicastPeer := []string{}
+	for _, peer := range peers {
+		if peer.Id == agent.Id {
+			continue
+		}
+		if peer.IP == "" {
+			useUnicast = false
+			log.Warningf("agent %s(%s) has no ip, use multicast vrrp", peer.Name, peer.Id)
+			break
+		}
+		unicastPeer = append(unicastPeer, peer.IP)
+	}
+
 	agentParams, err := agentmodels.NewAgentParams(agent)
 	if err != nil {
 		log.Errorf("agent params prepare failure: %s", err)
 		return false
 	}
-	if !agentParams.Equal(h.agentParams) {
+	agentParams.SetVrrpParams("notify_script", h.haStateProvider.StateScript())
+	if useUnicast {
+		agentParams.SetVrrpParams("unicast_peer", unicastPeer)
+	}
+	if !agentParams.Equals(h.agentParams) {
+		log.Infof("use unicast vrrp from %s to %s", agent.IP, strings.Join(unicastPeer, ","))
 		h.agentParams = agentParams
 		return true
 	}
@@ -293,3 +391,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

@@ -3,11 +3,37 @@ package models
 import (
 	"encoding/base64"
 	"fmt"
+	"reflect"
 	"text/template"
 
+	"yunion.io/x/pkg/utils"
+
 	"yunion.io/x/onecloud/pkg/mcclient/models"
 )
 
+func dataFromParams(p interface{}) map[string]interface{} {
+	rv := reflect.ValueOf(p)
+	if rv.Kind() != reflect.Struct {
+		panic(fmt.Sprintf("unexpected kind: %#v", p))
+	}
+	rt := rv.Type()
+
+	r := map[string]interface{}{}
+	for i := 0; i < rv.NumField(); i++ {
+		f := rt.Field(i)
+		fn := utils.CamelSplit(f.Name, "_")
+		if fn == "" {
+			continue
+		}
+		v := rv.Field(i)
+		if !v.IsValid() {
+			continue
+		}
+		r[fn] = v.Interface()
+	}
+	return r
+}
+
 type AgentParams struct {
 	AgentModel           *models.LoadbalancerAgent
 	KeepalivedConfigTmpl *template.Template
@@ -37,34 +63,13 @@ func NewAgentParams(agent *models.LoadbalancerAgent) (*AgentParams, error) {
 	dataAgent := map[string]interface{}{
 		"id":   agent.Id,
 		"name": agent.Name,
-	}
-	dataVrrp := map[string]interface{}{
-		"priority":            agent.Params.Vrrp.Priority,
-		"virtual_router_id":   agent.Params.Vrrp.VirtualRouterId,
-		"garp_master_refresh": agent.Params.Vrrp.GarpMasterRefresh,
-		"preempt":             agent.Params.Vrrp.Preempt,
-		"interface":           agent.Params.Vrrp.Interface,
-		"advert_int":          agent.Params.Vrrp.AdvertInt,
-		"pass":                agent.Params.Vrrp.Pass,
-	}
-	dataHaproxy := map[string]interface{}{
-		"global_log":      agent.Params.Haproxy.GlobalLog,
-		"global_nbthread": agent.Params.Haproxy.GlobalNbthread,
-		"log_http":        agent.Params.Haproxy.LogHttp,
-		"log_tcp":         agent.Params.Haproxy.LogTcp,
-		"log_normal":      agent.Params.Haproxy.LogNormal,
-	}
-	dataTelegraf := map[string]interface{}{
-		"influx_db_output_url":        agent.Params.Telegraf.InfluxDbOutputUrl,
-		"influx_db_output_name":       agent.Params.Telegraf.InfluxDbOutputName,
-		"influx_db_output_unsafe_ssl": agent.Params.Telegraf.InfluxDbOutputUnsafeSsl,
-		"haproxy_input_interval":      agent.Params.Telegraf.HaproxyInputInterval,
+		"ip":   agent.IP,
 	}
 	data := map[string]map[string]interface{}{
 		"agent":    dataAgent,
-		"vrrp":     dataVrrp,
-		"haproxy":  dataHaproxy,
-		"telegraf": dataTelegraf,
+		"vrrp":     dataFromParams(agent.Params.Vrrp),
+		"haproxy":  dataFromParams(agent.Params.Haproxy),
+		"telegraf": dataFromParams(agent.Params.Telegraf),
 	}
 	agentParams := &AgentParams{
 		AgentModel:           agent,
@@ -76,7 +81,7 @@ func NewAgentParams(agent *models.LoadbalancerAgent) (*AgentParams, error) {
 	return agentParams, nil
 }
 
-func (p *AgentParams) Equal(p2 *AgentParams) bool {
+func (p *AgentParams) Equals(p2 *AgentParams) bool {
 	if p == nil && p2 == nil {
 		return true
 	}
@@ -88,6 +93,14 @@ func (p *AgentParams) Equal(p2 *AgentParams) bool {
 	if agentP.Params != agentP2.Params {
 		return false
 	}
+	keys := []string{"notify_script", "unicast_peer"}
+	for _, key := range keys {
+		v := p.GetVrrpParams(key)
+		v2 := p.GetVrrpParams(key)
+		if !reflect.DeepEqual(v, v2) {
+			return false
+		}
+	}
 	return true
 }
 
@@ -104,10 +117,22 @@ func (p *AgentParams) setXxParams(xx, k string, v interface{}) map[string]interf
 	return dt
 }
 
+func (p *AgentParams) getXxParams(xx, k string) interface{} {
+	dt, ok := p.Data[xx]
+	if !ok {
+		return nil
+	}
+	return dt[k]
+}
+
 func (p *AgentParams) SetVrrpParams(k string, v interface{}) map[string]interface{} {
 	return p.setXxParams("vrrp", k, v)
 }
 
+func (p *AgentParams) GetVrrpParams(k string) interface{} {
+	return p.getXxParams("vrrp", k)
+}
+
 func (p *AgentParams) SetHaproxyParams(k string, v interface{}) map[string]interface{} {
 	return p.setXxParams("haproxy", k, v)
 }

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 {

pkg/util/netutils2/netutils.go

@@ -40,6 +40,22 @@ func IsTcpPortUsed(addr string, port int) bool {
 	}
 }
 
+// MyIP returns source ip used to communicate with udp:114.114.114.114:53
+func MyIP() (ip string, err error) {
+	conn, err := net.Dial("udp4", "114.114.114.114:53")
+	if err != nil {
+		return
+	}
+	defer conn.Close()
+	addr, ok := conn.LocalAddr().(*net.UDPAddr)
+	if !ok {
+		err = fmt.Errorf("not a net.UDPAddr: %#v", conn.LocalAddr())
+		return
+	}
+	ip = addr.IP.String()
+	return
+}
+
 func GetPrivatePrefixes(privatePrefixes []string) []string {
 	if privatePrefixes != nil {
 		return privatePrefixes