Update PodGroup status.

pkg/apis/scheduling/v1alpha1/types.go

@@ -17,18 +17,10 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// Event represent the phase of PodGroup, e.g. pod-failed.
-type Event string
-
-const (
-	EvictEvent            Event = "Evict"
-	UnschedulableEvent    Event = "Unschedulable"
-	FailedSchedulingEvent Event = "FailedScheduling"
-)
-
 // PodGroupPhase is the phase of a pod group at the current time.
 type PodGroupPhase string
 
@@ -41,35 +33,39 @@ const (
 	// PodRunning means `spec.minMember` pods of PodGroups has been in running phase.
 	PodGroupRunning PodGroupPhase = "Running"
 
-	// PodGroupRecovering means part of `spec.minMember` pods have exception, e.g. killed; scheduler will
-	// wait for related controller to recover it.
-	PodGroupRecovering PodGroupPhase = "Recovering"
-
-	// PodGroupUnschedulable means part of `spec.minMember` pods are running but the other part can not
+	// PodGroupUnknown means part of `spec.minMember` pods are running but the other part can not
 	// be scheduled, e.g. not enough resource; scheduler will wait for related controller to recover it.
-	PodGroupUnschedulable PodGroupPhase = "Unschedulable"
+	PodGroupUnknown PodGroupPhase = "Unknown"
 )
 
