Merge pull request #468 from adam-marek/detailed-resource-info

Detailed 'unschedulable' events

Author: k8s-ci-robot

pkg/scheduler/actions/allocate/allocate.go

@@ -102,6 +102,13 @@ func (alloc *allocateAction) Execute(ssn *framework.Session) {
 			glog.V(3).Infof("There are <%d> nodes for Job <%v/%v>",
 				len(ssn.Nodes), job.Namespace, job.Name)
 
+			//any task that doesn't fit will be the last processed
+			//within this loop context so any existing contents of
+			//NodesFitDelta are for tasks that eventually did fit on a
+			//node
+			if len(job.NodesFitDelta) > 0 {
+				job.NodesFitDelta = make(api.NodeResourceMap)
+			}
 			for _, node := range ssn.Nodes {
 				glog.V(3).Infof("Considering Task <%v/%v> on node <%v>: <%v> vs. <%v>",
 					task.Namespace, task.Name, node.Name, task.Resreq, node.Idle)
@@ -124,6 +131,10 @@ func (alloc *allocateAction) Execute(ssn *framework.Session) {
 					}
 					assigned = true
 					break
+				} else {
+					//store information about missing resources
+					job.NodesFitDelta[node.Name] = node.Idle.Clone()
+					job.NodesFitDelta[node.Name].FitDelta(task.Resreq)
 				}
 
 				// Allocate releasing resource to the task if any.

pkg/scheduler/api/job_info.go

@@ -26,6 +26,8 @@ import (
 	"github.com/kubernetes-sigs/kube-batch/cmd/kube-batch/app/options"
 	arbcorev1 "github.com/kubernetes-sigs/kube-batch/pkg/apis/scheduling/v1alpha1"
 	"github.com/kubernetes-sigs/kube-batch/pkg/apis/utils"
+	"sort"
+	"strings"
 )
 
 type TaskID types.UID
@@ -73,9 +75,8 @@ func NewTaskInfo(pod *v1.Pod) *TaskInfo {
 		NodeName:  pod.Spec.NodeName,
 		Status:    getTaskStatus(pod),
 		Priority:  1,
-
-		Pod:    pod,
-		Resreq: req,
+		Pod:       pod,
+		Resreq:    req,
 	}
 
 	if pod.Spec.Priority != nil {
@@ -109,6 +110,8 @@ type JobID types.UID
 
 type tasksMap map[TaskID]*TaskInfo
 
+type NodeResourceMap map[string]*Resource
+
 type JobInfo struct {
 	UID JobID
 
@@ -122,6 +125,8 @@ type JobInfo struct {
 	NodeSelector map[string]string
 	MinAvailable int32
 
+	NodesFitDelta NodeResourceMap
+
 	// All tasks of the Job.
 	TaskStatusIndex map[TaskStatus]tasksMap
 	Tasks           tasksMap
@@ -140,11 +145,11 @@ func NewJobInfo(uid JobID) *JobInfo {
 	return &JobInfo{
 		UID: uid,
 
-		MinAvailable: 0,
-		NodeSelector: make(map[string]string),
-
-		Allocated:    EmptyResource(),
-		TotalRequest: EmptyResource(),
+		MinAvailable:  0,
+		NodeSelector:  make(map[string]string),
+		NodesFitDelta: make(NodeResourceMap),
+		Allocated:     EmptyResource(),
+		TotalRequest:  EmptyResource(),
 
 		TaskStatusIndex: map[TaskStatus]tasksMap{},
 		Tasks:           tasksMap{},
@@ -278,10 +283,11 @@ func (ji *JobInfo) Clone() *JobInfo {
 		Namespace: ji.Namespace,
 		Queue:     ji.Queue,
 
-		MinAvailable: ji.MinAvailable,
-		NodeSelector: map[string]string{},
-		Allocated:    ji.Allocated.Clone(),
-		TotalRequest: ji.TotalRequest.Clone(),
+		MinAvailable:  ji.MinAvailable,
+		NodeSelector:  map[string]string{},
+		Allocated:     ji.Allocated.Clone(),
+		TotalRequest:  ji.TotalRequest.Clone(),
+		NodesFitDelta: make(NodeResourceMap),
 
 		PDB:      ji.PDB,
 		PodGroup: ji.PodGroup,
@@ -314,3 +320,36 @@ func (ji JobInfo) String() string {
 
 	return fmt.Sprintf("Job (%v): name %v, minAvailable %d", ji.UID, ji.Name, ji.MinAvailable) + res
 }
+
+// Error returns detailed information on why a job's task failed to fit on
+// each available node
+func (f *JobInfo) FitError() string {
+	if len(f.NodesFitDelta) == 0 {
+		reasonMsg := fmt.Sprintf("0 nodes are available")
+		return reasonMsg
+	}
+
+	reasons := make(map[string]int)
+	for _, v := range f.NodesFitDelta {
+		if v.Get(v1.ResourceCPU) < 0 {
+			reasons["cpu"]++
+		}
+		if v.Get(v1.ResourceMemory) < 0 {
+			reasons["memory"]++
+		}
+		if v.Get(GPUResourceName) < 0 {
+			reasons["GPU"]++
+		}
+	}
+
+	sortReasonsHistogram := func() []string {
+		reasonStrings := []string{}
+		for k, v := range reasons {
+			reasonStrings = append(reasonStrings, fmt.Sprintf("%v insufficient %v", v, k))
+		}
+		sort.Strings(reasonStrings)
+		return reasonStrings
+	}
+	reasonMsg := fmt.Sprintf("0/%v nodes are available, %v.", len(f.NodesFitDelta), strings.Join(sortReasonsHistogram(), ", "))
+	return reasonMsg
+}

pkg/scheduler/api/job_info_test.go

@@ -79,7 +79,8 @@ func TestAddTaskInfo(t *testing.T) {
 						case01_task4.UID: case01_task4,
 					},
 				},
-				NodeSelector: make(map[string]string),
+				NodeSelector:  make(map[string]string),
+				NodesFitDelta: make(NodeResourceMap),
 			},
 		},
 	}
@@ -143,7 +144,8 @@ func TestDeleteTaskInfo(t *testing.T) {
 					Pending: {case01_task1.UID: case01_task1},
 					Running: {case01_task3.UID: case01_task3},
 				},
-				NodeSelector: make(map[string]string),
+				NodeSelector:  make(map[string]string),
+				NodesFitDelta: make(NodeResourceMap),
 			},
 		},
 		{
@@ -168,7 +170,8 @@ func TestDeleteTaskInfo(t *testing.T) {
 						case02_task3.UID: case02_task3,
 					},
 				},
-				NodeSelector: make(map[string]string),
+				NodeSelector:  make(map[string]string),
+				NodesFitDelta: make(NodeResourceMap),
 			},
 		},
 	}

