You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by wi...@apache.org on 2024/05/14 07:01:46 UTC
(yunikorn-k8shim) branch master updated: [YUNIKORN-2531] Create unit tests for AsyncRMCallback (#832)
This is an automated email from the ASF dual-hosted git repository.
wilfreds pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 4f8da23f [YUNIKORN-2531] Create unit tests for AsyncRMCallback (#832)
4f8da23f is described below
commit 4f8da23f4f482ccfa1c57a1f943777a72b3c07dd
Author: Peter Bacsko <pb...@cloudera.com>
AuthorDate: Tue May 14 17:01:18 2024 +1000
[YUNIKORN-2531] Create unit tests for AsyncRMCallback (#832)
Closes: #832
Signed-off-by: Wilfred Spiegelenburg <wi...@apache.org>
---
pkg/cache/context.go | 5 -
pkg/cache/scheduler_callback_test.go | 629 +++++++++++++++++++++++++++++++++++
pkg/common/events/recorder.go | 6 -
3 files changed, 629 insertions(+), 11 deletions(-)
diff --git a/pkg/cache/context.go b/pkg/cache/context.go
index b97b9289..c58b7914 100644
--- a/pkg/cache/context.go
+++ b/pkg/cache/context.go
@@ -664,11 +664,6 @@ func (ctx *Context) IsPodFitNode(name, node string, allocate bool) error {
}
func (ctx *Context) IsPodFitNodeViaPreemption(name, node string, allocations []string, startIndex int) (int, bool) {
- // assume minimal pods need killing if running in testing mode
- if ctx.apiProvider.IsTestingMode() {
- return startIndex, false
- }
-
ctx.lock.RLock()
defer ctx.lock.RUnlock()
if pod := ctx.schedulerCache.GetPod(name); pod != nil {
diff --git a/pkg/cache/scheduler_callback_test.go b/pkg/cache/scheduler_callback_test.go
new file mode 100644
index 00000000..ce2633ac
--- /dev/null
+++ b/pkg/cache/scheduler_callback_test.go
@@ -0,0 +1,629 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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 cache
+
+import (
+ "encoding/json"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "gotest.tools/v3/assert"
+ v1 "k8s.io/api/core/v1"
+ apis "k8s.io/apimachinery/pkg/apis/meta/v1"
+ k8sEvents "k8s.io/client-go/tools/events"
+ "k8s.io/kubernetes/pkg/scheduler/framework"
+
+ "github.com/apache/yunikorn-k8shim/pkg/client"
+ "github.com/apache/yunikorn-k8shim/pkg/common/constants"
+ "github.com/apache/yunikorn-k8shim/pkg/common/events"
+ "github.com/apache/yunikorn-k8shim/pkg/common/test"
+ "github.com/apache/yunikorn-k8shim/pkg/common/utils"
+ "github.com/apache/yunikorn-k8shim/pkg/dispatcher"
+ "github.com/apache/yunikorn-k8shim/pkg/plugin/predicates"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
+)
+
+func TestUpdateAllocation_NewTask(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ New: []*si.Allocation{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ NodeID: fakeNodeName,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ assert.Assert(t, context.schedulerCache.IsAssumedPod(taskUID1))
+ task := context.getTask(appID, taskUID1)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Bound
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "task has not transitioned to Bound state")
+}
+
+func TestUpdateAllocation_NewTask_TaskNotFound(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ pod := context.getTask(appID, taskUID1).pod
+ context.RemoveTask(appID, taskUID1)
+ context.DeletePod(pod)
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ New: []*si.Allocation{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ NodeID: fakeNodeName,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+}
+
+func TestUpdateAllocation_NewTask_AssumePodFails(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ binder := test.NewVolumeBinderMock()
+ const errMsg = "error assuming pod volumes"
+ binder.SetAssumePodVolumesError(errMsg)
+ setVolumeBinder(context, binder)
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ New: []*si.Allocation{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ NodeID: fakeNodeName,
+ },
+ },
+ })
+ assert.Error(t, err, errMsg)
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(taskUID1))
+ task := context.getTask(appID, taskUID1)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "task has not transitioned to Failed state")
+}
+
+func TestUpdateAllocation_NewTask_PodAlreadyAssigned(t *testing.T) {
+ callback, context := initCallbackTest(t, true, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ New: []*si.Allocation{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ NodeID: fakeNodeName,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ assert.Assert(t, context.schedulerCache.IsAssumedPod(taskUID1))
+ task := context.getTask(appID, taskUID1)
+ assert.Equal(t, TaskStates().Bound, task.GetTaskState())
+ assert.Equal(t, fakeNodeName, task.nodeName)
+ assert.Equal(t, TaskSchedAllocated, task.schedulingState)
+}
+
+func TestUpdateAllocation_AskRejected(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ Rejected: []*si.RejectedAllocationAsk{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ task := context.getTask(appID, taskUID1)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "task has not transitioned to Failed state")
+}
+
+func TestUpdateAllocation_AllocationRejected(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ RejectedAllocations: []*si.RejectedAllocation{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ task := context.getTask(appID, taskUID1)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "task has not transitioned to Failed state")
+}
+
+func TestUpdateAllocation_AllocationReleased(t *testing.T) {
+ // release allocation where terminationType != TerminationType_STOPPED_BY_RM
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ err := context.AssumePod(taskUID1, fakeNodeName)
+ assert.NilError(t, err, "could not assume pod")
+ app := context.getApplication(appID)
+ assert.Assert(t, app != nil)
+ app.sm.SetState(ApplicationStates().Running)
+ var deleteCalled atomic.Bool
+ context.apiProvider.(*client.MockedAPIProvider).MockDeleteFn(func(pod *v1.Pod) error { //nolint:errcheck
+ deleteCalled.Store(true)
+ return nil
+ })
+ task := context.getTask(appID, taskUID1)
+ task.allocationKey = taskUID1
+
+ err = callback.UpdateAllocation(&si.AllocationResponse{
+ Released: []*si.AllocationRelease{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ TerminationType: si.TerminationType_PREEMPTED_BY_SCHEDULER,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(taskUID1))
+ err = utils.WaitForCondition(deleteCalled.Load, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "pod has not been deleted")
+}
+
+func TestUpdateAllocation_AllocationReleased_StoppedByRM(t *testing.T) {
+ // release allocation where terminationType == TerminationType_STOPPED_BY_RM
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ err := context.AssumePod(taskUID1, fakeNodeName)
+ assert.NilError(t, err, "could not assume pod")
+ app := context.getApplication(appID)
+ assert.Assert(t, app != nil)
+ app.sm.SetState(ApplicationStates().Running)
+ var deleteCalled atomic.Bool
+ context.apiProvider.(*client.MockedAPIProvider).MockDeleteFn(func(pod *v1.Pod) error { //nolint:errcheck
+ deleteCalled.Store(true)
+ return nil
+ })
+
+ err = callback.UpdateAllocation(&si.AllocationResponse{
+ Released: []*si.AllocationRelease{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ TerminationType: si.TerminationType_STOPPED_BY_RM,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(taskUID1))
+ err = utils.WaitForCondition(deleteCalled.Load, 10*time.Millisecond, 500*time.Millisecond)
+ assert.Error(t, err, "timeout waiting for condition") // pod is not expected to be deleted
+}
+
+func TestUpdateAllocation_AskReleased(t *testing.T) {
+ callback, context := initCallbackTest(t, false, true)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Running)
+ var deleteCalled atomic.Bool
+ context.apiProvider.(*client.MockedAPIProvider).MockDeleteFn(func(pod *v1.Pod) error { //nolint:errcheck
+ deleteCalled.Store(true)
+ return nil
+ })
+
+ err := callback.UpdateAllocation(&si.AllocationResponse{
+ ReleasedAsks: []*si.AllocationAskRelease{
+ {
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ TerminationType: si.TerminationType_TIMEOUT,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating allocation")
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(taskUID1))
+ err = utils.WaitForCondition(deleteCalled.Load, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "pod has not been deleted")
+}
+
+func TestUpdateApplication_Accepted(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Submitted)
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Accepted: []*si.AcceptedApplication{
+ {
+ ApplicationID: appID,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+ err = utils.WaitForCondition(func() bool {
+ return app.sm.Current() == ApplicationStates().Accepted
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "application has not transitioned to Accepted state")
+}
+
+func TestUpdateApplication_Rejected(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Submitted)
+ NewPlaceholderManager(context.apiProvider.GetAPIs())
+ recorder := k8sEvents.NewFakeRecorder(1024)
+ events.SetRecorder(recorder)
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Rejected: []*si.RejectedApplication{
+ {
+ ApplicationID: appID,
+ Reason: "test failure",
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+ err = utils.WaitForCondition(func() bool {
+ return app.sm.Current() == ApplicationStates().Failed
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "application has not transitioned to Failed state")
+ assert.Equal(t, 1, len(recorder.Events), "no K8s event received")
+ event := <-recorder.Events
+ assert.Assert(t, strings.Contains(event, "test failure"), "event does not contain 'test failure': %s", event)
+}
+
+func TestUpdateApplication_Completed(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Updated: []*si.UpdatedApplication{
+ {
+ ApplicationID: appID,
+ State: ApplicationStates().Completed,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+ assert.Assert(t, context.getApplication(appID) == nil, "application was not removed")
+}
+
+func TestUpdateApplication_Resuming_AppReserving(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Reserving)
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Updated: []*si.UpdatedApplication{
+ {
+ ApplicationID: appID,
+ State: ApplicationStates().Resuming,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+ err = utils.WaitForCondition(func() bool {
+ return app.sm.Current() == ApplicationStates().Resuming
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "application has not transitioned to Resuming state")
+}
+
+func TestUpdateApplication_Resuming_AppNotReserving(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Submitted) // not in Reserving - nothing should happen
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Updated: []*si.UpdatedApplication{
+ {
+ ApplicationID: appID,
+ State: ApplicationStates().Resuming,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+ time.Sleep(time.Second) // wait and make sure that the app is still submitted
+ assert.Equal(t, ApplicationStates().Submitted, app.GetApplicationState(), "application state has been changed unexpectedly")
+}
+
+func TestUpdateApplication_Resuming_AppNotFound(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Reserving)
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Updated: []*si.UpdatedApplication{
+ {
+ ApplicationID: "nonExisting",
+ State: ApplicationStates().Resuming,
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+}
+
+func TestUpdateApplication_Failing(t *testing.T) {
+ testUpdateApplicationFailure(t, ApplicationStates().Failing)
+}
+
+func TestUpdateApplication_Failed(t *testing.T) {
+ testUpdateApplicationFailure(t, ApplicationStates().Failed)
+}
+
+func testUpdateApplicationFailure(t *testing.T, state string) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ app := context.getApplication(appID)
+ app.sm.SetState(ApplicationStates().Running)
+ NewPlaceholderManager(context.apiProvider.GetAPIs())
+ recorder := k8sEvents.NewFakeRecorder(1024)
+ events.SetRecorder(recorder)
+
+ err := callback.UpdateApplication(&si.ApplicationResponse{
+ Updated: []*si.UpdatedApplication{
+ {
+ ApplicationID: appID,
+ State: state,
+ Message: "test failure",
+ },
+ },
+ })
+ assert.NilError(t, err, "error updating application")
+ err = utils.WaitForCondition(func() bool {
+ return app.sm.Current() == ApplicationStates().Failing
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "application has not transitioned to %s state", state)
+ assert.Equal(t, 1, len(recorder.Events), "no K8s event received")
+ event := <-recorder.Events
+ assert.Assert(t, strings.Contains(event, "test failure"), "event does not contain 'test failure': %s", event)
+}
+
+func TestUpdateNode_Accepted(t *testing.T) {
+ testUpdateNode(t, "NodeAccepted", &si.NodeResponse{
+ Accepted: []*si.AcceptedNode{
+ {
+ NodeID: "testNode",
+ },
+ },
+ })
+}
+
+func TestUpdateNode_Rejected(t *testing.T) {
+ testUpdateNode(t, "NodeRejected", &si.NodeResponse{
+ Rejected: []*si.RejectedNode{
+ {
+ NodeID: "testNode",
+ },
+ },
+ })
+}
+
+func testUpdateNode(t *testing.T, expectedEvent string, response *si.NodeResponse) {
+ callback, _ := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ var nodeName atomic.Value
+ dispatcher.RegisterEventHandler("testNodeHandler", dispatcher.EventTypeNode, func(event interface{}) {
+ nodeEvent := event.(CachedSchedulerNodeEvent) //nolint:errcheck
+ if nodeEvent.GetEvent() == expectedEvent {
+ nodeName.Store(nodeEvent.GetNodeID())
+ }
+ })
+
+ err := callback.UpdateNode(response)
+ assert.NilError(t, err, "error updating node")
+ err = utils.WaitForCondition(func() bool {
+ if val, ok := nodeName.Load().(string); ok {
+ return val == "testNode"
+ }
+ return false
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err)
+}
+
+func TestPredicates(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ context.predManager = &mockPredicateManager{}
+
+ // pod not found
+ err := callback.Predicates(&si.PredicatesArgs{AllocationKey: "unknown", NodeID: fakeNodeName, Allocate: true})
+ assert.Error(t, err, "predicates were not run because pod was not found in cache")
+
+ // pod found
+ err = callback.Predicates(&si.PredicatesArgs{AllocationKey: taskUID1, NodeID: fakeNodeName, Allocate: true})
+ assert.NilError(t, err)
+}
+
+func TestPreemptionPredicates(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ context.predManager = &mockPredicateManager{}
+
+ // pod not found
+ resp := callback.PreemptionPredicates(&si.PreemptionPredicatesArgs{AllocationKey: "unknown", NodeID: fakeNodeName, StartIndex: 0})
+ assert.Assert(t, !resp.Success, "response should have failed")
+
+ // pod found
+ resp = callback.PreemptionPredicates(&si.PreemptionPredicatesArgs{AllocationKey: taskUID1, NodeID: fakeNodeName, StartIndex: 0, PreemptAllocationKeys: []string{taskUID1}})
+ assert.Assert(t, resp.Success, "response should have succeeded")
+ assert.Equal(t, int32(0), resp.Index)
+}
+
+func TestSendEvent(t *testing.T) {
+ callback, _ := initCallbackTest(t, false, false)
+ recorder := k8sEvents.NewFakeRecorder(1024)
+ events.SetRecorder(recorder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ callback.SendEvent([]*si.EventRecord{
+ {
+ Type: si.EventRecord_REQUEST,
+ ObjectID: taskUID1,
+ ReferenceID: appID,
+ Message: "request test message",
+ },
+ {
+ Type: si.EventRecord_NODE,
+ EventChangeType: si.EventRecord_ADD,
+ EventChangeDetail: si.EventRecord_DETAILS_NONE,
+ ObjectID: fakeNodeName,
+ Message: "node test message",
+ },
+ })
+
+ assert.Equal(t, 2, len(recorder.Events))
+ event := <-recorder.Events
+ assert.Assert(t, strings.Contains(event, "request test message"), "event does not contain 'request test message': %s", event)
+ event = <-recorder.Events
+ assert.Assert(t, strings.Contains(event, "node test message"), "event does not contain 'node test message': %s", event)
+}
+
+func TestUpdateContainerSchedulingState(t *testing.T) {
+ callback, context := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ callback.UpdateContainerSchedulingState(&si.UpdateContainerSchedulingStateRequest{
+ State: si.UpdateContainerSchedulingStateRequest_FAILED,
+ ApplicationID: appID,
+ AllocationKey: taskUID1,
+ })
+
+ task := context.getTask(appID, taskUID1)
+ assert.Equal(t, TaskSchedFailed, task.GetTaskSchedulingState())
+}
+
+func TestCallbackGetStateDump(t *testing.T) {
+ callback, _ := initCallbackTest(t, false, false)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ stateDumpStr, err := callback.GetStateDump()
+ assert.NilError(t, err, "unable to get state dump")
+ stateDump := make(map[string]interface{})
+ err = json.Unmarshal([]byte(stateDumpStr), &stateDump)
+ assert.NilError(t, err)
+ assert.Assert(t, stateDump["cache"] != nil, "no 'cache' entry in the state dump")
+}
+
+var _ predicates.PredicateManager = &mockPredicateManager{}
+
+type mockPredicateManager struct{}
+
+func (m *mockPredicateManager) EventsToRegister(_ framework.QueueingHintFn) []framework.ClusterEventWithHint {
+ return nil
+}
+
+func (m *mockPredicateManager) Predicates(_ *v1.Pod, _ *framework.NodeInfo, _ bool) (plugin string, error error) {
+ return "", nil
+}
+
+func (m *mockPredicateManager) PreemptionPredicates(_ *v1.Pod, _ *framework.NodeInfo, _ []*v1.Pod, _ int) (index int, ok bool) {
+ return 0, true
+}
+
+func initCallbackTest(t *testing.T, podAssigned, placeholder bool) (*AsyncRMCallback, *Context) {
+ context, apiProvider := initContextAndAPIProviderForTest()
+ dispatcher.Start()
+ dispatcher.RegisterEventHandler("TestAppHandler", dispatcher.EventTypeApp, context.ApplicationEventHandler())
+ dispatcher.RegisterEventHandler("TestTaskHandler", dispatcher.EventTypeTask, context.TaskEventHandler())
+ callback := NewAsyncRMCallback(context)
+ apiProvider.MockSchedulerAPIUpdateNodeFn(func(request *si.NodeRequest) error {
+ for _, node := range request.Nodes {
+ dispatcher.Dispatch(CachedSchedulerNodeEvent{
+ NodeID: node.NodeID,
+ Event: NodeAccepted,
+ })
+ }
+ return nil
+ })
+ node1 := v1.Node{
+ ObjectMeta: apis.ObjectMeta{
+ Name: fakeNodeName,
+ Namespace: "default",
+ UID: "uid_0001",
+ },
+ }
+ context.addNode(&node1)
+ pod := &v1.Pod{
+ TypeMeta: apis.TypeMeta{
+ Kind: "Pod",
+ APIVersion: "v1",
+ },
+ ObjectMeta: apis.ObjectMeta{
+ Name: "yunikorn-test-00001",
+ UID: taskUID1,
+ Annotations: map[string]string{
+ constants.AnnotationApplicationID: appID,
+ },
+ },
+ Spec: v1.PodSpec{SchedulerName: "yunikorn"},
+ }
+ if podAssigned {
+ pod.Spec.NodeName = fakeNodeName
+ }
+ if placeholder {
+ pod.Annotations[constants.AnnotationPlaceholderFlag] = constants.True
+ }
+ context.AddPod(pod)
+ task := context.getTask(appID, taskUID1)
+ assert.Assert(t, task != nil)
+ task.sm.SetState(TaskStates().Scheduling)
+
+ return callback, context
+}
diff --git a/pkg/common/events/recorder.go b/pkg/common/events/recorder.go
index 161c0eef..48a06fa7 100644
--- a/pkg/common/events/recorder.go
+++ b/pkg/common/events/recorder.go
@@ -59,9 +59,3 @@ func SetRecorder(recorder events.EventRecorder) {
eventRecorder = recorder
once.Do(func() {}) // make sure Do() doesn't fire elsewhere
}
-
-func SetRecorderForTest(recorder events.EventRecorder) {
- lock.Lock()
- defer lock.Unlock()
- eventRecorder = recorder
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org