mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-10-31 05:40:42 +00:00
4e8e4a574c
Automatic merge from submit-queue use monotonic now in TestDelNode Fixes https://github.com/kubernetes/kubernetes/issues/24971. Briefly, the rate_limited_queue uses a `container/heap` to store values, and use this data structure to ensure we can always fetch the value with the minimum `processAt`. However, in some extreme condition, the continuous call to `time.Now()` would get the same value, which causes some unpredictable order in the queue, this fix uses a monotonic `now()` to avoid that. @smarterclayton please take a look.
284 lines
8.0 KiB
Go
284 lines
8.0 KiB
Go
/*
|
|
Copyright 2015 The Kubernetes Authors All rights reserved.
|
|
|
|
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 node
|
|
|
|
import (
|
|
"reflect"
|
|
"testing"
|
|
"time"
|
|
|
|
"k8s.io/kubernetes/pkg/util/flowcontrol"
|
|
"k8s.io/kubernetes/pkg/util/sets"
|
|
)
|
|
|
|
func CheckQueueEq(lhs []string, rhs TimedQueue) bool {
|
|
for i := 0; i < len(lhs); i++ {
|
|
if rhs[i].Value != lhs[i] {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func CheckSetEq(lhs, rhs sets.String) bool {
|
|
return lhs.HasAll(rhs.List()...) && rhs.HasAll(lhs.List()...)
|
|
}
|
|
|
|
func TestAddNode(t *testing.T) {
|
|
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
|
|
queuePattern := []string{"first", "second", "third"}
|
|
if len(evictor.queue.queue) != len(queuePattern) {
|
|
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
|
}
|
|
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
|
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
|
}
|
|
|
|
setPattern := sets.NewString("first", "second", "third")
|
|
if len(evictor.queue.set) != len(setPattern) {
|
|
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
|
}
|
|
if !CheckSetEq(setPattern, evictor.queue.set) {
|
|
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
|
}
|
|
}
|
|
|
|
func TestDelNode(t *testing.T) {
|
|
defer func() { now = time.Now }()
|
|
var tick int64
|
|
now = func() time.Time {
|
|
t := time.Unix(tick, 0)
|
|
tick++
|
|
return t
|
|
}
|
|
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
evictor.Remove("first")
|
|
|
|
queuePattern := []string{"second", "third"}
|
|
if len(evictor.queue.queue) != len(queuePattern) {
|
|
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
|
}
|
|
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
|
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
|
}
|
|
|
|
setPattern := sets.NewString("second", "third")
|
|
if len(evictor.queue.set) != len(setPattern) {
|
|
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
|
}
|
|
if !CheckSetEq(setPattern, evictor.queue.set) {
|
|
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
|
}
|
|
|
|
evictor = NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
evictor.Remove("second")
|
|
|
|
queuePattern = []string{"first", "third"}
|
|
if len(evictor.queue.queue) != len(queuePattern) {
|
|
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
|
}
|
|
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
|
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
|
}
|
|
|
|
setPattern = sets.NewString("first", "third")
|
|
if len(evictor.queue.set) != len(setPattern) {
|
|
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
|
}
|
|
if !CheckSetEq(setPattern, evictor.queue.set) {
|
|
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
|
}
|
|
|
|
evictor = NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
evictor.Remove("third")
|
|
|
|
queuePattern = []string{"first", "second"}
|
|
if len(evictor.queue.queue) != len(queuePattern) {
|
|
t.Fatalf("Queue %v should have length %d", evictor.queue.queue, len(queuePattern))
|
|
}
|
|
if !CheckQueueEq(queuePattern, evictor.queue.queue) {
|
|
t.Errorf("Invalid queue. Got %v, expected %v", evictor.queue.queue, queuePattern)
|
|
}
|
|
|
|
setPattern = sets.NewString("first", "second")
|
|
if len(evictor.queue.set) != len(setPattern) {
|
|
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
|
}
|
|
if !CheckSetEq(setPattern, evictor.queue.set) {
|
|
t.Errorf("Invalid map. Got %v, expected %v", evictor.queue.set, setPattern)
|
|
}
|
|
}
|
|
|
|
func TestTry(t *testing.T) {
|
|
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
evictor.Remove("second")
|
|
|
|
deletedMap := sets.NewString()
|
|
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
|
deletedMap.Insert(value.Value)
|
|
return true, 0
|
|
})
|
|
|
|
setPattern := sets.NewString("first", "third")
|
|
if len(deletedMap) != len(setPattern) {
|
|
t.Fatalf("Map %v should have length %d", evictor.queue.set, len(setPattern))
|
|
}
|
|
if !CheckSetEq(setPattern, deletedMap) {
|
|
t.Errorf("Invalid map. Got %v, expected %v", deletedMap, setPattern)
|
|
}
|
|
}
|
|
|
|
func TestTryOrdering(t *testing.T) {
|
|
defer func() { now = time.Now }()
|
|
current := time.Unix(0, 0)
|
|
delay := 0
|
|
// the current time is incremented by 1ms every time now is invoked
|
|
now = func() time.Time {
|
|
if delay > 0 {
|
|
delay--
|
|
} else {
|
|
current = current.Add(time.Millisecond)
|
|
}
|
|
t.Logf("time %d", current.UnixNano())
|
|
return current
|
|
}
|
|
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
|
|
order := []string{}
|
|
count := 0
|
|
hasQueued := false
|
|
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
|
count++
|
|
t.Logf("eviction %d", count)
|
|
if value.ProcessAt.IsZero() {
|
|
t.Fatalf("processAt should not be zero")
|
|
}
|
|
switch value.Value {
|
|
case "first":
|
|
if !value.AddedAt.Equal(time.Unix(0, time.Millisecond.Nanoseconds())) {
|
|
t.Fatalf("added time for %s is %d", value.Value, value.AddedAt)
|
|
}
|
|
|
|
case "second":
|
|
if !value.AddedAt.Equal(time.Unix(0, 2*time.Millisecond.Nanoseconds())) {
|
|
t.Fatalf("added time for %s is %d", value.Value, value.AddedAt)
|
|
}
|
|
if hasQueued {
|
|
if !value.ProcessAt.Equal(time.Unix(0, 6*time.Millisecond.Nanoseconds())) {
|
|
t.Fatalf("process time for %s is %d", value.Value, value.ProcessAt)
|
|
}
|
|
break
|
|
}
|
|
hasQueued = true
|
|
delay = 1
|
|
t.Logf("going to delay")
|
|
return false, 2 * time.Millisecond
|
|
|
|
case "third":
|
|
if !value.AddedAt.Equal(time.Unix(0, 3*time.Millisecond.Nanoseconds())) {
|
|
t.Fatalf("added time for %s is %d", value.Value, value.AddedAt)
|
|
}
|
|
}
|
|
order = append(order, value.Value)
|
|
return true, 0
|
|
})
|
|
if !reflect.DeepEqual(order, []string{"first", "third"}) {
|
|
t.Fatalf("order was wrong: %v", order)
|
|
}
|
|
if count != 3 {
|
|
t.Fatalf("unexpected iterations: %d", count)
|
|
}
|
|
}
|
|
|
|
func TestTryRemovingWhileTry(t *testing.T) {
|
|
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
|
|
processing := make(chan struct{})
|
|
wait := make(chan struct{})
|
|
order := []string{}
|
|
count := 0
|
|
queued := false
|
|
|
|
// while the Try function is processing "second", remove it from the queue
|
|
// we should not see "second" retried.
|
|
go func() {
|
|
<-processing
|
|
evictor.Remove("second")
|
|
close(wait)
|
|
}()
|
|
|
|
evictor.Try(func(value TimedValue) (bool, time.Duration) {
|
|
count++
|
|
if value.AddedAt.IsZero() {
|
|
t.Fatalf("added should not be zero")
|
|
}
|
|
if value.ProcessAt.IsZero() {
|
|
t.Fatalf("next should not be zero")
|
|
}
|
|
if !queued && value.Value == "second" {
|
|
queued = true
|
|
close(processing)
|
|
<-wait
|
|
return false, time.Millisecond
|
|
}
|
|
order = append(order, value.Value)
|
|
return true, 0
|
|
})
|
|
|
|
if !reflect.DeepEqual(order, []string{"first", "third"}) {
|
|
t.Fatalf("order was wrong: %v", order)
|
|
}
|
|
if count != 3 {
|
|
t.Fatalf("unexpected iterations: %d", count)
|
|
}
|
|
}
|
|
|
|
func TestClear(t *testing.T) {
|
|
evictor := NewRateLimitedTimedQueue(flowcontrol.NewFakeAlwaysRateLimiter())
|
|
evictor.Add("first")
|
|
evictor.Add("second")
|
|
evictor.Add("third")
|
|
|
|
evictor.Clear()
|
|
|
|
if len(evictor.queue.queue) != 0 {
|
|
t.Fatalf("Clear should remove all elements from the queue.")
|
|
}
|
|
}
|