From c774bd2ca3198ec776d928018b2bb6bc3feb7046 Mon Sep 17 00:00:00 2001 From: Jian Qiu Date: Thu, 1 Dec 2022 01:23:51 +0800 Subject: [PATCH] fix: lbagent switch to backup (#15466) Co-authored-by: Qiu Jian --- cmd/lbagent/main.go | 67 +--------------- pkg/compute/models/loadbalancernetworks.go | 2 +- pkg/lbagent/api.go | 18 +++-- pkg/lbagent/options.go | 2 +- pkg/lbagent/ovn.go | 13 +++- pkg/lbagent/service.go | 88 ++++++++++++++++++++++ 6 files changed, 117 insertions(+), 73 deletions(-) create mode 100644 pkg/lbagent/service.go diff --git a/cmd/lbagent/main.go b/cmd/lbagent/main.go index 8d9b6f743b..f1e5ba1db9 100644 --- a/cmd/lbagent/main.go +++ b/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() } diff --git a/pkg/compute/models/loadbalancernetworks.go b/pkg/compute/models/loadbalancernetworks.go index 6c5be2776f..8ff719bff3 100644 --- a/pkg/compute/models/loadbalancernetworks.go +++ b/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 { diff --git a/pkg/lbagent/api.go b/pkg/lbagent/api.go index cc3fcd3400..ee58e1029c 100644 --- a/pkg/lbagent/api.go +++ b/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{ diff --git a/pkg/lbagent/options.go b/pkg/lbagent/options.go index 96f9744efe..cc4b035889 100644 --- a/pkg/lbagent/options.go +++ b/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"` diff --git a/pkg/lbagent/ovn.go b/pkg/lbagent/ovn.go index 54fc19ea12..c20dbf8cab 100644 --- a/pkg/lbagent/ovn.go +++ b/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() diff --git a/pkg/lbagent/service.go b/pkg/lbagent/service.go new file mode 100644 index 0000000000..2dc8e1c9fd --- /dev/null +++ b/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() + } +}