Add queue controller #128
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,278 @@ | ||
| /* | ||
| Copyright 2019 The Volcano Authors. | ||
|
|
||
| Licensed under the Apache License, Version 2.0 (the "License"); | ||
| you may not use this file except in compliance with the License. | ||
| You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package queue | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/golang/glog" | ||
| "k8s.io/apimachinery/pkg/api/errors" | ||
| "k8s.io/apimachinery/pkg/util/wait" | ||
| "k8s.io/client-go/kubernetes" | ||
| "k8s.io/client-go/tools/cache" | ||
| "k8s.io/client-go/util/workqueue" | ||
|
|
||
| kbv1alpha1 "github.com/kubernetes-sigs/kube-batch/pkg/apis/scheduling/v1alpha1" | ||
| kbclientset "github.com/kubernetes-sigs/kube-batch/pkg/client/clientset/versioned" | ||
| kbinformerfactory "github.com/kubernetes-sigs/kube-batch/pkg/client/informers/externalversions" | ||
| kbinformer "github.com/kubernetes-sigs/kube-batch/pkg/client/informers/externalversions/scheduling/v1alpha1" | ||
| kblister "github.com/kubernetes-sigs/kube-batch/pkg/client/listers/scheduling/v1alpha1" | ||
| ) | ||
|
|
||
| // Controller manages queue status. | ||
| type Controller struct { | ||
| kubeClient kubernetes.Interface | ||
| kbClient kbclientset.Interface | ||
|
|
||
| // informer | ||
| queueInformer kbinformer.QueueInformer | ||
| pgInformer kbinformer.PodGroupInformer | ||
|
|
||
| // queueLister | ||
| queueLister kblister.QueueLister | ||
| queueSynced cache.InformerSynced | ||
|
|
||
| // podGroup lister | ||
| pgLister kblister.PodGroupLister | ||
| pgSynced cache.InformerSynced | ||
|
|
||
| // queues that need to be updated. | ||
| queue workqueue.RateLimitingInterface | ||
|
|
||
| pgMutex sync.RWMutex | ||
| podGroups map[string]map[string]struct{} | ||
| } | ||
|
|
||
| // NewQueueController creates a QueueController | ||
| func NewQueueController( | ||
| kubeClient kubernetes.Interface, | ||
| kbClient kbclientset.Interface, | ||
| ) *Controller { | ||
| factory := kbinformerfactory.NewSharedInformerFactory(kbClient, 0) | ||
| queueInformer := factory.Scheduling().V1alpha1().Queues() | ||
| pgInformer := factory.Scheduling().V1alpha1().PodGroups() | ||
| c := &Controller{ | ||
| kubeClient: kubeClient, | ||
| kbClient: kbClient, | ||
|
|
||
| queueInformer: queueInformer, | ||
| pgInformer: pgInformer, | ||
|
|
||
| queueLister: queueInformer.Lister(), | ||
| queueSynced: queueInformer.Informer().HasSynced, | ||
|
|
||
| pgLister: pgInformer.Lister(), | ||
| pgSynced: pgInformer.Informer().HasSynced, | ||
|
|
||
| queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()), | ||
| podGroups: make(map[string]map[string]struct{}), | ||
| } | ||
|
|
||
| queueInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ | ||
| AddFunc: c.addQueue, | ||
| DeleteFunc: c.deleteQueue, | ||
| }) | ||
|
|
||
| pgInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ | ||
| AddFunc: c.addPodGroup, | ||
| UpdateFunc: c.updatePodGroup, | ||
| DeleteFunc: c.deletePodGroup, | ||
| }) | ||
|
|
||
| return c | ||
| } | ||
|
|
||
| // Run starts QueueController | ||
| func (c *Controller) Run(stopCh <-chan struct{}) { | ||
|
|
||
| go c.queueInformer.Informer().Run(stopCh) | ||
| go c.pgInformer.Informer().Run(stopCh) | ||
|
|
||
| if !cache.WaitForCacheSync(stopCh, c.queueSynced, c.pgSynced) { | ||
| glog.Errorf("unable to sync caches for queue controller") | ||
| return | ||
| } | ||
|
|
||
| go wait.Until(c.worker, 0, stopCh) | ||
| glog.Infof("QueueController is running ...... ") | ||
| } | ||
|
|
||
| // worker runs a worker thread that just dequeues items, processes them, and | ||
| // marks them done. You may run as many of these in parallel as you wish; the | ||
| // workqueue guarantees that they will not end up processing the same `queue` | ||
| // at the same time. | ||
| func (c *Controller) worker() { | ||
| for c.processNextWorkItem() { | ||
| } | ||
| } | ||
|
|
||
| func (c *Controller) processNextWorkItem() bool { | ||
| eKey, quit := c.queue.Get() | ||
| if quit { | ||
| return false | ||
| } | ||
| defer c.queue.Done(eKey) | ||
|
|
||
| if err := c.syncQueue(eKey.(string)); err != nil { | ||
| glog.V(2).Infof("Error syncing queues %q, retrying. Error: %v", eKey, err) | ||
| c.queue.AddRateLimited(eKey) | ||
| return true | ||
| } | ||
|
|
||
| c.queue.Forget(eKey) | ||
| return true | ||
| } | ||
|
|
||
| func (c *Controller) syncQueue(key string) error { | ||
| glog.V(4).Infof("Begin sync queue %s", key) | ||
|
|
||
| var pending, running, unknown int32 | ||
| c.pgMutex.RLock() | ||
| if c.podGroups[key] == nil { | ||
| c.pgMutex.RUnlock() | ||
| glog.V(2).Infof("queue %s has not been seen or deleted", key) | ||
| return nil | ||
| } | ||
| podGroups := make([]string, 0, len(c.podGroups[key])) | ||
| for pgKey := range c.podGroups[key] { | ||
|
There was a problem hiding this comment. prefer to handle this in Cache: Queue -> Job -> PodGroup. There was a problem hiding this comment. I donot want to overlap with job controller. For queue controller, it just care about the podgroups, not care about Job or anything else. podGroups is keyed by queue name and value is the podgroups that are using this queue. There was a problem hiding this comment.
There was a problem hiding this comment. IC, then we need to record jobs too. There was a problem hiding this comment. But have a doubt: why would user delete their queues before jobs? Is there a use case? If there is a malicious user who does this, we should not delete the running jobs by any reason. There was a problem hiding this comment. For this, I can think of a more simple way to do this: I remember there will be an admission controller to validate queue existence before create jobs. Then we can also set the job's owner reference. Then when the queue is deleted, k8s garbage collector can reclaim all children resources automatically. So we have two choice:
There was a problem hiding this comment. Personally prefer not adding this into cache even we need considering deleting jobs when queue deleted. There was a problem hiding this comment.
owerReference should be reserved for There was a problem hiding this comment. OwnerReferences is an array, can be more than one.
Sorry, there is no There was a problem hiding this comment. not sure whether GC will check controlled owner reference.
so "reserved" it. |
||
| podGroups = append(podGroups, pgKey) | ||
| } | ||
| c.pgMutex.RUnlock() | ||
|
|
||
| for _, pgKey := range podGroups { | ||
| // Ignore error here, tt can not occur. | ||
| ns, name, _ := cache.SplitMetaNamespaceKey(pgKey) | ||
|
|
||
| pg, err := c.pgLister.PodGroups(ns).Get(name) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| switch pg.Status.Phase { | ||
| case kbv1alpha1.PodGroupPending: | ||
| pending++ | ||
| case kbv1alpha1.PodGroupRunning: | ||
| running++ | ||
| case kbv1alpha1.PodGroupUnknown: | ||
| unknown++ | ||
| } | ||
| } | ||
|
|
||
| queue, err := c.queueLister.Get(key) | ||
| if err != nil { | ||
| if errors.IsNotFound(err) { | ||
| glog.V(2).Infof("queue %s has been deleted", queue) | ||
| return nil | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| glog.V(4).Infof("queue %s jobs pending %d, running %d, unknown %d", key, pending, running, unknown) | ||
| // ignore update when status doesnot change | ||
| if pending == queue.Status.Pending && running == queue.Status.Running && unknown == queue.Status.Unknown { | ||
| return nil | ||
| } | ||
|
|
||
| newQueue := queue.DeepCopy() | ||
| newQueue.Status.Pending = pending | ||
| newQueue.Status.Running = running | ||
| newQueue.Status.Unknown = unknown | ||
|
|
||
| if _, err := c.kbClient.SchedulingV1alpha1().Queues().UpdateStatus(newQueue); err != nil { | ||
| glog.Errorf("Failed to update status of Queue %s: %v", newQueue.Name, err) | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (c *Controller) addQueue(obj interface{}) { | ||
| queue := obj.(*kbv1alpha1.Queue) | ||
| c.queue.Add(queue.Name) | ||
| } | ||
|
|
||
| func (c *Controller) deleteQueue(obj interface{}) { | ||
| queue, ok := obj.(*kbv1alpha1.Queue) | ||
| if !ok { | ||
| tombstone, ok := obj.(cache.DeletedFinalStateUnknown) | ||
| if !ok { | ||
| glog.Errorf("Couldn't get object from tombstone %#v", obj) | ||
| return | ||
| } | ||
| queue, ok = tombstone.Obj.(*kbv1alpha1.Queue) | ||
| if !ok { | ||
| glog.Errorf("Tombstone contained object that is not a Queue: %#v", obj) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| c.pgMutex.Lock() | ||
| delete(c.podGroups, queue.Name) | ||
| c.pgMutex.Unlock() | ||
| } | ||
|
|
||
| func (c *Controller) addPodGroup(obj interface{}) { | ||
| pg := obj.(*kbv1alpha1.PodGroup) | ||
| key, _ := cache.MetaNamespaceKeyFunc(obj) | ||
|
|
||
| c.pgMutex.Lock() | ||
| if c.podGroups[pg.Spec.Queue] == nil { | ||
| c.podGroups[pg.Spec.Queue] = make(map[string]struct{}) | ||
| } | ||
| c.podGroups[pg.Spec.Queue][key] = struct{}{} | ||
| c.pgMutex.Unlock() | ||
|
|
||
| // enqueue | ||
| c.queue.Add(pg.Spec.Queue) | ||
| } | ||
|
|
||
| func (c *Controller) updatePodGroup(old, new interface{}) { | ||
| oldPG := old.(*kbv1alpha1.PodGroup) | ||
| newPG := new.(*kbv1alpha1.PodGroup) | ||
|
|
||
| // Note: we have no use case update PodGroup.Spec.Queue | ||
| // So do not consider it here. | ||
| if oldPG.Status.Phase != newPG.Status.Phase { | ||
| // enqueue | ||
| c.queue.Add(newPG.Spec.Queue) | ||
| } | ||
|
|
||
| } | ||
|
|
||
| func (c *Controller) deletePodGroup(obj interface{}) { | ||
| pg, ok := obj.(*kbv1alpha1.PodGroup) | ||
| if !ok { | ||
| tombstone, ok := obj.(cache.DeletedFinalStateUnknown) | ||
| if !ok { | ||
| glog.Errorf("Couldn't get object from tombstone %#v", obj) | ||
| return | ||
| } | ||
| pg, ok = tombstone.Obj.(*kbv1alpha1.PodGroup) | ||
| if !ok { | ||
| glog.Errorf("Tombstone contained object that is not a PodGroup: %#v", obj) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| key, _ := cache.MetaNamespaceKeyFunc(obj) | ||
|
|
||
| c.pgMutex.Lock() | ||
| delete(c.podGroups[pg.Spec.Queue], key) | ||
| c.pgMutex.Unlock() | ||
|
|
||
| c.queue.Add(pg.Spec.Queue) | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What is user agent?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It is just for http server to know the client identity.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A http header