in brooklyn-server/policy/src/main/java/org/apache/brooklyn/policy/loadbalancing/BalancingStrategy.java [157:342]
protected boolean balanceItemsOnHotNode(NodeType node, double nodeWorkrate, boolean gonnaGrow) {
double originalNodeWorkrate = nodeWorkrate;
int migrationCount = 0;
int iterationCount = 0;
Set<ItemType> itemsMoved = new LinkedHashSet<ItemType>();
Set<NodeType> nodesChecked = new LinkedHashSet<NodeType>();
// if (nodeFinder.config.COUNT_BACKLOG_AS_EXTRA_WORKRATE) {
// int backlog = nodeFinder.getBacklogQueueLength(questionedNode);
// if (backlog>0) {
// Level l = backlog>1000 ? Level.INFO : backlog>10 ? Level.FINE : Level.FINER;
// if (LOG.isLoggable(l)) {
// LOG.log( l, MessageFormat.format(
// "policy "+getDataProvider().getAbbr()+" detected queue at node "+questionedNode+", " +
// "inflating workrate {0,number,#.##} + "+backlog, questionedNodeTotalWorkrate) );
// }
// questionedNodeTotalWorkrate += backlog;
// }
// }
Double highThreshold = model.getHighThreshold(node);
if (highThreshold == -1) {
// node presumably has been removed; TODO log
return false;
}
while (nodeWorkrate > highThreshold && migrationCount < getMaxMigrationsPerBalancingNode()) {
iterationCount++;
if (LOG.isDebugEnabled()) {
LOG.debug(MessageFormat.format(
"policy "+getDataProvider().getName()+" considering balancing hot node "+node+" " +
"(workrate {0,number,#.##}); iteration "+iterationCount, nodeWorkrate) );
}
// move from hot node, to coldest
NodeType coldNode = helper.findColdestContainer(nodesChecked);
if (coldNode == null) {
if (LOG.isDebugEnabled()) {
LOG.debug( MessageFormat.format(
"policy "+getDataProvider().getName()+" not balancing hot node "+node+" " +
"(workrate {0,number,#.##}); no coldest node available", nodeWorkrate) );
}
break;
}
if (coldNode.equals(node)) {
if (LOG.isDebugEnabled()) {
LOG.debug( MessageFormat.format(
"policy "+getDataProvider().getName()+" not balancing hot node "+node+" " +
"(workrate {0,number,#.##}); it is also the coldest modifiable node", nodeWorkrate) );
}
break;
}
double coldNodeWorkrate = getDataProvider().getTotalWorkrate(coldNode);
boolean emergencyLoadBalancing = coldNodeWorkrate < nodeWorkrate*2/3;
double coldNodeHighThreshold = model.getHighThreshold(coldNode);
if (coldNodeWorkrate >= coldNodeHighThreshold && !emergencyLoadBalancing) {
//don't balance if all nodes are approx equally hot (and very hot)
//for now, stop warning if it is a recurring theme!
// Level level = loggedColdestTooHigh ? Level.FINER : Level.INFO;
// LOG.log(level, MessageFormat.format(
// "policy "+getDataProvider().getAbbr()+" not balancing hot node "+questionedNode+" " +
// "(workrate {0,number,#.##}); coldest node "+coldNode+" has workrate {1,number,#.##} also too high"+
// (loggedColdestTooHigh ? "" : " (future cases will be logged at finer)"),
// questionedNodeTotalWorkrate, coldNodeWorkrate) );
// loggedColdestTooHigh = true;
break;
}
double poolLowWatermark = Double.MAX_VALUE; // TODO
if (gonnaGrow && (coldNodeWorkrate >= poolLowWatermark && !emergencyLoadBalancing)) {
//if we're growing the pool, refuse to balance unless the cold node is indeed very cold, or hot node very hot
//for now, stop warning if it is a recurring theme!
// Level level = loggedColdestTooHigh ? Level.FINER : Level.INFO;
// LOG.log(level, MessageFormat.format(
// "policy "+getDataProvider().getAbbr()+" not balancing hot node "+questionedNode+" " +
// "(workrate {0,number,#.##}); coldest node "+coldNode+" has workrate {1,number,#.##} also too high to accept while pool is growing" +
// (loggedColdestTooHigh ? "" : " (future cases will be logged at finer)"),
// questionedNodeTotalWorkrate, coldNodeWorkrate) );
// loggedColdestTooHigh = true;
break;
}
String questionedNodeName = getDataProvider().getName(node);
String coldNodeName = getDataProvider().getName(coldNode);
Location coldNodeLocation = getDataProvider().getLocation(coldNode);
if (LOG.isDebugEnabled()) {
LOG.debug( MessageFormat.format(
"policy "+getDataProvider().getName()+" balancing hot node "+questionedNodeName+" " +
"("+node+", workrate {0,number,#.##}), " +
"considering target "+coldNodeName+" ("+coldNode+", workrate {1,number,#.##})",
nodeWorkrate, coldNodeWorkrate) );
}
double idealSizeToMove = (nodeWorkrate - coldNodeWorkrate) / 2;
//if the 'ideal' amount to move would cause cold to be too hot, then reduce ideal amount
if (idealSizeToMove + coldNodeWorkrate > coldNodeHighThreshold)
idealSizeToMove = coldNodeHighThreshold - coldNodeWorkrate;
double maxSizeToMoveIdeally = Math.min(
nodeWorkrate/2 + 0.00001,
//permit it to exceed node high if there is no alternative (this is 'max' not 'ideal'),
//so long as it still gives a significant benefit
// getConfiguration().nodeHighWaterMark - coldNodeWorkrate,
(nodeWorkrate - coldNodeWorkrate)*0.9);
double maxSizeToMoveIfNoSmallButLarger = nodeWorkrate*3/4;
Map<ItemType, Double> questionedNodeItems = getDataProvider().getItemWorkrates(node);
if (questionedNodeItems == null) {
if (LOG.isDebugEnabled())
LOG.debug(MessageFormat.format(
"policy "+getDataProvider().getName()+" balancing hot node "+questionedNodeName+" " +
"("+node+", workrate {0,number,#.##}), abandoned; " +
"item report for " + questionedNodeName + " unavailable",
nodeWorkrate));
break;
}
ItemType itemToMove = findBestItemToMove(questionedNodeItems, idealSizeToMove, maxSizeToMoveIdeally,
maxSizeToMoveIfNoSmallButLarger, itemsMoved, coldNodeLocation);
if (itemToMove == null) {
if (LOG.isDebugEnabled())
LOG.debug(MessageFormat.format(
"policy "+getDataProvider().getName()+" balancing hot node "+questionedNodeName+" " +
"("+node+", workrate {0,number,#.##}), ending; " +
"no suitable segment found " +
"(ideal transition item size {1,number,#.##}, max {2,number,#.##}, " +
"moving to coldest node "+coldNodeName+" ("+coldNode+", workrate {3,number,#.##}); available items: {4}",
nodeWorkrate, idealSizeToMove, maxSizeToMoveIdeally, coldNodeWorkrate, questionedNodeItems) );
break;
}
itemsMoved.add(itemToMove);
double itemWorkrate = questionedNodeItems.get(itemToMove);
// if (LOG.isLoggable(Level.FINE))
// LOG.fine( MessageFormat.format(
// "policy "+getDataProvider().getAbbr()+" balancing hot node "+questionedNodeName+" " +
// "(workrate {0,number,#.##}, too high), transitioning " + itemToMove +
// " to "+coldNodeName+" (workrate {1,number,#.##}, now += {2,number,#.##})",
// questionedNodeTotalWorkrate, coldNodeWorkrate, segmentRate) );
nodeWorkrate -= itemWorkrate;
coldNodeWorkrate += itemWorkrate;
moveItem(itemToMove, node, coldNode);
++migrationCount;
}
if (LOG.isDebugEnabled()) {
if (iterationCount == 0) {
if (LOG.isTraceEnabled())
LOG.trace( MessageFormat.format(
"policy "+getDataProvider().getName()+" balancing if hot finished at node "+node+"; " +
"workrate {0,number,#.##} not hot",
originalNodeWorkrate) );
}
else if (itemsMoved.isEmpty()) {
if (LOG.isTraceEnabled())
LOG.trace( MessageFormat.format(
"policy "+getDataProvider().getName()+" balancing finished at hot node "+node+" " +
"(workrate {0,number,#.##}); no way to improve it",
originalNodeWorkrate) );
} else {
LOG.debug( MessageFormat.format(
"policy "+getDataProvider().getName()+" balancing finished at hot node "+node+"; " +
"workrate from {0,number,#.##} to {1,number,#.##} (report now says {2,number,#.##}) " +
"by moving off {3}",
originalNodeWorkrate,
nodeWorkrate,
getDataProvider().getTotalWorkrate(node),
itemsMoved
) );
}
}
return !itemsMoved.isEmpty();
}