Skip to content

Commit

Permalink
Merge pull request #645 from TrekkieCoder/main
Browse files Browse the repository at this point in the history 
PR : Support for LB end-point selection persistence with timeout
  • Loading branch information
@UltraInstinct14
UltraInstinct14 authored Apr 21, 2024
2 parents 95eaa93 + b6fa16c commit 82eac5e
Show file tree
Hide file tree
Showing 6 changed files with 48 additions and 28 deletions.
44 changes: 23 additions & 21 deletions api/loxinlp/ipvs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,12 +45,13 @@ type ipvsEndPoint struct {
}

type ipVSEntry struct {
Key ipVSKey
key ipVSKey
sel cmn.EpSelect
mode cmn.LBMode
pType string
InValid bool
EndPoints []ipvsEndPoint
timeout uint32
inValid bool
endPoints []ipvsEndPoint
}

type IPVSH struct {
Expand Down Expand Up @@ -85,6 +86,7 @@ func (ctx *IPVSH) buildIPVSDB() []*ipVSEntry {

newEntry.sel = cmn.LbSelRr
newEntry.pType = ""
newEntry.timeout = svc.Timeout
if svc.Flags&0x1 == 0x1 {
newEntry.sel = cmn.LbSelRrPersist
}
Expand All @@ -110,18 +112,18 @@ func (ctx *IPVSH) buildIPVSDB() []*ipVSEntry {

key := ipVSKey{Address: svc.Address.String(), Protocol: proto, Port: svc.Port}
for _, endPoint := range endPoints {
newEntry.EndPoints = append(newEntry.EndPoints, ipvsEndPoint{EpIP: endPoint.Address.String(), EpPort: endPoint.Port, Weight: uint8(endPoint.Weight)})
newEntry.endPoints = append(newEntry.endPoints, ipvsEndPoint{EpIP: endPoint.Address.String(), EpPort: endPoint.Port, Weight: uint8(endPoint.Weight)})
}

if len(newEntry.EndPoints) != 0 {
if len(newEntry.endPoints) != 0 {
if eEnt := ctx.RMap[key]; eEnt != nil {
if reflect.DeepEqual(eEnt.EndPoints, newEntry.EndPoints) {
eEnt.InValid = false
if reflect.DeepEqual(eEnt.endPoints, newEntry.endPoints) {
eEnt.inValid = false
continue
}
}

newEntry.Key = key
newEntry.key = key
ipVSList = append(ipVSList, &newEntry)
}
}
Expand All @@ -136,38 +138,38 @@ func IPVSSync() {
case <-ipVSCtx.ticker.C:

for _, ent := range ipVSCtx.RMap {
ent.InValid = true
ent.inValid = true
}

ipVSList := ipVSCtx.buildIPVSDB()

for _, ent := range ipVSCtx.RMap {
if ent.InValid {
name := fmt.Sprintf("ipvs_%s:%d-%s", ent.Key.Address, ent.Key.Port, ent.Key.Protocol)
lbrule := cmn.LbRuleMod{Serv: cmn.LbServiceArg{ServIP: ent.Key.Address, ServPort: ent.Key.Port, Proto: ent.Key.Protocol, Sel: ent.sel, Mode: ent.mode, Name: name, ProbeType: ent.pType}}
if ent.inValid {
name := fmt.Sprintf("ipvs_%s:%d-%s", ent.key.Address, ent.key.Port, ent.key.Protocol)
lbrule := cmn.LbRuleMod{Serv: cmn.LbServiceArg{ServIP: ent.key.Address, ServPort: ent.key.Port, Proto: ent.key.Protocol, Sel: ent.sel, Mode: ent.mode, Name: name, ProbeType: ent.pType}}
_, err := hooks.NetLbRuleDel(&lbrule)
if err != nil {
tk.LogIt(tk.LogError, "IPVS LB %v delete failed\n", ent.Key)
tk.LogIt(tk.LogError, "IPVS LB %v delete failed\n", ent.key)
}
tk.LogIt(tk.LogInfo, "IPVS ent %v deleted\n", ent.Key)
delete(ipVSCtx.RMap, ent.Key)
tk.LogIt(tk.LogInfo, "IPVS ent %v deleted\n", ent.key)
delete(ipVSCtx.RMap, ent.key)
}
}

for _, newEnt := range ipVSList {
name := fmt.Sprintf("ipvs_%s:%d-%s", newEnt.Key.Address, newEnt.Key.Port, newEnt.Key.Protocol)
lbrule := cmn.LbRuleMod{Serv: cmn.LbServiceArg{ServIP: newEnt.Key.Address, ServPort: newEnt.Key.Port, Proto: newEnt.Key.Protocol, Sel: newEnt.sel, Mode: newEnt.mode, Name: name, ProbeType: newEnt.pType}}
for _, ep := range newEnt.EndPoints {
name := fmt.Sprintf("ipvs_%s:%d-%s", newEnt.key.Address, newEnt.key.Port, newEnt.key.Protocol)
lbrule := cmn.LbRuleMod{Serv: cmn.LbServiceArg{ServIP: newEnt.key.Address, ServPort: newEnt.key.Port, Proto: newEnt.key.Protocol, Sel: newEnt.sel, Mode: newEnt.mode, Name: name, ProbeType: newEnt.pType, PersistTimeout: newEnt.timeout}}
for _, ep := range newEnt.endPoints {
lbrule.Eps = append(lbrule.Eps, cmn.LbEndPointArg{EpIP: ep.EpIP, EpPort: ep.EpPort, Weight: 1})
}

_, err := hooks.NetLbRuleAdd(&lbrule)
if err != nil {
tk.LogIt(tk.LogError, "IPVS LB %v add failed\n", newEnt.Key)
tk.LogIt(tk.LogError, "IPVS LB %v add failed\n", newEnt.key)
continue
}
ipVSCtx.RMap[newEnt.Key] = newEnt
tk.LogIt(tk.LogError, "IPVS ent %v added\n", newEnt.Key)
ipVSCtx.RMap[newEnt.key] = newEnt
tk.LogIt(tk.LogError, "IPVS ent %v added\n", newEnt.key)
}
}
}
Expand Down
2 changes: 2 additions & 0 deletions common/common.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -527,6 +527,8 @@ type LbServiceArg struct {
ProbeRetries int `json:"probeRetries"`
// Name - Service name
Name string `json:"name"`
// PersistTimeout - Persistence timeout in seconds
PersistTimeout uint32 `json:"persistTimeout"`
}

// LbEndPointArg - Information related to load-balancer end-point
Expand Down
2 changes: 1 addition & 1 deletion loxilb-ebpf
Submodule loxilb-ebpf updated 2 files
+3 −0 common/llb_dpapi.h
+18 −2 kernel/llb_kern_natlbfwd.c
1 change: 1 addition & 0 deletions pkg/loxinet/dpbroker.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -288,6 +288,7 @@ type NatDpWorkQ struct {
NatType NatT
EpSel NatSel
InActTo uint64
PersistTo uint64
endPoints []NatEP
secIP []net.IP
}
Expand Down
2 changes: 2 additions & 0 deletions pkg/loxinet/dpebpf_linux.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -940,6 +940,8 @@ func DpNatLbRuleMod(w *NatDpWorkQ) int {

// seconds to nanoseconds
dat.ito = C.uint64_t(w.InActTo * 1000000000)
dat.pto = C.uint64_t(w.PersistTo * 1000000000)
dat.base_to = 0

/*dat.npmhh = 2
dat.pmhh[0] = 0x64646464
Expand Down
25 changes: 19 additions & 6 deletions pkg/loxinet/rules.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ const (
EndPointCheckerDuration = 2 // Duration at which ep-helpers will run
MaxEndPointSweeps = 20 // Maximum end-point sweeps per round
VIPSweepDuration = 30 // Duration of periodic VIP maintenance
DefaultPersistTimeOut = 10800 // Default persistent LB session timeout
)

type ruleTType uint
Expand Down Expand Up @@ -275,7 +276,8 @@ type ruleEnt struct {
bgp bool
addrRslv bool
sT time.Time
iTo uint32
iTO uint32
pTO uint32
act ruleAct
secIP []ruleNatSIP
stat ruleStat
Expand Down Expand Up @@ -759,7 +761,7 @@ func (R *RuleH) GetNatLbRule() ([]cmn.LbRuleMod, error) {
ret.Serv.Sel = data.act.action.(*ruleNatActs).sel
ret.Serv.Mode = data.act.action.(*ruleNatActs).mode
ret.Serv.Monitor = data.hChk.actChk
ret.Serv.InactiveTimeout = data.iTo
ret.Serv.InactiveTimeout = data.iTO
ret.Serv.Bgp = data.bgp
ret.Serv.BlockNum = data.tuples.pref
ret.Serv.Managed = data.managed
Expand Down Expand Up @@ -1402,7 +1404,8 @@ func (R *RuleH) AddNatLbRule(serv cmn.LbServiceArg, servSecIPs []cmn.LbSecIPArg,
}

if eRule.hChk.prbType != serv.ProbeType || eRule.hChk.prbPort != serv.ProbePort ||
eRule.hChk.prbReq != serv.ProbeReq || eRule.hChk.prbResp != serv.ProbeResp {
eRule.hChk.prbReq != serv.ProbeReq || eRule.hChk.prbResp != serv.ProbeResp ||
eRule.pTO != serv.PersistTimeout {
ruleChg = true
}

Expand All @@ -1417,6 +1420,7 @@ func (R *RuleH) AddNatLbRule(serv cmn.LbServiceArg, servSecIPs []cmn.LbSecIPArg,
eRule.hChk.prbResp = serv.ProbeResp
eRule.hChk.prbRetries = serv.ProbeRetries
eRule.hChk.prbTimeo = serv.ProbeTimeout
eRule.pTO = serv.PersistTimeout
eRule.act.action.(*ruleNatActs).sel = natActs.sel
eRule.act.action.(*ruleNatActs).endPoints = eEps
eRule.act.action.(*ruleNatActs).mode = natActs.mode
Expand All @@ -1427,7 +1431,7 @@ func (R *RuleH) AddNatLbRule(serv cmn.LbServiceArg, servSecIPs []cmn.LbSecIPArg,
R.electEPSrc(eRule)

eRule.sT = time.Now()
eRule.iTo = serv.InactiveTimeout
eRule.iTO = serv.InactiveTimeout
tk.LogIt(tk.LogDebug, "nat lb-rule updated - %s:%s\n", eRule.tuples.String(), eRule.act.String())
eRule.DP(DpCreate)

Expand Down Expand Up @@ -1462,9 +1466,17 @@ func (R *RuleH) AddNatLbRule(serv cmn.LbServiceArg, servSecIPs []cmn.LbSecIPArg,
return RuleAllocErr, errors.New("rule-hwm error")
}
r.sT = time.Now()
r.iTo = serv.InactiveTimeout
r.iTO = serv.InactiveTimeout
r.bgp = serv.Bgp
r.ci = cmn.CIDefault
r.pTO = 0
if serv.Sel == cmn.LbSelRrPersist {
if serv.PersistTimeout == 0 || serv.PersistTimeout > 24*60*60 {
r.pTO = DefaultPersistTimeOut
} else {
r.pTO = serv.PersistTimeout
}
}

R.foldRecursiveEPs(r)

Expand Down Expand Up @@ -2350,7 +2362,8 @@ func (r *ruleEnt) Nat2DP(work DpWorkT) int {
nWork.Proto = r.tuples.l4Prot.val
nWork.Mark = int(r.ruleNum)
nWork.BlockNum = r.tuples.pref
nWork.InActTo = uint64(r.iTo)
nWork.InActTo = uint64(r.iTO)
nWork.PersistTo = uint64(r.pTO)

if r.act.actType == RtActDnat {
nWork.NatType = DpDnat
Expand Down

0 comments on commit 82eac5e

Please sign in to comment.