feat: aggregate samples into histogram per tflops

Author: knave

internal/autoscaler/autoscaler_test.go

@@ -33,6 +33,15 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+// tflops add all samples, like cpu in vpa
+// Consider gpu allocator, check if enough tflops or vram to allocate
+// cron scheduler stragegy
+// Add AutoSetResources to schedulingconfigtemplate and make it more configurable
+// refactor main, setup database may not put in leader election runnable group
+// scale to zero when query data if no usage, need carl to support
+// add recommendation to workload
+// resolve conversation on github, thanks for reviews
+
 var _ = Describe("Autoscaler", func() {
 	Context("when creating an autoscaler", func() {
 		It("should return an error if there is no client", func() {
@@ -49,8 +58,8 @@ var _ = Describe("Autoscaler", func() {
 			scaler.LoadHistoryMetrics(ctx)
 			metrics := scaler.MetricsProvider.GetHistoryMetrics()
 			for _, m := range metrics {
-				Expect(scaler.WorkloadStates).To(HaveKey(m.Workload))
-				Expect(scaler.WorkerStates).To(HaveKey(m.Worker))
+				Expect(scaler.WorkloadStates).To(HaveKey(m.WorkloadName))
+				Expect(scaler.WorkerStates).To(HaveKey(m.WorkerName))
 			}
 		})
 	})
@@ -101,7 +110,7 @@ var _ = Describe("Autoscaler", func() {
 	})
 
 	Context("when loading real time metrics", func() {
-		It("should update the state of workloads and workers", func() {
+		FIt("should update the state of workloads and workers", func() {
 			tfEnv := NewTensorFusionEnvBuilder().
 				AddPoolWithNodeCount(1).SetGpuCountPerNode(1).
 				Build()
@@ -117,17 +126,16 @@ var _ = Describe("Autoscaler", func() {
 			scaler.LoadWorkloads(ctx)
 			ws := scaler.WorkloadStates[workload.Name]
 			metrics := &WorkerMetrics{
-				Workload:    workload.Name,
-				Worker:      worker,
-				TflopsUsage: ResourceAmount(12.0),
-				VramUsage:   9000,
-				Timestamp:   time.Now(),
+				WorkloadName: workload.Name,
+				WorkerName:   worker,
+				TflopsUsage:  ResourceAmount(12.0),
+				VramUsage:    9000,
+				Timestamp:    time.Now(),
 			}
 
 			scaler.MetricsProvider = &FakeMetricsProvider{[]*WorkerMetrics{metrics}}
 			scaler.LoadRealTimeMetrics(ctx)
 
-			Expect(scaler.WorkerStates[worker].TflopsPeak).To(Equal(metrics.TflopsUsage))
 			Expect(scaler.WorkerStates[worker].LastTflopsSampleTime).To(Equal(metrics.Timestamp))
 			Expect(ws.TflopsHistogram.IsEmpty()).To(BeFalse())
 			Expect(scaler.WorkerStates[worker].VramPeak).To(Equal(metrics.VramUsage))
@@ -302,11 +310,11 @@ func (f *FakeMetricsProvider) GetHistoryMetrics() []*WorkerMetrics {
 		for hour := 0; hour < 24; hour++ {
 			idx := day*24 + hour
 			metrics = append(metrics, &WorkerMetrics{
-				Workload:    "workload-0",
-				Worker:      fmt.Sprintf("worker-%d", idx),
-				TflopsUsage: ResourceAmount(10.0 + float64(idx%10)),
-				VramUsage:   1 * 1024 * 1024 * 1024,
-				Timestamp:   startTime.Add(time.Duration(day*24+hour) * time.Hour),
+				WorkloadName: "workload-0",
+				WorkerName:   fmt.Sprintf("worker-%d", idx),
+				TflopsUsage:  ResourceAmount(10.0 + float64(idx%10)),
+				VramUsage:    1 * 1024 * 1024 * 1024,
+				Timestamp:    startTime.Add(time.Duration(day*24+hour) * time.Hour),
 			})
 		}
 	}

internal/autoscaler/workerstate.go

@@ -7,9 +7,7 @@ import (
 type WorkerState struct {
 	Name                 string
 	Workload             string
-	TflopsPeak           ResourceAmount
 	LastTflopsSampleTime time.Time
-	TflopsWindowEnd      time.Time
 
 	VramPeak           ResourceAmount
 	LastVramSampleTime time.Time
@@ -21,41 +19,17 @@ func NewWorkerState(name string, workload string) *WorkerState {
 		Name:                 name,
 		Workload:             workload,
 		LastTflopsSampleTime: time.Time{},
-		TflopsWindowEnd:      time.Time{},
 		LastVramSampleTime:   time.Time{},
 		VramWindowEnd:        time.Time{},
 	}
 }
 
 func (w *WorkerState) AddTflopsSample(workload *WorkloadState, metrics *WorkerMetrics) bool {
-	ts := metrics.Timestamp
-	if ts.Before(w.LastTflopsSampleTime) {
+	if metrics.Timestamp.Before(w.LastTflopsSampleTime) {
 		return false
 	}
-	w.LastTflopsSampleTime = ts
-	if w.TflopsWindowEnd.IsZero() {
-		w.TflopsWindowEnd = ts
-	}
-
-	addNewPeak := false
-	if ts.Before(w.TflopsWindowEnd) {
-		if w.TflopsPeak != 0 && metrics.TflopsUsage > w.TflopsPeak {
-			workload.TflopsHistogram.SubtractSample(float64(w.TflopsPeak), 1.0, w.TflopsWindowEnd)
-			addNewPeak = true
-		}
-	} else {
-		aggregationInteval := DefaultAggregationInterval
-		shift := ts.Sub(w.TflopsWindowEnd).Truncate(aggregationInteval) + aggregationInteval
-		w.TflopsWindowEnd = w.TflopsWindowEnd.Add(shift)
-		w.TflopsPeak = 0
-		addNewPeak = true
-	}
-
-	if addNewPeak {
-		workload.TflopsHistogram.AddSample(float64(metrics.TflopsUsage), 1.0, metrics.Timestamp)
-		w.TflopsPeak = metrics.TflopsUsage
-	}
-
+	workload.TflopsHistogram.AddSample(float64(metrics.TflopsUsage), minSampleWeight, metrics.Timestamp)
+	w.LastTflopsSampleTime = metrics.Timestamp
 	return true
 }