Merge pull request volcano-sh#21 from qi-min/agent-scheduler-clean

clean unused functions in agent-scheduler

Author: JesseStutler

pkg/agentscheduler/cache/cache.go

@@ -96,7 +96,6 @@ type SchedulerCache struct {
 	nodeInformer infov1.NodeInformer
 
 	Binder        Binder
-	Evictor       Evictor
 	StatusUpdater StatusUpdater
 
 	Recorder record.EventRecorder
@@ -106,8 +105,6 @@ type SchedulerCache struct {
 
 	taskCache *TaskCache
 
-	NamespaceCollection map[string]*schedulingapi.NamespaceCollection
-
 	errTasks  workqueue.TypedRateLimitingInterface[string]
 	nodeQueue workqueue.TypedRateLimitingInterface[string]
 
@@ -361,16 +358,15 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
 	)
 
 	sc := &SchedulerCache{
-		Nodes:               make(map[string]*schedulingapi.NodeInfo),
-		errTasks:            workqueue.NewTypedRateLimitingQueue[string](errTaskRateLimiter),
-		nodeQueue:           workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
-		kubeClient:          kubeClient,
-		vcClient:            vcClient,
-		restConfig:          config,
-		schedulerNames:      schedulerNames,
-		nodeSelectorLabels:  make(map[string]sets.Empty),
-		NamespaceCollection: make(map[string]*schedulingapi.NamespaceCollection),
-		imageStates:         make(map[string]*imageState),
+		Nodes:              make(map[string]*schedulingapi.NodeInfo),
+		errTasks:           workqueue.NewTypedRateLimitingQueue[string](errTaskRateLimiter),
+		nodeQueue:          workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
+		kubeClient:         kubeClient,
+		vcClient:           vcClient,
+		restConfig:         config,
+		schedulerNames:     schedulerNames,
+		nodeSelectorLabels: make(map[string]sets.Empty),
+		imageStates:        make(map[string]*imageState),
 
 		NodeList:    []string{},
 		nodeWorkers: nodeWorkers,
@@ -395,11 +391,6 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
 	}
 	sc.Binder = GetBindMethod()
 
-	sc.Evictor = &defaultEvictor{
-		kubeclient: sc.kubeClient,
-		recorder:   sc.Recorder,
-	}
-
 	sc.StatusUpdater = &defaultStatusUpdater{
 		kubeclient: sc.kubeClient,
 		vcclient:   sc.vcClient,
@@ -572,9 +563,6 @@ func (sc *SchedulerCache) Run(stopCh <-chan struct{}) {
 	// Re-sync error tasks.
 	go wait.Until(sc.processResyncTask, 0, stopCh)
 
-	// Cleanup jobs.
-	go wait.Until(sc.processCleanupJob, 0, stopCh)
-
 	go wait.Until(sc.processBindTask, time.Millisecond*20, stopCh)
 
 	sc.ConflictAwareBinder.Run(stopCh)
@@ -586,43 +574,6 @@ func (sc *SchedulerCache) WaitForCacheSync(stopCh <-chan struct{}) {
 	sc.vcInformerFactory.WaitForCacheSync(stopCh)
 }
 
-// Evict will evict the pod.
-//
-// If error occurs both task and job are guaranteed to be in the original state.
-func (sc *SchedulerCache) Evict(task *schedulingapi.TaskInfo, reason string) error {
-	sc.Mutex.Lock()
-	defer sc.Mutex.Unlock()
-
-	node, found := sc.Nodes[task.NodeName]
-	if !found {
-		return fmt.Errorf("failed to evict Task %v from host %v, host does not exist",
-			task.UID, task.NodeName)
-	}
-
-	originalStatus := task.Status
-
-	// Add new task to node.
-	if err := node.UpdateTask(task); err != nil {
-		// After failing to update task to a node we need to revert task status from Releasing,
-		// otherwise task might be stuck in the Releasing state indefinitely.
-		klog.Errorf("Task <%s/%s> will be resynchronized after failing to revert status "+
-			"from %s to %s after failing to update Task on Node <%s>: %v",
-			task.Namespace, task.Name, task.Status, originalStatus, node.Name, err)
-		sc.resyncTask(task)
-		return err
-	}
-
-	p := task.Pod
-
-	go func() {
-		err := sc.Evictor.Evict(p, reason)
-		if err != nil {
-			sc.resyncTask(task)
-		}
-	}()
-	return nil
-}
-
 // Bind binds task to the target host.
 func (sc *SchedulerCache) Bind(ctx context.Context, bindContexts []*vcache.BindContext, preBinders map[string]PreBinder) {
 	readyToBindTasks := make([]*schedulingapi.TaskInfo, len(bindContexts))
@@ -747,10 +698,6 @@ func (sc *SchedulerCache) TaskUnschedulable(task *schedulingapi.TaskInfo, reason
 	return nil
 }
 
-func (sc *SchedulerCache) processCleanupJob() {
-	// TODO process clean up Job
-}
-
 // GetTaskInfo retrieves a task by its ID.
 func (sc *SchedulerCache) GetTaskInfo(taskID schedulingapi.TaskID) (*schedulingapi.TaskInfo, bool) {
 	return sc.taskCache.Get(taskID)
@@ -772,15 +719,7 @@ func (sc *SchedulerCache) DeleteTaskInfo(taskID schedulingapi.TaskID) {
 }
 
 func (sc *SchedulerCache) resyncTask(task *schedulingapi.TaskInfo) {
-	key := sc.generateErrTaskKey(task)
-	sc.errTasks.AddRateLimited(key)
-}
-
-func (sc *SchedulerCache) generateErrTaskKey(task *schedulingapi.TaskInfo) string {
-	// Job UID is namespace + / +name, for example: theNs/theJob
-	// Task UID is derived from the Pod UID, for example: d336abea-4f14-42c7-8a6b-092959a31407
-	// In the example above, the key ultimately becomes: theNs/theJob/d336abea-4f14-42c7-8a6b-092959a31407
-	return fmt.Sprintf("%s/%s", task.Job, task.UID)
+	sc.errTasks.AddRateLimited(string(task.UID))
 }
 
 func (sc *SchedulerCache) parseErrTaskKey(key string) (*schedulingapi.TaskInfo, error) {
@@ -990,17 +929,15 @@ func (sc *SchedulerCache) BindTask() {
 	sc.bindCache = make([]*vcache.BindContext, 0)
 }
 
-// Snapshot returns the complete snapshot of the cluster from cache
+// Snapshot returns the complete snapshot of the cluster from cache, used for dump purpose only
 func (sc *SchedulerCache) Snapshot() *schedulingapi.ClusterInfo {
 	sc.Mutex.Lock()
 	defer sc.Mutex.Unlock()
 
 	snapshot := &schedulingapi.ClusterInfo{
-		Nodes:          make(map[string]*schedulingapi.NodeInfo),
-		RealNodesSet:   make(map[string]sets.Set[string]),
-		NamespaceInfo:  make(map[schedulingapi.NamespaceName]*schedulingapi.NamespaceInfo),
-		RevocableNodes: make(map[string]*schedulingapi.NodeInfo),
-		NodeList:       make([]string, len(sc.NodeList)),
+		Nodes:        make(map[string]*schedulingapi.NodeInfo),
+		RealNodesSet: make(map[string]sets.Set[string]),
+		NodeList:     make([]string, len(sc.NodeList)),
 	}
 
 	// TODO add agent scheduler cache
@@ -1016,17 +953,8 @@ func (sc *SchedulerCache) Snapshot() *schedulingapi.ClusterInfo {
 		}
 
 		snapshot.Nodes[value.Name] = value.Clone()
-
-		if value.RevocableZone != "" {
-			snapshot.RevocableNodes[value.Name] = snapshot.Nodes[value.Name]
-		}
 	}
-
-	for _, value := range sc.NamespaceCollection {
-		info := value.Snapshot()
-		snapshot.NamespaceInfo[info.Name] = info
-	}
-	klog.V(3).InfoS("SnapShot for scheduling", "NodeNum", "NamespaceNum", "RevocableNodesNum", len(snapshot.Nodes), len(snapshot.NamespaceInfo), len(snapshot.RevocableNodes))
+	klog.V(3).InfoS("SnapShot for scheduling", "NodeNum", len(snapshot.Nodes))
 	return snapshot
 }
 
@@ -1060,14 +988,6 @@ func (sc *SchedulerCache) String() string {
 		}
 	}
 
-	if len(sc.NamespaceCollection) != 0 {
-		str += "Namespaces:\n"
-		for _, ns := range sc.NamespaceCollection {
-			info := ns.Snapshot()
-			str += fmt.Sprintf("\t Namespace(%s)\n", info.Name)
-		}
-	}
-
 	if len(sc.NodeList) != 0 {
 		str += fmt.Sprintf("NodeList: %v\n", sc.NodeList)
 	}

pkg/agentscheduler/cache/cache_mock.go

@@ -42,17 +42,14 @@ import (
 // NewCustomMockSchedulerCache returns a mock scheduler cache with custom interface
 func NewCustomMockSchedulerCache(schedulerName string,
 	binder Binder,
-	evictor Evictor,
 	statusUpdater StatusUpdater,
-	PodGroupBinder BatchBinder,
 	recorder record.EventRecorder,
 ) *SchedulerCache {
 	msc := newMockSchedulerCache(schedulerName)
 	// add all events handlers
 	msc.addEventHandler()
 	msc.Recorder = recorder
 	msc.Binder = binder
-	msc.Evictor = evictor
 	msc.StatusUpdater = statusUpdater
 	checkAndSetDefaultInterface(msc)
 	return msc
@@ -79,12 +76,6 @@ func checkAndSetDefaultInterface(sc *SchedulerCache) {
 			recorder:   sc.Recorder,
 		}
 	}
-	if sc.Evictor == nil {
-		sc.Evictor = &defaultEvictor{
-			kubeclient: sc.kubeClient,
-			recorder:   sc.Recorder,
-		}
-	}
 	if sc.StatusUpdater == nil {
 		sc.StatusUpdater = &defaultStatusUpdater{
 			kubeclient: sc.kubeClient,
@@ -107,16 +98,15 @@ func getNodeWorkers() uint32 {
 // newMockSchedulerCache init the mock scheduler cache structure
 func newMockSchedulerCache(schedulerName string) *SchedulerCache {
 	msc := &SchedulerCache{
-		Nodes:               make(map[string]*schedulingapi.NodeInfo),
-		errTasks:            workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
-		nodeQueue:           workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
-		kubeClient:          fake.NewSimpleClientset(),
-		vcClient:            fakevcClient.NewSimpleClientset(),
-		restConfig:          nil,
-		schedulerNames:      []string{schedulerName},
-		nodeSelectorLabels:  make(map[string]sets.Empty),
-		NamespaceCollection: make(map[string]*schedulingapi.NamespaceCollection),
-		imageStates:         make(map[string]*imageState),
+		Nodes:              make(map[string]*schedulingapi.NodeInfo),
+		errTasks:           workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
+		nodeQueue:          workqueue.NewTypedRateLimitingQueue[string](workqueue.DefaultTypedControllerRateLimiter[string]()),
+		kubeClient:         fake.NewSimpleClientset(),
+		vcClient:           fakevcClient.NewSimpleClientset(),
+		restConfig:         nil,
+		schedulerNames:     []string{schedulerName},
+		nodeSelectorLabels: make(map[string]sets.Empty),
+		imageStates:        make(map[string]*imageState),
 
 		NodeList:       []string{},
 		binderRegistry: NewBinderRegistry(),

pkg/agentscheduler/cache/dumper.go

@@ -79,13 +79,6 @@ func (d *Dumper) dumpAll() {
 		klog.Info(d.printNodeInfo(nodeInfo))
 	}
 
-	klog.Info("Dump of jobs info in scheduler cache")
-	for _, jobInfo := range snapshot.Jobs {
-		klog.Info(d.printJobInfo(jobInfo))
-	}
-
-	klog.Info("Dump of hyperNodes info in scheduler cache")
-
 	d.displaySchedulerMemStats()
 }
 
@@ -103,14 +96,6 @@ func (d *Dumper) printNodeInfo(node *api.NodeInfo) string {
 	return data.String()
 }
 
-func (d *Dumper) printJobInfo(jobInfo *api.JobInfo) string {
-	var data strings.Builder
-	data.WriteString("\n")
-	data.WriteString(jobInfo.String())
-	data.WriteString("\n")
-	return data.String()
-}
-
 // ListenForSignal starts a goroutine that will respond when process
 // receives SIGUSER1/SIGUSER2 signal.
 func (d *Dumper) ListenForSignal(stopCh <-chan struct{}) {

pkg/agentscheduler/cache/event_handlers.go

@@ -506,80 +506,3 @@ func (sc *SchedulerCache) nodeCanAddCache(node *v1.Node) bool {
 	klog.Infof("node %s ignore add/update/delete into schedulerCache", node.Name)
 	return false
 }
-
-func (sc *SchedulerCache) updateResourceQuota(quota *v1.ResourceQuota) {
-	collection, ok := sc.NamespaceCollection[quota.Namespace]
-	if !ok {
-		collection = schedulingapi.NewNamespaceCollection(quota.Namespace)
-		sc.NamespaceCollection[quota.Namespace] = collection
-	}
-
-	collection.Update(quota)
-}
-
-func (sc *SchedulerCache) deleteResourceQuota(quota *v1.ResourceQuota) {
-	collection, ok := sc.NamespaceCollection[quota.Namespace]
-	if !ok {
-		return
-	}
-
-	collection.Delete(quota)
-}
-
-// DeleteResourceQuota delete ResourceQuota from the scheduler cache
-func (sc *SchedulerCache) DeleteResourceQuota(obj interface{}) {
-	var r *v1.ResourceQuota
-	switch t := obj.(type) {
-	case *v1.ResourceQuota:
-		r = t
-	case cache.DeletedFinalStateUnknown:
-		var ok bool
-		r, ok = t.Obj.(*v1.ResourceQuota)
-		if !ok {
-			klog.Errorf("Cannot convert to *v1.ResourceQuota: %v", t.Obj)
-			return
-		}
-	default:
-		klog.Errorf("Cannot convert to *v1.ResourceQuota: %v", t)
-		return
-	}
-
-	sc.Mutex.Lock()
-	defer sc.Mutex.Unlock()
-
-	klog.V(3).Infof("Delete ResourceQuota <%s/%v> in cache", r.Namespace, r.Name)
-	sc.deleteResourceQuota(r)
-}
-
-// UpdateResourceQuota update ResourceQuota to scheduler cache
-func (sc *SchedulerCache) UpdateResourceQuota(oldObj, newObj interface{}) {
-	newR, ok := newObj.(*v1.ResourceQuota)
-	if !ok {
-		klog.Errorf("Cannot convert newObj to *v1.ResourceQuota: %v", newObj)
-		return
-	}
-
-	sc.Mutex.Lock()
-	defer sc.Mutex.Unlock()
-
-	klog.V(3).Infof("Update ResourceQuota <%s/%v> in cache, with spec: %v.", newR.Namespace, newR.Name, newR.Spec.Hard)
-	sc.updateResourceQuota(newR)
-}
-
-// AddResourceQuota add ResourceQuota to scheduler cache
-func (sc *SchedulerCache) AddResourceQuota(obj interface{}) {
-	var r *v1.ResourceQuota
-	switch t := obj.(type) {
-	case *v1.ResourceQuota:
-		r = t
-	default:
-		klog.Errorf("Cannot convert to *v1.ResourceQuota: %v", t)
-		return
-	}
-
-	sc.Mutex.Lock()
-	defer sc.Mutex.Unlock()
-
-	klog.V(3).Infof("Add ResourceQuota <%s/%v> in cache, with spec: %v.", r.Namespace, r.Name, r.Spec.Hard)
-	sc.updateResourceQuota(r)
-}

pkg/agentscheduler/cache/interface.go

@@ -54,8 +54,6 @@ type Cache interface {
 	// TODO(jinzhej): clean up expire Tasks.
 	AddBindTask(bindCtx *vcache.BindContext) error
 
-	// Evict evicts the task to release resources.
-	Evict(task *api.TaskInfo, reason string) error
 	// Client returns the kubernetes clientSet, which can be used by plugins
 	Client() kubernetes.Interface
 
@@ -103,21 +101,11 @@ type Binder interface {
 	Bind(kubeClient kubernetes.Interface, tasks []*api.TaskInfo) map[api.TaskID]string
 }
 
-// Evictor interface for evict pods
-type Evictor interface {
-	Evict(pod *v1.Pod, reason string) error
-}
-
 // StatusUpdater updates pod with given PodCondition
 type StatusUpdater interface {
 	UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error)
 }
 
-// BatchBinder updates podgroup or job information
-type BatchBinder interface {
-	Bind(job *api.JobInfo, cluster string) (*api.JobInfo, error)
-}
-
 type PreBinder interface {
 	PreBind(ctx context.Context, bindCtx *vcache.BindContext) error
 

pkg/agentscheduler/plugins/factory.go

@@ -22,7 +22,6 @@ import (
 )
 
 func init() {
-	// Plugins for Jobs
 	framework.RegisterPluginBuilder(predicates.PluginName, predicates.New)
 	// framework.RegisterPluginBuilder(nodeorder.PluginName, nodeorder.New)
 }