Merge pull request loxilb-io#515 from TrekkieCoder/main

PR : loxilb-iogh-499 relaxed garabge collector characteristics

api/restapi/handler/prometheus.go

@@ -28,7 +28,7 @@ import (
 
 func ConfigGetPrometheusCounter(params operations.GetMetricsParams) middleware.Responder {
 	tk.LogIt(tk.LogDebug, "[API] Prometheus %s API called. url : %s\n", params.HTTPRequest.Method, params.HTTPRequest.URL)
-	if options.Opts.NoPrometheus {
+	if !options.Opts.Prometheus {
 		return operations.NewGetMetricsOK().WithPayload("Prometheus option is disabled.")
 	}
 	return CustomResponder(func(w http.ResponseWriter, _ runtime.Producer) {

loxinet/cluster.go

@@ -68,7 +68,7 @@ type CIStateH struct {
 
 func kaSpawn() {
 	url := fmt.Sprintf("http://127.0.0.1:%d/config/params", opts.Opts.Port)
-	for true {
+	for {
 		if IsLoxiAPIActive(url) {
 			break
 		}
@@ -234,7 +234,7 @@ func (h *CIStateH) CIVipGet(inst string) (net.IP, error) {
 			return ci.Vip, nil
 		}
 	}
-	return net.IPv4zero, errors.New("Not found")
+	return net.IPv4zero, errors.New("not found")
 }
 
 // IsCIKAMode - routine to get HA state
@@ -255,7 +255,7 @@ func (h *CIStateH) CIStateUpdate(cm cmn.HASMod) (int, error) {
 	ci, found := h.ClusterMap[cm.Instance]
 	if !found {
 		tk.LogIt(tk.LogError, "[CLUSTER] New Instance %s find error\n", cm.Instance)
-		return -1, errors.New("Cluster instance not found")
+		return -1, errors.New("cluster instance not found")
 	}
 
 	if ci.StateStr == cm.State {
@@ -278,7 +278,7 @@ func (h *CIStateH) CIStateUpdate(cm cmn.HASMod) (int, error) {
 	}
 
 	tk.LogIt(tk.LogError, "[CLUSTER] Invalid State: %s\n", cm.State)
-	return ci.State, errors.New("Invalid Cluster state")
+	return ci.State, errors.New("invalid cluster-state")
 
 }
 
@@ -288,7 +288,7 @@ func (h *CIStateH) ClusterNodeAdd(node cmn.ClusterNodeMod) (int, error) {
 	cNode := h.NodeMap[node.Addr.String()]
 
 	if cNode != nil {
-		return -1, errors.New("Exisitng Cnode")
+		return -1, errors.New("exisitng cnode")
 	}
 
 	cNode = new(ClusterNode)
@@ -306,7 +306,7 @@ func (h *CIStateH) ClusterNodeDelete(node cmn.ClusterNodeMod) (int, error) {
 	cNode := h.NodeMap[node.Addr.String()]
 
 	if cNode == nil {
-		return -1, errors.New("No such Cnode")
+		return -1, errors.New("no such cnode")
 	}
 
 	delete(h.NodeMap, node.Addr.String())

loxinet/dpbroker.go

@@ -410,6 +410,7 @@ type DpHookInterface interface {
 	DpTableGet(w *TableDpWorkQ) (DpRetT, error)
 	DpCtAdd(w *DpCtInfo) int
 	DpCtDel(w *DpCtInfo) int
+	DpTableGC()
 	DpCtGetAsync()
 	DpGetLock()
 	DpRelLock()
@@ -461,7 +462,7 @@ func (dp *DpH) WaitXsyncReady(who string) {
 		if dp.DpXsyncInSync() {
 			return
 		}
-		if time.Duration(time.Now().Sub(begin).Seconds()) >= 90 {
+		if time.Duration(time.Since(begin).Seconds()) >= 90 {
 			return
 		}
 		tk.LogIt(tk.LogDebug, "%s:waiting for Xsync..\n", who)
@@ -840,5 +841,7 @@ func (dp *DpH) DpMapGetCt4() []cmn.CtInfo {
 		}
 	}
 
+	dp.DpHooks.DpTableGC()
+
 	return CtInfoArr
 }

loxinet/dpebpf_linux.go

@@ -94,7 +94,8 @@ const (
 
 // constants
 const (
-	DpEbpfLinuxTiVal     = 10
+	dpEbpfLinuxTiVal     = 10
+	ctGCTiValDefault     = 45
 	ctiDeleteSyncRetries = 3
 	blkCtiMaxLen         = 8192
 	mapNotifierChLen     = 8096
@@ -139,6 +140,9 @@ type (
 type DpEbpfH struct {
 	ticker  *time.Ticker
 	tDone   chan bool
+	trigGC  chan bool
+	gcTS    time.Time
+	gcTiVal uint
 	ctBcast chan bool
 	nID     uint
 	tbN     uint
@@ -183,6 +187,10 @@ func dpEbpfTicker() {
 			tk.LogIt(tk.LogDebug, "CT Bcast\n")
 			dpCTMapBcast()
 			continue
+		case <-mh.dpEbpf.trigGC:
+			C.llb_age_map_entries(C.LL_DP_CT_MAP)
+			C.llb_age_map_entries(C.LL_DP_FCV4_MAP)
+			mh.dpEbpf.gcTS = time.Now()
 		case t := <-mh.dpEbpf.ticker.C:
 			sel := mh.dpEbpf.tbN % tLen
 			tk.LogIt(-1, "DP Tick at for selector %v:%d\n", t, sel)
@@ -196,8 +204,11 @@ func dpEbpfTicker() {
 			/* No need to fetch all stats in this fashion */
 			//C.llb_collect_map_stats(C.int(C.LL_DP_CT_STATS_MAP))
 			/* Per entry stats will be fetched in C.ll_ct_map_ent_has_aged */
-			C.llb_age_map_entries(C.LL_DP_CT_MAP)
-			C.llb_age_map_entries(C.LL_DP_FCV4_MAP)
+			if mh.dpEbpf.gcTiVal == 0 || time.Duration(time.Since(mh.dpEbpf.gcTS).Seconds()) > time.Duration(mh.dpEbpf.gcTiVal) {
+				C.llb_age_map_entries(C.LL_DP_CT_MAP)
+				C.llb_age_map_entries(C.LL_DP_FCV4_MAP)
+				mh.dpEbpf.gcTS = time.Now()
+			}
 
 			// This means around 10s from start
 			if !mh.dpEbpf.CtSync {
@@ -291,12 +302,15 @@ func DpEbpfInit(clusterEn bool, nodeNum int, rssEn bool, egrHooks bool, logLevel
 
 	ne := new(DpEbpfH)
 	ne.tDone = make(chan bool)
+	ne.trigGC = make(chan bool)
+	ne.gcTS = time.Now()
+	ne.gcTiVal = ctGCTiValDefault
 	ne.ToMapCh = make(chan interface{}, mapNotifierChLen)
 	for i := 0; i < mapNotifierWorkers; i++ {
 		ne.ToFinCh[i] = make(chan int)
 	}
 	ne.ctBcast = make(chan bool)
-	ne.ticker = time.NewTicker(DpEbpfLinuxTiVal * time.Second)
+	ne.ticker = time.NewTicker(dpEbpfLinuxTiVal * time.Second)
 	ne.ctMap = make(map[string]*DpCtInfo)
 	ne.RssEn = rssEn
 	ne.nID = uint((C.LLB_CT_MAP_ENTRIES / C.LLB_MAX_LB_NODES) * nodeNum)
@@ -377,14 +391,14 @@ func (e *DpEbpfH) loadEbpfPgm(name string) int {
 		C.llb_dp_link_attach(ifStr, xSection, C.LL_BPF_MOUNT_XDP, 0)
 	}
 	section := C.CString(string(C.TC_LL_SEC_DEFAULT))
-	ret := C.llb_dp_link_attach(ifStr, section, C.LL_BPF_MOUNT_TC, 0)
+	C.llb_dp_link_attach(ifStr, section, C.LL_BPF_MOUNT_TC, 0)
 
 	filters, err := nlp.FilterList(link, nlp.HANDLE_MIN_INGRESS)
 	if err != nil {
 		tk.LogIt(tk.LogWarning, "[DP] Filter on %s not found\n", name)
 		return -1
 	}
-	ret = -1
+	ret := -1
 	for _, f := range filters {
 		if t, ok := f.(*nlp.BpfFilter); ok {
 			if strings.Contains(t.Name, C.TC_LL_SEC_DEFAULT) {
@@ -2040,3 +2054,8 @@ func (e *DpEbpfH) DpGetLock() {
 func (e *DpEbpfH) DpRelLock() {
 	C.llb_xh_unlock()
 }
+
+// DpTableGC - Work on table garbage collection
+func (e *DpEbpfH) DpTableGC() {
+	e.trigGC <- true
+}

loxinet/loxinet.go

@@ -266,10 +266,10 @@ func loxiNetInit() {
 			}
 		}
 	} else {
-		// If bgp peer mode is enable then bgp flag has to be set by default
+		// If bgp peer mode is enabled then bgp flag has to be set by default
 		opts.Opts.Bgp = true
 		//opts.Opts.NoNlp = true
-		opts.Opts.NoPrometheus = true
+		opts.Opts.Prometheus = false
 	}
 
 	// Initialize goBgp client
@@ -296,7 +296,7 @@ func loxiNetInit() {
 	}
 
 	// Initialize the Prometheus subsystem
-	if !opts.Opts.NoPrometheus {
+	if opts.Opts.Prometheus {
 		prometheus.PrometheusRegister(NetAPIInit(opts.Opts.BgpPeerMode))
 		prometheus.Init()
 	}

loxinet/loxinettest.go

@@ -30,7 +30,7 @@ func TestLoxinet(t *testing.T) {
 	opts.Opts.NoNlp = true
 	opts.Opts.NoAPI = true
 	opts.Opts.CPUProfile = "none"
-	opts.Opts.NoPrometheus = true
+	opts.Opts.Prometheus = false
 	opts.Opts.K8sApi = "none"
 
 	fmt.Printf("LoxiLB Net Unit-Test \n")

loxinet/route.go

@@ -187,6 +187,13 @@ func (r *RtH) RtAdd(Dst net.IPNet, Zone string, Ra RtAttr, Na []RtNhAttr) (int,
 	key := RtKey{Dst.String(), Zone}
 	nhLen := len(Na)
 
+	if opts.Opts.FallBack {
+		if Dst.IP.IsUnspecified() {
+			tk.LogIt(tk.LogError, "rt add - %s:%s skipped - fallback\n", Dst.String(), Zone)
+			return 0, nil
+		}
+	}
+
 	if nhLen > 1 {
 		tk.LogIt(tk.LogError, "rt add - %s:%s ecmp not supported\n", Dst.String(), Zone)
 		return RtNhErr, errors.New("ecmp-rt error not supported")
@@ -207,7 +214,7 @@ func (r *RtH) RtAdd(Dst net.IPNet, Zone string, Ra RtAttr, Na []RtNhAttr) (int,
 			}
 		}
 
-		if rtMod == true {
+		if rtMod {
 			ret, _ := r.RtDelete(Dst, Zone)
 			if ret != 0 {
 				tk.LogIt(tk.LogError, "rt add - %s:%s del failed on mod\n", Dst.String(), Zone)
@@ -231,7 +238,7 @@ func (r *RtH) RtAdd(Dst net.IPNet, Zone string, Ra RtAttr, Na []RtNhAttr) (int,
 	if len(Na) != 0 {
 		rt.TFlags |= RtTypeInd
 
-		if Ra.HostRoute == true {
+		if Ra.HostRoute {
 			rt.TFlags |= RtTypeHost
 		}
 
@@ -326,8 +333,15 @@ func (rt *Rt) rtRemoveDepObj(i int) []RtDepObj {
 func (r *RtH) RtDelete(Dst net.IPNet, Zone string) (int, error) {
 	key := RtKey{Dst.String(), Zone}
 
+	if opts.Opts.FallBack {
+		if Dst.IP.IsUnspecified() {
+			tk.LogIt(tk.LogError, "rt delete - %s:%s skipped - fallback\n", Dst.String(), Zone)
+			return 0, nil
+		}
+	}
+
 	rt, found := r.RtMap[key]
-	if found == false {
+	if !found {
 		tk.LogIt(tk.LogError, "rt delete - %s:%s not found\n", Dst.String(), Zone)
 		return RtNoEntErr, errors.New("no such route")
 	}
@@ -470,13 +484,6 @@ func (rt *Rt) DP(work DpWorkT) int {
 		return -1
 	}
 
-	if opts.Opts.FallBack {
-		if rtNet.IP.IsUnspecified() {
-			fmt.Printf("FALL BACK MODE\n\n\n\n")
-			return 0
-		}
-	}
-
 	if work == DpStatsGet {
 		nStat := new(StatDpWorkQ)
 		nStat.Work = work

options/options.go

@@ -20,7 +20,7 @@ var Opts struct {
 	ClusterSelf       int            `long:"self" description:"annonation of self in cluster" default:"0"`
 	LogLevel          string         `long:"loglevel" description:"One of debug,info,error,warning,notice,critical,emergency,alert" default:"debug"`
 	CPUProfile        string         `long:"cpuprofile" description:"Enable cpu profiling and specify file to use" default:"none" env:"CPUPROF"`
-	NoPrometheus      bool           `short:"p" long:"nopro" description:" Do not run prometheus thread"`
+	Prometheus        bool           `short:"p" long:"prometheus" description:"Run prometheus thread"`
 	CSumDisable       bool           `long:"disable-csum" description:"Disable checksum update(experimental)"`
 	PassiveEPProbe    bool           `long:"passive-probe" description:"Enable passive liveness probes(experimental)"`
 	RssEnable         bool           `long:"rss-enable" description:"Enable rss optimization(experimental)"`