diff --git a/core/src/main/java/hudson/slaves/NodeProvisioner.java b/core/src/main/java/hudson/slaves/NodeProvisioner.java index c63a16180a..23ef37dca9 100644 --- a/core/src/main/java/hudson/slaves/NodeProvisioner.java +++ b/core/src/main/java/hudson/slaves/NodeProvisioner.java @@ -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 pendingLaunches = new ArrayList(); 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 strategies = Jenkins.getInstance().getExtensionList(Strategy.class); + for (Strategy strategy : strategies.isEmpty() + ? Arrays.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 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 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 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 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. - * - *

- * 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.) - * - *

- * 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. - * - *

- * 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 — the chance gets better and better as the number of current total - * capacity increases. - * - *

- * Therefore, we compute the threshold margin as follows: - * - * 

-     *   M(t) = M* + (M0 - M*) alpha ^ t
-     *
- * - * ... where: - * - *
    - *
  • M* is the ultimate margin value that M(t) converges to with t->inf, - *
  • M0 is the value of M(0), the initial value. - *
  • alpha is the decay factor in (0,1). M(t) converges to M* faster if alpha is smaller. - *
- */ - 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. + *

+ *

+ * 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.) + *

+ *

+ * 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. + *

+ *

+ * 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 — the chance gets better and better as the number of current total + * capacity increases. + *

+ *

+ * Therefore, we compute the threshold margin as follows: + *

+ * 

+         *   M(t) = M* + (M0 - M*) alpha ^ t
+         *
+ *

+ * ... where: + *

+ *

    + *
  • M* is the ultimate margin value that M(t) converges to with t->inf, + *
  • M0 is the value of M(0), the initial value. + *
  • alpha is the decay factor in (0,1). M(t) converges to M* faster if alpha is smaller. + *
+ */ + 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; + } } /**