fix: lbagent switch to backup (#15465)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

cmd/lbagent/main.go

@@ -16,77 +16,16 @@ package main
 
 import (
 	"context"
-	"os"
-	"os/signal"
-	"sync"
-	"syscall"
 
-	"yunion.io/x/log"
-
-	app_common "yunion.io/x/onecloud/pkg/cloudcommon/app"
-	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
 	"yunion.io/x/onecloud/pkg/lbagent"
 	"yunion.io/x/onecloud/pkg/util/atexit"
+	"yunion.io/x/onecloud/pkg/util/procutils"
 )
 
 func main() {
 	defer atexit.Handle()
 
-	opts := &lbagent.Options{}
-	commonOpts := &opts.CommonOptions
-	{
-		common_options.ParseOptions(opts, os.Args, "lbagent.conf", "lbagent")
-		app_common.InitAuth(commonOpts, func() {
-			log.Infof("auth finished ok")
-		})
-	}
-	if err := opts.ValidateThenInit(); err != nil {
-		log.Fatalf("opts validate: %s", err)
-	}
+	go procutils.WaitZombieLoop(context.TODO())
 
-	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 {
-			log.Fatalf("init haproxy helper failed: %s", err)
-		}
-	}
-	{
-		apiHelper, err = lbagent.NewApiHelper(opts)
-		if err != nil {
-			log.Fatalf("init api helper failed: %s", err)
-		}
-		apiHelper.SetHaStateProvider(haStateWatcher)
-	}
-
-	{
-		wg := &sync.WaitGroup{}
-		cmdChan := make(chan *lbagent.LbagentCmd) // internal
-		ctx, cancelFunc := context.WithCancel(context.Background())
-		ctx = context.WithValue(ctx, "wg", wg)
-		ctx = context.WithValue(ctx, "cmdChan", cmdChan)
-		wg.Add(3)
-		go haStateWatcher.Run(ctx)
-		go haproxyHelper.Run(ctx)
-		go apiHelper.Run(ctx)
-
-		go func() {
-			sigChan := make(chan os.Signal)
-			signal.Notify(sigChan, syscall.SIGINT)
-			signal.Notify(sigChan, syscall.SIGTERM)
-			sig := <-sigChan
-			log.Infof("signal received: %s", sig)
-			cancelFunc()
-		}()
-		wg.Wait()
-	}
+	lbagent.StartService()
 }

pkg/compute/models/loadbalancernetworks.go

