apps/openlan
1
0
Fork 0
mirror of https://github.com/luscis/openlan.git synced 2025-10-14 04:43:52 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: change nextgroup to findhop.

Browse Source
This commit is contained in:
Daniel Ding
2024-07-19 14:21:01 +08:00
parent 0b5ea13af0
commit b8c64bb822
11 changed files with 554 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
dist/rootfs/etc/openlan/switch/nextgroup/nextgroup.json.example → dist/rootfs/etc/openlan/switch/findhop/findhop.example vendored
View File

@@ -1,7 +1,7 @@
{
"ng1": {
"hop1": {
"check": "ping",
"ping": {
"params": {
"count": 5,
"loss": 2
},

4
dist/rootfs/etc/openlan/switch/network/nextgroup.json.example → dist/rootfs/etc/openlan/switch/network/findhop.json.example vendored
View File

@@ -11,8 +11,8 @@
},
"routes": [
{
"prefix": "172.32.10.0/24"
"nextgroup": "ng1"
"prefix": "172.32.10.0/24",
"findhop": "hop1"
}
],
"openvpn": {

38
pkg/config/findhop.go Normal file
View File

@@ -0,0 +1,38 @@
package config
type FindHop struct {
Check string `json:"check"`
Params PingParams `json:"params"`
Mode string `json:"mode,omitempty"`
NextHop []string `json:"nexthop"`
Available []MultiPath `json:"available,omitempty"`
}
func (ng *FindHop) Correct() {
if ng.Available == nil {
ng.Available = []MultiPath{}
}
if ng.Check == "" {
ng.Check = "ping"
}
if ng.Mode == "" {
ng.Mode = "active-backup"
}
ng.Params.Correct()
}
type PingParams struct {
Count int `json:"count"`
Loss int `json:"loss,omitempty"`
Rtt int `json:"rtt,omitempty"`
Interval int `json:"interval,omitempty"`
}
func (pp *PingParams) Correct() {
if pp.Count == 0 {
pp.Count = 3
}
if pp.Loss == 0 {
pp.Loss = 2
}
}

82
pkg/config/network.go
View File

@@ -8,25 +8,25 @@ import (
)
type Network struct {
ConfDir string `json:"-"`
File string `json:"file"`
Alias string `json:"-"`
Name string `json:"name"`
Provider string `json:"provider,omitempty"`
Bridge *Bridge `json:"bridge,omitempty"`
Subnet *Subnet `json:"subnet,omitempty"`
OpenVPN *OpenVPN `json:"openvpn,omitempty"`
Links []Point `json:"links,omitempty"`
Hosts []HostLease `json:"hosts,omitempty"`
Routes []PrefixRoute `json:"routes,omitempty"`
Acl string `json:"acl,omitempty"`
Specifies interface{} `json:"specifies,omitempty"`
Dhcp string `json:"dhcp,omitempty"`
Outputs []*Output `json:"outputs,omitempty"`
ZTrust string `json:"ztrust,omitempty"`
Qos string `json:"qos,omitempty"`
Namespace string `json:"namespace,omitempty"`
NextGroup map[string]NextGroup `json:"nextgroup,omitempty"`
ConfDir string `json:"-"`
File string `json:"file"`
Alias string `json:"-"`
Name string `json:"name"`
Provider string `json:"provider,omitempty"`
Bridge *Bridge `json:"bridge,omitempty"`
Subnet *Subnet `json:"subnet,omitempty"`
OpenVPN *OpenVPN `json:"openvpn,omitempty"`
Links []Point `json:"links,omitempty"`
Hosts []HostLease `json:"hosts,omitempty"`
Routes []PrefixRoute `json:"routes,omitempty"`
Acl string `json:"acl,omitempty"`
Specifies interface{} `json:"specifies,omitempty"`
Dhcp string `json:"dhcp,omitempty"`
Outputs []*Output `json:"outputs,omitempty"`
ZTrust string `json:"ztrust,omitempty"`
Qos string `json:"qos,omitempty"`
Namespace string `json:"namespace,omitempty"`
FindHop map[string]*FindHop `json:"findhop,omitempty"`
}
func (n *Network) NewSpecifies() interface{} {
@@ -82,9 +82,9 @@ func (n *Network) Correct(sw *Switch) {
n.OpenVPN.Correct(sw)
}
for key, value := range n.NextGroup {
for key, value := range n.FindHop {
value.Correct()
n.NextGroup[key] = value
n.FindHop[key] = value
}
}
@@ -95,37 +95,29 @@ func (n *Network) Dir(elem ...string) string {
func (n *Network) LoadLink() {
file := n.Dir("link", n.Name+".json")
if err := libol.FileExist(file); err == nil {
if err := libol.UnmarshalLoad(&n.Links, file); err != nil {
libol.Error("Network.LoadLink... %n", err)
}
if err := libol.UnmarshalLoad(&n.Links, file); err != nil {
libol.Error("Network.LoadLink... %n", err)
}
}
func (n *Network) LoadRoute() {
file := n.Dir("route", n.Name+".json")
if err := libol.FileExist(file); err == nil {
if err := libol.UnmarshalLoad(&n.Routes, file); err != nil {
libol.Error("Network.LoadRoute... %n", err)
}
if err := libol.UnmarshalLoad(&n.Routes, file); err != nil {
libol.Error("Network.LoadRoute... %n", err)
}
}
func (n *Network) LoadOutput() {
file := n.Dir("output", n.Name+".json")
if err := libol.FileExist(file); err == nil {
if err := libol.UnmarshalLoad(&n.Outputs, file); err != nil {
libol.Error("Network.LoadOutput... %n", err)
}
if err := libol.UnmarshalLoad(&n.Outputs, file); err != nil {
libol.Error("Network.LoadOutput... %n", err)
}
}
func (n *Network) LoadNextGroup() {
file := n.Dir("nextgroup", n.Name+".json")
if err := libol.FileExist(file); err == nil {
if err := libol.UnmarshalLoad(&n.NextGroup, file); err != nil {
libol.Error("Network.LoadNextGroup... %n", err)
}
func (n *Network) LoadFindHop() {
file := n.Dir("findhop", n.Name+".json")
if err := libol.UnmarshalLoad(&n.FindHop, file); err != nil {
libol.Error("Network.LoadFindHop... %n", err)
}
}
@@ -140,7 +132,7 @@ func (n *Network) Save() {
n.SaveRoute()
n.SaveLink()
n.SaveOutput()
n.SaveNextGroup()
n.SaveFindHop()
}
func (n *Network) SaveRoute() {
@@ -173,13 +165,13 @@ func (n *Network) SaveOutput() {
}
}
func (n *Network) SaveNextGroup() {
file := n.Dir("nextgroup", n.Name+".json")
if n.NextGroup == nil {
func (n *Network) SaveFindHop() {
file := n.Dir("findhop", n.Name+".json")
if n.FindHop == nil {
return
}
if err := libol.MarshalSave(n.NextGroup, file, true); err != nil {
libol.Error("Network.SaveNextGroup %s %s", n.Name, err)
if err := libol.MarshalSave(n.FindHop, file, true); err != nil {
libol.Error("Network.SaveFindHop %s %s", n.Name, err)
}
}

23
pkg/config/nextgroup.go
View File

@@ -1,23 +0,0 @@
package config
type NextGroup struct {
Check string `json:"check"`
Ping PingParams `json:"ping"`
Mode string `json:"mode,omitempty"`
NextHop []string `json:"nexthop"`
AvailableNextHop []MultiPath `json:"availableNexthop,omitempty"`
}
func (ng *NextGroup) Correct() {
if ng.AvailableNextHop == nil {
ng.AvailableNextHop = []MultiPath{}
}
}
type PingParams struct {
Count int `json:"count"`
Loss int `json:"loss,omitempty"`
Rtt int `json:"rtt,omitempty"`
CheckFrequency int `json:"checkFrequency,omitempty"`
}

3
pkg/config/subnet.go
View File

@@ -30,7 +30,7 @@ type PrefixRoute struct {
MultiPath []MultiPath `json:"multipath,omitempty"`
Metric int `json:"metric"`
Mode string `json:"forward,omitempty"` // route or snat
NextGroup string `json:"nextgroup,omitempty"`
FindHop string `json:"findhop,omitempty"`
}
func (r *PrefixRoute) String() string {
@@ -60,7 +60,6 @@ func (r *PrefixRoute) CorrectRoute(nexthop string) {
if r.Mode == "" {
r.Mode = "snat"
}
}
func CorrectRoutes(routes []PrefixRoute, nexthop string) {

2
pkg/config/switch.go
View File

@@ -189,7 +189,7 @@ func (s *Switch) LoadNetworkWithData(data []byte) (*Network, error) {
obj.LoadLink()
obj.LoadRoute()
obj.LoadOutput()
obj.LoadNextGroup()
obj.LoadFindHop()
s.Network[obj.Name] = obj
return obj, nil
}

2
pkg/libol/utils.go
View File

@@ -155,7 +155,7 @@ func Unmarshal(v interface{}, contents []byte) error {
func UnmarshalLoad(v interface{}, file string) error {
if err := FileExist(file); err != nil {
return NewErr("%s %s", file, err)
return nil
}
contents, err := LoadWithoutAnn(file)
if err != nil {

446
pkg/switch/findhop.go Normal file
View File

@@ -0,0 +1,446 @@
package cswitch
import (
"fmt"
"net"
"os/exec"
"regexp"
"sort"
"strconv"
"time"
"github.com/luscis/openlan/pkg/cache"
co "github.com/luscis/openlan/pkg/config"
"github.com/luscis/openlan/pkg/libol"
"github.com/luscis/openlan/pkg/models"
nl "github.com/vishvananda/netlink"
)
type FindHop struct {
Network string
cfg map[string]*co.FindHop
drivers map[string]FindHopDriver
out *libol.SubLogger
}
func NewFindHop(network string, cfg map[string]*co.FindHop) *FindHop {
drivers := make(map[string]FindHopDriver, 32)
for key, ng := range cfg {
drv := newCheckDriver(key, network, ng)
if drv != nil {
drivers[key] = drv
}
}
return &FindHop{
Network: network,
cfg: cfg,
drivers: drivers,
out: libol.NewSubLogger("findhop"),
}
}
func newCheckDriver(name string, network string, cfg *co.FindHop) FindHopDriver {
switch cfg.Check {
case "ping":
return NewPingDriver(name, network, cfg)
}
return nil
}
func (ng *FindHop) Start() {
ng.out.Info("FindHop.Start: findhop, drivers size: %d", len(ng.drivers))
if len(ng.drivers) > 0 {
for _, checker := range ng.drivers {
checker.Start()
}
}
}
func (ng *FindHop) Stop() {
if len(ng.drivers) > 0 {
for _, driver := range ng.drivers {
driver.Stop()
}
}
}
// for add findhop dynamicly
func (ng *FindHop) AddFindHop(name string, cfg *co.FindHop) {
if _, ok := ng.drivers[name]; ok {
ng.out.Error("FindHop.addFindHop: checker already exists %s", name)
return
}
driver := newCheckDriver(name, ng.Network, cfg)
if driver != nil {
ng.drivers[name] = driver
} else {
ng.out.Error("FindHop.AddFindHop: don't support this driver %s", name)
}
driver.Start()
}
// for del findhop dynamicly
func (ng *FindHop) DelFindHop(name string, cfg co.FindHop) {
if driver, ok := ng.drivers[name]; !ok {
ng.out.Error("FindHop.addFindHop: checker not exists %s", name)
return
} else {
if driver.HasRoute() {
ng.out.Error("FindHop.delFindHop: checker has route %s", name)
return
}
driver.Stop()
delete(ng.drivers, name)
}
}
func (ng *FindHop) LoadRoute(findhop string, nlr *nl.Route) {
if driver, ok := ng.drivers[findhop]; ok {
ng.out.Debug("FindHop.loadRoute: %v", nlr)
driver.LoadRoute(nlr)
} else {
ng.out.Error("FindHop.loadRoute: checker not found %s", findhop)
}
}
func (ng *FindHop) UnloadRoute(findhop string, nlr *nl.Route) {
if driver, ok := ng.drivers[findhop]; ok {
ng.out.Debug("FindHop.unloadRoute: %v", nlr)
driver.UnloadRoute(nlr)
} else {
ng.out.Error("FindHop.unloadRoute: checker not found %s", findhop)
}
}
type FindHopDriver interface {
Name() string
Check([]string) []co.MultiPath
Start()
Stop()
ReloadRoute()
LoadRoute(route *nl.Route)
UnloadRoute(route *nl.Route)
HasRoute() bool
}
type FindHopDriverImpl struct {
Network string
routes []*nl.Route
cfg *co.FindHop
out *libol.SubLogger
}
func (c *FindHopDriverImpl) Name() string {
return "common"
}
func (c *FindHopDriverImpl) Start() {
}
func (c *FindHopDriverImpl) Stop() {
}
func (c *FindHopDriverImpl) HasRoute() bool {
return len(c.routes) > 0
}
func (c *FindHopDriverImpl) Check(ipList []string) []co.MultiPath {
return nil
}
func (c *FindHopDriverImpl) UpdateAvailable(mp []co.MultiPath) bool {
if c.cfg.Mode == "load-balance" {
if !compareMultiPaths(mp, c.cfg.Available) {
c.cfg.Available = mp
c.out.Info("FindHopDriverImpl.UpdateAvailable: available %v", c.cfg.Available)
return true
}
} else {
var newPath co.MultiPath
if len(mp) > 0 {
newPath = mp[0]
}
var oldPath co.MultiPath
if len(c.cfg.Available) > 0 {
oldPath = c.cfg.Available[0]
}
if !newPath.CompareEqual(oldPath) {
c.cfg.Available = []co.MultiPath{newPath}
c.out.Info("FindHopDriverImpl.UpdateAvailable: available %v", c.cfg.Available)
return true
}
}
return false
}
func (c *FindHopDriverImpl) ReloadRoute() {
c.out.Debug("FindHopDriverImpl.ReloadRoute: route reload %d", len(c.routes))
for _, rt := range c.routes {
c.updateRoute(rt)
}
}
func (c *FindHopDriverImpl) modelMultiPath() []models.MultiPath {
var modelMultiPath []models.MultiPath
for _, mp := range c.cfg.Available {
modelMultiPath = append(modelMultiPath, models.MultiPath{
NextHop: mp.NextHop,
Weight: mp.Weight,
})
}
return modelMultiPath
}
func (c *FindHopDriverImpl) buildNexthopInfos() []*nl.NexthopInfo {
multiPath := make([]*nl.NexthopInfo, 0, len(c.cfg.Available))
if len(c.cfg.Available) > 0 {
for _, mr := range c.cfg.Available {
nxhe := &nl.NexthopInfo{
Hops: mr.Weight,
Gw: net.ParseIP(mr.NextHop),
}
multiPath = append(multiPath, nxhe)
}
}
return multiPath
}
func (c *FindHopDriverImpl) updateRoute(nlr *nl.Route) {
c.out.Debug("FindHopDriverImpl.updateRoute: %v ", nlr)
multiPath := c.buildNexthopInfos()
modelMultiPath := c.modelMultiPath()
nlr.MultiPath = multiPath
cache.Network.UpdateRoute(c.Network, co.PrefixRoute{
Prefix: nlr.Dst.String(),
}, func(obj *models.Route) {
obj.MultiPath = modelMultiPath
})
promise := libol.NewPromise()
promise.Go(func() error {
if err := nl.RouteReplace(nlr); err != nil {
c.out.Warn("FindHopDriverImpl.updateRoute: %v %s", nlr, err)
return err
}
c.out.Info("FindHopDriverImpl.updateRoute: %s success", nlr.String())
return nil
})
}
func (c *FindHopDriverImpl) LoadRoute(nlr *nl.Route) {
c.out.Debug("FindHopDriverImpl.LoadRoute: %v", nlr)
c.routes = append(c.routes, nlr)
nlr.MultiPath = c.buildNexthopInfos()
nlr.Gw = nil
if len(nlr.MultiPath) == 0 {
c.out.Debug("ignored if no nexthop")
} else {
c.updateRoute(nlr)
}
}
func (c *FindHopDriverImpl) UnloadRoute(rt *nl.Route) {
c.out.Debug("FindHopDriverImpl.UnLoadRoute: %v", rt)
//find route in routes
var nlr *nl.Route
for i, r := range c.routes {
if r.Dst == rt.Dst && r.Table == rt.Table {
nlr = r
c.routes = append(c.routes[:i], c.routes[i+1:]...)
break
}
}
if nlr != nil {
if err := nl.RouteDel(nlr); err != nil {
c.out.Warn("FindHopDriverImpl.UnLoadRoute: %s", err)
return
}
}
}
type PingResult struct {
Ip string
Latency float64
Loss int
}
type PingDriver struct {
*FindHopDriverImpl
CfgName string
Running bool
PingParams *co.PingParams
}
func NewPingDriver(name string, network string, cfg *co.FindHop) *PingDriver {
return &PingDriver{
CfgName: name,
FindHopDriverImpl: &FindHopDriverImpl{
Network: network,
cfg: cfg,
out: libol.NewSubLogger(cfg.Check + "_" + name),
},
PingParams: &cfg.Params,
}
}
func (pc *PingDriver) Name() string {
return "ping"
}
func filter(results []PingResult, condition func(PingResult) bool) []PingResult {
var filteredResults []PingResult
for _, result := range results {
if condition(result) {
filteredResults = append(filteredResults, result)
}
}
return filteredResults
}
func compareMultiPaths(a []co.MultiPath, b []co.MultiPath) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if !a[i].CompareEqual(b[i]) {
return false
}
}
return true
}
// check the ipList and return the available NextHops
func (pc *PingDriver) Check(ipList []string) []co.MultiPath {
count := pc.PingParams.Count
loss := pc.PingParams.Loss
pc.out.Debug("PingDriver.Check: start check ips")
var resultIps []PingResult
for _, ip := range ipList {
avgLatency, packetLoss, err := pc.ping(ip, count)
if err != nil {
continue
}
resultIps = append(resultIps, PingResult{
Ip: ip,
Latency: avgLatency,
Loss: packetLoss,
})
}
// filter loss
filterResultIps := filter(resultIps, func(result PingResult) bool {
return result.Loss <= loss
})
sort.Slice(filterResultIps, func(i, j int) bool {
ii := filterResultIps[i]
jj := filterResultIps[j]
if ii.Loss != jj.Loss {
return ii.Loss < jj.Loss
}
return ii.Ip < jj.Ip
})
var sortedIPs []co.MultiPath
for _, result := range filterResultIps {
sortedIPs = append(sortedIPs, co.MultiPath{
NextHop: result.Ip,
Weight: 1,
})
pc.out.Debug("PingDriver.Check: available %s loss: %d ", result.Ip, result.Loss)
}
return sortedIPs
}
func (pc *PingDriver) updateNextHop() {
ipList := pc.Check(pc.cfg.NextHop)
if pc.UpdateAvailable(ipList) {
pc.ReloadRoute()
}
}
func (pc *PingDriver) update() {
frequency := pc.cfg.Params.Interval
if frequency <= 0 {
frequency = 5
}
//wait tun device start
time.Sleep(time.Second * time.Duration(2))
for pc.Running = true; pc.Running; {
pc.updateNextHop()
time.Sleep(time.Second * time.Duration(frequency))
}
}
func (pc *PingDriver) Start() {
libol.Go(pc.update)
}
func (pc *PingDriver) Stop() {
pc.Running = false
}
func (pc *PingDriver) ping(ip string, count int) (float64, int, error) {
pingPath, err := exec.LookPath("ping")
if err != nil {
pc.out.Warn("PingDriver.Ping: cmd not found :", err)
}
output, err := libol.Exec(pingPath, ip, "-c", strconv.Itoa(count))
if err != nil {
pc.out.Debug("PingDriver.Ping: exec ping %s, error: %s", ip, err)
return 0, 0, err
}
avgLatency := pc.extractLatency(output)
LossRate, err := pc.extractLoss(output)
if err != nil {
pc.out.Error("PingDriver.Ping:parse loss error : %s", err)
return 0, 0, err
}
packetLoss := int(LossRate * float64(count) / 100)
pc.out.Debug("PingDriver.Ping: ping ip[%s] loss:%.f%", ip, avgLatency, LossRate)
return avgLatency, packetLoss, nil
}
func (pc *PingDriver) extractLatency(outputStr string) float64 {
pattern := `rtt min/avg/max/mdev = (\d+\.*\d*)/(\d+\.*\d*)/(\d+\.*\d*)/(\d+\.*\d*) ms`
re := regexp.MustCompile(pattern)
subMatches := re.FindStringSubmatch(outputStr)
if len(subMatches) != 5 {
pc.out.Error("PingDriver.Ping: Cannot extract average delay.")
return 0
}
avgLatencyStr := subMatches[2]
avgLatency, err := strconv.ParseFloat(avgLatencyStr, 64)
if err != nil {
pc.out.Error("PingDriver.Ping: parse float error : %s", err)
return 0
}
return avgLatency
}
func (pc *PingDriver) extractLoss(outputStr string) (float64, error) {
re := regexp.MustCompile(`(\d+)% packet loss`)
match := re.FindStringSubmatch(outputStr)
if len(match) < 2 {
return 0, fmt.Errorf("PingDriver.Ping: packet loss parse error")
}
lossRate, err := strconv.ParseFloat(match[1], 64)
if err != nil {
return 0, err
}
return lossRate, nil
}

52
pkg/switch/network.go
View File

@@ -41,21 +41,21 @@ func SplitCombined(value string) (string, string) {
}
type WorkerImpl struct {
uuid string
cfg *co.Network
out *libol.SubLogger
dhcp *Dhcp
fire *cn.FireWallTable
setR *cn.IPSet
setV *cn.IPSet
vpn *OpenVPN
ztrust *ZTrust
qos *QosCtrl
vrf *cn.VRF
table int
br cn.Bridger
acl *ACL
nextgroup *NextGroup
uuid string
cfg *co.Network
out *libol.SubLogger
dhcp *Dhcp
fire *cn.FireWallTable
setR *cn.IPSet
setV *cn.IPSet
vpn *OpenVPN
ztrust *ZTrust
qos *QosCtrl
vrf *cn.VRF
table int
br cn.Bridger
acl *ACL
findhop *FindHop
}
func NewWorkerApi(c *co.Network) *WorkerImpl {
@@ -98,7 +98,7 @@ func (w *WorkerImpl) Initialize() {
w.acl = NewACL(cfg.Name)
w.acl.Initialize()
w.nextgroup = NewNextGroup(cfg.Name, cfg.NextGroup)
w.findhop = NewFindHop(cfg.Name, cfg.FindHop)
n := models.Network{
Name: cfg.Name,
@@ -294,7 +294,7 @@ func (w *WorkerImpl) loadRoute(rt co.PrefixRoute) {
if err != nil {
return
}
if ifAddr == rt.NextHop && rt.MultiPath == nil && rt.NextGroup == "" {
if ifAddr == rt.NextHop && rt.MultiPath == nil && rt.FindHop == "" {
// route's next-hop is local not install again.
return
}
@@ -313,9 +313,9 @@ func (w *WorkerImpl) loadRoute(rt co.PrefixRoute) {
nlr.Gw = net.ParseIP(rt.NextHop)
nlr.Priority = rt.Metric
}
if rt.NextGroup != "" {
w.out.Info("WorkerImpl.loadRoute: %s , ng: %s", nlr.String(), rt.NextGroup)
w.nextgroup.LoadRoute(rt.NextGroup, &nlr)
if rt.FindHop != "" {
w.out.Info("WorkerImpl.loadRoute: %s, findhop: %s", nlr.String(), rt.FindHop)
w.findhop.LoadRoute(rt.FindHop, &nlr)
return
}
w.out.Info("WorkerImpl.loadRoute: %s", nlr.String())
@@ -401,7 +401,7 @@ func (w *WorkerImpl) Start(v api.Switcher) {
w.out.Info("WorkerImpl.Start")
w.nextgroup.Start()
w.findhop.Start()
w.loadVRF()
w.loadRoutes()
@@ -520,8 +520,8 @@ func (w *WorkerImpl) unloadRoute(rt co.PrefixRoute) {
nlr.Priority = rt.Metric
}
if rt.NextGroup != "" {
w.nextgroup.UnloadRoute(rt.NextGroup, &nlr)
if rt.FindHop != "" {
w.findhop.UnloadRoute(rt.FindHop, &nlr)
return
}
w.out.Debug("WorkerImpl.UnLoadRoute: %s", nlr.String())
@@ -549,7 +549,7 @@ func (w *WorkerImpl) Stop() {
w.out.Info("WorkerImpl.Stop")
w.fire.Stop()
w.nextgroup.Stop()
w.findhop.Stop()
w.unloadRoutes()
if !(w.vpn == nil) {
@@ -905,7 +905,7 @@ func (w *WorkerImpl) delCacheRoute(rt co.PrefixRoute) {
if rt.Metric > 0 {
rte.Metric = rt.Metric
}
if rt.NextGroup == "" && rt.NextHop != "" {
if rt.FindHop == "" && rt.NextHop != "" {
rte.Origin = rt.NextHop
}
@@ -924,7 +924,7 @@ func (w *WorkerImpl) addCacheRoute(rt co.PrefixRoute) {
rte.Metric = rt.Metric
}
if rt.NextGroup == "" && rt.NextHop != "" {
if rt.FindHop == "" && rt.NextHop != "" {
rte.Origin = rt.NextHop
}

461
pkg/switch/nextgroup.go
View File

@@ -1,461 +0,0 @@
package cswitch
import (
"fmt"
"net"
"os/exec"
"regexp"
"sort"
"strconv"
"time"
"github.com/luscis/openlan/pkg/cache"
co "github.com/luscis/openlan/pkg/config"
"github.com/luscis/openlan/pkg/libol"
"github.com/luscis/openlan/pkg/models"
nl "github.com/vishvananda/netlink"
)
type NextGroup struct {
Network string
cfg map[string]co.NextGroup
strategies map[string]NextGroupStrategy
out *libol.SubLogger
}
func NewNextGroup(network string, cfg map[string]co.NextGroup) *NextGroup {
strategies := make(map[string]NextGroupStrategy, 32)
for key, ng := range cfg {
strategy := newCheckStrategy(key, network, ng)
if strategy != nil {
strategies[key] = strategy
}
}
return &NextGroup{
Network: network,
cfg: cfg,
strategies: strategies,
out: libol.NewSubLogger("nextgroup"),
}
}
func newCheckStrategy(name string, network string, cfg co.NextGroup) NextGroupStrategy {
switch cfg.Check {
case "ping":
return NewPingStrategy(name, network, &cfg)
}
return nil
}
func (ng *NextGroup) Start() {
ng.out.Info("NextGroup.Start: nextgroup, ng strategies size: %d", len(ng.strategies))
if len(ng.strategies) > 0 {
for _, checker := range ng.strategies {
checker.Start()
}
}
}
func (ng *NextGroup) Stop() {
if len(ng.strategies) > 0 {
for _, strategy := range ng.strategies {
strategy.Stop()
}
}
}
// for add nextgrou dynamicly
func (ng *NextGroup) AddNextGroup(name string, cfg co.NextGroup) {
if _, ok := ng.strategies[name]; ok {
ng.out.Error("NextGroup.addNextGroup: checker already exists %s", name)
return
}
strategy := newCheckStrategy(name, ng.Network, cfg)
if strategy != nil {
ng.strategies[name] = strategy
} else {
ng.out.Error("NextGroup.AddNextGroup: don't support this strategy %s", name)
}
strategy.Start()
}
// for del nextgrou dynamicly
func (ng *NextGroup) DelNextGroup(name string, cfg co.NextGroup) {
if strategy, ok := ng.strategies[name]; !ok {
ng.out.Error("NextGroup.addNextGroup: checker not exists %s", name)
return
} else {
if strategy.HasRoute() {
ng.out.Error("NextGroup.delNextGroup: checker has route %s", name)
return
}
strategy.Stop()
delete(ng.strategies, name)
}
}
func (ng *NextGroup) LoadRoute(nextgroup string, nlr *nl.Route) {
if strategy, ok := ng.strategies[nextgroup]; ok {
ng.out.Debug("NextGroup.loadRoute: %v", nlr)
strategy.LoadRoute(nlr)
} else {
ng.out.Error("NextGroup.loadRoute: checker not found %s", nextgroup)
}
}
func (ng *NextGroup) UnloadRoute(nextgroup string, nlr *nl.Route) {
if strategy, ok := ng.strategies[nextgroup]; ok {
ng.out.Debug("NextGroup.unloadRoute: %v", nlr)
strategy.UnloadRoute(nlr)
} else {
ng.out.Error("NextGroup.unloadRoute: checker not found %s", nextgroup)
}
}
type NextGroupStrategy interface {
Name() string
Check([]string) []co.MultiPath
Start()
Stop()
ReloadRoute()
LoadRoute(route *nl.Route)
UnloadRoute(route *nl.Route)
HasRoute() bool
}
type NextGroupStrategyImpl struct {
Network string
routes []*nl.Route
cfg *co.NextGroup
out *libol.SubLogger
}
func (c *NextGroupStrategyImpl) Name() string {
return "common"
}
func (c *NextGroupStrategyImpl) Start() {
}
func (c *NextGroupStrategyImpl) Stop() {
}
func (c *NextGroupStrategyImpl) HasRoute() bool {
return len(c.routes) > 0
}
func (c *NextGroupStrategyImpl) Check(ipList []string) []co.MultiPath {
return nil
}
func (c *NextGroupStrategyImpl) UpdateAvailableNexthops(mp []co.MultiPath) bool {
if c.cfg.Mode == "active-backup" {
var newPath co.MultiPath
if len(mp) > 0 {
newPath = mp[0]
}
var oldPath co.MultiPath
if len(c.cfg.AvailableNextHop) > 0 {
oldPath = c.cfg.AvailableNextHop[0]
}
if !newPath.CompareEqual(oldPath) {
c.cfg.AvailableNextHop = []co.MultiPath{newPath}
c.out.Debug("NextGroupStrategyImpl.UpdateAvailableNexthops: final available nexthops %v", c.cfg.AvailableNextHop)
return true
}
} else if c.cfg.Mode == "load-balance" {
if !compareMultiPaths(mp, c.cfg.AvailableNextHop) {
c.cfg.AvailableNextHop = mp
c.out.Debug("NextGroupStrategyImpl.UpdateAvailableNexthops: final available nexthops %v", c.cfg.AvailableNextHop)
return true
}
} else {
c.cfg.AvailableNextHop = []co.MultiPath{}
c.out.Debug("NextGroupStrategyImpl.UpdateAvailableNexthops: final available nexthops %v", c.cfg.AvailableNextHop)
//ignore
return true
}
return false
}
func (c *NextGroupStrategyImpl) ReloadRoute() {
c.out.Debug("NextGroupStrategyImpl.ReloadRoute: route reload %d", len(c.routes))
for _, rt := range c.routes {
c.updateRoute(rt)
}
}
func (c *NextGroupStrategyImpl) modelMultiPath() []models.MultiPath {
var modelMultiPath []models.MultiPath
for _, mp := range c.cfg.AvailableNextHop {
modelMultiPath = append(modelMultiPath, models.MultiPath{
NextHop: mp.NextHop,
Weight: mp.Weight,
})
}
return modelMultiPath
}
func (c *NextGroupStrategyImpl) buildNexthopInfos() []*nl.NexthopInfo {
multiPath := make([]*nl.NexthopInfo, 0, len(c.cfg.AvailableNextHop))
if len(c.cfg.AvailableNextHop) > 0 {
for _, mr := range c.cfg.AvailableNextHop {
nxhe := &nl.NexthopInfo{
Hops: mr.Weight,
Gw: net.ParseIP(mr.NextHop),
}
multiPath = append(multiPath, nxhe)
}
}
return multiPath
}
func (c *NextGroupStrategyImpl) updateRoute(nlr *nl.Route) {
c.out.Debug("NextGroupStrategyImpl.updateRoute: %v ", nlr)
multiPath := c.buildNexthopInfos()
modelMultiPath := c.modelMultiPath()
nlr.MultiPath = multiPath
cache.Network.UpdateRoute(c.Network, co.PrefixRoute{
Prefix: nlr.Dst.String(),
}, func(obj *models.Route) {
obj.MultiPath = modelMultiPath
})
promise := libol.NewPromise()
promise.Go(func() error {
if err := nl.RouteReplace(nlr); err != nil {
c.out.Warn("NextGroupStrategyImpl.updateRoute: %v %s", nlr, err)
return err
}
c.out.Info("NextGroupStrategyImpl.updateRoute: %v success", nlr.String())
return nil
})
}
func (c *NextGroupStrategyImpl) LoadRoute(nlr *nl.Route) {
c.out.Debug("NextGroupStrategyImpl.LoadRoute: %v", nlr)
c.routes = append(c.routes, nlr)
nlr.MultiPath = c.buildNexthopInfos()
nlr.Gw = nil
if len(nlr.MultiPath) == 0 {
c.out.Debug("ignored if no nexthop")
} else {
c.updateRoute(nlr)
}
}
func (c *NextGroupStrategyImpl) UnloadRoute(rt *nl.Route) {
c.out.Debug("NextGroupStrategyImpl.UnLoadRoute: %v", rt)
//find route in routes
var nlr *nl.Route
for i, r := range c.routes {
if r.Dst == rt.Dst && r.Table == rt.Table {
nlr = r
c.routes = append(c.routes[:i], c.routes[i+1:]...)
break
}
}
if nlr != nil {
if err := nl.RouteDel(nlr); err != nil {
c.out.Warn("NextGroupStrategyImpl.UnLoadRoute: %s", err)
return
}
}
}
type PingCheckResult struct {
Ip string
AvgLatency float64
PacketLoss int
}
type PingCheckStrategy struct {
*NextGroupStrategyImpl
CfgName string
Running bool
PingParams *co.PingParams
out *libol.SubLogger
}
func NewPingStrategy(name string, network string, cfg *co.NextGroup) *PingCheckStrategy {
return &PingCheckStrategy{
CfgName: name,
NextGroupStrategyImpl: &NextGroupStrategyImpl{
Network: network,
cfg: cfg,
out: libol.NewSubLogger(cfg.Check + "_common_" + name),
},
PingParams: &cfg.Ping,
out: libol.NewSubLogger(cfg.Check + "_" + name),
}
}
func (pc *PingCheckStrategy) Name() string {
return "ping"
}
func filter(results []PingCheckResult, condition func(PingCheckResult) bool) []PingCheckResult {
var filteredResults []PingCheckResult
for _, result := range results {
if condition(result) {
filteredResults = append(filteredResults, result)
}
}
return filteredResults
}
func compareMultiPaths(a []co.MultiPath, b []co.MultiPath) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if !a[i].CompareEqual(b[i]) {
return false
}
}
return true
}
// check the ipList and return the available NextHops
func (pc *PingCheckStrategy) Check(ipList []string) []co.MultiPath {
count := pc.PingParams.Count
loss := pc.PingParams.Loss
pc.out.Debug("PingCheckStrategy.Check: start check ips")
var resultIps []PingCheckResult
for _, ip := range ipList {
avgLatency, packetLoss, err := pc.ping(ip, count)
if err != nil {
continue
}
resultIps = append(resultIps, PingCheckResult{
Ip: ip,
AvgLatency: avgLatency,
PacketLoss: packetLoss,
})
}
// filter loss
filterResultIps := filter(resultIps, func(result PingCheckResult) bool {
return result.PacketLoss <= loss
})
sort.Slice(filterResultIps, func(i, j int) bool {
if filterResultIps[i].PacketLoss != filterResultIps[j].PacketLoss {
return filterResultIps[i].PacketLoss < filterResultIps[j].PacketLoss
}
return filterResultIps[i].AvgLatency < filterResultIps[j].AvgLatency
})
var sortedIPs []co.MultiPath
for _, result := range filterResultIps {
sortedIPs = append(sortedIPs, co.MultiPath{
NextHop: result.Ip,
Weight: 1,
})
pc.out.Debug("PingCheckStrategy.Check: available ip : %s , rtt:%.4f, loss: %d ", result.Ip, result.AvgLatency, result.PacketLoss)
}
return sortedIPs
}
func (pc *PingCheckStrategy) updateNextHop() {
ipList := pc.Check(pc.cfg.NextHop)
if pc.UpdateAvailableNexthops(ipList) {
pc.ReloadRoute()
}
}
func (pc *PingCheckStrategy) update() {
frequency := pc.cfg.Ping.CheckFrequency
if frequency <= 0 {
frequency = 10
}
//wait tun device start
time.Sleep(time.Second * time.Duration(2))
for pc.Running = true; pc.Running; {
pc.updateNextHop()
time.Sleep(time.Second * time.Duration(frequency))
}
}
func (pc *PingCheckStrategy) Start() {
libol.Go(pc.update)
}
func (pc *PingCheckStrategy) Stop() {
pc.Running = false
}
func (pc *PingCheckStrategy) ping(ip string, count int) (float64, int, error) {
pingPath, err := exec.LookPath("ping")
if err != nil {
pc.out.Warn("PingCheckStrategy.Ping: cmd not found :", err)
}
output, err := libol.Exec(pingPath, ip, "-c", strconv.Itoa(count))
if err != nil {
pc.out.Debug("PingCheckStrategy.Ping: exec ping ip: %s, error: %s", ip, err)
return 0, 0, err
}
avgLatency := pc.extractAvgLatency(output)
packetLossRate, err := pc.extractPacketLoss(output)
if err != nil {
pc.out.Error("PingCheckStrategy.Ping:parse loss error : %s", err)
return 0, 0, err
}
packetLoss := int(packetLossRate * float64(count) / 100)
pc.out.Debug("PingCheckStrategy.Ping: ping ip[%s], rtt:%.4f, loss: %.f%%", ip, avgLatency, packetLossRate)
return avgLatency, packetLoss, nil
}
func (pc *PingCheckStrategy) extractAvgLatency(outputStr string) float64 {
pattern := `rtt min/avg/max/mdev = (\d+\.*\d*)/(\d+\.*\d*)/(\d+\.*\d*)/(\d+\.*\d*) ms`
re := regexp.MustCompile(pattern)
subMatches := re.FindStringSubmatch(outputStr)
if len(subMatches) != 5 {
pc.out.Error("PingCheckStrategy.Ping: Cannot extract average delay.")
return 0
}
avgLatencyStr := subMatches[2]
avgLatency, err := strconv.ParseFloat(avgLatencyStr, 64)
if err != nil {
pc.out.Error("PingCheckStrategy.Ping: parse float error : %s", err)
return 0
}
return avgLatency
}
func (pc *PingCheckStrategy) extractPacketLoss(outputStr string) (float64, error) {
re := regexp.MustCompile(`(\d+)% packet loss`)
match := re.FindStringSubmatch(outputStr)
if len(match) < 2 {
return 0, fmt.Errorf("PingCheckStrategy.Ping: packet loss parse error")
}
lossRate, err := strconv.ParseFloat(match[1], 64)
if err != nil {
return 0, err
}
return lossRate, nil
}