Skip to content

[BugFix] Fix EngineConn uneven distribution with weighted random selection (#5322) #5469

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhang-arvin wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[BugFix] Fix EngineConn uneven distribution with weighted random selection (#5322) #5469

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 55 additions & 2 deletions ...on-manager/src/main/scala/org/apache/linkis/manager/am/selector/DefaultNodeSelector.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,16 @@ package org.apache.linkis.manager.am.selector

import org.apache.linkis.common.utils.{Logging, Utils}
import org.apache.linkis.manager.am.selector.rule.NodeSelectRule
import org.apache.linkis.manager.common.entity.node.Node
import org.apache.linkis.manager.common.entity.node.{Node, RMNode}
import org.apache.linkis.manager.common.utils.ResourceUtils

import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

import java.util

import scala.collection.JavaConverters._
import scala.util.Random

@Service
class DefaultNodeSelector extends NodeSelector with Logging {
Expand All @@ -37,6 +39,8 @@ class DefaultNodeSelector extends NodeSelector with Logging {
/**
* Select the most suitable node from a series of nodes through selection rules
* 1. Rule processing logic, defaults to the last priority
* 2. After rule filtering, use weighted random selection based on available resources
* to avoid always picking the same best node, thereby improving EC distribution balance
*
* @param nodes
* @return
Expand All @@ -56,11 +60,60 @@ class DefaultNodeSelector extends NodeSelector with Logging {
if (resultNodes.isEmpty) {
None
} else {
Some(resultNodes(0))
Some(selectNodeWeightedRandom(resultNodes))
}
}
}

/**
* Weighted random selection based on node resource availability.
* Nodes with more remaining resources have higher probability of being selected,
* which helps distribute EngineConns more evenly across ECMs.
*/
private def selectNodeWeightedRandom(nodes: Array[Node]): Node = {
if (nodes.length <= 1) {
return nodes(0)
}

val weights = nodes.map { node =>
Utils.tryCatch {
node match {
case rmNode: RMNode =>
val nodeResource = rmNode.getNodeResource
if (nodeResource != null && nodeResource.getLeftResource != null &&
nodeResource.getMaxResource != null) {
val rate = ResourceUtils.getLoadInstanceResourceRate(
nodeResource.getLeftResource,
nodeResource.getMaxResource
)
// Ensure minimum weight of 1.0 to avoid zero-weight nodes being excluded
Math.max(rate.toDouble, 1.0)
} else {
1.0
}
case _ => 1.0
}
} { case _: Throwable =>
1.0
}
}

val totalWeight = weights.sum
if (totalWeight <= 0) {
return nodes(0)
}

val random = Random.nextDouble() * totalWeight
var cumulativeWeight = 0.0
for (i <- nodes.indices) {
cumulativeWeight += weights(i)
if (cumulativeWeight >= random) {
return nodes(i)
}
}
nodes(0)
}

override def getNodeSelectRules(): Array[NodeSelectRule] = {
if (null != ruleList) ruleList.asScala.toArray
else Array.empty[NodeSelectRule]
Expand Down