pkg/scheduler/api/resource_info.go

@@ -109,6 +109,25 @@ func (r *Resource) Sub(rr *Resource) *Resource {
 		r, rr))
 }
 
+//Computes the delta between a resource oject representing available
+//resources an operand representing resources being requested.  Any
+//field that is less than 0 after the operation represents an
+//insufficient resource.
+func (r *Resource) FitDelta(rr *Resource) *Resource {
+	if rr.MilliCPU > 0 {
+		r.MilliCPU -= rr.MilliCPU + minMilliCPU
+	}
+
+	if rr.Memory > 0 {
+		r.Memory -= rr.Memory + minMemory
+	}
+
+	if rr.MilliGPU > 0 {
+		r.MilliGPU -= rr.MilliGPU + minMilliGPU
+	}
+	return r
+}
+
 func (r *Resource) Multi(ratio float64) *Resource {
 	r.MilliCPU = r.MilliCPU * ratio
 	r.Memory = r.Memory * ratio

pkg/scheduler/cache/cache.go

@@ -550,5 +550,11 @@ func (sc *SchedulerCache) Backoff(job *arbapi.JobInfo, event arbcorev1.Event, re
 		return fmt.Errorf("no scheduling specification for job")
 	}
 
+	for _, tasks := range job.TaskStatusIndex {
+		for _, t := range tasks {
+			sc.recorder.Eventf(t.Pod, v1.EventTypeWarning, string(event), reason)
+		}
+	}
+
 	return nil
 }

pkg/scheduler/plugins/gang/gang.go

@@ -17,6 +17,7 @@ limitations under the License.
 package gang
 
 import (
+	"fmt"
 	"github.com/golang/glog"
 
 	arbcorev1 "github.com/kubernetes-sigs/kube-batch/pkg/apis/scheduling/v1alpha1"
@@ -144,8 +145,10 @@ func (gp *gangPlugin) OnSessionOpen(ssn *framework.Session) {
 
 func (gp *gangPlugin) OnSessionClose(ssn *framework.Session) {
 	for _, job := range ssn.Jobs {
-		if len(job.TaskStatusIndex[api.Allocated]) != 0 {
-			ssn.Backoff(job, arbcorev1.UnschedulableEvent, "not enough resource for job")
+		if len(job.TaskStatusIndex[api.Pending]) != 0 {
+			glog.V(3).Infof("Gang: <%v/%v> allocated: %v, pending: %v", job.Namespace, job.Name, len(job.TaskStatusIndex[api.Allocated]), len(job.TaskStatusIndex[api.Pending]))
+			msg := fmt.Sprintf("%v/%v tasks in gang unschedulable: %v", len(job.TaskStatusIndex[api.Pending]), len(job.Tasks), job.FitError())
+			ssn.Backoff(job, arbcorev1.UnschedulableEvent, msg)
 		}
 	}
 }