VPA AggregateContainerState serialization.

vertical-pod-autoscaler/pkg/apis/poc.autoscaling.k8s.io/v1alpha1/types.go

@@ -243,6 +243,9 @@ type VerticalPodAutoscalerCheckpointList struct {
 type VerticalPodAutoscalerCheckpointSpec struct {
 	// Name of the VPA object that stored VerticalPodAutoscalerCheckpoint object.
 	VPAObjectName string `json:"vpaObjectName,omitempty" protobuf:"bytes,1,opt,name=vpaObjectName"`
+
+	// Name of the checkpointed container.
+	ContainerName string `json:"containerName,omitempty" protobuf:"bytes,2,opt,name=containerName"`
 }
 
 // VerticalPodAutoscalerCheckpointStatus contains data of the checkpoint.
@@ -258,19 +261,25 @@ type VerticalPodAutoscalerCheckpointStatus struct {
 
 	// Checkpoint of histogram for consumption of memory.
 	MemoryHistogram HistogramCheckpoint `json:"memoryHistogram,omitempty" protobuf:"bytes,4,rep,name=memoryHistogram"`
+
+	// Timestamp of the fist sample from the histograms.
+	FirstSampleStart metav1.Time `json:"firstSampleStart,omitempty" protobuf:"bytes,5,opt,name=firstSampleStart"`
+
+	// Timestamp of the last sample from the histograms.
+	LastSampleStart metav1.Time `json:"lastSampleStart,omitempty" protobuf:"bytes,6,opt,name=lastSampleStart"`
+
+	// Total number of samples in the histograms.
+	TotalSamplesCount int `json:"totalSamplesCount,omitempty" protobuf:"bytes,7,opt,name=totalSamplesCount"`
 }
 
 // HistogramCheckpoint contains data needed to reconstruct the histogram.
 type HistogramCheckpoint struct {
-	// Name of the resource that this HistogramCheckpoint refers to.
-	Resource string `json:"resource,omitempty" protobuf:"bytes,1,opt,name=resource"`
-
 	// Reference timestamp for samples collected within this histogram.
-	ReferenceTimestamp metav1.Time `json:"referenceTimestamp,omitempty" protobuf:"bytes,2,opt,name=referenceTimestamp"`
+	ReferenceTimestamp metav1.Time `json:"referenceTimestamp,omitempty" protobuf:"bytes,1,opt,name=referenceTimestamp"`
 
 	// Map from bucket index to bucket weight.
-	BucketWeights map[int]uint32 `json:"bucketWeights,omitempty" protobuf:"bytes,3,opt,name=bucketWeights"`
+	BucketWeights map[int]uint32 `json:"bucketWeights,omitempty" protobuf:"bytes,2,opt,name=bucketWeights"`
 
 	// Sum of samples to be used as denominator for weights from BucketWeights.
-	TotalWeight float64 `json:"totalWeight,omitempty" protobuf:"bytes,4,opt,name=totalWeight"`
+	TotalWeight float64 `json:"totalWeight,omitempty" protobuf:"bytes,3,opt,name=totalWeight"`
 }

vertical-pod-autoscaler/pkg/apis/poc.autoscaling.k8s.io/v1alpha1/zz_generated.deepcopy.go

@@ -293,6 +293,8 @@ func (in *VerticalPodAutoscalerCheckpointStatus) DeepCopyInto(out *VerticalPodAu
 	in.LastUpdateTime.DeepCopyInto(&out.LastUpdateTime)
 	in.CPUHistogram.DeepCopyInto(&out.CPUHistogram)
 	in.MemoryHistogram.DeepCopyInto(&out.MemoryHistogram)
+	in.FirstSampleStart.DeepCopyInto(&out.FirstSampleStart)
+	in.LastSampleStart.DeepCopyInto(&out.LastSampleStart)
 	return
 }
 

vertical-pod-autoscaler/recommender/logic/recommender.go

@@ -17,12 +17,20 @@ limitations under the License.
 package logic
 
 import (
+	"fmt"
 	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	vpa_types "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/poc.autoscaling.k8s.io/v1alpha1"
 	"k8s.io/autoscaler/vertical-pod-autoscaler/recommender/model"
 	"k8s.io/autoscaler/vertical-pod-autoscaler/recommender/util"
 )
 
+const (
+	// SupportedCheckpointVersion is the tag of the supported version of serialized checkpoints.
+	SupportedCheckpointVersion = "v1"
+)
+
 // PodResourceRecommender computes resource recommendation for a Vpa object.
 type PodResourceRecommender interface {
 	GetRecommendedPodResources(vpa *model.Vpa) RecommendedPodResources
@@ -132,3 +140,44 @@ func (r *podResourceRecommender) getRecommendedContainerResources(s *AggregateCo
 		r.upperBoundEstimator.GetResourceEstimation(s),
 	}
 }
+
+// SaveToCheckpoint serializes AggregateContainerState as VerticalPodAutoscalerCheckpointStatus.
+// The serialization may result in loss of precission of the histograms.
+func (a *AggregateContainerState) SaveToCheckpoint() (*vpa_types.VerticalPodAutoscalerCheckpointStatus, error) {
+	memory, err := a.aggregateMemoryPeaks.SaveToChekpoint()
+	if err != nil {
+		return nil, err
+	}
+	cpu, err := a.aggregateCPUUsage.SaveToChekpoint()
+	if err != nil {
+		return nil, err
+	}
+	return &vpa_types.VerticalPodAutoscalerCheckpointStatus{
+		FirstSampleStart:  metav1.NewTime(a.firstSampleStart),
+		LastSampleStart:   metav1.NewTime(a.lastSampleStart),
+		TotalSamplesCount: a.totalSamplesCount,
+		MemoryHistogram:   *memory,
+		CPUHistogram:      *cpu,
+		Version:           SupportedCheckpointVersion,
+	}, nil
+}
+
+// LoadFromCheckpoint deserializes data from VerticalPodAutoscalerCheckpointStatus
+// into the AggregateContainerState.
+func (a *AggregateContainerState) LoadFromCheckpoint(checkpoint *vpa_types.VerticalPodAutoscalerCheckpointStatus) error {
+	if checkpoint.Version != SupportedCheckpointVersion {
+		return fmt.Errorf("Unssuported checkpoint version %s", checkpoint.Version)
+	}
+	a.totalSamplesCount = checkpoint.TotalSamplesCount
+	a.firstSampleStart = checkpoint.FirstSampleStart.Time
+	a.lastSampleStart = checkpoint.LastSampleStart.Time
+	err := a.aggregateMemoryPeaks.LoadFromCheckpoint(&checkpoint.MemoryHistogram)
+	if err != nil {
+		return err
+	}
+	err = a.aggregateCPUUsage.LoadFromCheckpoint(&checkpoint.CPUHistogram)
+	if err != nil {
+		return err
+	}
+	return nil
+}

vertical-pod-autoscaler/recommender/logic/recommender_test.go

@@ -21,6 +21,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	vpa_types "k8s.io/autoscaler/vertical-pod-autoscaler/pkg/apis/poc.autoscaling.k8s.io/v1alpha1"
 	"k8s.io/autoscaler/vertical-pod-autoscaler/recommender/model"
 	"k8s.io/autoscaler/vertical-pod-autoscaler/recommender/util"
 )