-// PodGroupState contains details for the current state of this pod group.
-type PodGroupState struct {
-	// Current phase of PodGroup.
-	Phase PodGroupPhase `json:"phase,omitempty" protobuf:"bytes,1,opt,name=phase"`
+type PodGroupConditionType string
 
-	// Last time we probed to this Phase.
-	// +optional
-	LastProbeTime metav1.Time `json:"lastProbeTime,omitempty" protobuf:"bytes,2,opt,name=lastProbeTime"`
+const (
+	PodGroupUnschedulableType PodGroupConditionType = "Unschedulable"
+)
+
+// PodGroupCondition contains details for the current state of this pod group.
+type PodGroupCondition struct {
+	// Type is the type of the condition
+	Type PodGroupConditionType `json:"type,omitempty" protobuf:"bytes,1,opt,name=type"`
+
+	// Status is the status of the condition.
+	Status v1.ConditionStatus `json:"status,omitempty" protobuf:"bytes,2,opt,name=status"`
+
+	// The ID of condition transition.
+	TransitionID string `json:"transitionID,omitempty" protobuf:"bytes,3,opt,name=transitionID"`
 
 	// Last time the phase transitioned from another to current phase.
 	// +optional
-	LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty" protobuf:"bytes,3,opt,name=lastTransitionTime"`
+	LastTransitionTime metav1.Time `json:"lastTransitionTime,omitempty" protobuf:"bytes,4,opt,name=lastTransitionTime"`
 
 	// Unique, one-word, CamelCase reason for the phase's last transition.
 	// +optional
-	Reason string `json:"reason,omitempty" protobuf:"bytes,4,opt,name=reason"`
+	Reason string `json:"reason,omitempty" protobuf:"bytes,5,opt,name=reason"`
 
 	// Human-readable message indicating details about last transition.
 	// +optional
-	Message string `json:"message,omitempty" protobuf:"bytes,5,opt,name=message"`
+	Message string `json:"message,omitempty" protobuf:"bytes,6,opt,name=message"`
 }
 
 const (
@@ -122,21 +118,24 @@ type PodGroupSpec struct {
 
 // PodGroupStatus represents the current state of a pod group.
 type PodGroupStatus struct {
-	// The state of PodGroup.
+	// Current phase of PodGroup.
+	Phase PodGroupPhase `json:"phase,omitempty" protobuf:"bytes,1,opt,name=phase"`
+
+	// The conditions of PodGroup.
 	// +optional
-	State PodGroupState `json:"state,omitempty" protobuf:"bytes,1,opt,name=state,casttype=State"`
+	Conditions []PodGroupCondition `json:"conditions,omitempty" protobuf:"bytes,2,opt,name=conditions"`
 
 	// The number of actively running pods.
 	// +optional
-	Running int32 `json:"running,omitempty" protobuf:"bytes,2,opt,name=running"`
+	Running int32 `json:"running,omitempty" protobuf:"bytes,3,opt,name=running"`
 
 	// The number of pods which reached phase Succeeded.
 	// +optional
-	Succeeded int32 `json:"succeeded,omitempty" protobuf:"bytes,3,opt,name=succeeded"`
+	Succeeded int32 `json:"succeeded,omitempty" protobuf:"bytes,4,opt,name=succeeded"`
 
 	// The number of pods which reached phase Failed.
 	// +optional
-	Failed int32 `json:"failed,omitempty" protobuf:"bytes,4,opt,name=failed"`
+	Failed int32 `json:"failed,omitempty" protobuf:"bytes,5,opt,name=failed"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/scheduler/actions/allocate/allocate_test.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/record"
 
 	kbv1 "github.com/kubernetes-sigs/kube-batch/pkg/apis/scheduling/v1alpha1"
 	"github.com/kubernetes-sigs/kube-batch/pkg/scheduler/api"
@@ -113,12 +114,17 @@ func (fb *fakeBinder) Bind(p *v1.Pod, hostname string) error {
 	return nil
 }
 
-type fakeTaskStatusUpdater struct {
+type fakeStatusUpdater struct {
 }
 
-func (ftsu *fakeTaskStatusUpdater) Update(pod *v1.Pod, podCondition *v1.PodCondition) error {
+func (ftsu *fakeStatusUpdater) UpdatePod(pod *v1.Pod, podCondition *v1.PodCondition) (*v1.Pod, error) {
 	// do nothing here
-	return nil
+	return nil, nil
+}
+
+func (ftsu *fakeStatusUpdater) UpdatePodGroup(pg *kbv1.PodGroup) (*kbv1.PodGroup, error) {
+	// do nothing here
+	return nil, nil
 }
 
 type fakeVolumeBinder struct {
@@ -239,12 +245,14 @@ func TestAllocate(t *testing.T) {
 			c:     make(chan string),
 		}
 		schedulerCache := &cache.SchedulerCache{
-			Nodes:             make(map[string]*api.NodeInfo),
-			Jobs:              make(map[api.JobID]*api.JobInfo),
-			Queues:            make(map[api.QueueID]*api.QueueInfo),
-			Binder:            binder,
-			TaskStatusUpdater: &fakeTaskStatusUpdater{},
-			VolumeBinder:      &fakeVolumeBinder{},
+			Nodes:         make(map[string]*api.NodeInfo),
+			Jobs:          make(map[api.JobID]*api.JobInfo),
+			Queues:        make(map[api.QueueID]*api.QueueInfo),
+			Binder:        binder,
+			StatusUpdater: &fakeStatusUpdater{},
+			VolumeBinder:  &fakeVolumeBinder{},
+
+			Recorder: record.NewFakeRecorder(100),
 		}
 		for _, node := range test.nodes {
 			schedulerCache.AddNode(node)

pkg/scheduler/api/helpers.go

@@ -99,6 +99,5 @@ func MergeErrors(errs ...error) error {
 // JobTerminated checkes whether job was terminated.
 func JobTerminated(job *JobInfo) bool {
 	return job.PodGroup == nil &&
-		job.PDB == nil &&
 		len(job.Tasks) == 0
 }

pkg/scheduler/api/job_info.go

@@ -139,9 +139,6 @@ type JobInfo struct {
 
 	CreationTimestamp metav1.Time
 	PodGroup          *arbcorev1.PodGroup
-
-	// TODO(k82cn): keep backward compatbility, removed it when v1alpha1 finalized.
-	PDB *policyv1.PodDisruptionBudget
 }
 
 func NewJobInfo(uid JobID) *JobInfo {
@@ -196,11 +193,6 @@ func (ji *JobInfo) SetPDB(pdb *policyv1.PodDisruptionBudget) {
 	}
 
 	ji.CreationTimestamp = pdb.GetCreationTimestamp()
-	ji.PDB = pdb
-}
-
-func (ji *JobInfo) UnsetPDB() {
-	ji.PDB = nil
 }
 
 func (ji *JobInfo) GetTasks(statuses ...TaskStatus) []*TaskInfo {
@@ -292,7 +284,6 @@ func (ji *JobInfo) Clone() *JobInfo {
 		TotalRequest:  ji.TotalRequest.Clone(),
 		NodesFitDelta: make(NodeResourceMap),
 
-		PDB:      ji.PDB,
 		PodGroup: ji.PodGroup,
 
 		TaskStatusIndex: map[TaskStatus]tasksMap{},
@@ -326,14 +317,14 @@ func (ji JobInfo) String() string {
 
 // 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 {
+func (ji *JobInfo) FitError() string {
+	if len(ji.NodesFitDelta) == 0 {
 		reasonMsg := fmt.Sprintf("0 nodes are available")
 		return reasonMsg
 	}
 
 	reasons := make(map[string]int)
-	for _, v := range f.NodesFitDelta {
+	for _, v := range ji.NodesFitDelta {
 		if v.Get(v1.ResourceCPU) < 0 {
 			reasons["cpu"]++
 		}
@@ -353,6 +344,6 @@ func (f *JobInfo) FitError() string {
 		sort.Strings(reasonStrings)
 		return reasonStrings
 	}
-	reasonMsg := fmt.Sprintf("0/%v nodes are available, %v.", len(f.NodesFitDelta), strings.Join(sortReasonsHistogram(), ", "))
+	reasonMsg := fmt.Sprintf("0/%v nodes are available, %v.", len(ji.NodesFitDelta), strings.Join(sortReasonsHistogram(), ", "))
 	return reasonMsg
 }

pkg/scheduler/api/types.go

@@ -88,6 +88,15 @@ type CompareFn func(interface{}, interface{}) int
 // ValidateFn is the func declaration used to check object's status.
 type ValidateFn func(interface{}) bool
 
+//
+type ValidateResult struct {
+	Pass    bool
+	Reason  string
+	Message string
+}
+
+type ValidateExFn func(interface{}) *ValidateResult
+
 // PredicateFn is the func declaration used to predicate node for task.
 type PredicateFn func(*TaskInfo, *NodeInfo) error