pbacsko commented on code in PR #734:
URL: https://github.com/apache/yunikorn-k8shim/pull/734#discussion_r1440660876
##########
pkg/cache/context.go:
##########
@@ -48,17 +52,19 @@ import (
"github.com/apache/yunikorn-scheduler-interface/lib/go/si"
)
+const registerNodeContextHandler = "RegisterNodeContextHandler"
+
// context maintains scheduling state, like apps and apps' tasks.
type Context struct {
applications map[string]*Application // apps
- nodes *schedulerNodes // nodes
schedulerCache *schedulercache.SchedulerCache // external cache
apiProvider client.APIProvider // apis to interact with
api-server, scheduler-core, etc
predManager predicates.PredicateManager // K8s predicates
pluginMode bool // true if we are
configured as a scheduler plugin
namespace string // yunikorn namespace
configMaps []*v1.ConfigMap // cached yunikorn
configmaps
lock *sync.RWMutex // lock
+ txnID atomic.Uint64 // transaction ID counter
Review Comment:
This counter is referenced only at a single place and it's never
incremented. I guess it was used in a previous version of the code.
##########
pkg/cache/context.go:
##########
@@ -126,64 +115,101 @@ func (ctx *Context) AddSchedulingEventHandlers() {
UpdateFn: ctx.updatePriorityClass,
DeleteFn: ctx.deletePriorityClass,
})
+ ctx.apiProvider.AddEventHandler(&client.ResourceEventHandlers{
+ Type: client.NodeInformerHandlers,
+ AddFn: ctx.addNode,
+ UpdateFn: ctx.updateNode,
+ DeleteFn: ctx.deleteNode,
+ })
+ ctx.apiProvider.AddEventHandler(&client.ResourceEventHandlers{
+ Type: client.PodInformerHandlers,
+ AddFn: ctx.AddPod,
+ UpdateFn: ctx.UpdatePod,
+ DeleteFn: ctx.DeletePod,
+ })
}
func (ctx *Context) IsPluginMode() bool {
return ctx.pluginMode
}
func (ctx *Context) addNode(obj interface{}) {
+ ctx.updateNode(nil, obj)
+}
+
+func (ctx *Context) updateNode(_, obj interface{}) {
ctx.lock.Lock()
defer ctx.lock.Unlock()
-
node, err := convertToNode(obj)
if err != nil {
log.Log(log.ShimContext).Error("node conversion failed",
zap.Error(err))
return
}
-
- // add node to secondary scheduler cache
- log.Log(log.ShimContext).Warn("adding node to cache",
zap.String("NodeName", node.Name))
- ctx.schedulerCache.AddNode(node)
-
- // add node to internal cache
- ctx.nodes.addNode(node)
-
- // post the event
- events.GetRecorder().Eventf(node.DeepCopy(), nil, v1.EventTypeNormal,
"NodeAccepted", "NodeAccepted",
- fmt.Sprintf("node %s is accepted by the scheduler", node.Name))
+ ctx.updateNodeInternal(node, true)
}
-func (ctx *Context) updateNode(oldObj, newObj interface{}) {
- ctx.lock.Lock()
- defer ctx.lock.Unlock()
+func (ctx *Context) updateNodeInternal(node *v1.Node, register bool) {
+ // update scheduler cache
+ if prevNode, adoptedPods := ctx.schedulerCache.UpdateNode(node);
prevNode == nil {
+ // newly added node
+
+ // if requested, register this node with the scheduler core.
this is optional to allow for bulk registration
+ // during scheduler initialization.
+ if register {
+ if err := ctx.registerNode(node); err != nil {
+ // remove from secondary cache and return
Review Comment:
Do we even have primary/secondary caches anymore?
##########
pkg/cache/external/scheduler_cache.go:
##########
@@ -403,21 +484,20 @@ func (cache *SchedulerCache) updatePod(pod *v1.Pod) {
// assign to node
nodeInfo, ok := cache.nodesMap[pod.Spec.NodeName]
if !ok {
- // node doesn't exist, create a synthetic one for now
- nodeInfo = framework.NewNodeInfo()
- cache.nodesMap[pod.Spec.NodeName] = nodeInfo
- // work around a crash bug in NodeInfo.RemoveNode()
when Node is unset
- nodeInfo.SetNode(&v1.Node{ObjectMeta:
metav1.ObjectMeta{Name: pod.Spec.NodeName}})
- }
- nodeInfo.AddPod(pod)
- cache.assignedPods[key] = pod.Spec.NodeName
- if podWithAffinity(pod) {
- cache.nodesInfoPodsWithAffinity = nil
- }
- if podWithRequiredAntiAffinity(pod) {
- cache.nodesInfoPodsWithReqAntiAffinity = nil
+ // node doesn't exist, so this is an orphaned pod
+ cache.orphanedPods[key] = pod
Review Comment:
Just like in case of pod adoption in `Context.updateNodeInternal()`, logging
is useful here
##########
pkg/cache/context.go:
##########
@@ -126,64 +115,101 @@ func (ctx *Context) AddSchedulingEventHandlers() {
UpdateFn: ctx.updatePriorityClass,
DeleteFn: ctx.deletePriorityClass,
})
+ ctx.apiProvider.AddEventHandler(&client.ResourceEventHandlers{
+ Type: client.NodeInformerHandlers,
+ AddFn: ctx.addNode,
+ UpdateFn: ctx.updateNode,
+ DeleteFn: ctx.deleteNode,
+ })
+ ctx.apiProvider.AddEventHandler(&client.ResourceEventHandlers{
+ Type: client.PodInformerHandlers,
+ AddFn: ctx.AddPod,
+ UpdateFn: ctx.UpdatePod,
+ DeleteFn: ctx.DeletePod,
+ })
}
func (ctx *Context) IsPluginMode() bool {
return ctx.pluginMode
}
func (ctx *Context) addNode(obj interface{}) {
+ ctx.updateNode(nil, obj)
+}
+
+func (ctx *Context) updateNode(_, obj interface{}) {
ctx.lock.Lock()
defer ctx.lock.Unlock()
-
node, err := convertToNode(obj)
if err != nil {
log.Log(log.ShimContext).Error("node conversion failed",
zap.Error(err))
return
}
-
- // add node to secondary scheduler cache
- log.Log(log.ShimContext).Warn("adding node to cache",
zap.String("NodeName", node.Name))
- ctx.schedulerCache.AddNode(node)
-
- // add node to internal cache
- ctx.nodes.addNode(node)
-
- // post the event
- events.GetRecorder().Eventf(node.DeepCopy(), nil, v1.EventTypeNormal,
"NodeAccepted", "NodeAccepted",
- fmt.Sprintf("node %s is accepted by the scheduler", node.Name))
+ ctx.updateNodeInternal(node, true)
}
-func (ctx *Context) updateNode(oldObj, newObj interface{}) {
- ctx.lock.Lock()
- defer ctx.lock.Unlock()
+func (ctx *Context) updateNodeInternal(node *v1.Node, register bool) {
+ // update scheduler cache
+ if prevNode, adoptedPods := ctx.schedulerCache.UpdateNode(node);
prevNode == nil {
+ // newly added node
+
+ // if requested, register this node with the scheduler core.
this is optional to allow for bulk registration
+ // during scheduler initialization.
+ if register {
+ if err := ctx.registerNode(node); err != nil {
+ // remove from secondary cache and return
+ log.Log(log.ShimContext).Error("node
registration failed", zap.Error(err))
+ ctx.schedulerCache.RemoveNode(node)
+ return
+ }
+ }
- // we only trigger update when resource changes
- oldNode, err := convertToNode(oldObj)
- if err != nil {
- log.Log(log.ShimContext).Error("old node conversion failed",
- zap.Error(err))
- return
- }
+ // iterate newly adopted pods and register them with the
scheduler
+ for _, pod := range adoptedPods {
Review Comment:
I would definitely log something here. IMO it's very informative to see
orphaned pods getting assigned to nodes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
