Make node provisioning strategy plugable

This commit is contained in:
Stephen Connolly committed 2014-10-17 12:21:44 +01:00
1 parent 7acbba330c
commit 9fff4118d4
1 file changed
+378 -92
@@ -23,14 +23,18 @@
*/
package hudson.slaves;
import hudson.ExtensionPoint;
import hudson.model.*;
import jenkins.model.Jenkins;
import static hudson.model.LoadStatistics.DECAY;
import hudson.model.MultiStageTimeSeries.TimeScale;
import hudson.Extension;
import net.jcip.annotations.GuardedBy;
import javax.annotation.Nonnull;
import java.awt.Color;
import java.util.Arrays;
import java.util.concurrent.Future;
import java.util.concurrent.ExecutionException;
import java.util.List;
@@ -116,6 +120,7 @@ public class NodeProvisioner {
*/
private final Label label;
@GuardedBy("this")
private List<PlannedNode> pendingLaunches = new ArrayList<PlannedNode>();
private transient volatile long lastSuggestedReview;
@@ -182,7 +187,9 @@ public class NodeProvisioner {
cl.onComplete(f,node);
hudson.addNode(node);
LOGGER.info(f.displayName+" provisioning successfully completed. We have now "+hudson.getComputers().length+" computer(s)");
LOGGER.log(Level.INFO,
"{0} provisioning successfully completed. We have now {1,number,integer} computer(s)",
new Object[]{f.displayName, hudson.getComputers().length});
} catch (InterruptedException e) {
throw new AssertionError(e); // since we confirmed that the future is already done
} catch (ExecutionException e) {
@@ -204,6 +211,260 @@ public class NodeProvisioner {
float plannedCapacity = plannedCapacitySnapshot;
plannedCapacitiesEMA.update(plannedCapacity);
int idleSnapshot = stat.computeIdleExecutors();
int queueLengthSnapshot = stat.computeQueueLength();
if (queueLengthSnapshot <= idleSnapshot) {
LOGGER.log(Level.FINE,
"Queue length {0} is less than the idle capacity {1}. No provisioning strategy required",
new Object[]{queueLengthSnapshot, idleSnapshot});
} else {
StrategyState state =
new StrategyState(queueLengthSnapshot, label, idleSnapshot, stat.computeTotalExecutors(),
plannedCapacitySnapshot);
List<Strategy> strategies = Jenkins.getInstance().getExtensionList(Strategy.class);
for (Strategy strategy : strategies.isEmpty()
? Arrays.<Strategy>asList(new StandardStrategyImpl())
: strategies) {
LOGGER.log(Level.FINER, "Consulting {0} provisioning strategy with state {1}",
new Object[]{strategy, state});
if (StrategyDecision.PROVISIONING_COMPLETED == strategy.apply(state)) {
LOGGER.log(Level.FINER, "Provisioning strategy {0} declared provisioning complete",
strategy);
break;
}
}
}
}
/**
* Represents the decision taken by an individual {@link hudson.slaves.NodeProvisioner.Strategy}.
* @since 1.586
*/
public static enum StrategyDecision {
/**
* This decision is the default decision and indicates that the {@link hudson.slaves.NodeProvisioner.Strategy}
* either could not provision sufficient resources or did not take any action. Any remaining strategies
* will be able to contribute to the ultimate decision.
*/
CONSULT_REMAINING_STRATEGIES,
/**
* This decision indicates that the {@link hudson.slaves.NodeProvisioner.Strategy} has taken sufficient
* action so as to ensure that the required resources are available, and therefore there is no requirement
* to consult the remaining strategies. Only return this decision when you are certain that
*/
PROVISIONING_COMPLETED
}
/**
* Extension point for node provisioning strategies.
* @since 1.586
*/
public static abstract class Strategy implements ExtensionPoint {
/**
* Applies the strategy to the specified state. Any provisioning activities should be recorded by calling
* {@link NodeProvisioner.StrategyState#recordPendingLaunches(java.util.Collection)}
* @param state the current state.
* @return the decision.
*/
@Nonnull
public abstract StrategyDecision apply(@Nonnull StrategyState state);
}
/**
* Parameter object for {@link NodeProvisioner.Strategy#apply(NodeProvisioner.StrategyState)}
* @since 1.586
*/
public final class StrategyState {
/**
* The label under consideration.
*/
private final Label label;
/**
* The number of items in the queue requiring this {@link #label}.
*/
private final int queueLengthSnapshot;
/**
* The planned capacity for this {@link #label}.
*/
private final int plannedCapacitySnapshot;
/**
* The number of idle executors for this {@link #label}
*/
private final int idleSnapshot;
/**
* The total number of executors for this {@link #label}
*/
private final int totalSnapshot;
/**
* The additional planned capacity for this {@link #label} and provisioned by previous strategies during the
* current updating of the {@link NodeProvisioner}.
*/
@GuardedBy("this")
private int additionalPlannedCapacity;
/**
* Should only be instantiated by {@link NodeProvisioner#update()}
* @param queueLengthSnapshot the queue length.
* @param label the label.
* @param idleSnapshot the idle executor count.
* @param totalSnapshot the totoal executor count.
* @param plannedCapacitySnapshot the planned executor count.
*/
private StrategyState(int queueLengthSnapshot, Label label, int idleSnapshot, int totalSnapshot,
int plannedCapacitySnapshot) {
this.queueLengthSnapshot = queueLengthSnapshot;
this.label = label;
this.idleSnapshot = idleSnapshot;
this.totalSnapshot = totalSnapshot;
this.plannedCapacitySnapshot = plannedCapacitySnapshot;
}
/**
* The label under consideration.
*/
public Label getLabel() {
return label;
}
/**
* The number of items in the queue requiring this {@link #getLabel()}.
*/
public int getQueueLengthSnapshot() {
return queueLengthSnapshot;
}
/**
* The planned capacity for this {@link #getLabel()}.
*/
public int getPlannedCapacitySnapshot() {
return plannedCapacitySnapshot;
}
/**
* The number of idle executors for this {@link #getLabel()}
*/
public int getIdleSnapshot() {
return idleSnapshot;
}
/**
* The total number of executors for this {@link #getLabel()}
*/
public int getTotalSnapshot() {
return totalSnapshot;
}
/**
* The additional planned capacity for this {@link #getLabel()} and provisioned by previous strategies during
* the current updating of the {@link NodeProvisioner}.
*/
public synchronized int getAdditionalPlannedCapacity() {
return additionalPlannedCapacity;
}
/**
* The time series average number of items in the queue requiring this {@link #getLabel()}.
*/
public float getQueueLengthLatest() {
return stat.queueLength.getLatest(TIME_SCALE);
}
/**
* The time series average planned capacity for this {@link #getLabel()}.
*/
public float getPlannedCapacityLatest() {
return plannedCapacitiesEMA.getLatest(TIME_SCALE);
}
/**
* The time series average number of idle executors for this {@link #getLabel()}
*/
public float getIdleLatest() {
return stat.getLatestIdleExecutors(TIME_SCALE);
}
/**
* The time series average total number of executors for this {@link #getLabel()}
*/
public float getTotalLatest() {
return stat.totalExecutors.getLatest(TIME_SCALE);
}
/**
* If a {@link hudson.slaves.NodeProvisioner.Strategy} takes some provisioning action, it should record
* and {@link hudson.slaves.NodeProvisioner.PlannedNode} instances by calling this method.
*
* @param plannedNodes the {@link hudson.slaves.NodeProvisioner.PlannedNode} instances.
*/
public void recordPendingLaunches(PlannedNode... plannedNodes) {
recordPendingLaunches(Arrays.asList(plannedNodes));
}
/**
* If a {@link hudson.slaves.NodeProvisioner.Strategy} takes some provisioning action, it should record
* and {@link hudson.slaves.NodeProvisioner.PlannedNode} instances by calling this method.
*
* @param plannedNodes the {@link hudson.slaves.NodeProvisioner.PlannedNode} instances.
*/
public void recordPendingLaunches(Collection<PlannedNode> plannedNodes) {
int additionalPlannedCapacity = 0;
for (PlannedNode f : plannedNodes) {
if (f.future.isDone()) {
// if done we should use the actual delivered capacity
try {
Node node = f.future.get();
if (node != null) {
additionalPlannedCapacity += node.getNumExecutors();
}
} catch (InterruptedException e) {
// ignore, this will be caught by others later
} catch (ExecutionException e) {
// ignore, this will be caught by others later
}
} else {
additionalPlannedCapacity += f.numExecutors;
}
}
synchronized (NodeProvisioner.this) {
pendingLaunches.addAll(plannedNodes);
}
if (additionalPlannedCapacity > 0) {
synchronized (this) {
this.additionalPlannedCapacity += additionalPlannedCapacity;
}
}
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("StrategyState{");
sb.append("label=").append(label);
sb.append(", queueLengthSnapshot=").append(queueLengthSnapshot);
sb.append(", plannedCapacitySnapshot=").append(plannedCapacitySnapshot);
sb.append(", idleSnapshot=").append(idleSnapshot);
sb.append(", totalSnapshot=").append(totalSnapshot);
sb.append(", additionalPlannedCapacity=").append(additionalPlannedCapacity);
sb.append('}');
return sb.toString();
}
}
/**
* The default strategy.
*
* @since 1.586
*/
@Extension
public static class StandardStrategyImpl extends Strategy {
/** {@inheritDoc} */
@Nonnull
@Override
public StrategyDecision apply(@Nonnull StrategyState state) {
/*
Here we determine how many additional slaves we need to keep up with the load (if at all),
which involves a simple math.
@@ -236,107 +497,132 @@ public class NodeProvisioner {
estimate won't create a starvation.
*/
int idleSnapshot = stat.computeIdleExecutors();
int totalSnapshot = stat.computeTotalExecutors();
boolean needSomeWhenNoneAtAll = (idleSnapshot==0) && (totalSnapshot + plannedCapacitySnapshot == 0) && (stat.computeQueueLength() > 0);
float idle = Math.max(stat.getLatestIdleExecutors(TIME_SCALE), idleSnapshot);
if(idle<MARGIN || needSomeWhenNoneAtAll) {
// make sure the system is fully utilized before attempting any new launch.
boolean needSomeWhenNoneAtAll = (state.getIdleSnapshot() == 0)
&& (state.getTotalSnapshot() + state.getPlannedCapacitySnapshot() + state.getAdditionalPlannedCapacity() == 0)
&& (state.getQueueLengthSnapshot() > 0);
float idle = Math.max(state.getIdleLatest(), state.getIdleSnapshot());
if (idle < MARGIN || needSomeWhenNoneAtAll) {
// make sure the system is fully utilized before attempting any new launch.
// this is the amount of work left to be done
float qlen = Math.min(stat.queueLength.getLatest(TIME_SCALE), stat.computeQueueLength());
// this is the amount of work left to be done
float qlen = Math.min(state.getQueueLengthLatest(), state.getQueueLengthSnapshot());
// ... and this is the additional executors we've already provisioned.
plannedCapacity = Math.max(plannedCapacitiesEMA.getLatest(TIME_SCALE),plannedCapacity);
// ... and this is the additional executors we've already provisioned.
float plannedCapacity = Math.max(state.getPlannedCapacityLatest(), state.getPlannedCapacitySnapshot())
+ state.getAdditionalPlannedCapacity();
float excessWorkload = qlen - plannedCapacity;
if (needSomeWhenNoneAtAll && excessWorkload < 1) {
// in this specific exceptional case we should just provision right now
// the exponential smoothing will delay the build unnecessarily
excessWorkload = 1;
}
float m = calcThresholdMargin(totalSnapshot);
if(excessWorkload>1-m) {// and there's more work to do...
LOGGER.fine("Excess workload "+excessWorkload+" detected. (planned capacity="+plannedCapacity+",Qlen="+qlen+",idle="+idle+"&"+idleSnapshot+",total="+totalSnapshot+"m,="+m+")");
float excessWorkload = qlen - plannedCapacity;
if (needSomeWhenNoneAtAll && excessWorkload < 1) {
// in this specific exceptional case we should just provision right now
// the exponential smoothing will delay the build unnecessarily
excessWorkload = 1;
}
float m = calcThresholdMargin(state.getTotalSnapshot());
if (excessWorkload > 1 - m) {// and there's more work to do...
LOGGER.log(Level.FINE, "Excess workload {0,number,#.###} detected. "
+ "(planned capacity={1,number,#.###},"
+ "Qlen={2,number,#.###},idle={3,number,#.###}&{4,number,integer},"
+ "total={5,number,integer},m={6,number,#.###})",
new Object[]{
excessWorkload, plannedCapacity, qlen, idle, state.getIdleSnapshot(),
state.getTotalSnapshot(), m
});
CLOUD:
for( Cloud c : hudson.clouds ) {
if(excessWorkload<0) break; // enough slaves allocated
// Make sure this cloud actually can provision for this label.
if (c.canProvision(label)) {
// provisioning a new node should be conservative --- for example if exceeWorkload is 1.4,
// we don't want to allocate two nodes but just one.
// OTOH, because of the exponential decay, even when we need one slave, excess workload is always
// something like 0.95, in which case we want to allocate one node.
// so the threshold here is 1-MARGIN, and hence floor(excessWorkload+MARGIN) is needed to handle this.
int workloadToProvision = (int) Math.round(Math.floor(excessWorkload + m));
for (CloudProvisioningListener cl : CloudProvisioningListener.all())
// consider displaying reasons in a future cloud ux
if (cl.canProvision(c,label,workloadToProvision) != null)
break CLOUD;
Collection<PlannedNode> additionalCapacities = c.provision(label, workloadToProvision);
for (CloudProvisioningListener cl : CloudProvisioningListener.all())
cl.onStarted(c, label, additionalCapacities);
for (PlannedNode ac : additionalCapacities) {
excessWorkload -= ac.numExecutors;
LOGGER.info("Started provisioning "+ac.displayName+" from "+c.name+" with "+ac.numExecutors+" executors. Remaining excess workload:"+excessWorkload);
CLOUD:
for (Cloud c : Jenkins.getInstance().clouds) {
if (excessWorkload < 0) {
break; // enough slaves allocated
}
// Make sure this cloud actually can provision for this label.
if (c.canProvision(state.getLabel())) {
// provisioning a new node should be conservative --- for example if excessWorkload is 1.4,
// we don't want to allocate two nodes but just one.
// OTOH, because of the exponential decay, even when we need one slave,
// excess workload is always
// something like 0.95, in which case we want to allocate one node.
// so the threshold here is 1-MARGIN, and hence floor(excessWorkload+MARGIN) is needed to
// handle this.
int workloadToProvision = (int) Math.round(Math.floor(excessWorkload + m));
for (CloudProvisioningListener cl : CloudProvisioningListener.all())
// consider displaying reasons in a future cloud ux
{
if (cl.canProvision(c, state.getLabel(), workloadToProvision) != null) {
break CLOUD;
}
}
Collection<PlannedNode> additionalCapacities =
c.provision(state.getLabel(), workloadToProvision);
for (CloudProvisioningListener cl : CloudProvisioningListener.all()) {
cl.onStarted(c, state.getLabel(), additionalCapacities);
}
for (PlannedNode ac : additionalCapacities) {
excessWorkload -= ac.numExecutors;
LOGGER.log(Level.INFO, "Started provisioning {0} from {1} with {2,number,integer} "
+ "executors. Remaining excess workload: {3,number,#.###}",
new Object[]{ac.displayName, c.name, ac.numExecutors, excessWorkload});
}
state.recordPendingLaunches(additionalCapacities);
}
pendingLaunches.addAll(additionalCapacities);
}
// we took action, only pass on to other strategies if our action was insufficient
return excessWorkload > 1 - m ? StrategyDecision.CONSULT_REMAINING_STRATEGIES : StrategyDecision.PROVISIONING_COMPLETED;
}
}
// if we reach here then the standard strategy obviously decided to do nothing, so let any other strategies
// take their considerations.
return StrategyDecision.CONSULT_REMAINING_STRATEGIES;
}
}
/**
* Computes the threshold for triggering an allocation.
*
* <p>
* Because the excessive workload value is EMA, even when the snapshot value of the excessive
* workload is 1, the value never really gets to 1. So we need to introduce a notion of the margin M,
* where we provision a new node if the EMA of the excessive workload goes beyond 1-M (where M is a small value
* in the (0,1) range.)
*
* <p>
* M effectively controls how long Hudson waits until allocating a new node, in the face of workload.
* This delay is justified for absorbing temporary ups and downs, and can be interpreted as Hudson
* holding off provisioning in the hope that one of the existing nodes will become available.
*
* <p>
* M can be a constant value, but there's a benefit in adjusting M based on the total current capacity,
* based on the above justification; that is, if there's no existing capacity at all, holding off
* an allocation doesn't make much sense, as there won't be any executors available no matter how long we wait.
* On the other hand, if we have a large number of existing executors, chances are good that some
* of them become available &mdash; the chance gets better and better as the number of current total
* capacity increases.
*
* <p>
* Therefore, we compute the threshold margin as follows:
*
* <pre>
* M(t) = M* + (M0 - M*) alpha ^ t
* </pre>
*
* ... where:
*
* <ul>
* <li>M* is the ultimate margin value that M(t) converges to with t->inf,
* <li>M0 is the value of M(0), the initial value.
* <li>alpha is the decay factor in (0,1). M(t) converges to M* faster if alpha is smaller.
* </ul>
*/
private float calcThresholdMargin(int totalSnapshot) {
float f = (float) (MARGIN + (MARGIN0 - MARGIN) * Math.pow(MARGIN_DECAY, totalSnapshot));
// defensively ensure that the threshold margin is in (0,1)
f = Math.max(f,0);
f = Math.min(f,1);
return f;
/**
* Computes the threshold for triggering an allocation.
* <p/>
* <p/>
* Because the excessive workload value is EMA, even when the snapshot value of the excessive
* workload is 1, the value never really gets to 1. So we need to introduce a notion of the margin M,
* where we provision a new node if the EMA of the excessive workload goes beyond 1-M (where M is a small value
* in the (0,1) range.)
* <p/>
* <p/>
* M effectively controls how long Hudson waits until allocating a new node, in the face of workload.
* This delay is justified for absorbing temporary ups and downs, and can be interpreted as Hudson
* holding off provisioning in the hope that one of the existing nodes will become available.
* <p/>
* <p/>
* M can be a constant value, but there's a benefit in adjusting M based on the total current capacity,
* based on the above justification; that is, if there's no existing capacity at all, holding off
* an allocation doesn't make much sense, as there won't be any executors available no matter how long we wait.
* On the other hand, if we have a large number of existing executors, chances are good that some
* of them become available &mdash; the chance gets better and better as the number of current total
* capacity increases.
* <p/>
* <p/>
* Therefore, we compute the threshold margin as follows:
* <p/>
* <pre>
* M(t) = M* + (M0 - M*) alpha ^ t
* </pre>
* <p/>
* ... where:
* <p/>
* <ul>
* <li>M* is the ultimate margin value that M(t) converges to with t->inf,
* <li>M0 is the value of M(0), the initial value.
* <li>alpha is the decay factor in (0,1). M(t) converges to M* faster if alpha is smaller.
* </ul>
*/
private float calcThresholdMargin(int totalSnapshot) {
float f = (float) (MARGIN + (MARGIN0 - MARGIN) * Math.pow(MARGIN_DECAY, totalSnapshot));
// defensively ensure that the threshold margin is in (0,1)
f = Math.max(f, 0);
f = Math.min(f, 1);
return f;
}
}
/**