Merge pull request #24382 from smarterclayton/fix_rlqt

Automatic merge from submit-queue

RateLimitedQueue TestTryOrdering could fail under load

Remove the possibility of contention in the test by providing a
synthetic Now() function.

Fixes #24125

pkg/controller/node/rate_limited_queue.go

@@ -164,7 +164,7 @@ func (q *RateLimitedTimedQueue) Try(fn ActionFunc) {
 	for ok {
 		// rate limit the queue checking
 		if !q.limiter.TryAccept() {
-			glog.V(10).Info("Try rate limitted...")
+			glog.V(10).Info("Try rate limited...")
 			// Try again later
 			break
 		}

pkg/controller/node/rate_limited_queue_test.go

@@ -152,6 +152,19 @@ func TestTry(t *testing.T) {
 }
 
 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")
@@ -159,18 +172,38 @@ func TestTryOrdering(t *testing.T) {
 
 	order := []string{}
 	count := 0
-	queued := false
+	hasQueued := false
 	evictor.Try(func(value TimedValue) (bool, time.Duration) {
 		count++
-		if value.AddedAt.IsZero() {
+		t.Logf("eviction %d", count)
-			t.Fatalf("added should not be zero")
-		}
 		if value.ProcessAt.IsZero() {
-			t.Fatalf("next should not be zero")
+			t.Fatalf("processAt should not be zero")
 		}
-		if !queued && value.Value == "second" {
+		switch value.Value {
-			queued = true
+		case "first":
-			return false, time.Millisecond
+			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