@@ -100,3 +102,73 @@ func TestBuildAggregateResourcesMap(t *testing.T) {
 	assert.True(t, expectedCPUHistogram.Equals(actualCPUHistogram), "Expected:\n%s\nActual:\n%s", expectedCPUHistogram, actualCPUHistogram)
 	assert.True(t, expectedMemoryHistogram.Equals(actualMemoryHistogram), "Expected:\n%s\nActual:\n%s", expectedMemoryHistogram, actualMemoryHistogram)
 }
+
+func TestAggregateContainerStateSaveToCheckpoint(t *testing.T) {
+	location, _ := time.LoadLocation("UTC")
+	cs := newAggregateContainerState()
+	t1, t2 := time.Date(2018, time.January, 1, 2, 3, 4, 0, location), time.Date(2018, time.February, 1, 2, 3, 4, 0, location)
+	cs.firstSampleStart = t1
+	cs.lastSampleStart = t2
+	cs.totalSamplesCount = 10
+
+	cs.aggregateCPUUsage.AddSample(1, 33, t2)
+	cs.aggregateMemoryPeaks.AddSample(1, 55, t1)
+	cs.aggregateMemoryPeaks.AddSample(10000000, 55, t1)
+	checkpoint, err := cs.SaveToCheckpoint()
+
+	assert.NoError(t, err)
+	assert.Equal(t, t1, checkpoint.FirstSampleStart.Time)
+	assert.Equal(t, t2, checkpoint.LastSampleStart.Time)
+	assert.Equal(t, 10, checkpoint.TotalSamplesCount)
+
+	assert.Equal(t, SupportedCheckpointVersion, checkpoint.Version)
+
+	// Basic check that serialization of histograms happened.
+	// Full tests are part of the Histogram.
+	assert.Len(t, checkpoint.CPUHistogram.BucketWeights, 1)
+	assert.Len(t, checkpoint.MemoryHistogram.BucketWeights, 2)
+}
+
+func TestAggregateContainerStateLoadFromCheckpointFailsForVersionMismatch(t *testing.T) {
+	checkpoint := vpa_types.VerticalPodAutoscalerCheckpointStatus{
+		Version: "foo",
+	}
+	cs := newAggregateContainerState()
+	err := cs.LoadFromCheckpoint(&checkpoint)
+	assert.Error(t, err)
+}
+
+func TestAggregateContainerStateLoadFromCheckpoint(t *testing.T) {
+	location, _ := time.LoadLocation("UTC")
+
+	t1, t2 := time.Date(2018, time.January, 1, 2, 3, 4, 0, location), time.Date(2018, time.February, 1, 2, 3, 4, 0, location)
+
+	checkpoint := vpa_types.VerticalPodAutoscalerCheckpointStatus{
+		Version:           SupportedCheckpointVersion,
+		FirstSampleStart:  metav1.NewTime(t1),
+		LastSampleStart:   metav1.NewTime(t2),
+		TotalSamplesCount: 20,
+		MemoryHistogram: vpa_types.HistogramCheckpoint{
+			BucketWeights: map[int]uint32{
+				0: 10,
+			},
+			TotalWeight: 33.0,
+		},
+		CPUHistogram: vpa_types.HistogramCheckpoint{
+			BucketWeights: map[int]uint32{
+				0: 10,
+			},
+			TotalWeight: 44.0,
+		},
+	}
+
+	cs := newAggregateContainerState()
+	err := cs.LoadFromCheckpoint(&checkpoint)
+	assert.NoError(t, err)
+
+	assert.Equal(t, t1, cs.firstSampleStart)
+	assert.Equal(t, t2, cs.lastSampleStart)
+	assert.Equal(t, 20, cs.totalSamplesCount)
+	assert.False(t, cs.aggregateCPUUsage.IsEmpty())
+	assert.False(t, cs.aggregateMemoryPeaks.IsEmpty())
+}