vertical-pod-autoscaler/pkg/recommender/routines/recommender.go (149 lines of code) (raw):

/* Copyright 2017 The Kubernetes Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ package routines import ( "context" "flag" "time" "k8s.io/klog/v2" vpa_api "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/client/clientset/versioned/typed/autoscaling.k8s.io/v1" "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/recommender/checkpoint" "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/recommender/input" "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/recommender/logic" "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/recommender/model" controllerfetcher "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/target/controller_fetcher" metrics_recommender "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/utils/metrics/recommender" vpa_utils "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/utils/vpa" ) var ( checkpointsWriteTimeout = flag.Duration("checkpoints-timeout", time.Minute, `Timeout for writing checkpoints since the start of the recommender's main loop`) minCheckpointsPerRun = flag.Int("min-checkpoints", 10, "Minimum number of checkpoints to write per recommender's main loop") ) // Recommender recommend resources for certain containers, based on utilization periodically got from metrics api. type Recommender interface { // RunOnce performs one iteration of recommender duties followed by update of recommendations in VPA objects. RunOnce() // GetClusterState returns ClusterState used by Recommender GetClusterState() model.ClusterState // GetClusterStateFeeder returns ClusterStateFeeder used by Recommender GetClusterStateFeeder() input.ClusterStateFeeder // UpdateVPAs computes recommendations and sends VPAs status updates to API Server UpdateVPAs() // MaintainCheckpoints stores current checkpoints in API Server and garbage collect old ones // MaintainCheckpoints writes at least minCheckpoints if there are more checkpoints to write. // Checkpoints are written until ctx permits or all checkpoints are written. MaintainCheckpoints(ctx context.Context, minCheckpoints int) } type recommender struct { clusterState model.ClusterState clusterStateFeeder input.ClusterStateFeeder checkpointWriter checkpoint.CheckpointWriter checkpointsGCInterval time.Duration controllerFetcher controllerfetcher.ControllerFetcher lastCheckpointGC time.Time vpaClient vpa_api.VerticalPodAutoscalersGetter podResourceRecommender logic.PodResourceRecommender useCheckpoints bool lastAggregateContainerStateGC time.Time recommendationPostProcessor []RecommendationPostProcessor } func (r *recommender) GetClusterState() model.ClusterState { return r.clusterState } func (r *recommender) GetClusterStateFeeder() input.ClusterStateFeeder { return r.clusterStateFeeder } // UpdateVPAs update VPA CRD objects' status. func (r *recommender) UpdateVPAs() { cnt := metrics_recommender.NewObjectCounter() defer cnt.Observe() for _, observedVpa := range r.clusterState.ObservedVPAs() { key := model.VpaID{ Namespace: observedVpa.Namespace, VpaName: observedVpa.Name, } vpa, found := r.clusterState.VPAs()[key] if !found { continue } resources := r.podResourceRecommender.GetRecommendedPodResources(GetContainerNameToAggregateStateMap(vpa)) had := vpa.HasRecommendation() listOfResourceRecommendation := logic.MapToListOfRecommendedContainerResources(resources) for _, postProcessor := range r.recommendationPostProcessor { listOfResourceRecommendation = postProcessor.Process(observedVpa, listOfResourceRecommendation) } vpa.UpdateRecommendation(listOfResourceRecommendation) if vpa.HasRecommendation() && !had { metrics_recommender.ObserveRecommendationLatency(vpa.Created) } hasMatchingPods := vpa.PodCount > 0 vpa.UpdateConditions(hasMatchingPods) if err := r.clusterState.RecordRecommendation(vpa, time.Now()); err != nil { klog.V(0).InfoS("", "err", err) if klog.V(4).Enabled() { pods := r.clusterState.GetMatchingPods(vpa) if len(pods) != vpa.PodCount { klog.ErrorS(nil, "ClusterState pod count and matching pods disagree for VPA", "vpa", klog.KRef(vpa.ID.Namespace, vpa.ID.VpaName), "podCount", vpa.PodCount, "matchingPods", pods) } klog.InfoS("VPA dump", "vpa", vpa, "hasMatchingPods", hasMatchingPods, "podCount", vpa.PodCount, "matchingPods", pods) } } cnt.Add(vpa) _, err := vpa_utils.UpdateVpaStatusIfNeeded( r.vpaClient.VerticalPodAutoscalers(vpa.ID.Namespace), vpa.ID.VpaName, vpa.AsStatus(), &observedVpa.Status) if err != nil { klog.ErrorS(err, "Cannot update VPA", "vpa", klog.KRef(vpa.ID.Namespace, vpa.ID.VpaName)) } } } func (r *recommender) MaintainCheckpoints(ctx context.Context, minCheckpointsPerRun int) { now := time.Now() if r.useCheckpoints { if err := r.checkpointWriter.StoreCheckpoints(ctx, now, minCheckpointsPerRun); err != nil { klog.V(0).InfoS("Failed to store checkpoints", "err", err) } if time.Since(r.lastCheckpointGC) > r.checkpointsGCInterval { r.lastCheckpointGC = now r.clusterStateFeeder.GarbageCollectCheckpoints(ctx) } } } func (r *recommender) RunOnce() { timer := metrics_recommender.NewExecutionTimer() defer timer.ObserveTotal() ctx := context.Background() klog.V(3).InfoS("Recommender Run") r.clusterStateFeeder.LoadVPAs(ctx) timer.ObserveStep("LoadVPAs") r.clusterStateFeeder.LoadPods() timer.ObserveStep("LoadPods") r.clusterStateFeeder.LoadRealTimeMetrics(ctx) timer.ObserveStep("LoadMetrics") klog.V(3).InfoS("ClusterState is tracking", "pods", len(r.clusterState.Pods()), "vpas", len(r.clusterState.VPAs())) r.UpdateVPAs() timer.ObserveStep("UpdateVPAs") stepCtx, cancelFunc := context.WithDeadline(ctx, time.Now().Add(*checkpointsWriteTimeout)) defer cancelFunc() r.MaintainCheckpoints(stepCtx, *minCheckpointsPerRun) timer.ObserveStep("MaintainCheckpoints") r.clusterState.RateLimitedGarbageCollectAggregateCollectionStates(ctx, time.Now(), r.controllerFetcher) timer.ObserveStep("GarbageCollect") klog.V(3).InfoS("ClusterState is tracking", "aggregateContainerStates", r.clusterState.StateMapSize()) } // RecommenderFactory makes instances of Recommender. type RecommenderFactory struct { ClusterState model.ClusterState ClusterStateFeeder input.ClusterStateFeeder ControllerFetcher controllerfetcher.ControllerFetcher CheckpointWriter checkpoint.CheckpointWriter PodResourceRecommender logic.PodResourceRecommender VpaClient vpa_api.VerticalPodAutoscalersGetter RecommendationPostProcessors []RecommendationPostProcessor CheckpointsGCInterval time.Duration UseCheckpoints bool } // Make creates a new recommender instance, // which can be run in order to provide continuous resource recommendations for containers. func (c RecommenderFactory) Make() Recommender { recommender := &recommender{ clusterState: c.ClusterState, clusterStateFeeder: c.ClusterStateFeeder, checkpointWriter: c.CheckpointWriter, checkpointsGCInterval: c.CheckpointsGCInterval, controllerFetcher: c.ControllerFetcher, useCheckpoints: c.UseCheckpoints, vpaClient: c.VpaClient, podResourceRecommender: c.PodResourceRecommender, recommendationPostProcessor: c.RecommendationPostProcessors, lastAggregateContainerStateGC: time.Now(), lastCheckpointGC: time.Now(), } klog.V(3).InfoS("New Recommender created", "recommender", recommender) return recommender }