wilfred-s commented on code in PR #810:
URL: https://github.com/apache/yunikorn-k8shim/pull/810#discussion_r1548190346
##########
pkg/cache/context.go:
##########
@@ -792,45 +792,54 @@ func (ctx *Context) AssumePod(name string, node string)
error {
// assume pod volumes before assuming the pod
// this will update scheduler cache with essential
PV/PVC binding info
var allBound = true
- // volume builder might be null in UTs
- if ctx.apiProvider.GetAPIs().VolumeBinder != nil {
- var err error
- // retrieve the volume claims
- podVolumeClaims, err :=
ctx.apiProvider.GetAPIs().VolumeBinder.GetPodVolumeClaims(ctx.klogger, pod)
- if err != nil {
- log.Log(log.ShimContext).Error("Failed
to get pod volume claims",
- zap.String("podName",
assumedPod.Name),
- zap.Error(err))
- return err
- }
+ task := ctx.getTask(applicationID, name)
+ if task == nil {
+ log.Log(log.ShimContext).Error("BUG: task not
found", zap.String("taskID", name))
+ return fmt.Errorf("task not found: %s", name)
+ }
+ task.setAllocationID(allocationID)
Review Comment:
This can be prevented if we handle the error returned in UpdateAllocation to
fail the task etc.
##########
pkg/cache/context.go:
##########
@@ -1152,9 +1161,13 @@ func (ctx *Context) RemoveTask(appID, taskID string) {
app.RemoveTask(taskID)
}
-func (ctx *Context) getTask(appID string, taskID string) *Task {
+func (ctx *Context) GetTask(appID string, taskID string) *Task {
ctx.lock.RLock()
defer ctx.lock.RUnlock()
+ return ctx.getTask(appID, taskID)
+}
+
+func (ctx *Context) getTask(appID, taskID string) *Task {
Review Comment:
This is not needed if you first retrieve the task using `getTask` before
calling `AssumePod` in the tests. Also becomes unneeded if we handle the fail
in the `UpdateAllocation`
##########
pkg/cache/context_test.go:
##########
@@ -2138,6 +2142,168 @@ func TestTaskRemoveOnCompletion(t *testing.T) {
assert.Error(t, err, "task task00001 doesn't exist in application
app01")
}
+func TestAssumePod(t *testing.T) {
+ context := initAssumePodTest(test.NewVolumeBinderMock())
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.NilError(t, err)
+ assert.Assert(t, context.schedulerCache.ArePodVolumesAllBound(pod1UID))
+ assumedPod, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, assumedPod.Spec.NodeName, fakeNodeName)
+ assert.Assert(t, context.schedulerCache.IsAssumedPod(pod1UID))
+}
+
+func TestAssumePod_TaskNotFound(t *testing.T) {
+ context := initAssumePodTest(nil)
+ context.RemoveTask(appID, pod1UID)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, "task not found: task00001")
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_GetPodVolumeClaimsError(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ const errMsg = "error getting volume claims"
+ binder.EnableVolumeClaimsError(errMsg)
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, errMsg)
+ task := context.getTask(appID, pod1UID)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 100*time.Millisecond, time.Second)
+ assert.NilError(t, err)
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_FindPodVolumesError(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ const errMsg = "error getting pod volumes"
+ binder.EnableFindPodVolumesError(errMsg)
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, errMsg)
+ task := context.getTask(appID, pod1UID)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 100*time.Millisecond, time.Second)
+ assert.NilError(t, err)
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_ConflictingVolumes(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ binder.SetConflictReasons("reason1", "reason2")
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, "pod my-pod-1 has conflicting volume claims:
reason1, reason2")
+ task := context.getTask(appID, pod1UID)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 100*time.Millisecond, time.Second)
+ assert.NilError(t, err)
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_AssumePodVolumesError(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ const errMsg = "error assuming pod volumes"
+ binder.SetAssumePodVolumesError(errMsg)
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, errMsg)
+ task := context.getTask(appID, pod1UID)
Review Comment:
swap order getTask before AssumePod
##########
pkg/cache/context_test.go:
##########
@@ -2138,6 +2142,168 @@ func TestTaskRemoveOnCompletion(t *testing.T) {
assert.Error(t, err, "task task00001 doesn't exist in application
app01")
}
+func TestAssumePod(t *testing.T) {
+ context := initAssumePodTest(test.NewVolumeBinderMock())
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.NilError(t, err)
+ assert.Assert(t, context.schedulerCache.ArePodVolumesAllBound(pod1UID))
+ assumedPod, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, assumedPod.Spec.NodeName, fakeNodeName)
+ assert.Assert(t, context.schedulerCache.IsAssumedPod(pod1UID))
+}
+
+func TestAssumePod_TaskNotFound(t *testing.T) {
+ context := initAssumePodTest(nil)
+ context.RemoveTask(appID, pod1UID)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, "task not found: task00001")
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_GetPodVolumeClaimsError(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ const errMsg = "error getting volume claims"
+ binder.EnableVolumeClaimsError(errMsg)
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, errMsg)
+ task := context.getTask(appID, pod1UID)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 100*time.Millisecond, time.Second)
+ assert.NilError(t, err)
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_FindPodVolumesError(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ const errMsg = "error getting pod volumes"
+ binder.EnableFindPodVolumesError(errMsg)
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, errMsg)
+ task := context.getTask(appID, pod1UID)
+ err = utils.WaitForCondition(func() bool {
+ return task.GetTaskState() == TaskStates().Failed
+ }, 100*time.Millisecond, time.Second)
+ assert.NilError(t, err)
+ assert.Assert(t, !context.schedulerCache.IsAssumedPod(pod1UID))
+ podInCache, ok := context.schedulerCache.GetPod(pod1UID)
+ assert.Assert(t, ok, "pod not found in cache")
+ assert.Equal(t, podInCache.Spec.NodeName, "", "NodeName in pod spec was
set unexpectedly")
+}
+
+func TestAssumePod_ConflictingVolumes(t *testing.T) {
+ binder := test.NewVolumeBinderMock()
+ binder.SetConflictReasons("reason1", "reason2")
+ context := initAssumePodTest(binder)
+ defer dispatcher.UnregisterAllEventHandlers()
+ defer dispatcher.Stop()
+
+ err := context.AssumePod(pod1UID, appID, "alloc-0", fakeNodeName)
+ assert.Error(t, err, "pod my-pod-1 has conflicting volume claims:
reason1, reason2")
+ task := context.getTask(appID, pod1UID)
Review Comment:
swap order getTask before AssumePod
##########
pkg/cache/external/scheduler_cache.go:
##########
@@ -450,7 +450,7 @@ func (cache *SchedulerCache) StartPodAllocation(podKey
string, nodeID string) bo
}
// return if pod is assumed in cache, avoid nil
-func (cache *SchedulerCache) isAssumedPod(podKey string) bool {
+func (cache *SchedulerCache) IsAssumedPod(podKey string) bool {
Review Comment:
No need to export, callers have not changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
