✨ persistence (#23)

Author: amitai-devops

README.md

@@ -46,6 +46,8 @@ cpuFactor: 1
 memoryFactor: 1
 logLevel: info
 concurrentWorkers: 10
+persistence:
+  enabled: true
 ```
 ## Prerequisites
 - The metrics server must be deployed in your cluster. Read more about [Metrics Server](https://github.com/kubernetes-sigs/metrics-server). This controller uses the **metrics.k8s.io** extension API group (apis/metrics.k8s.io/v1beta1)
@@ -55,53 +57,48 @@ concurrentWorkers: 10
 **kube-reqsizer** has primary custom flags:
 
 ```bash
+# Enable a annotation filter for pod scraping. 
+# True will only set requests of controllers of which PODS or NAMESPACE 
+# have the annotation set to "true".
+# If "false", will ignore annotations and work 
+# on all pods in the cluster unless
+# they have "false".
+
+# reqsizer.jatalocks.github.io/optimize=true
+# reqsizer.jatalocks.github.io/optimize=false
 --annotation-filter bool (default true)
-    
-    Enable a annotation filter for pod scraping. 
-    Enabling this will ensure that the controller 
-    only sets requests of controllers of which PODS or NAMESPACE 
-    have the annotation set to "true".
-    If "false", will ignore annotations and work on all pods in the cluster.
-
-    # reqsizer.jatalocks.github.io/optimize=true
-    # reqsizer.jatalocks.github.io/optimize=false
 
+# The sample size to create an average from when reconciling.
 --sample-size int (default 1)
 
-    The sample size to create an average from when reconciling.
-
+# Minimum seconds between pod restart.
+# This ensures the controller will not restart a pod if the minimum time
+# has not passed since it has started.
 --min-seconds float (default 1)
 
-    Minimum seconds between pod restart.
-    This ensures the controller will not restart a pod if the minimum time
-    has not passed since it has started.
-
+# Allow controller to reduce/increase requests
 --enable-increase (default true)
-    Enables the controller to increase pod requests
-
 --enable-reduce (default true)
-    Enables the controller to reduce pod requests
 
+# Min and Max CPU (m) and Memory (Mi) the controller can set a pod request to. 0 is infinite
 --max-cpu int (default 0)
-    Maximum CPU in (m) that the controller can set a pod request to. 0 is infinite 
-
 --max-memory int (default 0)
-    Maximum memory in (Mi) that the controller can set a pod request to. 0 is infinite
-
 --min-cpu int (default 0)
-    Minimum CPU in (m) that the controller can set a pod request to. 0 is infinite
-
 --min-memory int (default 0)
-    Minimum memory in (Mi) that the controller can set a pod request to. 0 is infinite
 
+# Multiply requests when reconciling
 --cpu-factor float (default 1)
-    A factor to multiply CPU requests when reconciling.
-
 --memory-factor float (default 1)
-    A factor to multiply Memory requests when reconciling.
 
+# How many pods to sample in parallel. This may affect the controllers stability.
 --concurrent-workers (default 10)
-    How many pods to sample in parallel. This may affect the controller's stability.
+
+# Persistence using Redis
+--enable-persistence (default false)
+--redis-host (default "localhost")
+--redis-port (default "6379")
+--redis-password (default "")
+--redis-db (default 0)
 ```
 
 ### Annotations 
@@ -140,8 +137,6 @@ reqsizer.jatalocks.github.io/mode=min       # Sets the request to the MINIMUM of
 ## Limitations
 
 - Does not work with CRD controllers (such as Argo Rollouts)
-- Does not have persistent cache. Will reset cache on controller restart
-
 # Development
 ## Getting Started
 You’ll need a Kubernetes cluster to run against. You can use [KIND](https://sigs.k8s.io/kind) to get a local cluster for testing, or run against a remote cluster.

charts/kube-reqsizer/Chart.yaml

@@ -13,4 +13,9 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.8.55
+version: 0.9.0
+dependencies:
+- name: redis
+  condition: persistence.enabled
+  repository: https://charts.bitnami.com/bitnami
+  version: 17.4.0

charts/kube-reqsizer/templates/deployment.yaml

@@ -36,7 +36,9 @@ spec:
         - --min-memory={{.Values.minMemory}}
         - --cpu-factor={{.Values.cpuFactor}}
         - --memory-factor={{.Values.memoryFactor}}        
-        - --concurrent-workers={{.Values.concurrentWorkers}}        
+        - --concurrent-workers={{.Values.concurrentWorkers}}
+        - --enable-persistence={{.Values.persistence.enabled}}        
+        - --redis-host={{.Release.Name}}-redis-master
         resources:
           {{- toYaml .Values.controllerManager.manager.resources | nindent 10 }}
         command:

charts/kube-reqsizer/values.yaml

@@ -11,7 +11,8 @@ cpuFactor: 1
 memoryFactor: 1
 logLevel: info
 concurrentWorkers: 10
-
+persistence:
+  enabled: true
 serviceMonitor: false
 
 controllerManager:

controllers/pod_controller.go

@@ -22,10 +22,16 @@ import (
 	"math"
 	"time"
 
+	"github.com/go-logr/logr"
+	"github.com/jatalocks/kube-reqsizer/pkg/cache/localcache"
+	"github.com/jatalocks/kube-reqsizer/pkg/cache/rediscache"
+	"github.com/jatalocks/kube-reqsizer/types"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -34,13 +40,34 @@ import (
 
 // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;update;patch
 
+// PodReconciler reconciles a Pod object
+type PodReconciler struct {
+	client.Client
+	Log                         logr.Logger
+	Scheme                      *runtime.Scheme
+	ClientSet                   *kubernetes.Clientset
+	SampleSize                  int
+	EnableAnnotation            bool
+	MinSecondsBetweenPodRestart float64
+	EnableIncrease              bool
+	EnableReduce                bool
+	MaxMemory                   int64
+	MaxCPU                      int64
+	MinMemory                   int64
+	MinCPU                      int64
+	CPUFactor                   float64
+	MemoryFactor                float64
+	RedisClient                 rediscache.RedisClient
+	EnablePersistence           bool
+}
+
 const (
 	operatorAnnotation     = "reqsizer.jatalocks.github.io/optimize"
 	operatorModeAnnotation = "reqsizer.jatalocks.github.io/mode"
 )
 
 func cacheKeyFunc(obj interface{}) (string, error) {
-	return obj.(PodRequests).Name + obj.(PodRequests).Namespace, nil
+	return obj.(types.PodRequests).Name + "-" + obj.(types.PodRequests).Namespace, nil
 }
 
 var cacheStore = cache.NewStore(cacheKeyFunc)
@@ -88,18 +115,31 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 		if err != nil {
 			deploymentName = pod.Name
 		}
-		SumPodRequest := PodRequests{Name: deploymentName, Namespace: pod.Namespace, ContainerRequests: []ContainerRequests{}}
+		SumPodRequest := types.PodRequests{Name: deploymentName, Namespace: pod.Namespace, ContainerRequests: []types.ContainerRequests{}}
 
 		SumPodRequest.ContainerRequests = PodUsageData.ContainerRequests
-
-		LatestPodRequest, err := fetchFromCache(cacheStore, deploymentName+pod.Namespace)
+		var LatestPodRequest types.PodRequests
+		if r.EnablePersistence {
+			LatestPodRequest, err = r.RedisClient.FetchFromCache(deploymentName + "-" + pod.Namespace)
+		} else {
+			LatestPodRequest, err = localcache.FetchFromCache(cacheStore, deploymentName+"-"+pod.Namespace)
+		}
 		if err != nil {
 			SumPodRequest.Sample = 0
 			log.Info(fmt.Sprint("Adding cache sample ", SumPodRequest.Sample))
-			addToCache(cacheStore, SumPodRequest)
-			log.Info(fmt.Sprint("Items in Cache: ", len(cacheStore.List())))
+			if r.EnablePersistence {
+				r.RedisClient.AddToCache(SumPodRequest)
+				log.Info(fmt.Sprint("Items in Cache: ", r.RedisClient.CacheSize()))
+			} else {
+				localcache.AddToCache(cacheStore, SumPodRequest)
+				log.Info(fmt.Sprint("Items in Cache: ", len(cacheStore.List())))
+			}
 		} else {
-			log.Info(fmt.Sprint("Items in Cache: ", len(cacheStore.List())))
+			if r.EnablePersistence {
+				log.Info(fmt.Sprint("Items in Cache: ", r.RedisClient.CacheSize()))
+			} else {
+				log.Info(fmt.Sprint("Items in Cache: ", len(cacheStore.List())))
+			}
 			SumPodRequest.Sample = LatestPodRequest.Sample + 1
 
 			log.Info(fmt.Sprint("Updating cache sample ", SumPodRequest.Sample))
@@ -128,19 +168,27 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 				}
 			}
 
-			if err := deleteFromCache(cacheStore, LatestPodRequest); err != nil {
-				log.Error(err, err.Error())
-			}
-
-			if err = addToCache(cacheStore, SumPodRequest); err != nil {
-				log.Error(err, err.Error())
+			if r.EnablePersistence {
+				if err := r.RedisClient.DeleteFromCache(LatestPodRequest); err != nil {
+					log.Error(err, err.Error())
+				}
+				if err = r.RedisClient.AddToCache(SumPodRequest); err != nil {
+					log.Error(err, err.Error())
+				}
+			} else {
+				if err := localcache.DeleteFromCache(cacheStore, LatestPodRequest); err != nil {
+					log.Error(err, err.Error())
+				}
+				if err = localcache.AddToCache(cacheStore, SumPodRequest); err != nil {
+					log.Error(err, err.Error())
+				}
 			}
 		}
 		log.Info(fmt.Sprint(SumPodRequest))
 		if (SumPodRequest.Sample >= r.SampleSize) && r.MinimumUptimeOfPodInParent(pod, ctx) {
 			log.Info("Sample Size and Minimum Time have been reached")
 			PodChange := false
-			Requests := []NewContainerRequests{}
+			Requests := []types.NewContainerRequests{}
 			for _, c := range SumPodRequest.ContainerRequests {
 				AverageUsageCPU := c.CPU / int64(SumPodRequest.Sample)
 				AverageUsageMemory := c.Memory / int64(SumPodRequest.Sample)
@@ -195,7 +243,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 									}
 								}
 
-								Requests = append(Requests, NewContainerRequests{Name: c.Name, Requests: pod.Spec.Containers[i].Resources})
+								Requests = append(Requests, types.NewContainerRequests{Name: c.Name, Requests: pod.Spec.Containers[i].Resources})
 							}
 						}
 					}
@@ -222,9 +270,14 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 
 			}
 
-			err := deleteFromCache(cacheStore, SumPodRequest)
-			if err != nil {
-				log.Error(err, err.Error())
+			if r.EnablePersistence {
+				if err := r.RedisClient.DeleteFromCache(SumPodRequest); err != nil {
+					log.Error(err, err.Error())
+				}
+			} else {
+				if err := localcache.DeleteFromCache(cacheStore, LatestPodRequest); err != nil {
+					log.Error(err, err.Error())
+				}
 			}
 		}
 	}

controllers/pod_controller_functions.go

@@ -9,13 +9,12 @@ import (
 	"time"
 	"unicode/utf8"
 
+	"github.com/jatalocks/kube-reqsizer/types"
 	"github.com/labstack/gommon/log"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/klog"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -82,7 +81,7 @@ func (r *PodReconciler) UpdateKubeObject(pod client.Object, ctx context.Context)
 	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 }
 
-func UpdatePodController(podspec *corev1.PodSpec, Requests []NewContainerRequests, ctx context.Context) {
+func UpdatePodController(podspec *corev1.PodSpec, Requests []types.NewContainerRequests, ctx context.Context) {
 	for _, podContainer := range Requests {
 		for i, depContainer := range podspec.Containers {
 			if depContainer.Name == podContainer.Name {
@@ -149,8 +148,8 @@ func (r *PodReconciler) ValidateMemory(currentMemory, AverageUsageMemory int64)
 	return false
 }
 
-func GetPodRequests(pod corev1.Pod) PodRequests {
-	containerData := []ContainerRequests{}
+func GetPodRequests(pod corev1.Pod) types.PodRequests {
+	containerData := []types.ContainerRequests{}
 	for _, c := range pod.Spec.Containers {
 		nanoCores, _ := strconv.Atoi(RemoveLastChar(c.Resources.Requests.Cpu().String()))
 		memString := c.Resources.Requests.Memory().String()
@@ -166,49 +165,22 @@ func GetPodRequests(pod corev1.Pod) PodRequests {
 			giMemory, _ := strconv.Atoi(strings.ReplaceAll(memString, "Gi", ""))
 			miMemory = giMemory / 1000
 		}
-		containerData = append(containerData, ContainerRequests{Name: c.Name, CPU: int64(nanoCores), Memory: int64(miMemory)})
+		containerData = append(containerData, types.ContainerRequests{Name: c.Name, CPU: int64(nanoCores), Memory: int64(miMemory)})
 	}
-	return PodRequests{pod.Name, pod.Namespace, containerData, 0}
+	return types.PodRequests{pod.Name, pod.Namespace, containerData, 0}
 }
 
-func addToCache(cacheStore cache.Store, object PodRequests) error {
-	err := cacheStore.Add(object)
-	if err != nil {
-		klog.Errorf("failed to add key value to cache error", err)
-		return err
-	}
-	return nil
-}
-
-func fetchFromCache(cacheStore cache.Store, key string) (PodRequests, error) {
-	obj, exists, err := cacheStore.GetByKey(key)
-	if err != nil {
-		// klog.Errorf("failed to add key value to cache error", err)
-		return PodRequests{}, err
-	}
-	if !exists {
-		// klog.Errorf("object does not exist in the cache")
-		err = errors.New("object does not exist in the cache")
-		return PodRequests{}, err
-	}
-	return obj.(PodRequests), nil
-}
-
-func deleteFromCache(cacheStore cache.Store, object PodRequests) error {
-	return cacheStore.Delete(object)
-}
-
-func GeneratePodRequestsObjectFromRestData(restData []byte) PodRequests {
+func GeneratePodRequestsObjectFromRestData(restData []byte) types.PodRequests {
 	s := restData[:]
-	data := PodMetricsRestData{}
+	data := types.PodMetricsRestData{}
 	json.Unmarshal([]byte(s), &data)
-	containerData := []ContainerRequests{}
+	containerData := []types.ContainerRequests{}
 	for _, c := range data.Containers {
 		nanoCores, _ := strconv.Atoi(RemoveLastChar(c.Usage.CPU))
 		kiMemory, _ := strconv.Atoi(strings.ReplaceAll(c.Usage.Memory, "Ki", ""))
-		containerData = append(containerData, ContainerRequests{Name: c.Name, CPU: int64(nanoCores / 1000000), Memory: int64(kiMemory / 1000)})
+		containerData = append(containerData, types.ContainerRequests{Name: c.Name, CPU: int64(nanoCores / 1000000), Memory: int64(kiMemory / 1000)})
 	}
-	return PodRequests{data.Metadata.Name, data.Metadata.Namespace, containerData, 0}
+	return types.PodRequests{data.Metadata.Name, data.Metadata.Namespace, containerData, 0}
 }
 
 func (r *PodReconciler) MinimumUptimeOfPodInParent(pod corev1.Pod, ctx context.Context) bool {
@@ -252,7 +224,6 @@ func (r *PodReconciler) GetPodParentKind(pod corev1.Pod, ctx context.Context) (e
 		deployment, err := r.ClientSet.AppsV1().DaemonSets(pod.Namespace).Get(ctx, pod.OwnerReferences[0].Kind, metav1.GetOptions{})
 		return err, &deployment.Spec.Template.Spec, deployment, deployment.Name
 	case "StatefulSet":
-		log.Info("STATEFUL SET HAHAHHAHA")
 		deployment, err := r.ClientSet.AppsV1().StatefulSets(pod.Namespace).Get(ctx, pod.OwnerReferences[0].Kind, metav1.GetOptions{})
 		return err, &deployment.Spec.Template.Spec, deployment, deployment.Name
 	default:

go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.19.5 // indirect
 	github.com/go-openapi/swag v0.19.14 // indirect
+	github.com/go-redis/redis v6.15.9+incompatible // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.2.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect

go.sum

@@ -148,6 +148,8 @@ github.com/go-openapi/swag v0.19.14 h1:gm3vOOXfiuw5i9p5N9xJvfjvuofpyvLA9Wr6QfK5F
 github.com/go-openapi/swag v0.19.14/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
 github.com/go-openapi/swag v0.22.3 h1:yMBqmnQ0gyZvEb/+KzuWZOXgllrXT4SADYbvDaXHv/g=
 github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14=
+github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
+github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=