20 - Fix min seconds. Remove Code Dupe. Optimization. (#21)

Author: amitai-devops

README.md

@@ -45,7 +45,7 @@ minCPU: 0
 cpuFactor: 1
 memoryFactor: 1
 logLevel: info
-concurrentWorkers: 20
+concurrentWorkers: 10
 ```
 ## 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)
@@ -100,7 +100,7 @@ concurrentWorkers: 20
 --memory-factor float (default 1)
     A factor to multiply Memory requests when reconciling.
 
---concurrent-workers (default 20)
+--concurrent-workers (default 10)
     How many pods to sample in parallel. This may affect the controller's stability.
 ```
 

charts/kube-reqsizer/Chart.yaml

@@ -13,4 +13,4 @@ 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.1
+version: 0.8.2

charts/kube-reqsizer/values.yaml

@@ -10,7 +10,7 @@ minCPU: 0
 cpuFactor: 1
 memoryFactor: 1
 logLevel: info
-concurrentWorkers: 20
+concurrentWorkers: 10
 
 
 controllerManager:

controllers/pod_controller.go

@@ -26,10 +26,10 @@ import (
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
 
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;update;patch
@@ -77,7 +77,6 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 	}
 
 	if ((!r.EnableAnnotation) || (r.EnableAnnotation && annotation)) && !ignoreAnnotation {
-
 		data, err := r.ClientSet.RESTClient().Get().AbsPath(fmt.Sprintf("apis/metrics.k8s.io/v1beta1/namespaces/%v/pods/%v", pod.Namespace, pod.Name)).DoRaw(ctx)
 
 		if err != nil {
@@ -93,20 +92,12 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 		LatestPodRequest, err := fetchFromCache(cacheStore, pod.Name)
 		if err != nil {
 			SumPodRequest.Sample = 0
-			SumPodRequest.TimeSinceFirstSample = 0
-			SumPodRequest.Timestamp = time.Now()
 			log.Info(fmt.Sprint("Adding cache sample ", SumPodRequest.Sample))
 			addToCache(cacheStore, SumPodRequest)
 			log.Info(fmt.Sprint("Items in Cache: ", len(cacheStore.List())))
 		} else {
-
+			log.Info(fmt.Sprint("Items in Cache: ", len(cacheStore.List())))
 			SumPodRequest.Sample = LatestPodRequest.Sample + 1
-			if LatestPodRequest.Sample == 1 {
-				SumPodRequest.Timestamp = time.Now()
-				LatestPodRequest.Timestamp = SumPodRequest.Timestamp
-			}
-			log.Info(fmt.Sprint(time.Now(), LatestPodRequest.Timestamp))
-			SumPodRequest.TimeSinceFirstSample = time.Since(LatestPodRequest.Timestamp).Seconds()
 
 			log.Info(fmt.Sprint("Updating cache sample ", SumPodRequest.Sample))
 
@@ -143,7 +134,7 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 			}
 		}
 		log.Info(fmt.Sprint(SumPodRequest))
-		if (SumPodRequest.Sample >= r.SampleSize) && (SumPodRequest.TimeSinceFirstSample >= r.MinSecondsBetweenPodRestart) {
+		if (SumPodRequest.Sample >= r.SampleSize) && r.MinimumUptimeOfPodInParent(pod, ctx) {
 			log.Info("Sample Size and Minimum Time have been reached")
 			PodChange := false
 			Requests := []NewContainerRequests{}
@@ -155,49 +146,52 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 					if currentC.Name == c.Name {
 						for i, v := range pod.Spec.Containers {
 							if v.Name == c.Name {
-								log.Info(c.Name)
-								log.Info(fmt.Sprint("Comparing CPU: ", fmt.Sprintf("%dm", AverageUsageCPU), " <> ", fmt.Sprintf("%dm", currentC.CPU)))
-								log.Info(fmt.Sprint("Comparing Memory: ", fmt.Sprintf("%dMi", AverageUsageMemory), " <> ", fmt.Sprintf("%dMi", currentC.Memory)))
+								if AverageUsageCPU < r.MinCPU && (r.MinCPU > 0) {
+									AverageUsageCPU = r.MinCPU
+								}
+								if AverageUsageCPU > r.MaxCPU && (r.MaxCPU > 0) {
+									AverageUsageCPU = r.MaxCPU
+								}
+								if AverageUsageMemory < r.MinMemory && (r.MinMemory > 0) {
+									AverageUsageMemory = r.MinMemory
+								}
+								if AverageUsageMemory > r.MaxMemory && (r.MaxMemory > 0) {
+									AverageUsageMemory = r.MaxMemory
+								}
+								log.Info(fmt.Sprint(c.Name, " Comparing CPU: ", fmt.Sprintf("%dm", AverageUsageCPU), " <> ", fmt.Sprintf("%dm", currentC.CPU)))
+								log.Info(fmt.Sprint(c.Name, " Comparing Memory: ", fmt.Sprintf("%dMi", AverageUsageMemory), " <> ", fmt.Sprintf("%dMi", currentC.Memory)))
 								if pod.Spec.Containers[i].Resources.Requests != nil {
 									switch r.GetPodMode(pod, ctx) {
 									case "average":
 										if r.ValidateCPU(currentC.CPU, AverageUsageCPU) {
 											pod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%dm", int(float64(AverageUsageCPU)*r.CPUFactor)))
 											PodChange = true
 										}
+										if r.ValidateMemory(currentC.Memory, AverageUsageMemory) {
+											pod.Spec.Containers[i].Resources.Requests[v1.ResourceMemory] = resource.MustParse(fmt.Sprintf("%dMi", int(float64(AverageUsageMemory)*r.MemoryFactor)))
+											PodChange = true
+										}
 									case "min":
 										if r.ValidateCPU(currentC.CPU, c.MinCPU) {
 											pod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%dm", int(float64(c.MinCPU)*r.CPUFactor)))
 											PodChange = true
 										}
+										if r.ValidateMemory(currentC.Memory, c.MinMemory) {
+											pod.Spec.Containers[i].Resources.Requests[v1.ResourceMemory] = resource.MustParse(fmt.Sprintf("%dMi", int(float64(c.MinMemory)*r.MemoryFactor)))
+											PodChange = true
+										}
 									case "max":
 										if r.ValidateCPU(currentC.CPU, c.MaxCPU) {
 											pod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%dm", int(float64(c.MaxCPU)*r.CPUFactor)))
 											PodChange = true
 										}
-									}
-								}
-								if AverageUsageMemory > 0 {
-									if pod.Spec.Containers[i].Resources.Requests != nil {
-										switch r.GetPodMode(pod, ctx) {
-										case "average":
-											if r.ValidateMemory(currentC.Memory, AverageUsageMemory) {
-												pod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%dm", int(float64(AverageUsageMemory)*r.MemoryFactor)))
-												PodChange = true
-											}
-										case "min":
-											if r.ValidateMemory(currentC.Memory, c.MinMemory) {
-												pod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%dm", int(float64(c.MinMemory)*r.MemoryFactor)))
-												PodChange = true
-											}
-										case "max":
-											if r.ValidateMemory(currentC.Memory, c.MaxMemory) {
-												pod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse(fmt.Sprintf("%dm", int(float64(c.MaxMemory)*r.MemoryFactor)))
-												PodChange = true
-											}
+										if r.ValidateMemory(currentC.Memory, c.MaxMemory) {
+											pod.Spec.Containers[i].Resources.Requests[v1.ResourceMemory] = resource.MustParse(fmt.Sprintf("%dMi", int(float64(c.MaxMemory)*r.MemoryFactor)))
+											PodChange = true
 										}
 									}
 								}
+
 								Requests = append(Requests, NewContainerRequests{Name: c.Name, Requests: pod.Spec.Containers[i].Resources})
 							}
 						}
@@ -206,68 +200,25 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.R
 			}
 			if PodChange {
 				pod.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
+
 				log.Info("Pod Requests Will Change")
 
 				if len(pod.OwnerReferences) == 0 {
 					log.Info("Pod has no owner")
 					return r.UpdateKubeObject(&pod, ctx)
 				}
 
-				var ownerName string
-				switch pod.OwnerReferences[0].Kind {
-				case "ReplicaSet":
-					replica, err := r.ClientSet.AppsV1().ReplicaSets(pod.Namespace).Get(ctx, pod.OwnerReferences[0].Name, metav1.GetOptions{})
-					if err != nil {
-						log.Error(err, err.Error())
-						return ctrl.Result{}, err
-					}
-
-					ownerName = replica.OwnerReferences[0].Name
-					if replica.OwnerReferences[0].Kind == "Deployment" {
-						log.Info("Is Owned by Deployment")
-						deployment, err := r.ClientSet.AppsV1().Deployments(pod.Namespace).Get(ctx, ownerName, metav1.GetOptions{})
-						if err != nil {
-							log.Error(err, err.Error())
-							return ctrl.Result{}, err
-						}
-						UpdatePodController(&deployment.Spec.Template.Spec, Requests, ctx)
-						deployment.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
+				err, podSpec, deployment, _ := r.GetPodParentKind(pod, ctx)
+				if err != nil {
+					return ctrl.Result{}, nil
+				}
 
-						return r.UpdateKubeObject(deployment, ctx)
-					} else {
-						log.Info("Is Owned by Unknown CRD")
-						return ctrl.Result{}, nil
-					}
-				case "DaemonSet":
-					log.Info("Is Owned by DaemonSet")
-					ownerName = pod.OwnerReferences[0].Name
-
-					deployment, err := r.ClientSet.AppsV1().DaemonSets(pod.Namespace).Get(ctx, ownerName, metav1.GetOptions{})
-					if err != nil {
-						log.Error(err, err.Error())
-						return ctrl.Result{}, err
-					}
-					UpdatePodController(&deployment.Spec.Template.Spec, Requests, ctx)
-					deployment.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
-					return r.UpdateKubeObject(deployment, ctx)
-				case "StatefulSet":
-					log.Info("Is Owned by StatefulSet")
-					ownerName = pod.OwnerReferences[0].Name
-
-					deployment, err := r.ClientSet.AppsV1().StatefulSets(pod.Namespace).Get(ctx, ownerName, metav1.GetOptions{})
-					if err != nil {
-						log.Error(err, err.Error())
-						return ctrl.Result{}, err
-					}
+				UpdatePodController(podSpec, Requests, ctx)
 
-					UpdatePodController(&deployment.Spec.Template.Spec, Requests, ctx)
-					deployment.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
-					return r.UpdateKubeObject(deployment, ctx)
-				default:
-					fmt.Printf("Could not find resource manager for type %s\n", pod.OwnerReferences[0].Kind)
-				}
+				return r.UpdateKubeObject(deployment.(client.Object), ctx)
 
 			}
+
 			err := deleteFromCache(cacheStore, SumPodRequest)
 			if err != nil {
 				log.Error(err, err.Error())

controllers/pod_controller_functions.go

@@ -12,6 +12,7 @@ import (
 
 	"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"
@@ -168,7 +169,7 @@ func GetPodRequests(pod corev1.Pod) PodRequests {
 		}
 		containerData = append(containerData, ContainerRequests{Name: c.Name, CPU: int64(nanoCores), Memory: int64(miMemory)})
 	}
-	return PodRequests{pod.Name, containerData, 0, 0, time.Now()}
+	return PodRequests{pod.Name, containerData, 0}
 }
 
 func addToCache(cacheStore cache.Store, object PodRequests) error {
@@ -208,5 +209,56 @@ func GeneratePodRequestsObjectFromRestData(restData []byte) PodRequests {
 		kiMemory, _ := strconv.Atoi(strings.ReplaceAll(c.Usage.Memory, "Ki", ""))
 		containerData = append(containerData, ContainerRequests{Name: c.Name, CPU: int64(nanoCores / 1000000), Memory: int64(kiMemory / 1000)})
 	}
-	return PodRequests{data.Metadata.Name, containerData, 0, 0, time.Now()}
+	return PodRequests{data.Metadata.Name, containerData, 0}
+}
+
+func (r *PodReconciler) MinimumUptimeOfPodInParent(pod corev1.Pod, ctx context.Context) bool {
+
+	if len(pod.OwnerReferences) == 0 {
+		return time.Since(pod.CreationTimestamp.Time).Seconds() >= r.MinSecondsBetweenPodRestart
+	}
+	err, _, _, deploymentName := r.GetPodParentKind(pod, ctx)
+	if err != nil {
+		return false
+	}
+	options := metav1.ListOptions{
+		LabelSelector: "app=" + deploymentName,
+	}
+	podList, _ := r.ClientSet.CoreV1().Pods(pod.Namespace).List(ctx, options)
+	// List() returns a pointer to slice, derefernce it, before iterating
+	for _, podInfo := range (*podList).Items {
+		if time.Since(podInfo.CreationTimestamp.Time).Seconds() < r.MinSecondsBetweenPodRestart {
+			return false
+		}
+	}
+	return true
+
+}
+
+func (r *PodReconciler) GetPodParentKind(pod corev1.Pod, ctx context.Context) (error, *v1.PodSpec, interface{}, string) {
+	switch pod.OwnerReferences[0].Kind {
+	case "ReplicaSet":
+		replica, err := r.ClientSet.AppsV1().ReplicaSets(pod.Namespace).Get(ctx, pod.OwnerReferences[0].Name, metav1.GetOptions{})
+		if err != nil {
+			log.Error(err, err.Error())
+			return err, nil, nil, ""
+		}
+		deployment, err := r.ClientSet.AppsV1().Deployments(pod.Namespace).Get(ctx, replica.OwnerReferences[0].Name, metav1.GetOptions{})
+		deployment.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
+		if replica.OwnerReferences[0].Kind == "Deployment" {
+			return err, &deployment.Spec.Template.Spec, deployment, deployment.Name
+		} else {
+			return errors.New("Is Owned by Unknown CRD"), nil, nil, ""
+		}
+	case "DaemonSet":
+		deployment, err := r.ClientSet.AppsV1().DaemonSets(pod.Namespace).Get(ctx, pod.OwnerReferences[0].Kind, metav1.GetOptions{})
+		deployment.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
+		return err, &deployment.Spec.Template.Spec, deployment, deployment.Name
+	case "StatefulSet":
+		deployment, err := r.ClientSet.AppsV1().StatefulSets(pod.Namespace).Get(ctx, pod.OwnerReferences[0].Kind, metav1.GetOptions{})
+		deployment.Annotations["reqsizer.jatalocks.github.io/changed"] = "true"
+		return err, &deployment.Spec.Template.Spec, deployment, deployment.Name
+	default:
+		return errors.New("Is Owned by Unknown CRD"), nil, nil, ""
+	}
 }

controllers/pod_controller_types.go

@@ -31,11 +31,11 @@ type PodReconciler struct {
 }
 
 type PodRequests struct {
-	Name                 string
-	ContainerRequests    []ContainerRequests
-	Sample               int
-	TimeSinceFirstSample float64
-	Timestamp            time.Time
+	Name              string
+	ContainerRequests []ContainerRequests
+	Sample            int
+	// TimeSinceFirstSample float64
+	// Timestamp            time.Time
 }
 
 type ContainerRequests struct {

main.go

@@ -74,7 +74,7 @@ func main() {
 	flag.Float64Var(&cpuFactor, "cpu-factor", 1, "A factor to multiply CPU requests when reconciling. 1 By default.")
 	flag.Float64Var(&memoryFactor, "memory-factor", 1, "A factor to multiply Memory requests when reconciling. 1 By default.")
 
-	flag.UintVar(&concurrentWorkers, "concurrent-workers", 20, "How many pods to sample in parallel. This may affect the controller's stability.")
+	flag.UintVar(&concurrentWorkers, "concurrent-workers", 10, "How many pods to sample in parallel. This may affect the controller's stability.")
 
 	flag.Int64Var(&minMemory, "min-memory", 0, "Minimum memory in (Mi) that the controller can set a pod request to. 0 is infinite")
 	flag.Int64Var(&minCPU, "min-cpu", 0, "Minimum CPU in (m) that the controller can set a pod request to. 0 is infinite")