@@ -64,7 +64,7 @@ type SLoadbalancerNetwork struct {
 	LoadbalancerId string `width:"36" charset:"ascii" nullable:"false" list:"user"`
 	NetworkId      string `width:"36" charset:"ascii" nullable:"false" list:"user"`
 	IpAddr         string `width:"16" charset:"ascii" list:"user"`
-	MacAddr        string `width:"32" charset:"ascii" nullable:"false" list:"user"`
+	MacAddr        string `width:"32" charset:"ascii" nullable:"true" list:"user"`
 }
 
 func (manager *SLoadbalancernetworkManager) GetMasterFieldName() string {

pkg/lbagent/api.go

@@ -84,8 +84,14 @@ func (h *ApiHelper) Run(ctx context.Context) {
 		wg.Done()
 		log.Infof("api helper bye")
 	}()
-	h.startOvnWorker(ctx)
-	defer h.stopOvnWorker()
+
+	h.haState = <-h.haStateProvider.StateChannel()
+	log.Infof("initial haState: %s", h.haState)
+	switch h.haState {
+	case api.LB_HA_STATE_BACKUP:
+	default:
+		h.startOvnWorker(ctx)
+	}
 
 	wg.Add(1)
 	go h.apih.Start(ctx, nil, "")
@@ -95,7 +101,6 @@ func (h *ApiHelper) Run(ctx context.Context) {
 	defer hbTicker.Stop()
 	defer agentParamsSyncTicker.Stop()
 
-	h.haState = <-h.haStateProvider.StateChannel()
 	for {
 		select {
 		case <-hbTicker.C:
@@ -125,6 +130,7 @@ func (h *ApiHelper) Run(ctx context.Context) {
 				h.doUseCorpus(ctx)
 			}
 		case state := <-h.haStateProvider.StateChannel():
+			log.Infof("current state: %s ha_state: %s", h.haState, state)
 			switch state {
 			case api.LB_HA_STATE_BACKUP:
 				h.stopOvnWorker()
@@ -388,8 +394,10 @@ func (h *ApiHelper) doUseCorpus(ctx context.Context) {
 		log.Warningf("agent params nil")
 		return
 	}
-	if err := h.ovn.Refresh(ctx, h.corpus.ModelSets.Loadbalancers); err != nil {
-		log.Errorf("ovn refresh: %v", err)
+	if h.ovn != nil {
+		if err := h.ovn.Refresh(ctx, h.corpus.ModelSets.Loadbalancers); err != nil {
+			log.Errorf("ovn refresh: %v", err)
+		}
 	}
 	log.Infof("make effect new corpus and params")
 	cmdData := &LbagentCmdUseCorpusData{

pkg/lbagent/options.go

@@ -40,7 +40,7 @@ type LbagentOptions struct {
 	haproxyConfigDir string
 	haproxyRunDir    string
 	haproxyShareDir  string
-	haStateChan      chan string
+	// haStateChan      chan string
 
 	KeepalivedBin string `default:"keepalived"`
 	HaproxyBin    string `default:"haproxy"`

pkg/lbagent/ovn.go

@@ -558,6 +558,9 @@ func (ovnHost *OvnHost) Refresh(ctx context.Context, lb *agentmodels.Loadbalance
 func (ovnHost *OvnHost) refresh(ctx context.Context, lb *agentmodels.Loadbalancer) error {
 	// We may allow detch/attach network for loadbalancer in the future
 	lbnet := lb.LoadbalancerNetwork
+	if len(lbnet.MacAddr) == 0 {
+		return errors.Errorf("empty LoadbalancerNetwork MacAddr for lb %s(%s)? mismatch region version???", lb.Name, lb.Id)
+	}
 	lb.ListenAddress = ovnHost.vethAddrOuter
 	ovnHost.lb = lb
 	ovnHost.macaddr = lbnet.MacAddr
@@ -640,6 +643,7 @@ func NewOvnWorker() *OvnWorker {
 }
 
 func (ovn *OvnWorker) Start(ctx context.Context) {
+	log.Infof("ovn worker started")
 	ovn.Bar.Init()
 	defer ovn.Bar.Done()
 	defer log.Infof("ovn worker bye!")
@@ -649,6 +653,7 @@ func (ovn *OvnWorker) Start(ctx context.Context) {
 			err := ovn.refresh(ctx, cmd.lbs)
 			cmd.done <- err
 		case <-ovn.Bar.Cancelled():
+			log.Infof("ovn worker stop on cancel signal")
 			ovn.stop()
 			return
 		case <-ctx.Done():
@@ -658,8 +663,8 @@ func (ovn *OvnWorker) Start(ctx context.Context) {
 }
 
 func (ovn *OvnWorker) stop() {
-	for _, ovn := range ovn.lbMap {
-		ovn.Stop()
+	for _, o := range ovn.lbMap {
+		o.Stop()
 	}
 }
 
@@ -765,6 +770,10 @@ func (ovn *OvnWorker) refresh(ctx context.Context, lbs agentmodels.Loadbalancers
 		if lb.NetworkType != computeapis.LB_NETWORK_TYPE_VPC {
 			continue
 		}
+		if len(lb.LoadbalancerNetwork.MacAddr) == 0 {
+			log.Errorf("empty LoadbalancerNetwork MacAddr for lb %s(%s)? mismatch region version???", lb.Name, lb.Id)
+			continue
+		}
 		ovnHost, ok := m[lb.Id]
 		if !ok {
 			inner, outer, err := ovn.findTwoAddrIdx()

pkg/lbagent/service.go

@@ -0,0 +1,88 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lbagent
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"yunion.io/x/log"
+
+	app_common "yunion.io/x/onecloud/pkg/cloudcommon/app"
+	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
+)
+
+func StartService() {
+	opts := &Options{}
+	commonOpts := &opts.CommonOptions
+	{
+		common_options.ParseOptions(opts, os.Args, "lbagent.conf", "lbagent")
+		app_common.InitAuth(commonOpts, func() {
+			log.Infof("auth finished ok")
+		})
+	}
+	if err := opts.ValidateThenInit(); err != nil {
+		log.Fatalf("opts validate: %s", err)
+	}
+
+	var haproxyHelper *HaproxyHelper
+	var apiHelper *ApiHelper
+	var haStateWatcher *HaStateWatcher
+	var err error
+	{
+		haStateWatcher, err = NewHaStateWatcher(opts)
+		if err != nil {
+			log.Fatalf("init ha state watcher failed: %s", err)
+		}
+	}
+	{
+		haproxyHelper, err = NewHaproxyHelper(opts)
+		if err != nil {
+			log.Fatalf("init haproxy helper failed: %s", err)
+		}
+	}
+	{
+		apiHelper, err = NewApiHelper(opts)
+		if err != nil {
+			log.Fatalf("init api helper failed: %s", err)
+		}
+		apiHelper.SetHaStateProvider(haStateWatcher)
+	}
+
+	{
+		wg := &sync.WaitGroup{}
+		cmdChan := make(chan *LbagentCmd) // internal
+		ctx, cancelFunc := context.WithCancel(context.Background())
+		ctx = context.WithValue(ctx, "wg", wg)
+		ctx = context.WithValue(ctx, "cmdChan", cmdChan)
+		wg.Add(3)
+		go haStateWatcher.Run(ctx)
+		go haproxyHelper.Run(ctx)
+		go apiHelper.Run(ctx)
+
+		go func() {
+			sigChan := make(chan os.Signal)
+			signal.Notify(sigChan, syscall.SIGINT)
+			signal.Notify(sigChan, syscall.SIGTERM)
+			sig := <-sigChan
+			log.Infof("signal received: %s", sig)
+			cancelFunc()
+		}()
+		wg.Wait()
+	}
+}