Implement window

A window is implemented, where values are inserted and kept in memory
for a short time before updating service directory values.

Signed-off-by: Elis Lulja <[email protected]>

Author: asimpleidea

controllers/base.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/CloudNativeSDWAN/cnwan-operator/internal/types"
@@ -33,6 +34,29 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+type window struct {
+	values []*windowValue
+	lock   sync.Mutex
+}
+
+func (w *window) getHighest() int {
+	highest := 0
+	for _, val := range w.values {
+		if val.totalCount > highest {
+			highest = val.totalCount
+		}
+	}
+
+	return highest
+}
+
+type windowValue struct {
+	epsliceName  string
+	epsliceCount int
+	totalCount   int
+	timer        *time.Timer
+}
+
 // BaseReconciler is the base controller/reconciler upon which all other
 // controllers will be based.
 type BaseReconciler struct {
@@ -45,6 +69,7 @@ type BaseReconciler struct {
 	CountPodKey        string
 
 	epsliceCounter *counter
+	srvWindows     map[string]*window
 }
 
 // NewBaseReconciler returns a new instance of a base reconciler to be used
@@ -64,6 +89,7 @@ func NewBaseReconciler(cli client.Client, scheme *runtime.Scheme, broker *sr.Bro
 		CurrentNsPolicy:    currNsPolicy,
 		CountPodKey:        countPodKey,
 		epsliceCounter:     newCounter(),
+		srvWindows:         map[string]*window{},
 	}
 }
 

controllers/counters.go

@@ -50,6 +50,27 @@ func (c *counter) getSrvCount(nsName, srvName string) int {
 	return totalCount
 }
 
+func (c *counter) resetCounterTo(nsName, srvName string, data map[string]int) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	fullName := ktypes.NamespacedName{Namespace: nsName, Name: srvName}.String()
+	c.counts[fullName] = data
+}
+
+func (c *counter) getSrvData(nsName, srvName string) map[string]int {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	fullName := ktypes.NamespacedName{Namespace: nsName, Name: srvName}.String()
+	val, exists := c.counts[fullName]
+	if !exists {
+		return map[string]int{}
+	}
+
+	return val
+}
+
 func (c *counter) putSrvCount(nsName, srvName, epSliceName string, count int) {
 	c.lock.Lock()
 	defer c.lock.Unlock()

controllers/endpointslice_controller.go

@@ -19,6 +19,7 @@ package controllers
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"k8s.io/api/discovery/v1beta1"
 	discoveryv1beta1 "k8s.io/api/discovery/v1beta1"
@@ -66,19 +67,83 @@ func (r *EndpointSliceReconciler) Reconcile(req ctrl.Request) (ctrl.Result, erro
 		return ctrl.Result{}, nil
 	}
 
-	r.epsliceCounter.putSrvCount(req.Namespace, data.srv, req.Name, data.count)
+	srvname := ktypes.NamespacedName{Namespace: req.Namespace, Name: data.srv}
+	wind, exists := r.srvWindows[srvname.String()]
+	if !exists {
+		r.srvWindows[srvname.String()] = &window{values: []*windowValue{}}
+		wind = r.srvWindows[srvname.String()]
+	}
 
-	if r.srvRecon == nil {
-		l.Error(fmt.Errorf("service reconciler is nil"), "could not reconcile service from endpointslice")
-		return ctrl.Result{}, nil
+	wind.lock.Lock()
+	defer wind.lock.Unlock()
+
+	countBeforeUpd := r.epsliceCounter.getSrvCount(srvname.Namespace, srvname.Name)
+	r.epsliceCounter.putSrvCount(req.Namespace, srvname.Name, req.Name, data.count)
+	newCount := r.epsliceCounter.getSrvCount(srvname.Namespace, srvname.Name)
+
+	if len(wind.values) > 0 {
+		if newCount > wind.getHighest() {
+			l.Info("new count detected and is the highest in the window, updating service registry...", "highest", wind.getHighest(), "new-count", newCount)
+			r.srvRecon.cacheSrvWatch[srvname.String()] = true
+			r.srvRecon.Reconcile(ctrl.Request{NamespacedName: srvname})
+		} else {
+			l.Info("new count detected, but not highest in window: performing cooldown...", "highest", wind.getHighest(), "new-count", newCount)
+		}
+	} else {
+		if newCount > countBeforeUpd {
+			l.Info("new count detected and window is empty, updating service registry...", "old-val", countBeforeUpd, "new-val", newCount)
+			r.srvRecon.cacheSrvWatch[srvname.String()] = true
+			r.srvRecon.Reconcile(ctrl.Request{NamespacedName: srvname})
+		} else {
+			l.Info("new count detected, but not higher than current value, performing cooldown...", "old-val", countBeforeUpd, "new-val", newCount)
+		}
 	}
 
-	srvname := ktypes.NamespacedName{Namespace: req.Namespace, Name: data.srv}
-	r.srvRecon.lock.Lock()
-	r.srvRecon.cacheSrvWatch[srvname.String()] = true
-	r.srvRecon.lock.Unlock()
+	wind.values = append(wind.values, &windowValue{
+		epsliceName:  req.Name,
+		epsliceCount: data.count,
+		totalCount:   newCount,
+		timer: time.AfterFunc(time.Minute, func() {
+			r.exitWindow(srvname)
+		}),
+	})
+
+	return ctrl.Result{}, nil
+}
+
+func (r *EndpointSliceReconciler) exitWindow(srv ktypes.NamespacedName) {
+	l := r.Log.WithName("EndpointSliceReconciler").WithValues("Window", srv)
+
+	wind := r.srvWindows[srv.String()]
+	wind.lock.Lock()
+	defer wind.lock.Unlock()
+
+	if len(wind.values) == 0 {
+		l.V(2).Info("window is empty, returning...")
+		return
+	}
+
+	oldestVal := wind.values[0]
+	oldestVal.timer.Stop()
+	defer func() {
+		wind.values = wind.values[1:]
+	}()
+
+	highestVal := r.epsliceCounter.getSrvCount(srv.Namespace, srv.Name)
+	for i := 1; i < len(wind.values); i++ {
+		if wind.values[i].totalCount > highestVal {
+			highestVal = wind.values[i].totalCount
+		}
+	}
+
+	if oldestVal.totalCount <= highestVal && len(wind.values) > 1 {
+		l.Info("highest value isn't changed, returning...", "exiting-value", oldestVal.totalCount, "highest", highestVal)
+		return
+	}
 
-	return r.srvRecon.Reconcile(ctrl.Request{NamespacedName: srvname})
+	l.Info("updating service registry...", "highest-value", highestVal)
+	r.srvRecon.cacheSrvWatch[srv.String()] = true
+	r.srvRecon.Reconcile(ctrl.Request{NamespacedName: srv})
 }
 
 // SetServiceReconciler sets the service reconciler, so that the endpointslice