From c2f2ceafd4e7ff7df099a1e849ec6740b28b73ef Mon Sep 17 00:00:00 2001 From: Jian Qiu Date: Thu, 2 Feb 2023 21:32:30 +0800 Subject: [PATCH] feature: lbagent register (#15872) Co-authored-by: Qiu Jian --- cmd/climc/shell/compute/loadbalanceragents.go | 20 +- pkg/apis/compute/loadbalanceragent.go | 8 + pkg/apis/compute/zz_generated.model.go | 38 ++- pkg/cloudcommon/agent/agent.go | 36 +-- pkg/compute/models/loadbalanceragents.go | 303 +++++++++--------- .../models/loadbalanceragents_deploy.go | 13 +- .../models/loadbalancercluster_params.go | 38 +++ .../models/loadbalancerclusterresource.go | 2 +- pkg/compute/models/loadbalancerclusters.go | 208 +++++++++++- pkg/lbagent/api.go | 15 +- pkg/lbagent/haproxy.go | 3 + pkg/lbagent/models/agentparams.go | 31 +- pkg/lbagent/models/keepalived.go | 5 + pkg/lbagent/options.go | 17 +- pkg/lbagent/ovn.go | 4 +- pkg/lbagent/register.go | 89 +++++ pkg/lbagent/service.go | 26 +- .../modules/compute/mod_loadbalanceragents.go | 2 + .../options/compute/loadbalanceragents.go | 20 ++ pkg/util/netutils2/iface.go | 61 ++++ 20 files changed, 686 insertions(+), 253 deletions(-) create mode 100644 pkg/compute/models/loadbalancercluster_params.go create mode 100644 pkg/lbagent/register.go create mode 100644 pkg/util/netutils2/iface.go diff --git a/cmd/climc/shell/compute/loadbalanceragents.go b/cmd/climc/shell/compute/loadbalanceragents.go index 6143c4a02d..0b73abde52 100644 --- a/cmd/climc/shell/compute/loadbalanceragents.go +++ b/cmd/climc/shell/compute/loadbalanceragents.go @@ -132,7 +132,7 @@ func init() { printLbagent(lbagent) return nil }) - R(&options.LoadbalancerAgentActionDeployOptions{}, "lbagent-deploy", "Deploy lbagent", func(s *mcclient.ClientSession, opts *options.LoadbalancerAgentActionDeployOptions) error { + /*R(&options.LoadbalancerAgentActionDeployOptions{}, "lbagent-deploy", "Deploy lbagent", func(s *mcclient.ClientSession, opts *options.LoadbalancerAgentActionDeployOptions) error { params, err := opts.Params() if err != nil { return err @@ -151,5 +151,23 @@ func init() { } printLbagent(lbagent) return nil + })*/ + R(&options.LoadbalancerAgentJoinClusterOptions{}, "lbagent-join-cluster", "Agent join cluster", func(s *mcclient.ClientSession, opts *options.LoadbalancerAgentJoinClusterOptions) error { + params, _ := opts.Params() + lbagent, err := modules.LoadbalancerAgents.PerformAction(s, opts.ID, "join-cluster", params) + if err != nil { + return err + } + printLbagent(lbagent) + return nil + }) + R(&options.LoadbalancerAgentLeaveClusterOptions{}, "lbagent-leave-cluster", "Agent leave cluster", func(s *mcclient.ClientSession, opts *options.LoadbalancerAgentLeaveClusterOptions) error { + params, _ := opts.Params() + lbagent, err := modules.LoadbalancerAgents.PerformAction(s, opts.ID, "leave-cluster", params) + if err != nil { + return err + } + printLbagent(lbagent) + return nil }) } diff --git a/pkg/apis/compute/loadbalanceragent.go b/pkg/apis/compute/loadbalanceragent.go index 7a826ecf01..3bf2527169 100644 --- a/pkg/apis/compute/loadbalanceragent.go +++ b/pkg/apis/compute/loadbalanceragent.go @@ -31,3 +31,11 @@ type LoadbalancerAgentListInput struct { IP []string `json:"ip"` HaState []string `json:"ha_state"` } + +type LoadbalancerAgentJoinClusterInput struct { + ClusterId string `json:"cluster_id"` + Priority int `json:"priority"` +} + +type LoadbalancerAgentLeaveClusterInput struct { +} diff --git a/pkg/apis/compute/zz_generated.model.go b/pkg/apis/compute/zz_generated.model.go index 1985fc75f2..5c044af23f 100644 --- a/pkg/apis/compute/zz_generated.model.go +++ b/pkg/apis/compute/zz_generated.model.go @@ -787,9 +787,9 @@ type SElasticcache struct { NetworkId string `json:"network_id"` // 带宽 Bandwidth int `json:"bandwidth"` - // 内网DNS + // 内网DNS PrivateDNS string `json:"private_dns"` - // 内网IP地址 + // 内网IP地址 PrivateIpAddr string `json:"private_ip_addr"` // 内网访问端口 PrivateConnectPort int `json:"private_connect_port"` @@ -1632,9 +1632,11 @@ type SLoadbalancerAclResourceBase struct { type SLoadbalancerAgent struct { apis.SStandaloneResourceBase SLoadbalancerClusterResourceBase - Version string `json:"version"` - IP string `json:"ip"` - HaState string `json:"ha_state"` + Version string `json:"version"` + IP string `json:"ip"` + Interface string `json:"interface"` + Priority int `json:"priority"` + HaState string `json:"ha_state"` // LB_HA_STATE_UNKNOWN HbLastSeen time.Time `json:"hb_last_seen"` HbTimeout int `json:"hb_timeout"` @@ -1688,13 +1690,9 @@ type SLoadbalancerAgentParamsTelegraf struct { // SLoadbalancerAgentParamsVrrp is an autogenerated struct via yunion.io/x/onecloud/pkg/compute/models.SLoadbalancerAgentParamsVrrp. type SLoadbalancerAgentParamsVrrp struct { - Priority int `json:"priority"` - VirtualRouterId int `json:"virtual_router_id"` - GarpMasterRefresh int `json:"garp_master_refresh"` - Preempt bool `json:"preempt"` - Interface string `json:"interface"` - AdvertInt int `json:"advert_int"` - Pass string `json:"pass"` + SLoadbalancerClusterParams + Priority int `json:"priority"` + Interface string `json:"interface"` } // SLoadbalancerBackend is an autogenerated struct via yunion.io/x/onecloud/pkg/compute/models.SLoadbalancerBackend. @@ -1744,6 +1742,16 @@ type SLoadbalancerCluster struct { apis.SStandaloneResourceBase SZoneResourceBase SWireResourceBase + Params *SLoadbalancerClusterParams `json:"params"` +} + +// SLoadbalancerClusterParams is an autogenerated struct via yunion.io/x/onecloud/pkg/compute/models.SLoadbalancerClusterParams. +type SLoadbalancerClusterParams struct { + VirtualRouterId int `json:"virtual_router_id"` + Preempt bool `json:"preempt"` + AdvertInt int `json:"advert_int"` + Pass string `json:"pass"` + GarpMasterRefresh int `json:"garp_master_refresh"` } // SLoadbalancerClusterResourceBase is an autogenerated struct via yunion.io/x/onecloud/pkg/compute/models.SLoadbalancerClusterResourceBase. @@ -1804,9 +1812,9 @@ type SLoadbalancerHealthCheck struct { // 健康检查路径 / HealthCheckHttpCode string `json:"health_check_http_code"` HealthCheckRise int `json:"health_check_rise"` - // 健康检查健康阈值 3秒 + // 健康检查健康阈值 3秒 HealthCheckFall int `json:"health_check_fall"` - // 健康检查不健康阈值 15秒 + // 健康检查不健康阈值 15秒 HealthCheckTimeout int `json:"health_check_timeout"` // 健康检查超时时间 10秒 HealthCheckInterval int `json:"health_check_interval"` @@ -2429,7 +2437,7 @@ type SSchedtag struct { apis.SStandaloneResourceBase apis.SScopedResourceBase DefaultStrategy string `json:"default_strategy"` - // Column(VARCHAR(16, charset='ascii'), nullable=True, default=”) + // Column(VARCHAR(16, charset='ascii'), nullable=True, default='') ResourceType string `json:"resource_type"` } diff --git a/pkg/cloudcommon/agent/agent.go b/pkg/cloudcommon/agent/agent.go index f6f916b588..55e587a39f 100644 --- a/pkg/cloudcommon/agent/agent.go +++ b/pkg/cloudcommon/agent/agent.go @@ -32,6 +32,7 @@ import ( "yunion.io/x/onecloud/pkg/hostman/storageman" "yunion.io/x/onecloud/pkg/mcclient" modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute" + "yunion.io/x/onecloud/pkg/util/netutils2" ) type SZoneInfo struct { @@ -54,45 +55,14 @@ type SBaseAgent struct { stop bool } -func getIfaceIPs(iface *net.Interface) ([]net.IP, error) { - addrs, err := iface.Addrs() - if err != nil { - return nil, err - } - ips := make([]net.IP, 0) - for _, a := range addrs { - if ipnet, ok := a.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { - if ipnet.IP.To4() != nil { - ips = append(ips, ipnet.IP) - } - } - } - return ips, nil -} - func (agent *SBaseAgent) IAgent() iagent.IAgent { return agent.GetVirtualObject().(iagent.IAgent) } func (agent *SBaseAgent) Init(iagent iagent.IAgent, ifname string, cachePath string) error { - iface, err := net.InterfaceByName(ifname) + iface, ips, err := netutils2.WaitIfaceIps(ifname) if err != nil { - return err - } - var ips []net.IP - MAX := 60 - wait := 0 - for wait < MAX { - ips, err = getIfaceIPs(iface) - if err != nil { - return err - } - if len(ips) == 0 { - time.Sleep(2 * time.Second) - wait += 2 - } else { - break - } + return errors.Wrap(err, "WaitIfaceIps") } if len(ips) == 0 { return fmt.Errorf("Interface %s ip address not found", ifname) diff --git a/pkg/compute/models/loadbalanceragents.go b/pkg/compute/models/loadbalanceragents.go index 4a1147762c..98475849b5 100644 --- a/pkg/compute/models/loadbalanceragents.go +++ b/pkg/compute/models/loadbalanceragents.go @@ -16,8 +16,8 @@ package models import ( "context" + "database/sql" "encoding/base64" - "fmt" "net/url" "reflect" "text/template" @@ -33,11 +33,13 @@ import ( api "yunion.io/x/onecloud/pkg/apis/compute" identity_apis "yunion.io/x/onecloud/pkg/apis/identity" "yunion.io/x/onecloud/pkg/cloudcommon/db" + "yunion.io/x/onecloud/pkg/cloudcommon/db/lockman" "yunion.io/x/onecloud/pkg/cloudcommon/validators" "yunion.io/x/onecloud/pkg/compute/options" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" "yunion.io/x/onecloud/pkg/mcclient/auth" + "yunion.io/x/onecloud/pkg/util/logclient" "yunion.io/x/onecloud/pkg/util/stringutils2" ) @@ -70,10 +72,12 @@ func init() { // - agent configuration params type SLoadbalancerAgent struct { db.SStandaloneResourceBase - SLoadbalancerClusterResourceBase `width:"36" charset:"ascii" nullable:"false" list:"user" create:"required"` + SLoadbalancerClusterResourceBase `width:"36" charset:"ascii" nullable:"true" list:"user" update:"admin"` - Version string `width:"64" nullable:"true" list:"admin" update:"admin"` - IP string `width:"32" nullable:"true" list:"admin" update:"admin"` + Version string `width:"64" nullable:"true" list:"admin" create:"required" update:"admin"` + IP string `width:"32" charset:"ascii" nullable:"true" list:"admin" create:"required"` + Interface string `width:"17" charset:"ascii" nullable:"true" list:"admin" create:"required" update:"admin"` + Priority int `nullable:"true" list:"user" 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"` @@ -94,13 +98,10 @@ type SLoadbalancerAgent struct { } type SLoadbalancerAgentParamsVrrp struct { - Priority int `json:",omitzero"` - VirtualRouterId int `json:",omitzero"` - GarpMasterRefresh int `json:",omitzero"` - Preempt bool - Interface string - AdvertInt int `json:",omitzero"` - Pass string + SLoadbalancerClusterParams + + Priority int `json:",omitzero"` + Interface string } const ( @@ -134,7 +135,7 @@ type SLoadbalancerAgentParams struct { } func (p *SLoadbalancerAgentParamsVrrp) Validate(data *jsonutils.JSONDict) error { - if len(p.Interface) == 0 || len(p.Interface) > 16 { + /*if len(p.Interface) == 0 || len(p.Interface) > 16 { // TODO printable exclude white space return httperrors.NewInputParameterError("invalid vrrp interface %q", p.Interface) } @@ -150,10 +151,11 @@ func (p *SLoadbalancerAgentParamsVrrp) Validate(data *jsonutils.JSONDict) error } if p.AdvertInt < 1 || p.AdvertInt > 255 { return httperrors.NewInputParameterError("invalid vrrp advert_int %d: want [1,255]", p.AdvertInt) - } + }*/ return nil } +/* func (p *SLoadbalancerAgentParamsVrrp) validatePeer(pp *SLoadbalancerAgentParamsVrrp) error { if p.Priority == pp.Priority { return fmt.Errorf("vrrp priority of peer lbagents must be different, got %d", p.Priority) @@ -200,7 +202,7 @@ func (p *SLoadbalancerAgentParamsVrrp) initDefault(data *jsonutils.JSONDict) { p.VirtualRouterId = lbagentVrrpDefaultVrid } if !data.Contains("params", "vrrp", "advert_int") { - p.AdvertInt = 1 + p.AdvertInt = 5 } if !data.Contains("params", "vrrp", "garp_master_refresh") { p.GarpMasterRefresh = 27 @@ -208,7 +210,7 @@ func (p *SLoadbalancerAgentParamsVrrp) initDefault(data *jsonutils.JSONDict) { if !data.Contains("params", "vrrp", "pass") { p.Pass = "YunionLB" } -} +}*/ func (p *SLoadbalancerAgentParamsHaproxy) Validate(data *jsonutils.JSONDict) error { if p.GlobalNbthread < 1 { @@ -313,7 +315,7 @@ func (p *SLoadbalancerAgentParams) initDefault(data *jsonutils.JSONDict) { if p.TelegrafConfTmpl == "" { p.TelegrafConfTmpl = loadbalancerTelegrafConfTmplDefaultEncoded } - p.Vrrp.initDefault(data) + //p.Vrrp.initDefault(data) p.Haproxy.initDefault(data) p.Telegraf.initDefault(data) } @@ -329,9 +331,9 @@ func (p *SLoadbalancerAgentParams) Validate(data *jsonutils.JSONDict) error { if err := p.validateTmpl("telegraf_conf_tmpl", p.TelegrafConfTmpl); err != nil { return err } - if err := p.Vrrp.Validate(data); err != nil { + /*if err := p.Vrrp.Validate(data); err != nil { return err - } + }*/ if err := p.Haproxy.Validate(data); err != nil { return err } @@ -347,8 +349,7 @@ func (p *SLoadbalancerAgentParams) needsUpdatePeer(pp *SLoadbalancerAgentParams) p.TelegrafConfTmpl != pp.TelegrafConfTmpl { return true } - return p.Vrrp.needsUpdatePeer(&pp.Vrrp) || - p.Haproxy.needsUpdatePeer(&pp.Haproxy) || + return p.Haproxy.needsUpdatePeer(&pp.Haproxy) || p.Telegraf.needsUpdatePeer(&pp.Telegraf) } @@ -357,7 +358,7 @@ func (p *SLoadbalancerAgentParams) updateBy(pp *SLoadbalancerAgentParams) { p.HaproxyConfTmpl = pp.HaproxyConfTmpl p.TelegrafConfTmpl = pp.TelegrafConfTmpl - p.Vrrp.updateBy(&pp.Vrrp) + // p.Vrrp.updateBy(&pp.Vrrp) p.Haproxy.updateBy(&pp.Haproxy) p.Telegraf.updateBy(&pp.Telegraf) } @@ -403,11 +404,11 @@ func (man *SLoadbalancerAgentManager) GetPropertyDefaultParams(ctx context.Conte } func (man *SLoadbalancerAgentManager) ValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { - clusterV := validators.NewModelIdOrNameValidator("cluster", "loadbalancercluster", ownerId) + // clusterV := validators.NewModelIdOrNameValidator("cluster", "loadbalancercluster", ownerId) { keyV := map[string]validators.IValidator{ "hb_timeout": validators.NewNonNegativeValidator("hb_timeout").Default(3600), - "cluster": clusterV, + // "cluster": clusterV, } for _, v := range keyV { if err := v.Validate(data); err != nil { @@ -415,92 +416,6 @@ func (man *SLoadbalancerAgentManager) ValidateCreateData(ctx context.Context, us } } } - { - cluster := clusterV.Model.(*SLoadbalancerCluster) - lbagents, err := LoadbalancerClusterManager.getLoadbalancerAgents(cluster.Id) - if err != nil { - return nil, httperrors.NewGeneralError(err) - } - params := &SLoadbalancerAgentParams{} - { - if len(lbagents) > 0 { - peerLbagent := &lbagents[0] - peerParams := peerLbagent.Params - params.Vrrp.setByPeer(&peerParams.Vrrp) - } - if len(lbagents) == 0 && !data.Contains("params", "vrrp", "virtual_router_id") { - otherLbagents := []SLoadbalancerAgent{} - q := man.Query().GroupBy("cluster_id") - err := db.FetchModelObjects(LoadbalancerAgentManager, q, &otherLbagents) - if err != nil { - return nil, httperrors.NewInternalServerError("fetch lbagents of other clusters: %v", err) - } - maxVrid := -1 - for i := range otherLbagents { - lbagent := &otherLbagents[i] - if lbagent.ClusterId == cluster.Id { - continue - } - vrid := lbagent.Params.Vrrp.VirtualRouterId - if vrid > maxVrid { - maxVrid = vrid - } - } - if maxVrid > 0 && maxVrid < 255 { - params.Vrrp.VirtualRouterId = maxVrid + 1 - } else { - params.Vrrp.VirtualRouterId = lbagentVrrpDefaultVrid - } - } - if !data.Contains("params", "vrrp", "priority") { - // a backup to all existing members - minPrio := 256 - for i := range lbagents { - peerLbagent := &lbagents[i] - priority := peerLbagent.Params.Vrrp.Priority - if priority < minPrio { - minPrio = priority - } - } - if minPrio > 1 { - params.Vrrp.Priority = minPrio - 1 - } else { - params.Vrrp.Priority = lbagentVrrpDefaultPrio - } - } - var ( - oldVrrpParams jsonutils.JSONObject - err error - ) - if oldVrrpParams, err = data.Get("params", "vrrp"); err != nil { - oldVrrpParams = jsonutils.NewDict() - } - vrrpParams := jsonutils.Marshal(params.Vrrp).(*jsonutils.JSONDict) - vrrpParams.UpdateDefault(oldVrrpParams) - data.Add(vrrpParams, "params", "vrrp") - paramsV := validators.NewStructValidator("params", params) - if err := paramsV.Validate(data); err != nil { - return nil, err - } - } - for i := range lbagents { - peerLbagent := &lbagents[i] - peerParams := peerLbagent.Params - err := params.Vrrp.validatePeer(&peerParams.Vrrp) - if err != nil { - return nil, httperrors.NewConflictError("conflict with lbagent %s(%s): %v", peerLbagent.Name, peerLbagent.Id, err) - } - } - vrrpRouterId := params.Vrrp.VirtualRouterId - otherCluster, err := LoadbalancerClusterManager.findByVrrpRouterIdInZone(cluster.ZoneId, vrrpRouterId) - if err != nil { - return nil, err - } - if otherCluster != nil && otherCluster.Id != cluster.Id { - return nil, httperrors.NewConflictError("lbcluster %s(%s) already has virtual_router_id %d", - otherCluster.Name, otherCluster.Id, vrrpRouterId) - } - } input := apis.StandaloneResourceCreateInput{} err := data.Unmarshal(&input) @@ -515,6 +430,25 @@ func (man *SLoadbalancerAgentManager) ValidateCreateData(ctx context.Context, us return data, nil } +func (agent *SLoadbalancerAgent) PostCreate( + ctx context.Context, + userCred mcclient.TokenCredential, + ownerId mcclient.IIdentityProvider, + query jsonutils.JSONObject, + data jsonutils.JSONObject, +) { + params := SLoadbalancerAgentParams{} + params.initDefault(data.(*jsonutils.JSONDict)) + + _, err := db.Update(agent, func() error { + agent.Params = ¶ms + return nil + }) + if err != nil { + log.Errorf("init params fail: %s", err) + } +} + // 负载均衡Agent列表 func (man *SLoadbalancerAgentManager) ListItemFilter( ctx context.Context, @@ -626,6 +560,13 @@ func (lbagent *SLoadbalancerAgent) ValidateUpdateData(ctx context.Context, userC return data, nil } +func (agent *SLoadbalancerAgent) ValidateDeleteCondition(ctx context.Context, info jsonutils.JSONObject) error { + if len(agent.ClusterId) > 0 { + return errors.Wrap(httperrors.ErrResourceBusy, "agent join a cluster") + } + return agent.SStandaloneResourceBase.ValidateDeleteCondition(ctx, info) +} + func (manager *SLoadbalancerAgentManager) FetchCustomizeColumns( ctx context.Context, userCred mcclient.TokenCredential, @@ -692,11 +633,23 @@ func (man *SLoadbalancerAgentManager) getByClusterId(clusterId string) ([]SLoadb r := []SLoadbalancerAgent{} q := man.Query().Equals("cluster_id", clusterId) if err := db.FetchModelObjects(man, q, &r); err != nil { - return nil, err + return nil, errors.Wrap(err, "FetchModelObjects") } return r, nil } +func (lbagent *SLoadbalancerAgent) getCluster() *SLoadbalancerCluster { + if len(lbagent.ClusterId) == 0 { + return nil + } + clusterObj, err := LoadbalancerClusterManager.FetchById(lbagent.ClusterId) + if err != nil { + log.Errorf("SLoadbalancerAgent.getCluster error %s", err) + return nil + } + return clusterObj.(*SLoadbalancerCluster) +} + func (lbagent *SLoadbalancerAgent) PerformHb(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { ipV := validators.NewIPv4AddrValidator("ip") haStateV := validators.NewStringChoicesValidator("ha_state", api.LB_HA_STATES) @@ -749,6 +702,103 @@ func (lbagent *SLoadbalancerAgent) IsActive() bool { return true } +func (lbagent *SLoadbalancerAgent) PerformJoinCluster( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + input api.LoadbalancerAgentJoinClusterInput, +) (*jsonutils.JSONDict, error) { + if len(lbagent.ClusterId) > 0 { + return nil, errors.Wrap(httperrors.ErrConflict, "lbagent has been join cluster") + } + clusterObj, err := LoadbalancerClusterManager.FetchByIdOrName(userCred, input.ClusterId) + if err != nil { + if errors.Cause(err) == sql.ErrNoRows { + return nil, errors.Wrapf(httperrors.ErrNotFound, "%s %s", LoadbalancerClusterManager.Keyword(), input.ClusterId) + } else { + return nil, errors.Wrap(err, "LoadbalancerClusterManager.FetchById") + } + } + cluster := clusterObj.(*SLoadbalancerCluster) + + lockman.LockObject(ctx, cluster) + defer lockman.ReleaseObject(ctx, cluster) + + peerAgents, err := LoadbalancerAgentManager.getByClusterId(cluster.Id) + if err != nil { + return nil, errors.Wrap(err, "LoadbalancerAgentManager.getByClusterId") + } + if len(peerAgents) >= 2 { + return nil, errors.Wrap(httperrors.ErrTooLarge, "too many agents") + } + priority := 255 + if input.Priority > 0 { + for i := range peerAgents { + if input.Priority == peerAgents[i].Priority { + return nil, errors.Wrap(httperrors.ErrDuplicateId, "duplicate priority in same cluster") + } + } + priority = input.Priority + } else { + for i := range peerAgents { + if priority >= peerAgents[i].Priority { + priority = peerAgents[i].Priority - 1 + } + } + } + var params SLoadbalancerAgentParams + if lbagent.Params != nil { + params = *lbagent.Params + } + params.Vrrp.SLoadbalancerClusterParams = *cluster.Params + _, err = db.Update(lbagent, func() error { + lbagent.ClusterId = cluster.Id + lbagent.Priority = priority + lbagent.Params = ¶ms + return nil + }) + if err != nil { + return nil, errors.Wrap(err, "Update") + } else { + notes := struct { + ClusterId string + Priority int + }{ + ClusterId: cluster.Id, + Priority: priority, + } + logclient.AddActionLogWithContext(ctx, lbagent, logclient.ACT_ATTACH_HOST, notes, userCred, true) + } + return nil, nil +} + +func (lbagent *SLoadbalancerAgent) PerformLeaveCluster( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + data api.LoadbalancerAgentLeaveClusterInput, +) (*jsonutils.JSONDict, error) { + if len(lbagent.ClusterId) == 0 { + return nil, errors.Wrap(httperrors.ErrInvalidStatus, "lbagent not belong to any cluster") + } + oldClusterId := lbagent.ClusterId + _, err := db.Update(lbagent, func() error { + lbagent.ClusterId = "" + return nil + }) + if err != nil { + return nil, errors.Wrap(err, "Update") + } else { + notes := struct { + ClusterId string + }{ + ClusterId: oldClusterId, + } + logclient.AddActionLogWithContext(ctx, lbagent, logclient.ACT_DETACH_HOST, notes, userCred, true) + } + return nil, nil +} + func (lbagent *SLoadbalancerAgent) PerformParamsPatch(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { oldParams := lbagent.Params params := gotypes.DeepCopy(*lbagent.Params).(SLoadbalancerAgentParams) @@ -758,39 +808,6 @@ func (lbagent *SLoadbalancerAgent) PerformParamsPatch(ctx context.Context, userC if err := paramsV.Validate(d); err != nil { return nil, err } - // new vrrp virtual_router_id should be unique across clusters - if params.Vrrp.VirtualRouterId != oldParams.Vrrp.VirtualRouterId { - clusterM, err := LoadbalancerClusterManager.FetchById(lbagent.ClusterId) - if err != nil { - return nil, httperrors.NewGeneralError(err) - } - cluster := clusterM.(*SLoadbalancerCluster) - otherCluster, err := LoadbalancerClusterManager.findByVrrpRouterIdInZone(cluster.ZoneId, params.Vrrp.VirtualRouterId) - if err != nil { - return nil, err - } - if otherCluster != nil { - return nil, httperrors.NewConflictError("lbcluster %s(%s) already has virtual_router_id %d", - otherCluster.Name, otherCluster.Id, params.Vrrp.VirtualRouterId) - } - } - // new vrrp priority should be unique in the cluster - if params.Vrrp.Priority != oldParams.Vrrp.Priority { - lbagents, err := LoadbalancerClusterManager.getLoadbalancerAgents(lbagent.ClusterId) - if err != nil { - return nil, httperrors.NewGeneralError(err) - } - for i := range lbagents { - peerLbagent := &lbagents[i] - if peerLbagent.Id == lbagent.Id { - continue - } - if peerLbagent.Params.Vrrp.Priority == params.Vrrp.Priority { - return nil, httperrors.NewConflictError("peer lbagent %s(%s) already has vrrp priority %d", - peerLbagent.Name, peerLbagent.Id, params.Vrrp.Priority) - } - } - } { diff, err := db.Update(lbagent, func() error { lbagent.Params = ¶ms diff --git a/pkg/compute/models/loadbalanceragents_deploy.go b/pkg/compute/models/loadbalanceragents_deploy.go index 43ff902d6d..e7e53416b2 100644 --- a/pkg/compute/models/loadbalanceragents_deploy.go +++ b/pkg/compute/models/loadbalanceragents_deploy.go @@ -28,9 +28,7 @@ import ( "yunion.io/x/pkg/utils" compute_apis "yunion.io/x/onecloud/pkg/apis/compute" - identity_apis "yunion.io/x/onecloud/pkg/apis/identity" "yunion.io/x/onecloud/pkg/cloudcommon/db" - "yunion.io/x/onecloud/pkg/compute/options" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" "yunion.io/x/onecloud/pkg/mcclient/auth" @@ -38,7 +36,6 @@ import ( ansible_modules "yunion.io/x/onecloud/pkg/mcclient/modules/ansible" compute_modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute" "yunion.io/x/onecloud/pkg/util/ansible" - "yunion.io/x/onecloud/pkg/util/logclient" ) type SLoadbalancerAgentDeployment struct { @@ -295,7 +292,7 @@ func (lbagent *SLoadbalancerAgent) PerformDeploy( query jsonutils.JSONObject, input *compute_apis.LoadbalancerAgentDeployInput, ) (*compute_apis.LoadbalancerAgentDeployInput, error) { - host := input.Host + /*host := input.Host for _, k := range []string{"user", "pass", "proj"} { if v, ok := host.GetVar(k); !ok { return nil, httperrors.NewBadRequestError("host missing %s field", k) @@ -356,7 +353,8 @@ func (lbagent *SLoadbalancerAgent) PerformDeploy( }); err != nil { return nil, err } - return nil, err + return nil, err*/ + return nil, errors.Wrap(httperrors.ErrNotSupported, "deprecated") } func (lbagent *SLoadbalancerAgent) updateOrCreatePbModel(ctx context.Context, @@ -373,7 +371,7 @@ func (lbagent *SLoadbalancerAgent) updateOrCreatePbModel(ctx context.Context, } func (lbagent *SLoadbalancerAgent) PerformUndeploy(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { - deployment := lbagent.Deployment + /*deployment := lbagent.Deployment if deployment == nil || deployment.Host == "" { return nil, httperrors.NewConflictError("No previous deployment info available") } @@ -401,7 +399,8 @@ func (lbagent *SLoadbalancerAgent) PerformUndeploy(ctx context.Context, userCred }); err != nil { return nil, err } - return nil, nil + return nil, nil*/ + return nil, errors.Wrap(httperrors.ErrNotSupported, "deprecated") } const ( diff --git a/pkg/compute/models/loadbalancercluster_params.go b/pkg/compute/models/loadbalancercluster_params.go new file mode 100644 index 0000000000..e6e8949213 --- /dev/null +++ b/pkg/compute/models/loadbalancercluster_params.go @@ -0,0 +1,38 @@ +// 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 models + +import ( + "yunion.io/x/jsonutils" +) + +type SLoadbalancerClusterParams struct { + VirtualRouterId int `json:",omitzero"` + Preempt bool + AdvertInt int `json:",omitzero"` + Pass string + GarpMasterRefresh int `json:",omitzero"` +} + +func (p *SLoadbalancerClusterParams) String() string { + return jsonutils.Marshal(p).String() +} + +func (p *SLoadbalancerClusterParams) IsZero() bool { + if *p == (SLoadbalancerClusterParams{}) { + return true + } + return false +} diff --git a/pkg/compute/models/loadbalancerclusterresource.go b/pkg/compute/models/loadbalancerclusterresource.go index 3f34b89ddb..c3cfdad459 100644 --- a/pkg/compute/models/loadbalancerclusterresource.go +++ b/pkg/compute/models/loadbalancerclusterresource.go @@ -33,7 +33,7 @@ import ( type SLoadbalancerClusterResourceBase struct { // 归属LB集群 - ClusterId string `width:"36" charset:"ascii" nullable:"false" list:"user" update:"user" create:"optional" json:"cluster_id"` + ClusterId string `width:"36" charset:"ascii" nullable:"true" list:"user" update:"user" create:"optional" json:"cluster_id"` } type SLoadbalancerClusterResourceBaseManager struct { diff --git a/pkg/compute/models/loadbalancerclusters.go b/pkg/compute/models/loadbalancerclusters.go index 7eb43f9b5b..677c6dce6f 100644 --- a/pkg/compute/models/loadbalancerclusters.go +++ b/pkg/compute/models/loadbalancerclusters.go @@ -16,10 +16,13 @@ package models import ( "context" + "reflect" "yunion.io/x/jsonutils" "yunion.io/x/log" "yunion.io/x/pkg/errors" + "yunion.io/x/pkg/gotypes" + "yunion.io/x/pkg/util/seclib" "yunion.io/x/sqlchemy" "yunion.io/x/onecloud/pkg/apis" @@ -28,6 +31,7 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/validators" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" + "yunion.io/x/onecloud/pkg/util/logclient" "yunion.io/x/onecloud/pkg/util/stringutils2" ) @@ -40,6 +44,9 @@ type SLoadbalancerClusterManager struct { var LoadbalancerClusterManager *SLoadbalancerClusterManager func init() { + gotypes.RegisterSerializable(reflect.TypeOf(&SLoadbalancerClusterParams{}), func() gotypes.ISerializable { + return &SLoadbalancerClusterParams{} + }) LoadbalancerClusterManager = &SLoadbalancerClusterManager{ SStandaloneResourceBaseManager: db.NewStandaloneResourceBaseManager( SLoadbalancerCluster{}, @@ -55,6 +62,8 @@ type SLoadbalancerCluster struct { db.SStandaloneResourceBase SZoneResourceBase SWireResourceBase `width:"36" charset:"ascii" nullable:"true" list:"admin" create:"optional" update:"admin"` + + Params *SLoadbalancerClusterParams `nullable:"true" create:"optional" list:"admin" get:"admin"` } // 负载均衡集群列表 @@ -131,7 +140,13 @@ func (man *SLoadbalancerClusterManager) QueryDistinctExtraField(q *sqlchemy.SQue return q, httperrors.ErrNotFound } -func (man *SLoadbalancerClusterManager) ValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { +func (man *SLoadbalancerClusterManager) ValidateCreateData( + ctx context.Context, + userCred mcclient.TokenCredential, + ownerId mcclient.IIdentityProvider, + query jsonutils.JSONObject, + data *jsonutils.JSONDict, +) (*jsonutils.JSONDict, error) { zoneV := validators.NewModelIdOrNameValidator("zone", "zone", ownerId) wireV := validators.NewModelIdOrNameValidator("wire", "wire", ownerId) vs := []validators.IValidator{ @@ -168,7 +183,28 @@ func (man *SLoadbalancerClusterManager) ValidateCreateData(ctx context.Context, return data, nil } -func (lbc *SLoadbalancerCluster) ValidateUpdateData(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { +func (cluster *SLoadbalancerCluster) PostCreate( + ctx context.Context, + userCred mcclient.TokenCredential, + ownerId mcclient.IIdentityProvider, + query jsonutils.JSONObject, + data jsonutils.JSONObject, +) { + err := cluster.selfInitParams() + if err != nil { + log.Errorf("fail to generate cluster params: %s", err) + logclient.AddSimpleActionLog(cluster, logclient.ACT_CREATE, err, userCred, false) + } else { + logclient.AddSimpleActionLog(cluster, logclient.ACT_CREATE, cluster.Params, userCred, true) + } +} + +func (lbc *SLoadbalancerCluster) ValidateUpdateData( + ctx context.Context, + userCred mcclient.TokenCredential, + query jsonutils.JSONObject, + data *jsonutils.JSONDict, +) (*jsonutils.JSONDict, error) { wireV := validators.NewModelIdOrNameValidator("wire", "wire", lbc.GetOwnerId()) wireV.Optional(true) if err := wireV.Validate(data); err != nil { @@ -280,30 +316,57 @@ func (man *SLoadbalancerClusterManager) FindByZoneId(zoneId string) []SLoadbalan return r } -func (man *SLoadbalancerClusterManager) findByVrrpRouterIdInZone(zoneId string, routerId int) (*SLoadbalancerCluster, error) { - var r *SLoadbalancerCluster +func (man *SLoadbalancerClusterManager) findByVrrpRouterIdInZone(zoneId string, routerId int) ([]*SLoadbalancerCluster, error) { + r := make([]*SLoadbalancerCluster, 0) peerClusters := man.FindByZoneId(zoneId) for i := range peerClusters { peerCluster := &peerClusters[i] - peerClusterLbagents, err := man.getLoadbalancerAgents(peerCluster.Id) - if err != nil { - return nil, httperrors.NewGeneralError(err) - } - for j := range peerClusterLbagents { - peerClusterLbagent := &peerClusterLbagents[j] - if peerClusterLbagent.Params.Vrrp.VirtualRouterId == routerId { - if r != nil { - return nil, httperrors.NewInternalServerError("lbclusters %s(%s) and %s(%s) has conflict virtual_router_id: %d ", r.Name, r.Id, peerCluster.Name, peerCluster.Id, routerId) - } - r = peerCluster - break - } + if peerCluster.Params.VirtualRouterId == routerId { + r = append(r, peerCluster) } } return r, nil } +func (man *SLoadbalancerClusterManager) getVrrpRouterIdsInZone(zoneId string) ([]int, error) { + q := man.Query().IsNotEmpty("params").Equals("zone_id", zoneId) + clusters := make([]SLoadbalancerCluster, 0) + err := db.FetchModelObjects(man, q, &clusters) + if err != nil { + return nil, errors.Wrap(err, "initParams.FetchModelObjects") + } + ret := make([]int, 0) + for i := range clusters { + if clusters[i].Params.VirtualRouterId > 0 { + ret = append(ret, clusters[i].Params.VirtualRouterId) + } + } + return ret, nil +} + +func isInArray[K comparable](v K, arr []K) bool { + for i := range arr { + if arr[i] == v { + return true + } + } + return false +} + +func (man *SLoadbalancerClusterManager) newVrrpRouterIdsInZone(zoneId string) (int, error) { + idList, err := man.getVrrpRouterIdsInZone(zoneId) + if err != nil { + return -1, errors.Wrap(err, "getVrrpRouterIdsInZone") + } + for i := 17; i < 250; i++ { + if !isInArray(i, idList) { + return i, nil + } + } + return -1, errors.Wrapf(httperrors.ErrNotFound, "no available vrrp router id in zone %s", zoneId) +} + func (man *SLoadbalancerClusterManager) getLoadbalancerAgents(clusterId string) ([]SLoadbalancerAgent, error) { r := []SLoadbalancerAgent{} q := LoadbalancerAgentManager.Query().Equals("cluster_id", clusterId) @@ -391,6 +454,65 @@ func (man *SLoadbalancerClusterManager) InitializeData() error { } } + man.initParams() + + return nil +} + +func (man *SLoadbalancerClusterManager) initParams() error { + q := man.Query().IsNullOrEmpty("params") + clusters := make([]SLoadbalancerCluster, 0) + err := db.FetchModelObjects(man, q, &clusters) + if err != nil { + return errors.Wrap(err, "initParams.FetchModelObjects") + } + for i := range clusters { + err := clusters[i].initParams() + if err != nil { + return errors.Wrap(err, "cluster initParams") + } + } + return nil +} + +func (cluster *SLoadbalancerCluster) initParams() error { + lbagents, err := LoadbalancerClusterManager.getLoadbalancerAgents(cluster.Id) + if err != nil { + return errors.Wrap(err, "getLoadbalancerAgents") + } + if len(lbagents) == 0 { + // generate params + return cluster.selfInitParams() + } + _, err = db.Update(cluster, func() error { + cluster.Params = &lbagents[0].Params.Vrrp.SLoadbalancerClusterParams + return nil + }) + if err != nil { + return errors.Wrap(err, "UpdateWithLock") + } + return nil +} + +func (cluster *SLoadbalancerCluster) selfInitParams() error { + newRouterId, err := LoadbalancerClusterManager.newVrrpRouterIdsInZone(cluster.ZoneId) + if err != nil { + return errors.Wrap(err, "newVrrpRouterIdsInZone") + } + params := SLoadbalancerClusterParams{ + VirtualRouterId: newRouterId, + Preempt: false, + AdvertInt: 5, + Pass: seclib.RandomPassword(6), + GarpMasterRefresh: 29, + } + _, err = db.Update(cluster, func() error { + cluster.Params = ¶ms + return nil + }) + if err != nil { + return errors.Wrap(err, "UpdateWithLock") + } return nil } @@ -418,3 +540,55 @@ func (manager *SLoadbalancerClusterManager) ListItemExportKeys(ctx context.Conte } return q, nil } + +func (cluster *SLoadbalancerCluster) PerformParamsPatch(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) { + oldParams := cluster.Params + params := gotypes.DeepCopy(*cluster.Params).(SLoadbalancerClusterParams) + d := jsonutils.NewDict() + d.Set("params", data) + paramsV := validators.NewStructValidator("params", ¶ms) + if err := paramsV.Validate(d); err != nil { + return nil, err + } + // new vrrp virtual_router_id should be unique across clusters + if params.VirtualRouterId != oldParams.VirtualRouterId { + otherClusters, err := LoadbalancerClusterManager.findByVrrpRouterIdInZone(cluster.ZoneId, params.VirtualRouterId) + if err != nil { + return nil, errors.Wrap(err, "findByVrrpRouterIdInZone") + } + if len(otherClusters) > 0 { + return nil, httperrors.NewConflictError("lbcluster %s(%s) already has virtual_router_id %d", + otherClusters[0].Name, otherClusters[0].Id, params.VirtualRouterId) + } + } + { + diff, err := db.Update(cluster, func() error { + cluster.Params = ¶ms + return nil + }) + if err != nil { + return nil, errors.Wrap(err, "Update") + } + db.OpsLog.LogEvent(cluster, db.ACT_UPDATE, diff, userCred) + } + { + // populate changes to underlying lbagents + lbagents, err := LoadbalancerAgentManager.getByClusterId(cluster.Id) + if err != nil { + return nil, errors.Wrap(err, "getByClusterId") + } + for i := range lbagents { + lbagent := lbagents[i] + params := *lbagent.Params + params.Vrrp.SLoadbalancerClusterParams = *cluster.Params + _, err := db.Update(&lbagent, func() error { + lbagent.Params = ¶ms + return nil + }) + if err != nil { + return nil, errors.Wrap(err, "Update") + } + } + } + return nil, nil +} diff --git a/pkg/lbagent/api.go b/pkg/lbagent/api.go index ee58e1029c..0633156c5d 100644 --- a/pkg/lbagent/api.go +++ b/pkg/lbagent/api.go @@ -41,6 +41,8 @@ import ( type ApiHelper struct { opts *Options + lbagentId string + dataDirMan *agentutils.ConfigDirManager apih *apihelper.APIHelper corpus *agentmodels.LoadbalancerCorpus @@ -54,7 +56,7 @@ type ApiHelper struct { ovn *OvnWorker } -func NewApiHelper(opts *Options) (*ApiHelper, error) { +func NewApiHelper(opts *Options, lbagentId string) (*ApiHelper, error) { corpus := agentmodels.NewEmptyLoadbalancerCorpus() apiOpts := &apihelper.Options{ CommonOptions: opts.CommonOptions, @@ -69,6 +71,8 @@ func NewApiHelper(opts *Options) (*ApiHelper, error) { helper := &ApiHelper{ opts: opts, + lbagentId: lbagentId, + dataDirMan: agentutils.NewConfigDirManager(opts.apiDataStoreDir), apih: apih, corpus: corpus, @@ -178,7 +182,6 @@ func (h *ApiHelper) adminClientSession(ctx context.Context) *mcclient.ClientSess } region := h.opts.CommonOptions.Region - // apiVersion := "v2" h.mcclientSession = auth.GetAdminSession(ctx, region) return h.mcclientSession } @@ -187,7 +190,7 @@ func (h *ApiHelper) agentPeekOnce(ctx context.Context) (*computemodels.SLoadbala s := h.adminClientSession(ctx) params := jsonutils.NewDict() params.Set(api.LBAGENT_QUERY_ORIG_KEY, jsonutils.NewString(api.LBAGENT_QUERY_ORIG_VAL)) - data, err := modules.LoadbalancerAgents.Get(s, h.opts.ApiLbagentId, params) + data, err := modules.LoadbalancerAgents.Get(s, h.lbagentId, params) if err != nil { err := fmt.Errorf("agent get error: %s", err) return nil, err @@ -246,6 +249,7 @@ func (r *agentPeekResult) staleInFuture(s int) bool { return false } +/* func (h *ApiHelper) agentPeek(ctx context.Context) *agentPeekResult { doPeekWithLog := func() *computemodels.SLoadbalancerAgent { agent, err := h.agentPeekOnce(ctx) @@ -273,11 +277,12 @@ func (h *ApiHelper) agentPeek(ctx context.Context) *agentPeekResult { } return (*agentPeekResult)(agent) } +*/ func (h *ApiHelper) agentUpdateSeen(ctx context.Context) *computemodels.SLoadbalancerAgent { s := h.adminClientSession(ctx) params := h.corpus.MaxSeenUpdatedAtParams() - data, err := modules.LoadbalancerAgents.Update(s, h.opts.ApiLbagentId, params) + data, err := modules.LoadbalancerAgents.Update(s, h.lbagentId, params) if err != nil { log.Errorf("agent get error: %s", err) return nil @@ -317,7 +322,7 @@ func (h *ApiHelper) doHb(ctx context.Context) (*computemodels.SLoadbalancerAgent if err != nil { return nil, fmt.Errorf("heartbeat: making params: %s", err) } - data, err := modules.LoadbalancerAgents.PerformAction(s, h.opts.ApiLbagentId, "hb", params) + data, err := modules.LoadbalancerAgents.PerformAction(s, h.lbagentId, "hb", params) if err != nil { err := fmt.Errorf("heartbeat api error: %s", err) return nil, err diff --git a/pkg/lbagent/haproxy.go b/pkg/lbagent/haproxy.go index 12b7359141..71a658cf57 100644 --- a/pkg/lbagent/haproxy.go +++ b/pkg/lbagent/haproxy.go @@ -171,6 +171,7 @@ func (h *HaproxyHelper) handleUseCorpusCmd(ctx context.Context, cmd *LbagentCmd) LoadbalancersEnabled: genHaproxyConfigsResult.LoadbalancersEnabled, AgentParams: agentParams, } + log.Infof("GenKeepalivedConfigs %s", dir) err := corpus.GenKeepalivedConfigs(dir, opts) if err != nil { err = fmt.Errorf("generating keepalived config failed: %s", err) @@ -511,6 +512,8 @@ func (h *HaproxyHelper) reloadKeepalived(ctx context.Context) error { } func (h *HaproxyHelper) runCmd(args []string) error { + log.Debugf("run command %s", args) + name := args[0] args = args[1:] cmd := exec.Command(name, args...) diff --git a/pkg/lbagent/models/agentparams.go b/pkg/lbagent/models/agentparams.go index 5a0d25db82..53feff8113 100644 --- a/pkg/lbagent/models/agentparams.go +++ b/pkg/lbagent/models/agentparams.go @@ -20,32 +20,21 @@ import ( "reflect" "text/template" - "yunion.io/x/pkg/utils" + "yunion.io/x/jsonutils" + "yunion.io/x/log" compute_models "yunion.io/x/onecloud/pkg/compute/models" ) func dataFromParams(p interface{}) map[string]interface{} { - rv := reflect.ValueOf(p) - if rv.Kind() != reflect.Struct { - panic(fmt.Sprintf("unexpected kind: %#v", p)) + data := jsonutils.Marshal(p) + ret := make(map[string]interface{}) + err := data.Unmarshal(&ret) + if err != nil { + log.Errorf("unmarshal map[string]interface{} error %s", err) + return nil } - 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 + return ret } type AgentParams struct { @@ -79,6 +68,8 @@ func NewAgentParams(agent *compute_models.SLoadbalancerAgent) (*AgentParams, err "name": agent.Name, "ip": agent.IP, } + agent.Params.Vrrp.Interface = agent.Interface + agent.Params.Vrrp.Priority = agent.Priority data := map[string]map[string]interface{}{ "agent": dataAgent, "vrrp": dataFromParams(agent.Params.Vrrp), diff --git a/pkg/lbagent/models/keepalived.go b/pkg/lbagent/models/keepalived.go index 8a9e7d2f40..1a2e92922d 100644 --- a/pkg/lbagent/models/keepalived.go +++ b/pkg/lbagent/models/keepalived.go @@ -20,6 +20,9 @@ import ( "path/filepath" "text/template" + "yunion.io/x/jsonutils" + "yunion.io/x/log" + computeapi "yunion.io/x/onecloud/pkg/apis/compute" agentutils "yunion.io/x/onecloud/pkg/lbagent/utils" ) @@ -57,6 +60,8 @@ func (b *LoadbalancerCorpus) GenKeepalivedConfigs(dir string, opts *GenKeepalive if err != nil { return err } + log.Debugf("data: %s", jsonutils.Marshal(agentParams.Data)) + log.Debugf("output: %s", buf.String()) } { // write keepalived.conf diff --git a/pkg/lbagent/options.go b/pkg/lbagent/options.go index cc4b035889..52880dd1b4 100644 --- a/pkg/lbagent/options.go +++ b/pkg/lbagent/options.go @@ -23,10 +23,12 @@ import ( agentutils "yunion.io/x/onecloud/pkg/lbagent/utils" ) -type LbagentOptions struct { - ApiLbagentId string `require:"true"` - ApiLbagentHbInterval int `default:"10"` - ApiLbagentHbTimeoutRelaxation int `default:"120" help:"If agent is to stale out in specified seconds in the future, consider it staled to avoid race condition when doing incremental api data fetch"` +type LbagentCommonOptions struct { + common_options.CommonOptions + + // ApiLbagentId string `require:"true"` + ApiLbagentHbInterval int `default:"10"` + ApiLbagentHbTimeoutRelaxation int `default:"120" help:"If agent is to stale out in specified seconds in the future, consider it staled to avoid race condition when doing incremental api data fetch"` ApiSyncIntervalSeconds int `default:"10"` ApiRunDelayMilliseconds int `default:"10"` @@ -49,9 +51,12 @@ type LbagentOptions struct { } type Options struct { - common_options.CommonOptions + LbagentCommonOptions - LbagentOptions + CommonConfigFile string `help:"common config file for container"` + + ListenInterface string `help:"listening interface of lbagent" default:"eth0"` + AccessIp string `help:"access ip of lbagent, if there are multiple IPs on listen interface"` } func (opts *Options) ValidateThenInit() error { diff --git a/pkg/lbagent/ovn.go b/pkg/lbagent/ovn.go index c20dbf8cab..5575647131 100644 --- a/pkg/lbagent/ovn.go +++ b/pkg/lbagent/ovn.go @@ -183,8 +183,10 @@ func (ovnHost *OvnHost) cleanUpBridge() { "ovs-vsctl", "--", "--if-exists", "del-port", bridge, peer0, } + var cancelFunc context.CancelFunc ctx := context.Background() - ctx, _ = context.WithTimeout(ctx, time.Second*7) + ctx, cancelFunc = context.WithTimeout(ctx, time.Second*7) + defer cancelFunc() cmd := exec.CommandContext(ctx, args[0], args[1:]...) _, err := cmd.Output() if err != nil { diff --git a/pkg/lbagent/register.go b/pkg/lbagent/register.go new file mode 100644 index 0000000000..f70674df0c --- /dev/null +++ b/pkg/lbagent/register.go @@ -0,0 +1,89 @@ +// 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" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" + "yunion.io/x/pkg/util/version" + + api "yunion.io/x/onecloud/pkg/apis/compute" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/mcclient/auth" + modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute" + "yunion.io/x/onecloud/pkg/util/netutils2" +) + +func register(ctx context.Context, opts *Options) (string, error) { + _, ips, err := netutils2.WaitIfaceIps(opts.ListenInterface) + if err != nil { + return "", errors.Wrap(err, "netutils2.WaitIfaceIps") + } + + if len(ips) == 0 { + return "", errors.Wrapf(httperrors.ErrNotFound, "no valid ip on interface %s", opts.ListenInterface) + } + + if len(ips) > 1 && len(opts.AccessIp) == 0 { + return "", errors.Wrap(httperrors.ErrDuplicateResource, "multiple IPs, must specified a valid access IP") + } + + if len(opts.AccessIp) == 0 { + opts.AccessIp = ips[0].String() + } else { + find := false + for i := range ips { + if ips[i].String() == opts.AccessIp { + find = true + break + } + } + if !find { + return "", errors.Wrapf(httperrors.ErrConflict, "access IP %s not present on interface %s", opts.AccessIp, opts.ListenInterface) + } + } + + s := auth.GetAdminSession(ctx, opts.Region) + params := jsonutils.NewDict() + params.Set(api.LBAGENT_QUERY_ORIG_KEY, jsonutils.NewString(api.LBAGENT_QUERY_ORIG_VAL)) + params.Set("ip", jsonutils.NewString(opts.AccessIp)) + results, err := modules.LoadbalancerAgents.List(s, params) + if err != nil { + return "", errors.Wrap(err, "LoadbalancerAgents.List") + } + if len(results.Data) > 1 { + // multiple lbagent with ident IP? conflict + return "", errors.Wrapf(httperrors.ErrDuplicateResource, "multiple lbagent with same IP %s", opts.AccessIp) + } + if len(results.Data) == 0 { + // not found, to create a new one + createParams := jsonutils.NewDict() + createParams.Set("generate_name", jsonutils.NewString("lbagent")) + createParams.Set("ip", jsonutils.NewString(opts.AccessIp)) + createParams.Set("interface", jsonutils.NewString(opts.ListenInterface)) + createParams.Set("version", jsonutils.NewString(version.Get().GitVersion)) + data, err := modules.LoadbalancerAgents.Create(s, createParams) + if err != nil { + return "", errors.Wrap(err, "LoadbalancerAgents.Create") + } + results.Data = []jsonutils.JSONObject{data} + } + idstr, _ := results.Data[0].GetString("id") + log.Infof("register lbagent with ID %s", idstr) + return idstr, nil +} diff --git a/pkg/lbagent/service.go b/pkg/lbagent/service.go index c78d4b8ac3..1db38b498f 100644 --- a/pkg/lbagent/service.go +++ b/pkg/lbagent/service.go @@ -26,6 +26,7 @@ import ( app_common "yunion.io/x/onecloud/pkg/cloudcommon/app" common_options "yunion.io/x/onecloud/pkg/cloudcommon/options" + "yunion.io/x/onecloud/pkg/util/fileutils2" ) func StartService() { @@ -33,6 +34,16 @@ func StartService() { commonOpts := &opts.CommonOptions { common_options.ParseOptions(opts, os.Args, "lbagent.conf", "lbagent") + if len(opts.CommonConfigFile) > 0 && fileutils2.Exists(opts.CommonConfigFile) { + log.Infof("read common config file: %s", opts.CommonConfigFile) + commonCfg := &LbagentCommonOptions{} + commonCfg.Config = opts.CommonConfigFile + common_options.ParseOptions(commonCfg, []string{os.Args[0]}, "common.conf", "lbagent") + baseOpt := opts.BaseOptions.BaseOptions + opts.LbagentCommonOptions = *commonCfg + // keep base options + opts.BaseOptions.BaseOptions = baseOpt + } app_common.InitAuth(commonOpts, func() { log.Infof("auth finished ok") }) @@ -41,10 +52,17 @@ func StartService() { log.Fatalf("opts validate: %s", err) } + // register lbagent + ctx := context.Background() + ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_APPNAME, "lbagent") + lbagentId, err := register(ctx, opts) + if err != nil { + log.Fatalf("register lbagent failed: %s", err) + } + var haproxyHelper *HaproxyHelper var apiHelper *ApiHelper var haStateWatcher *HaStateWatcher - var err error { haStateWatcher, err = NewHaStateWatcher(opts) if err != nil { @@ -58,7 +76,7 @@ func StartService() { } } { - apiHelper, err = NewApiHelper(opts) + apiHelper, err = NewApiHelper(opts, lbagentId) if err != nil { log.Fatalf("init api helper failed: %s", err) } @@ -68,10 +86,10 @@ func StartService() { { wg := &sync.WaitGroup{} cmdChan := make(chan *LbagentCmd) // internal - ctx, cancelFunc := context.WithCancel(context.Background()) + var cancelFunc context.CancelFunc + ctx, cancelFunc = context.WithCancel(ctx) ctx = context.WithValue(ctx, "wg", wg) ctx = context.WithValue(ctx, "cmdChan", cmdChan) - ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_APPNAME, "lbagent") wg.Add(3) go haStateWatcher.Run(ctx) go haproxyHelper.Run(ctx) diff --git a/pkg/mcclient/modules/compute/mod_loadbalanceragents.go b/pkg/mcclient/modules/compute/mod_loadbalanceragents.go index e951622f3d..5299e1102c 100644 --- a/pkg/mcclient/modules/compute/mod_loadbalanceragents.go +++ b/pkg/mcclient/modules/compute/mod_loadbalanceragents.go @@ -35,6 +35,8 @@ func init() { []string{ "id", "name", + "cluster", + "cluster_id", "hb_last_seen", "hb_timeout", diff --git a/pkg/mcclient/options/compute/loadbalanceragents.go b/pkg/mcclient/options/compute/loadbalanceragents.go index a58f1056eb..cff84fc23b 100644 --- a/pkg/mcclient/options/compute/loadbalanceragents.go +++ b/pkg/mcclient/options/compute/loadbalanceragents.go @@ -182,3 +182,23 @@ type LoadbalancerAgentActionUndeployOptions struct { type LoadbalancerAgentDefaultParamsOptions struct { Cluster string } + +type LoadbalancerAgentJoinClusterOptions struct { + ID string `json:"-"` + + compute_apis.LoadbalancerAgentJoinClusterInput +} + +func (opts *LoadbalancerAgentJoinClusterOptions) Params() (*jsonutils.JSONDict, error) { + return jsonutils.Marshal(opts).(*jsonutils.JSONDict), nil +} + +type LoadbalancerAgentLeaveClusterOptions struct { + ID string `json:"-"` + + compute_apis.LoadbalancerAgentLeaveClusterInput +} + +func (opts *LoadbalancerAgentLeaveClusterOptions) Params() (*jsonutils.JSONDict, error) { + return jsonutils.Marshal(opts).(*jsonutils.JSONDict), nil +} diff --git a/pkg/util/netutils2/iface.go b/pkg/util/netutils2/iface.go new file mode 100644 index 0000000000..5398ed50ca --- /dev/null +++ b/pkg/util/netutils2/iface.go @@ -0,0 +1,61 @@ +// 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 netutils2 + +import ( + "net" + "time" + + "yunion.io/x/pkg/errors" +) + +func getIfaceIPs(iface *net.Interface) ([]net.IP, error) { + addrs, err := iface.Addrs() + if err != nil { + return nil, errors.Wrap(err, "iface.Addrs") + } + ips := make([]net.IP, 0) + for _, a := range addrs { + if ipnet, ok := a.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { + if ipnet.IP.To4() != nil { + ips = append(ips, ipnet.IP) + } + } + } + return ips, nil +} + +func WaitIfaceIps(ifname string) (*net.Interface, []net.IP, error) { + iface, err := net.InterfaceByName(ifname) + if err != nil { + return nil, nil, errors.Wrap(err, "net.InterfaceByName") + } + var ips []net.IP + MAX := 60 + wait := 0 + for wait < MAX { + ips, err = getIfaceIPs(iface) + if err != nil { + return nil, nil, errors.Wrap(err, "getIfaceIPs") + } + if len(ips) == 0 { + time.Sleep(2 * time.Second) + wait += 2 + } else { + break + } + } + return iface, ips, nil +}