You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by cc...@apache.org on 2023/07/27 16:09:27 UTC
[yunikorn-k8shim] branch master updated: [YUNIKORN-1874] Data race: unlocked access in Context.updatePodCondition() (#639)
This is an automated email from the ASF dual-hosted git repository.
ccondit 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 8442770c [YUNIKORN-1874] Data race: unlocked access in Context.updatePodCondition() (#639)
8442770c is described below
commit 8442770c2fd575a82c76292895c2d5ccfdb2fc6f
Author: Peter Bacsko <pb...@cloudera.com>
AuthorDate: Thu Jul 27 11:09:11 2023 -0500
[YUNIKORN-1874] Data race: unlocked access in Context.updatePodCondition() (#639)
Closes: #639
Signed-off-by: Craig Condit <cc...@apache.org>
---
pkg/cache/context.go | 24 +++++---------
pkg/cache/context_test.go | 52 +++++++++++++++++++++++++++++
pkg/cache/task.go | 30 +++++++++++++++--
pkg/cache/task_test.go | 83 +++++++++++++++++++++++++++++++++++++++++++++++
4 files changed, 170 insertions(+), 19 deletions(-)
diff --git a/pkg/cache/context.go b/pkg/cache/context.go
index 570b418c..ce401068 100644
--- a/pkg/cache/context.go
+++ b/pkg/cache/context.go
@@ -30,7 +30,6 @@ import (
v1 "k8s.io/api/core/v1"
schedulingv1 "k8s.io/api/scheduling/v1"
"k8s.io/client-go/tools/cache"
- podutil "k8s.io/kubernetes/pkg/api/v1/pod"
"k8s.io/kubernetes/pkg/scheduler/framework"
"k8s.io/kubernetes/pkg/scheduler/framework/plugins/volumebinding"
@@ -1014,24 +1013,17 @@ func (ctx *Context) updatePodCondition(task *Task, podCondition *v1.PodCondition
if task.GetTaskState() == TaskStates().Scheduling {
// only update the pod when pod condition changes
// minimize the overhead added to the api-server/etcd
- if !utils.PodUnderCondition(task.pod, podCondition) {
- log.Log(log.ShimContext).Debug("updating pod condition",
- zap.String("namespace", task.pod.Namespace),
- zap.String("name", task.pod.Name),
- zap.Any("podCondition", podCondition))
- // call api-server to do the pod condition update
- if podutil.UpdatePodCondition(&task.pod.Status, podCondition) {
- podCopy := task.pod.DeepCopy()
- _, err := ctx.apiProvider.GetAPIs().KubeClient.UpdateStatus(podCopy)
- if err == nil {
- return true
- }
- // only log the error here, no need to handle it if the update failed
- log.Log(log.ShimContext).Error("update pod condition failed",
- zap.Error(err))
+ if ok, podCopy := task.UpdatePodCondition(podCondition); ok {
+ _, err := ctx.apiProvider.GetAPIs().KubeClient.UpdateStatus(podCopy)
+ if err == nil {
+ return true
}
+ // only log the error here, no need to handle it if the update failed
+ log.Log(log.ShimContext).Error("update pod condition failed",
+ zap.Error(err))
}
}
+
return false
}
diff --git a/pkg/cache/context_test.go b/pkg/cache/context_test.go
index 279e0cc3..57a80754 100644
--- a/pkg/cache/context_test.go
+++ b/pkg/cache/context_test.go
@@ -1445,3 +1445,55 @@ func TestDeletePriorityClass(t *testing.T) {
result = context.schedulerCache.GetPriorityClass("pc-test")
assert.Assert(t, result == nil)
}
+
+func TestCtxUpdatePodCondition(t *testing.T) {
+ condition := v1.PodCondition{
+ Type: v1.ContainersReady,
+ Status: v1.ConditionTrue,
+ }
+ pod := &v1.Pod{
+ TypeMeta: apis.TypeMeta{
+ Kind: "Pod",
+ APIVersion: "v1",
+ },
+ ObjectMeta: apis.ObjectMeta{
+ Name: "pod-test-00001",
+ },
+ Status: v1.PodStatus{
+ Phase: v1.PodPending,
+ Conditions: []v1.PodCondition{
+ condition,
+ },
+ },
+ }
+ context := initContextForTest()
+ context.AddApplication(&interfaces.AddApplicationRequest{
+ Metadata: interfaces.ApplicationMetadata{
+ ApplicationID: "app00001",
+ QueueName: "root.a",
+ User: "test-user",
+ Tags: nil,
+ },
+ })
+ task := context.AddTask(&interfaces.AddTaskRequest{ //nolint:errcheck
+ Metadata: interfaces.TaskMetadata{
+ ApplicationID: "app00001",
+ TaskID: "task00001",
+ Pod: pod,
+ },
+ }).(*Task)
+
+ // task state is not Scheduling
+ updated := context.updatePodCondition(task, &condition)
+ assert.Equal(t, false, updated)
+
+ // no update
+ task.sm.SetState(TaskStates().Scheduling)
+ updated = context.updatePodCondition(task, &condition)
+ assert.Equal(t, false, updated)
+
+ // update status
+ condition.Status = v1.ConditionFalse
+ updated = context.updatePodCondition(task, &condition)
+ assert.Equal(t, true, updated)
+}
diff --git a/pkg/cache/task.go b/pkg/cache/task.go
index 22075968..bbe4d039 100644
--- a/pkg/cache/task.go
+++ b/pkg/cache/task.go
@@ -25,7 +25,10 @@ import (
"sync"
"time"
+ "github.com/looplab/fsm"
"go.uber.org/zap"
+ v1 "k8s.io/api/core/v1"
+ podutil "k8s.io/kubernetes/pkg/api/v1/pod"
"github.com/apache/yunikorn-k8shim/pkg/appmgmt/interfaces"
"github.com/apache/yunikorn-k8shim/pkg/common"
@@ -35,9 +38,6 @@ import (
"github.com/apache/yunikorn-k8shim/pkg/dispatcher"
"github.com/apache/yunikorn-k8shim/pkg/log"
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
-
- "github.com/looplab/fsm"
- v1 "k8s.io/api/core/v1"
)
type Task struct {
@@ -48,6 +48,7 @@ type Task struct {
allocationUUID string
resource *si.Resource
pod *v1.Pod
+ podStatus v1.PodStatus // pod status, maintained separately for efficiency reasons
context *Context
nodeName string
createTime time.Time
@@ -97,6 +98,7 @@ func createTaskInternal(tid string, app *Application, resource *si.Resource,
applicationID: app.GetApplicationID(),
application: app,
pod: pod,
+ podStatus: *pod.Status.DeepCopy(),
resource: resource,
createTime: pod.GetCreationTimestamp().Time,
placeholder: placeholder,
@@ -558,3 +560,25 @@ func (task *Task) sanityCheckBeforeScheduling() error {
}
return nil
}
+
+func (task *Task) UpdatePodCondition(podCondition *v1.PodCondition) (bool, *v1.Pod) {
+ task.lock.Lock()
+ defer task.lock.Unlock()
+
+ status := task.podStatus.DeepCopy()
+ pod := task.pod.DeepCopy()
+ pod.Status = *status
+ if !utils.PodUnderCondition(pod, podCondition) {
+ log.Log(log.ShimContext).Debug("updating pod condition",
+ zap.String("namespace", task.pod.Namespace),
+ zap.String("name", task.pod.Name),
+ zap.Any("podCondition", podCondition))
+ if podutil.UpdatePodCondition(&task.podStatus, podCondition) {
+ status = task.podStatus.DeepCopy()
+ pod.Status = *status
+ return true, pod
+ }
+ }
+
+ return false, pod
+}
diff --git a/pkg/cache/task_test.go b/pkg/cache/task_test.go
index f2b0c7c1..ff1cfb6f 100644
--- a/pkg/cache/task_test.go
+++ b/pkg/cache/task_test.go
@@ -658,3 +658,86 @@ func TestSimultaneousTaskCompleteAndAllocate(t *testing.T) {
assert.NilError(t, err, "failed to handle AllocateTask event")
assert.Equal(t, task1.GetTaskState(), TaskStates().Completed)
}
+
+func TestUpdatePodCondition(t *testing.T) {
+ condition := v1.PodCondition{
+ Type: v1.ContainersReady,
+ Status: v1.ConditionTrue,
+ Reason: v1.PodReasonSchedulingGated,
+ }
+
+ pod := &v1.Pod{
+ TypeMeta: apis.TypeMeta{
+ Kind: "Pod",
+ APIVersion: "v1",
+ },
+ ObjectMeta: apis.ObjectMeta{
+ Name: "pod-test-00001",
+ },
+ Status: v1.PodStatus{
+ Phase: v1.PodPending,
+ },
+ }
+ app := NewApplication(appID, "root.default", "user", testGroups, map[string]string{}, nil)
+ task := NewTask("pod-1", app, nil, pod)
+ updated, podCopy := task.UpdatePodCondition(&condition)
+ assert.Equal(t, true, updated)
+ assert.Equal(t, 1, len(podCopy.Status.Conditions))
+ assert.Equal(t, v1.ConditionTrue, podCopy.Status.Conditions[0].Status)
+ assert.Equal(t, v1.ContainersReady, podCopy.Status.Conditions[0].Type)
+ assert.Equal(t, v1.PodReasonSchedulingGated, podCopy.Status.Conditions[0].Reason)
+ assert.Equal(t, v1.PodPending, podCopy.Status.Phase)
+ assert.Equal(t, 1, len(task.podStatus.Conditions))
+ assert.Equal(t, v1.ConditionTrue, task.podStatus.Conditions[0].Status)
+ assert.Equal(t, v1.ContainersReady, task.podStatus.Conditions[0].Type)
+ assert.Equal(t, v1.PodReasonSchedulingGated, task.podStatus.Conditions[0].Reason)
+ assert.Equal(t, v1.PodPending, task.podStatus.Phase)
+
+ podWithCondition := &v1.Pod{
+ TypeMeta: apis.TypeMeta{
+ Kind: "Pod",
+ APIVersion: "v1",
+ },
+ ObjectMeta: apis.ObjectMeta{
+ Name: "pod-test-00001",
+ },
+ Status: v1.PodStatus{
+ Phase: v1.PodPending,
+ Conditions: []v1.PodCondition{
+ condition,
+ },
+ },
+ }
+ app = NewApplication(appID, "root.default", "user", testGroups, map[string]string{}, nil)
+ task = NewTask("pod-1", app, nil, podWithCondition)
+
+ // no update
+ updated, podCopy = task.UpdatePodCondition(&condition)
+ assert.Equal(t, false, updated)
+ assert.Equal(t, 1, len(task.podStatus.Conditions))
+ assert.Equal(t, v1.ConditionTrue, task.podStatus.Conditions[0].Status)
+ assert.Equal(t, v1.ContainersReady, task.podStatus.Conditions[0].Type)
+ assert.Equal(t, v1.PodPending, task.podStatus.Phase)
+ assert.Equal(t, v1.PodReasonSchedulingGated, task.podStatus.Conditions[0].Reason)
+ assert.Equal(t, 1, len(podCopy.Status.Conditions))
+ assert.Equal(t, v1.ConditionTrue, podCopy.Status.Conditions[0].Status)
+ assert.Equal(t, v1.ContainersReady, podCopy.Status.Conditions[0].Type)
+ assert.Equal(t, v1.PodPending, podCopy.Status.Phase)
+ assert.Equal(t, v1.PodReasonSchedulingGated, podCopy.Status.Conditions[0].Reason)
+
+ // update status & reason
+ condition.Status = v1.ConditionFalse
+ condition.Reason = v1.PodReasonUnschedulable
+ updated, podCopy = task.UpdatePodCondition(&condition)
+ assert.Equal(t, true, updated)
+ assert.Equal(t, 1, len(task.podStatus.Conditions))
+ assert.Equal(t, v1.ConditionFalse, task.podStatus.Conditions[0].Status)
+ assert.Equal(t, v1.ContainersReady, task.podStatus.Conditions[0].Type)
+ assert.Equal(t, v1.PodPending, task.podStatus.Phase)
+ assert.Equal(t, v1.PodReasonUnschedulable, task.podStatus.Conditions[0].Reason)
+ assert.Equal(t, 1, len(podCopy.Status.Conditions))
+ assert.Equal(t, v1.ConditionFalse, podCopy.Status.Conditions[0].Status)
+ assert.Equal(t, v1.ContainersReady, podCopy.Status.Conditions[0].Type)
+ assert.Equal(t, v1.PodPending, podCopy.Status.Phase)
+ assert.Equal(t, v1.PodReasonUnschedulable, podCopy.Status.Conditions[0].Reason)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org