This is an automated email from the ASF dual-hosted git repository.
pbacsko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new d46f216e [YUNIKORN-2643] utils.go WaitForCondition test coverage
improvement (#875)
d46f216e is described below
commit d46f216e2fa85a38edff11d48e82f481a0c03f3c
Author: mean-world <[email protected]>
AuthorDate: Thu Jun 6 18:04:37 2024 +0200
[YUNIKORN-2643] utils.go WaitForCondition test coverage improvement (#875)
Closes: #875
Signed-off-by: Peter Bacsko <[email protected]>
---
pkg/common/utils.go | 26 +++++-------
pkg/common/utils_test.go | 50 +++++++++++-------------
pkg/events/event_publisher_test.go | 20 ++++++----
pkg/events/event_system_test.go | 13 +++---
pkg/scheduler/health_checker_test.go | 17 +++++---
pkg/scheduler/objects/application_state_test.go | 12 +++---
pkg/scheduler/objects/application_test.go | 28 ++++++-------
pkg/scheduler/objects/queue_test.go | 4 +-
pkg/scheduler/partition_test.go | 4 +-
pkg/scheduler/tests/application_tracking_test.go | 2 +-
pkg/scheduler/tests/mock_rm_callback_test.go | 18 ++++-----
pkg/scheduler/tests/mockscheduler_test.go | 2 +-
pkg/scheduler/tests/plugin_test.go | 2 +-
pkg/scheduler/tests/smoke_test.go | 10 ++---
pkg/scheduler/tests/utilities_test.go | 18 ++++-----
pkg/webservice/handlers_test.go | 6 +--
16 files changed, 116 insertions(+), 116 deletions(-)
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index 51383001..cdf6fd45 100644
--- a/pkg/common/utils.go
+++ b/pkg/common/utils.go
@@ -19,6 +19,7 @@
package common
import (
+ "errors"
"fmt"
"os"
"strconv"
@@ -33,6 +34,11 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
+var (
+ // ErrorTimeout returned if waiting for a condition times out
+ ErrorTimeout = errors.New("timeout waiting for condition")
+)
+
func GetNormalizedPartitionName(partitionName string, rmID string) string {
if partitionName == "" {
partitionName = "default"
@@ -62,20 +68,6 @@ func GetPartitionNameWithoutClusterID(partitionName string)
string {
return partitionName
}
-func WaitFor(interval time.Duration, timeout time.Duration, condition func()
bool) error {
- deadline := time.Now().Add(timeout)
- for {
- if time.Now().After(deadline) {
- return fmt.Errorf("timeout waiting for condition")
- }
- if condition() {
- return nil
- }
- time.Sleep(interval)
- continue
- }
-}
-
// Generate a new uuid. The chance that we generate a collision is really
small.
// As long as we check the UUID before we communicate it back to the RM we can
still replace it without a problem.
func GetNewUUID() string {
@@ -223,15 +215,15 @@ func ZeroTimeInUnixNano(t time.Time) *int64 {
return &tInt
}
-func WaitForCondition(eval func() bool, interval time.Duration, timeout
time.Duration) error {
+func WaitForCondition(interval time.Duration, timeout time.Duration, eval
func() bool) error {
deadline := time.Now().Add(timeout)
for {
if eval() {
return nil
}
- if time.Now().After(deadline) {
- return fmt.Errorf("timeout waiting for condition")
+ if time.Now().After(deadline) ||
time.Now().Add(interval).After(deadline) {
+ return ErrorTimeout
}
time.Sleep(interval)
diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go
index c286e8c4..fb62e46e 100644
--- a/pkg/common/utils_test.go
+++ b/pkg/common/utils_test.go
@@ -234,36 +234,30 @@ func TestConvertSITimestamp(t *testing.T) {
assert.Equal(t, result, time.Time{})
}
-func TestWaitFor(t *testing.T) {
- var tests = []struct {
- testname string
- bound int
- ErrorExist bool
+func TestWaitForCondition(t *testing.T) {
+ target := false
+ eval := func() bool {
+ return target
+ }
+ tests := []struct {
+ input bool
+ interval time.Duration
+ timeout time.Duration
+ output error
}{
- {"Timeout case", 10000, true},
- {"Fullfilling case", 10, false},
+ {true, time.Duration(1) * time.Second, time.Duration(2) *
time.Second, nil},
+ {false, time.Duration(1) * time.Second, time.Duration(2) *
time.Second, ErrorTimeout},
+ {true, time.Duration(3) * time.Second, time.Duration(2) *
time.Second, nil},
+ {false, time.Duration(3) * time.Second, time.Duration(2) *
time.Second, ErrorTimeout},
}
- for _, tt := range tests {
- t.Run(tt.testname, func(t *testing.T) {
- count := 0
- err := WaitFor(time.Nanosecond, time.Millisecond,
func() bool {
- if count <= tt.bound {
- count++
- return false
- }
- return true
- })
- switch tt.ErrorExist {
- case true:
- if errorExist := (err != nil); !errorExist {
- t.Errorf("ErrorExist: got %v, expected
%v", errorExist, tt.ErrorExist)
- }
- case false:
- if errorExist := (err == nil); !errorExist {
- t.Errorf("ErrorExist: got %v, expected
%v", errorExist, tt.ErrorExist)
- }
- }
- })
+ for _, test := range tests {
+ target = test.input
+ get := WaitForCondition(test.interval, test.timeout, eval)
+ if test.output == nil {
+ assert.NilError(t, get)
+ } else {
+ assert.Equal(t, get.Error(), test.output.Error())
+ }
}
}
diff --git a/pkg/events/event_publisher_test.go
b/pkg/events/event_publisher_test.go
index 79fc95fc..78680b59 100644
--- a/pkg/events/event_publisher_test.go
+++ b/pkg/events/event_publisher_test.go
@@ -61,9 +61,12 @@ func TestNoFillWithoutEventPluginRegistered(t *testing.T) {
}
store.Store(event)
- err := common.WaitForCondition(func() bool {
- return store.CountStoredEvents() == 0
- }, time.Millisecond, time.Second)
+ err := common.WaitForCondition(time.Millisecond,
+ time.Second,
+ func() bool {
+ return store.CountStoredEvents() == 0
+ },
+ )
assert.NilError(t, err, "the Publisher should erase the store even if
no EventPlugin registered")
}
@@ -92,10 +95,13 @@ func TestPublisherSendsEvent(t *testing.T) {
store.Store(event)
var eventFromPlugin *si.EventRecord
- err := common.WaitForCondition(func() bool {
- eventFromPlugin = eventPlugin.GetNextEventRecord()
- return eventFromPlugin != nil
- }, time.Millisecond, time.Second)
+ err := common.WaitForCondition(time.Millisecond,
+ time.Second,
+ func() bool {
+ eventFromPlugin = eventPlugin.GetNextEventRecord()
+ return eventFromPlugin != nil
+ },
+ )
assert.NilError(t, err, "event was not received in time: %v", err)
assert.Equal(t, eventFromPlugin.ObjectID, "ask")
assert.Equal(t, eventFromPlugin.ReferenceID, "app")
diff --git a/pkg/events/event_system_test.go b/pkg/events/event_system_test.go
index 5d2f9d05..f2addd93 100644
--- a/pkg/events/event_system_test.go
+++ b/pkg/events/event_system_test.go
@@ -63,7 +63,7 @@ func TestSingleEventStoredCorrectly(t *testing.T) {
eventSystem.AddEvent(&event)
// wait for events to be processed
- err := common.WaitFor(time.Millisecond, time.Second, func() bool {
+ err := common.WaitForCondition(time.Millisecond, time.Second, func()
bool {
return eventSystem.Store.CountStoredEvents() == 1
})
assert.NilError(t, err, "the event should have been processed")
@@ -95,7 +95,7 @@ func TestGetEvents(t *testing.T) {
}
eventSystem.AddEvent(event)
}
- err := common.WaitFor(time.Millisecond, time.Second, func() bool {
+ err := common.WaitForCondition(time.Millisecond, time.Second, func()
bool {
return eventSystem.Store.CountStoredEvents() == 10
})
assert.NilError(t, err, "the events should have been processed")
@@ -132,9 +132,12 @@ func TestConfigUpdate(t *testing.T) {
configs.CMEventRingBufferCapacity:
strconv.FormatUint(newRingBufferCapacity, 10),
configs.CMEventRequestCapacity:
strconv.FormatUint(newRequestCapacity, 10),
})
- err := common.WaitForCondition(func() bool {
- return !eventSystem.IsEventTrackingEnabled()
- }, 10*time.Millisecond, 5*time.Second)
+ err := common.WaitForCondition(10*time.Millisecond,
+ 5*time.Second,
+ func() bool {
+ return !eventSystem.IsEventTrackingEnabled()
+ },
+ )
assert.NilError(t, err, "timed out waiting for config refresh")
assert.Equal(t, eventSystem.GetRingBufferCapacity(),
newRingBufferCapacity)
diff --git a/pkg/scheduler/health_checker_test.go
b/pkg/scheduler/health_checker_test.go
index 23fc8854..15fb4dd8 100644
--- a/pkg/scheduler/health_checker_test.go
+++ b/pkg/scheduler/health_checker_test.go
@@ -120,9 +120,12 @@ func TestConfigUpdate(t *testing.T) {
// update config to run every 100ms and wait for refresh
configs.SetConfigMap(map[string]string{configs.HealthCheckInterval:
"100ms"})
- err = common.WaitForCondition(func() bool {
- return healthChecker.GetPeriod() == 100*time.Millisecond
- }, 10*time.Millisecond, 5*time.Second)
+ err = common.WaitForCondition(10*time.Millisecond,
+ 5*time.Second,
+ func() bool {
+ return healthChecker.GetPeriod() == 100*time.Millisecond
+ },
+ )
assert.NilError(t, err, "timed out waiting for config refresh")
// clear results
@@ -134,9 +137,11 @@ func TestConfigUpdate(t *testing.T) {
// disable and wait for refresh
configs.SetConfigMap(map[string]string{configs.HealthCheckInterval:
"0"})
- err = common.WaitForCondition(func() bool {
- return !healthChecker.IsEnabled()
- }, 10*time.Millisecond, 5*time.Second)
+ err = common.WaitForCondition(10*time.Millisecond,
+ 5*time.Second, func() bool {
+ return !healthChecker.IsEnabled()
+ },
+ )
assert.NilError(t, err, "timed out waiting for config refresh")
// clear results
diff --git a/pkg/scheduler/objects/application_state_test.go
b/pkg/scheduler/objects/application_state_test.go
index 7a41fb36..3b64e3fe 100644
--- a/pkg/scheduler/objects/application_state_test.go
+++ b/pkg/scheduler/objects/application_state_test.go
@@ -49,7 +49,7 @@ func TestAcceptStateTransition(t *testing.T) {
// accepted to failed
err = app.HandleApplicationEvent(FailApplication)
assert.NilError(t, err, "no error expected accepted to failing")
- err = common.WaitFor(10*time.Microsecond, time.Millisecond*100,
app.IsFailing)
+ err = common.WaitForCondition(10*time.Microsecond,
time.Millisecond*100, app.IsFailing)
assert.NilError(t, err, "App should be in Failing state")
}
@@ -97,7 +97,7 @@ func TestRunStateTransition(t *testing.T) {
// run to failing
err = appInfo.HandleApplicationEvent(FailApplication)
assert.NilError(t, err, "no error expected running to failing")
- err = common.WaitFor(10*time.Microsecond, time.Millisecond*100,
appInfo.IsFailing)
+ err = common.WaitForCondition(10*time.Microsecond,
time.Millisecond*100, appInfo.IsFailing)
assert.NilError(t, err, "App should be in Failing state")
// run fails from failing
@@ -195,7 +195,7 @@ func TestFailedStateTransition(t *testing.T) {
// new to failing
err := appInfo.HandleApplicationEvent(FailApplication)
assert.NilError(t, err, "no error expected new to failing")
- err = common.WaitFor(10*time.Microsecond, time.Millisecond*100,
appInfo.IsFailing)
+ err = common.WaitForCondition(10*time.Microsecond,
time.Millisecond*100, appInfo.IsFailing)
assert.NilError(t, err, "App should be in Failing state")
// failing to failed
@@ -206,7 +206,7 @@ func TestFailedStateTransition(t *testing.T) {
// failed to rejected: error expected
err = appInfo.HandleApplicationEvent(RejectApplication)
assert.Assert(t, err != nil, "error expected failing to rejected")
- err = common.WaitFor(10*time.Microsecond, time.Millisecond*100,
appInfo.IsFailed)
+ err = common.WaitForCondition(10*time.Microsecond,
time.Millisecond*100, appInfo.IsFailed)
assert.NilError(t, err, "App should be in Failed state")
}
@@ -235,7 +235,7 @@ func TestAppStateTransitionEvents(t *testing.T) {
// run to failing
err = appInfo.HandleApplicationEvent(FailApplication)
assert.NilError(t, err, "no error expected new to failing")
- err = common.WaitFor(10*time.Microsecond, time.Millisecond*100,
appInfo.IsFailing)
+ err = common.WaitForCondition(10*time.Microsecond,
time.Millisecond*100, appInfo.IsFailing)
assert.NilError(t, err, "App should be in Failing state")
// failing to failed
@@ -255,7 +255,7 @@ func TestAppStateTransitionEvents(t *testing.T) {
assert.Assert(t, appInfo.IsResuming(), "App should be in Resuming
state")
// Verify application events
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
return eventSystem.Store.CountStoredEvents() == 8
})
diff --git a/pkg/scheduler/objects/application_test.go
b/pkg/scheduler/objects/application_test.go
index 04353a48..eb3df4bd 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -414,7 +414,7 @@ func TestAddAllocAsk(t *testing.T) {
// test add alloc ask event
noEvents := uint64(0)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
fmt.Printf("checking event length: %d\n",
eventSystem.Store.CountStoredEvents())
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 2
@@ -1033,10 +1033,10 @@ func TestCompleted(t *testing.T) {
assert.NilError(t, err, "no error expected accepted to completing
(completed test)")
assert.Assert(t, app.IsCompleting(), "App should be waiting")
// give it some time to run and progress
- err = common.WaitFor(10*time.Microsecond, time.Millisecond*200,
app.IsCompleted)
+ err = common.WaitForCondition(10*time.Microsecond,
time.Millisecond*200, app.IsCompleted)
assert.NilError(t, err, "Application did not progress into Completed
state")
- err = common.WaitFor(1*time.Millisecond, time.Millisecond*200,
app.IsExpired)
+ err = common.WaitForCondition(1*time.Millisecond, time.Millisecond*200,
app.IsExpired)
assert.NilError(t, err, "Application did not progress into Expired
state")
assert.Assert(t, app.sortedRequests == nil)
@@ -1157,13 +1157,13 @@ func TestRejected(t *testing.T) {
err := app.handleApplicationEventWithInfoLocking(RejectApplication,
rejectedMessage)
assert.NilError(t, err, "no error expected new to rejected")
- err = common.WaitFor(1*time.Millisecond, time.Millisecond*200,
app.IsRejected)
+ err = common.WaitForCondition(1*time.Millisecond, time.Millisecond*200,
app.IsRejected)
assert.NilError(t, err, "Application did not progress into Rejected
state")
assert.Assert(t, !app.FinishedTime().IsZero())
assert.Equal(t, app.rejectedMessage, rejectedMessage)
assert.Equal(t, app.GetRejectedMessage(), rejectedMessage)
- err = common.WaitFor(1*time.Millisecond, time.Millisecond*200,
app.IsExpired)
+ err = common.WaitForCondition(1*time.Millisecond, time.Millisecond*200,
app.IsExpired)
assert.NilError(t, err, "Application did not progress into Expired
state")
log := app.GetStateLog()
@@ -1422,7 +1422,7 @@ func runTimeoutPlaceholderTest(t *testing.T,
expectedState string, gangSchedulin
ph = newPlaceholderAlloc(appID1, nodeID1, res)
app.AddAllocation(ph)
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res,
2))
- err = common.WaitFor(10*time.Millisecond, 1*time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, 1*time.Second,
func() bool {
app.RLock()
defer app.RUnlock()
return app.placeholderTimer == nil
@@ -1499,7 +1499,7 @@ func TestTimeoutPlaceholderAllocReleased(t *testing.T) {
alloc := newAllocation(appID1, nodeID1, res)
app.AddAllocation(alloc)
assert.Assert(t, app.IsRunning(), "App should be in running state after
the first allocation")
- err = common.WaitFor(10*time.Millisecond, 1*time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, 1*time.Second,
func() bool {
return app.getPlaceholderTimer() == nil
})
assert.NilError(t, err, "Placeholder timeout cleanup did not trigger
unexpectedly")
@@ -1554,7 +1554,7 @@ func TestTimeoutPlaceholderCompleting(t *testing.T) {
assert.Assert(t, app.IsCompleting(), "App should be in completing state
all allocs have been removed")
assertUserGroupResource(t, getTestUserGroup(), resources.Multiply(res,
1))
// make sure the placeholders time out
- err = common.WaitFor(10*time.Millisecond, 1*time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, 1*time.Second,
func() bool {
return app.getPlaceholderTimer() == nil
})
assert.NilError(t, err, "Placeholder timer did not time out as
expected")
@@ -2032,7 +2032,7 @@ func TestAskEvents(t *testing.T) {
assert.NilError(t, err)
app.RemoveAllocationAsk(ask.allocationKey)
noEvents := uint64(0)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 3
})
@@ -2055,7 +2055,7 @@ func TestAskEvents(t *testing.T) {
err = app.AddAllocationAsk(ask3)
assert.NilError(t, err)
app.removeAsksInternal("", si.EventRecord_REQUEST_TIMEOUT)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 6
})
@@ -2104,7 +2104,7 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
app.RemoveAllocation(alloc1.GetAllocationKey(),
si.TerminationType_STOPPED_BY_RM)
app.RemoveAllocation(alloc2.GetAllocationKey(),
si.TerminationType_PLACEHOLDER_REPLACED)
noEvents := uint64(0)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 5
})
@@ -2138,7 +2138,7 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
app.AddAllocation(alloc1)
app.ReplaceAllocation(alloc1.GetAllocationKey())
noEvents = 0
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 2
})
@@ -2160,7 +2160,7 @@ func TestAllocationEvents(t *testing.T) { //nolint:funlen
app.AddAllocation(alloc1)
app.AddAllocation(alloc2)
app.RemoveAllAllocations()
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 4
})
@@ -2227,7 +2227,7 @@ func TestPlaceholderLargerEvent(t *testing.T) {
})
noEvents := uint64(0)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 4
})
diff --git a/pkg/scheduler/objects/queue_test.go
b/pkg/scheduler/objects/queue_test.go
index 2a36f50f..1b2c4d8c 100644
--- a/pkg/scheduler/objects/queue_test.go
+++ b/pkg/scheduler/objects/queue_test.go
@@ -2459,7 +2459,7 @@ func TestQueueEvents(t *testing.T) {
queue.AddApplication(app)
queue.RemoveApplication(app)
noEvents := uint64(0)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 5
})
@@ -2488,7 +2488,7 @@ func TestQueueEvents(t *testing.T) {
}
err = queue.ApplyConf(newConf)
assert.NilError(t, err)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 3
})
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index e99d5761..88ec551a 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -2660,7 +2660,7 @@ func completeApplicationAndWait(app *objects.Application,
pc *PartitionContext)
return err
}
- err = common.WaitFor(10*time.Millisecond,
time.Duration(1000)*time.Millisecond, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond,
time.Duration(1000)*time.Millisecond, func() bool {
newCount := len(pc.GetCompletedApplications())
return newCount == currentCount+1
})
@@ -3824,7 +3824,7 @@ func TestNewQueueEvents(t *testing.T) {
})
assert.NilError(t, err)
noEvents := uint64(0)
- err = common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = eventSystem.Store.CountStoredEvents()
return noEvents == 3
})
diff --git a/pkg/scheduler/tests/application_tracking_test.go
b/pkg/scheduler/tests/application_tracking_test.go
index 0d50497d..ea0deb6b 100644
--- a/pkg/scheduler/tests/application_tracking_test.go
+++ b/pkg/scheduler/tests/application_tracking_test.go
@@ -158,7 +158,7 @@ func TestApplicationHistoryTracking(t *testing.T) {
// make sure app transitions to Completing
app := ms.getApplication(appID1)
- err = common.WaitFor(time.Millisecond*10, time.Second, func() bool {
+ err = common.WaitForCondition(time.Millisecond*10, time.Second, func()
bool {
return app.IsCompleting()
})
assert.NilError(t, err, "timeout waiting for app state Completing")
diff --git a/pkg/scheduler/tests/mock_rm_callback_test.go
b/pkg/scheduler/tests/mock_rm_callback_test.go
index b3f5419a..eef3ad3f 100644
--- a/pkg/scheduler/tests/mock_rm_callback_test.go
+++ b/pkg/scheduler/tests/mock_rm_callback_test.go
@@ -124,7 +124,7 @@ func (m *mockRMCallback) getAllocations()
map[string]*si.Allocation {
}
func (m *mockRMCallback) waitForAcceptedApplication(tb testing.TB, appID
string, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
return m.acceptedApplications[appID]
@@ -135,7 +135,7 @@ func (m *mockRMCallback) waitForAcceptedApplication(tb
testing.TB, appID string,
}
func (m *mockRMCallback) waitForRejectedApplication(t *testing.T, appID
string, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
return m.rejectedApplications[appID]
@@ -144,7 +144,7 @@ func (m *mockRMCallback) waitForRejectedApplication(t
*testing.T, appID string,
}
func (m *mockRMCallback) waitForApplicationState(t *testing.T, appID, state
string, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
return m.appStates[appID] == state
@@ -153,7 +153,7 @@ func (m *mockRMCallback) waitForApplicationState(t
*testing.T, appID, state stri
}
func (m *mockRMCallback) waitForAcceptedNode(t *testing.T, nodeID string,
timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
return m.acceptedNodes[nodeID]
@@ -163,7 +163,7 @@ func (m *mockRMCallback) waitForAcceptedNode(t *testing.T,
nodeID string, timeou
func (m *mockRMCallback) waitForMinAcceptedNodes(tb testing.TB, minNumNode
int, timeoutMs int) {
var numNodes int
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
numNodes = len(m.acceptedNodes)
@@ -175,7 +175,7 @@ func (m *mockRMCallback) waitForMinAcceptedNodes(tb
testing.TB, minNumNode int,
}
func (m *mockRMCallback) waitForRejectedNode(t *testing.T, nodeID string,
timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
return m.rejectedNodes[nodeID]
@@ -185,7 +185,7 @@ func (m *mockRMCallback) waitForRejectedNode(t *testing.T,
nodeID string, timeou
func (m *mockRMCallback) waitForAllocations(t *testing.T, nAlloc int,
timeoutMs int) {
var allocLen int
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
allocLen = len(m.Allocations)
@@ -196,7 +196,7 @@ func (m *mockRMCallback) waitForAllocations(t *testing.T,
nAlloc int, timeoutMs
func (m *mockRMCallback) waitForMinAllocations(tb testing.TB, nAlloc int,
timeoutMs int) {
var allocLen int
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
allocLen = len(m.Allocations)
@@ -209,7 +209,7 @@ func (m *mockRMCallback) waitForMinAllocations(tb
testing.TB, nAlloc int, timeou
func (m *mockRMCallback) waitForReleasedPlaceholders(t *testing.T, releases
int, timeoutMs int) {
var releasesLen int
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
m.RLock()
defer m.RUnlock()
releasesLen = len(m.releasedPhs)
diff --git a/pkg/scheduler/tests/mockscheduler_test.go
b/pkg/scheduler/tests/mockscheduler_test.go
index 9031ff2d..51289c1a 100644
--- a/pkg/scheduler/tests/mockscheduler_test.go
+++ b/pkg/scheduler/tests/mockscheduler_test.go
@@ -62,7 +62,7 @@ func (m *mockScheduler) Init(config string, autoSchedule
bool, withWebapp bool)
m.mockRM = newMockRMCallbackHandler()
if withWebapp {
- err := common.WaitFor(500*time.Millisecond, 2*time.Second,
func() bool {
+ err := common.WaitForCondition(500*time.Millisecond,
2*time.Second, func() bool {
conn, err := net.DialTimeout("tcp",
net.JoinHostPort("127.0.0.1", "9080"), time.Second)
if err == nil {
defer conn.Close()
diff --git a/pkg/scheduler/tests/plugin_test.go
b/pkg/scheduler/tests/plugin_test.go
index 467ffebc..4eb4062e 100644
--- a/pkg/scheduler/tests/plugin_test.go
+++ b/pkg/scheduler/tests/plugin_test.go
@@ -128,7 +128,7 @@ partitions:
})
assert.NilError(t, err)
- err = common.WaitFor(100*time.Millisecond, 3000*time.Millisecond,
func() bool {
+ err = common.WaitForCondition(100*time.Millisecond,
3000*time.Millisecond, func() bool {
reqSent := csu.GetContainerUpdateRequest()
return reqSent != nil && reqSent.ApplicationID == appID1 &&
reqSent.GetState() ==
si.UpdateContainerSchedulingStateRequest_FAILED
diff --git a/pkg/scheduler/tests/smoke_test.go
b/pkg/scheduler/tests/smoke_test.go
index dcf6d8c1..8051d29a 100644
--- a/pkg/scheduler/tests/smoke_test.go
+++ b/pkg/scheduler/tests/smoke_test.go
@@ -118,7 +118,7 @@ partitions:
assert.NilError(t, err, "configuration reload failed")
// wait until configuration is reloaded
- if err = common.WaitFor(time.Second, 5*time.Second, func() bool {
+ if err = common.WaitForCondition(time.Second, 5*time.Second, func()
bool {
return configs.ConfigContext.Get("policygroup").Checksum !=
configChecksum
}); err != nil {
t.Errorf("timeout waiting for configuration to be reloaded:
%v", err)
@@ -1015,7 +1015,7 @@ partitions:
assert.NilError(t, err, "NodeRequest 2 failed")
- err = common.WaitFor(10*time.Millisecond, 10*time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, 10*time.Second,
func() bool {
return
!ms.scheduler.GetClusterContext().GetPartition(partition).GetNode(node1ID).IsSchedulable()
&&
ms.scheduler.GetClusterContext().GetPartition(partition).GetNode(node2ID).IsSchedulable()
})
@@ -1035,7 +1035,7 @@ partitions:
assert.NilError(t, err, "NodeRequest 3 failed")
- err = common.WaitFor(10*time.Millisecond, 10*time.Second, func() bool {
+ err = common.WaitForCondition(10*time.Millisecond, 10*time.Second,
func() bool {
return
ms.scheduler.GetClusterContext().GetPartition(partition).GetNode(node1ID).IsSchedulable()
&&
ms.scheduler.GetClusterContext().GetPartition(partition).GetNode(node2ID).IsSchedulable()
})
@@ -1629,10 +1629,10 @@ partitions:
// Check app to Completing status
assert.Equal(t, app01.CurrentState(), objects.Completing.String())
// the app changes from completing state to completed state
- err = common.WaitFor(1*time.Millisecond, time.Millisecond*200,
app.IsCompleted)
+ err = common.WaitForCondition(1*time.Millisecond, time.Millisecond*200,
app.IsCompleted)
assert.NilError(t, err, "App should be in Completed state")
// partition manager should be able to clean up the dynamically created
queue.
- if err = common.WaitFor(1*time.Millisecond, time.Second*11, func() bool
{
+ if err = common.WaitForCondition(1*time.Millisecond, time.Second*11,
func() bool {
return part.GetQueue(leafName) == nil
}); err != nil {
t.Errorf("timeout waiting for queue is cleared %v", err)
diff --git a/pkg/scheduler/tests/utilities_test.go
b/pkg/scheduler/tests/utilities_test.go
index 0c40370a..f6904c25 100644
--- a/pkg/scheduler/tests/utilities_test.go
+++ b/pkg/scheduler/tests/utilities_test.go
@@ -60,7 +60,7 @@ func caller() string {
}
func waitForPendingQueueResource(t *testing.T, queue *objects.Queue, memory
resources.Quantity, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
return queue.GetPendingResource().Resources[siCommon.Memory] ==
memory
})
if err != nil {
@@ -71,21 +71,21 @@ func waitForPendingQueueResource(t *testing.T, queue
*objects.Queue, memory reso
}
func waitForPendingAppResource(t *testing.T, app *objects.Application, memory
resources.Quantity, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
return app.GetPendingResource().Resources[siCommon.Memory] ==
memory
})
assert.NilError(t, err, "Failed to wait for pending resource, expected
%v, actual %v, called from: %s", memory,
app.GetPendingResource().Resources[siCommon.Memory], caller())
}
func waitForAllocatedAppResource(t *testing.T, app *objects.Application,
memory resources.Quantity, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
return app.GetAllocatedResource().Resources[siCommon.Memory] ==
memory
})
assert.NilError(t, err, "Failed to wait for pending resource, expected
%v, actual %v, called from: %s", memory,
app.GetPendingResource().Resources[siCommon.Memory], caller())
}
func waitForAllocatedQueueResource(t *testing.T, queue *objects.Queue, memory
resources.Quantity, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
return queue.GetAllocatedResource().Resources[siCommon.Memory]
== memory
})
assert.NilError(t, err, "Failed to wait for allocations on queue %s,
called from: %s", queue.QueuePath, caller())
@@ -93,7 +93,7 @@ func waitForAllocatedQueueResource(t *testing.T, queue
*objects.Queue, memory re
func waitForAllocatedNodeResource(t *testing.T, cc *scheduler.ClusterContext,
partitionName string, nodeIDs []string, allocatedMemory resources.Quantity,
timeoutMs int) {
var totalNodeResource resources.Quantity
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
totalNodeResource = 0
for _, nodeID := range nodeIDs {
totalNodeResource +=
cc.GetPartition(partitionName).GetNode(nodeID).GetAllocatedResource().Resources[siCommon.Memory]
@@ -105,7 +105,7 @@ func waitForAllocatedNodeResource(t *testing.T, cc
*scheduler.ClusterContext, pa
func waitForAvailableNodeResource(t *testing.T, cc *scheduler.ClusterContext,
partitionName string, nodeIDs []string, availableMemory resources.Quantity,
timeoutMs int) {
var totalNodeResource resources.Quantity
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
totalNodeResource = 0
for _, nodeID := range nodeIDs {
totalNodeResource +=
cc.GetPartition(partitionName).GetNode(nodeID).GetAvailableResource().Resources[siCommon.Memory]
@@ -116,7 +116,7 @@ func waitForAvailableNodeResource(t *testing.T, cc
*scheduler.ClusterContext, pa
}
func waitForNewNode(t *testing.T, cc *scheduler.ClusterContext, nodeID string,
partitionName string, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
node := cc.GetNode(nodeID, partitionName)
return node != nil
})
@@ -124,7 +124,7 @@ func waitForNewNode(t *testing.T, cc
*scheduler.ClusterContext, nodeID string, p
}
func waitForRemovedNode(t *testing.T, context *scheduler.ClusterContext,
nodeID string, partitionName string, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
node := context.GetNode(nodeID, partitionName)
return node == nil
})
@@ -132,7 +132,7 @@ func waitForRemovedNode(t *testing.T, context
*scheduler.ClusterContext, nodeID
}
func waitForUpdatePartitionResource(t *testing.T, pc
*scheduler.PartitionContext, resourcesName string, availableQuantity
resources.Quantity, timeoutMs int) {
- err := common.WaitFor(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond,
time.Duration(timeoutMs)*time.Millisecond, func() bool {
return pc.GetTotalPartitionResource().Resources[resourcesName]
== availableQuantity
})
assert.NilError(t, err, "Failed to wait for available resource %v with
target quantity %v, called from: %s", resourcesName, availableQuantity,
caller())
diff --git a/pkg/webservice/handlers_test.go b/pkg/webservice/handlers_test.go
index 8bb60695..dfbe729b 100644
--- a/pkg/webservice/handlers_test.go
+++ b/pkg/webservice/handlers_test.go
@@ -1398,7 +1398,7 @@ func addAppWithUserGroup(t *testing.T, id string, part
*scheduler.PartitionConte
currentCount := len(part.GetCompletedApplications())
err = app.HandleApplicationEvent(objects.CompleteApplication)
assert.NilError(t, err, "The app should have completed")
- err = common.WaitFor(10*time.Millisecond, time.Second, func()
bool {
+ err = common.WaitForCondition(10*time.Millisecond, time.Second,
func() bool {
newCount := len(part.GetCompletedApplications())
return newCount == currentCount+1
})
@@ -2303,7 +2303,7 @@ func TestGetStream_Limit(t *testing.T) {
go getStream(NewResponseRecorderWithDeadline(), req)
// wait until the StreamingLimiter.AddHost() calls
- err := common.WaitFor(time.Millisecond, time.Second, func() bool {
+ err := common.WaitForCondition(time.Millisecond, time.Second, func()
bool {
streamingLimiter.Lock()
defer streamingLimiter.Unlock()
return streamingLimiter.streams == 3
@@ -2426,7 +2426,7 @@ func addEvents(t *testing.T) (appEvent, nodeEvent,
queueEvent *si.EventRecord) {
}
ev.AddEvent(queueEvent)
noEvents := uint64(0)
- err := common.WaitFor(10*time.Millisecond, time.Second, func() bool {
+ err := common.WaitForCondition(10*time.Millisecond, time.Second, func()
bool {
noEvents = ev.Store.CountStoredEvents()
return noEvents == 3
})
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
