factor: check the metric interval for CPU and memory-based balance (#613)

Author: djshow832

pkg/balance/factor/factor_cpu.go

@@ -50,17 +50,21 @@ var (
 			if len(pairs) < 2 {
 				return model.SampleValue(math.NaN())
 			}
-			pair1 := pairs[len(pairs)-2]
-			pair2 := pairs[len(pairs)-1]
-			timeDiff := float64(pair2.Timestamp-pair1.Timestamp) / 1000.0
-			if timeDiff < 1e-4 {
-				return model.SampleValue(math.NaN())
-			}
-			// Maybe the backend just rebooted.
-			if pair1.Value > pair2.Value {
-				return model.SampleValue(math.NaN())
+			lastPair := pairs[len(pairs)-1]
+			for i := len(pairs) - 2; i >= 0; i-- {
+				pair := pairs[i]
+				// Skip this one if the interval is too short.
+				seconds := float64(lastPair.Timestamp-pair.Timestamp) / 1000.0
+				if seconds < 1 {
+					continue
+				}
+				// Maybe the backend just rebooted.
+				if pair.Value > lastPair.Value {
+					return model.SampleValue(math.NaN())
+				}
+				return (lastPair.Value - pair.Value) / model.SampleValue(seconds)
 			}
-			return (pair2.Value - pair1.Value) / model.SampleValue(timeDiff)
+			return model.SampleValue(math.NaN())
 		},
 		ResultType: model.ValMatrix,
 	}

pkg/balance/factor/factor_cpu_test.go

@@ -338,6 +338,14 @@ tidb_server_maxprocs 2
 			curValue:   1,
 			finalValue: model.SampleValue(math.NaN()),
 		},
+		{
+			text: `process_cpu_seconds_total 3
+tidb_server_maxprocs 2
+`,
+			timestamp:  model.Time(3500),
+			curValue:   1.5,
+			finalValue: model.SampleValue(math.NaN()),
+		},
 	}
 
 	historyPair := make([]model.SamplePair, 0)

pkg/balance/factor/factor_memory.go

@@ -204,9 +204,14 @@ func calcMemUsage(usageHistory []model.SamplePair) (latestUsage float64, timeToO
 			latestUsage = value
 			latestTime = usageHistory[i].Timestamp
 		} else {
-			diff := latestUsage - value
-			if diff > 0.0001 {
-				timeToOOM = time.Duration(float64(latestTime-usageHistory[i].Timestamp)*(oomMemoryUsage-latestUsage)/diff) * time.Millisecond
+			timeDiff := latestTime.Sub(usageHistory[i].Timestamp)
+			if timeDiff < 10*time.Second {
+				// Skip this one is the interval is too short.
+				continue
+			}
+			usageDiff := latestUsage - value
+			if usageDiff > 1e-4 {
+				timeToOOM = timeDiff * time.Duration((oomMemoryUsage-latestUsage)/usageDiff)
 			}
 			break
 		}

pkg/balance/factor/factor_memory_test.go

@@ -109,6 +109,53 @@ func TestMemoryScore(t *testing.T) {
 	}
 }
 
+func TestMemoryUsage(t *testing.T) {
+	tests := []struct {
+		memory    []float64
+		ts        []model.Time
+		lastUsage float64
+		timeToOOM time.Duration
+	}{
+		{
+			memory:    []float64{0.2, 0.3},
+			ts:        []model.Time{model.Time(15000), model.Time(30000)},
+			lastUsage: 0.3,
+			timeToOOM: 90 * time.Second,
+		},
+		{
+			memory:    []float64{0.2, 0.3, 0.3},
+			ts:        []model.Time{model.Time(15000), model.Time(30000), model.Time(31000)},
+			lastUsage: 0.3,
+			timeToOOM: 96 * time.Second,
+		},
+		{
+			memory:    []float64{0.3, 0.3},
+			ts:        []model.Time{model.Time(30000), model.Time(31000)},
+			lastUsage: 0.3,
+			timeToOOM: time.Duration(math.MaxInt),
+		},
+		{
+			memory:    []float64{0.3, 0.3},
+			ts:        []model.Time{model.Time(30000), model.Time(45000)},
+			lastUsage: 0.3,
+			timeToOOM: time.Duration(math.MaxInt),
+		},
+		{
+			memory:    []float64{0.3, 0.2},
+			ts:        []model.Time{model.Time(30000), model.Time(45000)},
+			lastUsage: 0.2,
+			timeToOOM: time.Duration(math.MaxInt),
+		},
+	}
+
+	for i, test := range tests {
+		pairs := createPairs(test.memory, test.ts)
+		latestUsage, timeToOOM := calcMemUsage(pairs)
+		require.Equal(t, test.lastUsage, latestUsage, "test index %d", i)
+		require.Equal(t, test.timeToOOM, timeToOOM, "test index %d", i)
+	}
+}
+
 func TestMemoryBalance(t *testing.T) {
 	tests := []struct {
 		memory   [][]float64

pkg/balance/factor/mock_test.go

@@ -150,6 +150,14 @@ func createSampleStream(values []float64, backendIdx int, curTime model.Time) *m
 	return &model.SampleStream{Metric: labelSet, Values: pairs}
 }
 
+func createPairs(values []float64, ts []model.Time) []model.SamplePair {
+	pairs := make([]model.SamplePair, 0, len(values))
+	for i, value := range values {
+		pairs = append(pairs, model.SamplePair{Timestamp: ts[i], Value: model.SampleValue(value)})
+	}
+	return pairs
+}
+
 func createSample(value float64, backendIdx int) *model.Sample {
 	host := strconv.Itoa(backendIdx)
 	labelSet := model.Metric{metricsreader.LabelNameInstance: model.LabelValue(host + ":10080")}