/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.brooklyn.policy.autoscaling;

import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.brooklyn.util.JavaGroovyEquivalents.groovyTruth;

import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.brooklyn.api.catalog.Catalog;
import org.apache.brooklyn.api.catalog.CatalogConfig;
import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.entity.EntityLocal;
import org.apache.brooklyn.api.mgmt.Task;
import org.apache.brooklyn.api.objs.HighlightTuple;
import org.apache.brooklyn.api.policy.PolicySpec;
import org.apache.brooklyn.api.sensor.AttributeSensor;
import org.apache.brooklyn.api.sensor.Sensor;
import org.apache.brooklyn.api.sensor.SensorEvent;
import org.apache.brooklyn.api.sensor.SensorEventListener;
import org.apache.brooklyn.config.ConfigKey;
import org.apache.brooklyn.core.config.BasicConfigKey;
import org.apache.brooklyn.core.entity.Entities;
import org.apache.brooklyn.core.entity.trait.Resizable;
import org.apache.brooklyn.core.entity.trait.Startable;
import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
import org.apache.brooklyn.core.policy.AbstractPolicy;
import org.apache.brooklyn.core.sensor.BasicNotificationSensor;
import org.apache.brooklyn.entity.group.DynamicCluster;
import org.apache.brooklyn.policy.autoscaling.SizeHistory.WindowSummary;
import org.apache.brooklyn.policy.loadbalancing.LoadBalancingPolicy;
import org.apache.brooklyn.util.collections.MutableMap;
import org.apache.brooklyn.util.core.flags.SetFromFlag;
import org.apache.brooklyn.util.core.flags.TypeCoercions;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.text.Strings;
import org.apache.brooklyn.util.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.reflect.TypeToken;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import groovy.lang.Closure;


/**
 * Policy that is attached to a {@link Resizable} entity and dynamically adjusts its size in response to
 * emitted {@code POOL_COLD} and {@code POOL_HOT} events. Alternatively, the policy can be configured to
 * keep a given metric within a required range.
 * <p>
 * This policy does not itself determine whether the pool is hot or cold, but instead relies on these
 * events being emitted by the monitored entity itself, or by another policy that is attached to it; see, 
 * for example, {@link LoadBalancingPolicy}.)
 */
@SuppressWarnings({"rawtypes", "unchecked"})
@Catalog(name="Auto-scaler", description="Policy that is attached to a Resizable entity and dynamically "
        + "adjusts its size in response to either keep a metric within a given range, or in response to "
        + "POOL_COLD and POOL_HOT events")
public class AutoScalerPolicy extends AbstractPolicy {
    
    private static final Logger LOG = LoggerFactory.getLogger(AutoScalerPolicy.class);

    public static Builder builder() {
        return new Builder();
    }
    
    public static class Builder {
        private String id;
        private String name;
        private AttributeSensor<? extends Number> metric;
        private Entity entityWithMetric;
        private Number metricUpperBound;
        private Number metricLowerBound;
        private int minPoolSize = 1;
        private int maxPoolSize = Integer.MAX_VALUE;
        private Integer resizeDownIterationIncrement;
        private Integer resizeDownIterationMax;
        private Integer resizeUpIterationIncrement;
        private Integer resizeUpIterationMax;
        private Duration minPeriodBetweenExecs;
        private Duration resizeUpStabilizationDelay;
        private Duration resizeDownStabilizationDelay;
        private ResizeOperator resizeOperator;
        private Function<Entity,Integer> currentSizeOperator;
        private BasicNotificationSensor<?> poolHotSensor;
        private BasicNotificationSensor<?> poolColdSensor;
        private BasicNotificationSensor<?> poolOkSensor;
        private BasicNotificationSensor<? super MaxPoolSizeReachedEvent> maxSizeReachedSensor;
        private Duration maxReachedNotificationDelay;
        
        public Builder id(String val) {
            this.id = val; return this;
        }
        public Builder name(String val) {
            this.name = val; return this;
        }
        public Builder metric(AttributeSensor<? extends Number> val) {
            this.metric = val; return this;
        }
        public Builder entityWithMetric(Entity val) {
            this.entityWithMetric = val; return this;
        }
        public Builder metricLowerBound(Number val) {
            this.metricLowerBound = val; return this;
        }
        public Builder metricUpperBound(Number val) {
            this.metricUpperBound = val; return this;
        }
        public Builder metricRange(Number min, Number max) {
            metricLowerBound = checkNotNull(min);
            metricUpperBound = checkNotNull(max);
            return this;
        }
        public Builder minPoolSize(int val) {
            this.minPoolSize = val; return this;
        }
        public Builder maxPoolSize(int val) {
            this.maxPoolSize = val; return this;
        }
        public Builder sizeRange(int min, int max) {
            minPoolSize = min;
            maxPoolSize = max;
            return this;
        }
        
        public Builder resizeUpIterationIncrement(Integer val) {
            this.resizeUpIterationIncrement = val; return this;
        }
        public Builder resizeUpIterationMax(Integer val) {
            this.resizeUpIterationMax = val; return this;
        }
        public Builder resizeDownIterationIncrement(Integer val) {
            this.resizeDownIterationIncrement = val; return this;
        }
        public Builder resizeDownIterationMax(Integer val) {
            this.resizeDownIterationMax = val; return this;
        }

        public Builder minPeriodBetweenExecs(Duration val) {
            this.minPeriodBetweenExecs = val; return this;
        }
        public Builder resizeUpStabilizationDelay(Duration val) {
            this.resizeUpStabilizationDelay = val; return this;
        }
        public Builder resizeDownStabilizationDelay(Duration val) {
            this.resizeDownStabilizationDelay = val; return this;
        }
        public Builder resizeOperator(ResizeOperator val) {
            this.resizeOperator = val; return this;
        }
        public Builder currentSizeOperator(Function<Entity, Integer> val) {
            this.currentSizeOperator = val; return this;
        }
        public Builder poolHotSensor(BasicNotificationSensor<?> val) {
            this.poolHotSensor = val; return this;
        }
        public Builder poolColdSensor(BasicNotificationSensor<?> val) {
            this.poolColdSensor = val; return this;
        }
        public Builder poolOkSensor(BasicNotificationSensor<?> val) {
            this.poolOkSensor = val; return this;
        }
        public Builder maxSizeReachedSensor(BasicNotificationSensor<? super MaxPoolSizeReachedEvent> val) {
            this.maxSizeReachedSensor = val; return this;
        }
        public Builder maxReachedNotificationDelay(Duration val) {
            this.maxReachedNotificationDelay = val; return this;
        }
        /**
         * @deprecated since 0.12.0; use {@link #buildSpec()}, or use {@link PolicySpec} directly
         */
        @Deprecated
        public AutoScalerPolicy build() {
            return new AutoScalerPolicy(toFlags());
        }
        public PolicySpec<AutoScalerPolicy> buildSpec() {
            return PolicySpec.create(AutoScalerPolicy.class)
                    .configure(toFlags());
        }
        private Map<String,?> toFlags() {
            return MutableMap.<String,Object>builder()
                    .putIfNotNull("id", id)
                    .putIfNotNull("name", name)
                    .putIfNotNull("metric", metric)
                    .putIfNotNull("entityWithMetric", entityWithMetric)
                    .putIfNotNull("metricUpperBound", metricUpperBound)
                    .putIfNotNull("metricLowerBound", metricLowerBound)
                    .putIfNotNull("minPoolSize", minPoolSize)
                    .putIfNotNull("maxPoolSize", maxPoolSize)
                    .putIfNotNull("resizeUpIterationMax", resizeUpIterationMax)
                    .putIfNotNull("resizeUpIterationIncrement", resizeUpIterationIncrement)
                    .putIfNotNull("resizeDownIterationMax", resizeDownIterationMax)
                    .putIfNotNull("resizeDownIterationIncrement", resizeDownIterationIncrement)
                    .putIfNotNull("minPeriodBetweenExecs", minPeriodBetweenExecs)
                    .putIfNotNull("resizeUpStabilizationDelay", resizeUpStabilizationDelay)
                    .putIfNotNull("resizeDownStabilizationDelay", resizeDownStabilizationDelay)
                    .putIfNotNull("resizeOperator", resizeOperator)
                    .putIfNotNull("currentSizeOperator", currentSizeOperator)
                    .putIfNotNull("poolHotSensor", poolHotSensor)
                    .putIfNotNull("poolColdSensor", poolColdSensor)
                    .putIfNotNull("poolOkSensor", poolOkSensor)
                    .putIfNotNull("maxSizeReachedSensor", maxSizeReachedSensor)
                    .putIfNotNull("maxReachedNotificationDelay", maxReachedNotificationDelay)
                    .build();
        }
    }
    
    // TODO Is there a nicer pattern for registering such type-coercions? 
    // Can't put it in the ResizeOperator interface, nor in core TypeCoercions class because interface is defined in policy/.
    static {
        TypeCoercions.registerAdapter(Closure.class, ResizeOperator.class, new Function<Closure,ResizeOperator>() {
            @Override
            public ResizeOperator apply(final Closure closure) {
                LOG.warn("Use of groovy.lang.Closure is deprecated in AutoScalerPolicy type-coercion Closure -> ResizeOperator");
                return new ResizeOperator() {
                    @Override public Integer resize(Entity entity, Integer input) {
                        return (Integer) closure.call(entity, input);
                    }
                };
            }
        });
    }
    
    // Pool workrate notifications.
    public static BasicNotificationSensor<Map> DEFAULT_POOL_HOT_SENSOR = new BasicNotificationSensor<Map>(
        Map.class, "resizablepool.hot", "Pool is over-utilized; it has insufficient resource for current workload");
    public static BasicNotificationSensor<Map> DEFAULT_POOL_COLD_SENSOR = new BasicNotificationSensor<Map>(
        Map.class, "resizablepool.cold", "Pool is under-utilized; it has too much resource for current workload");
    public static BasicNotificationSensor<Map> DEFAULT_POOL_OK_SENSOR = new BasicNotificationSensor<Map>(
        Map.class, "resizablepool.cold", "Pool utilization is ok; the available resources are fine for the current workload");

    /**
     * A convenience for policies that want to register a {@code builder.maxSizeReachedSensor(sensor)}.
     * Note that this "default" is not set automatically; the default is for no sensor to be used (so
     * no events emitted).
     */
    public static BasicNotificationSensor<MaxPoolSizeReachedEvent> DEFAULT_MAX_SIZE_REACHED_SENSOR = new BasicNotificationSensor<MaxPoolSizeReachedEvent>(
            MaxPoolSizeReachedEvent.class, "resizablepool.maxSizeReached", "Consistently wanted to resize the pool above the max allowed size");

    public static final String POOL_CURRENT_SIZE_KEY = "pool.current.size";
    public static final String POOL_HIGH_THRESHOLD_KEY = "pool.high.threshold";
    public static final String POOL_LOW_THRESHOLD_KEY = "pool.low.threshold";
    public static final String POOL_CURRENT_WORKRATE_KEY = "pool.current.workrate";
    
    @CatalogConfig(label = "Metric")
    @SuppressWarnings("serial")
    @SetFromFlag("metric")
    public static final ConfigKey<AttributeSensor<? extends Number>> METRIC = BasicConfigKey.builder(new TypeToken<AttributeSensor<? extends Number>>() {})
            .name("autoscaler.metric")
            .description("The (numeric) sensor to use for auto-scaling decisions, keeping it within the given bounds")
            .build();

    @SetFromFlag("entityWithMetric")
    public static final ConfigKey<Entity> ENTITY_WITH_METRIC = BasicConfigKey.builder(Entity.class)
            .name("autoscaler.entityWithMetric")
            .description("The Entity with the metric that will be monitored")
            .build();
    
    @CatalogConfig(label = "Metric lower bound")
    @SetFromFlag("metricLowerBound")
    public static final ConfigKey<Number> METRIC_LOWER_BOUND = BasicConfigKey.builder(Number.class)
            .name("autoscaler.metricLowerBound")
            .description("The lower bound of the monitored metric. Below this the policy will resize down")
            .reconfigurable(true)
            .build();
    
    @CatalogConfig(label = "Metric upper bound")
    @SetFromFlag("metricUpperBound")
    public static final ConfigKey<Number> METRIC_UPPER_BOUND = BasicConfigKey.builder(Number.class)
            .name("autoscaler.metricUpperBound")
            .description("The upper bound of the monitored metric. Above this the policy will resize up")
            .reconfigurable(true)
            .build();
    
    @SetFromFlag("resizeUpIterationIncrement")
    public static final ConfigKey<Integer> RESIZE_UP_ITERATION_INCREMENT = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.resizeUpIterationIncrement")
            .description("Batch size for resizing up; the size will be increased by a multiple of this value")
            .defaultValue(1)
            .reconfigurable(true)
            .build();
    
    @SetFromFlag("resizeUpIterationMax")
    public static final ConfigKey<Integer> RESIZE_UP_ITERATION_MAX = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.resizeUpIterationMax")
            .defaultValue(Integer.MAX_VALUE)
            .description("Maximum change to the size on a single iteration when scaling up")
            .reconfigurable(true)
            .build();
    
    @SetFromFlag("resizeDownIterationIncrement")
    public static final ConfigKey<Integer> RESIZE_DOWN_ITERATION_INCREMENT = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.resizeDownIterationIncrement")
            .description("Batch size for resizing down; the size will be decreased by a multiple of this value")
            .defaultValue(1)
            .reconfigurable(true)
            .build();
    
    @SetFromFlag("resizeDownIterationMax")
    public static final ConfigKey<Integer> RESIZE_DOWN_ITERATION_MAX = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.resizeDownIterationMax")
            .defaultValue(Integer.MAX_VALUE)
            .description("Maximum change to the size on a single iteration when scaling down")
            .reconfigurable(true)
            .build();

    @SetFromFlag("minPeriodBetweenExecs")
    public static final ConfigKey<Duration> MIN_PERIOD_BETWEEN_EXECS = BasicConfigKey.builder(Duration.class)
            .name("autoscaler.minPeriodBetweenExecs")
            .description("When re-evaluating the desired size, wait at least this duration before computing")
            .defaultValue(Duration.millis(100))
            .build();
    
    @SetFromFlag("resizeUpStabilizationDelay")
    public static final ConfigKey<Duration> RESIZE_UP_STABILIZATION_DELAY = BasicConfigKey.builder(Duration.class)
            .name("autoscaler.resizeUpStabilizationDelay")
            .description("The required duration of 'sustained load' before scaling up "
                    + "(i.e. the length of time the metric must be above its upper bound before acting)")
            .defaultValue(Duration.ZERO)
            .reconfigurable(true)
            .build();
    
    @SetFromFlag("resizeDownStabilizationDelay")
    public static final ConfigKey<Duration> RESIZE_DOWN_STABILIZATION_DELAY = BasicConfigKey.builder(Duration.class)
            .name("autoscaler.resizeDownStabilizationDelay")
            .description("The required duration of 'sustained low load' before scaling down "
                    + "(i.e. the length of time the metric must be belowe its lower bound before acting)")
            .defaultValue(Duration.ZERO)
            .reconfigurable(true)
            .build();

    @SetFromFlag("minPoolSize")
    public static final ConfigKey<Integer> MIN_POOL_SIZE = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.minPoolSize")
            .description("The minimum acceptable pool size (never scaling down below this size, and automatically scaling up to this min size if required)")
            .defaultValue(1)
            .reconfigurable(true)
            .build();
    
    @SetFromFlag("maxPoolSize")
    public static final ConfigKey<Integer> MAX_POOL_SIZE = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.maxPoolSize")
            .description("The maximum acceptable pool size (never scaling up above this size, and automatically scaling down to this min size if required)")
            .defaultValue(Integer.MAX_VALUE)
            .reconfigurable(true)
            .build();

    public static final ConfigKey<Integer> INSUFFICIENT_CAPACITY_HIGH_WATER_MARK = BasicConfigKey.builder(Integer.class)
            .name("autoscaler.insufficientCapacityHighWaterMark")
            .description("Level at which we either expect, or experienced, 'InsufficientCapacityException', "
                    + "so should not attempt to go above this size. This is set automatically if that exception is hit.")
            .defaultValue(null)
            .reconfigurable(true)
            .build();

    @SetFromFlag("resizeOperator")
    public static final ConfigKey<ResizeOperator> RESIZE_OPERATOR = BasicConfigKey.builder(ResizeOperator.class)
            .name("autoscaler.resizeOperator")
            .description("The operation to perform for resizing (defaults to calling resize(int) effector on a Resizable entity)")
            .defaultValue(new ResizeOperator() {
                    @Override
                    public Integer resize(Entity entity, Integer desiredSize) {
                        return ((Resizable)entity).resize(desiredSize);
                    }})
            .build();
    
    @SuppressWarnings("serial")
    @SetFromFlag("currentSizeOperator")
    public static final ConfigKey<Function<Entity,Integer>> CURRENT_SIZE_OPERATOR = BasicConfigKey.builder(new TypeToken<Function<Entity,Integer>>() {})
            .name("autoscaler.currentSizeOperator")
            .description("The operation to perform to calculate the current size (defaults to calling getCurrentSize() on a Resizable entity)")
            .defaultValue(new Function<Entity,Integer>() {
                    @Override
                    public Integer apply(Entity entity) {
                        return ((Resizable)entity).getCurrentSize();
                    }})
            .build();

    @SuppressWarnings("serial")
    @SetFromFlag("poolHotSensor")
    public static final ConfigKey<BasicNotificationSensor<? extends Map>> POOL_HOT_SENSOR = BasicConfigKey.builder(new TypeToken<BasicNotificationSensor<? extends Map>>() {})
            .name("autoscaler.poolHotSensor")
            .description("Sensor to subscribe to, for 'pool hot' events. This is an alternative mechanism to keeping the 'metric' within a given range.")
            .defaultValue(DEFAULT_POOL_HOT_SENSOR)
            .build();

    @SuppressWarnings("serial")
    @SetFromFlag("poolColdSensor")
    public static final ConfigKey<BasicNotificationSensor<? extends Map>> POOL_COLD_SENSOR = BasicConfigKey.builder(new TypeToken<BasicNotificationSensor<? extends Map>>() {})
            .name("autoscaler.poolColdSensor")
            .description("Sensor to subscribe to, for 'pool cold' events. This is an alternative mechanism to keeping the 'metric' within a given range.")
            .defaultValue(DEFAULT_POOL_COLD_SENSOR)
            .build();

    @SuppressWarnings("serial")
    @SetFromFlag("poolOkSensor")
    public static final ConfigKey<BasicNotificationSensor<? extends Map>> POOL_OK_SENSOR = BasicConfigKey.builder(new TypeToken<BasicNotificationSensor<? extends Map>>() {})
            .name("autoscaler.poolOkSensor")
            .description("Sensor to subscribe to, for 'pool ok' events. This is an alternative mechanism to keeping the 'metric' within a given range.")
            .defaultValue(DEFAULT_POOL_OK_SENSOR)
            .build();

    @SuppressWarnings("serial")
    @SetFromFlag("maxSizeReachedSensor")
    public static final ConfigKey<BasicNotificationSensor<? super MaxPoolSizeReachedEvent>> MAX_SIZE_REACHED_SENSOR = BasicConfigKey.builder(new TypeToken<BasicNotificationSensor<? super MaxPoolSizeReachedEvent>>() {})
            .name("autoscaler.maxSizeReachedSensor")
            .description("Sensor by which a notification will be emitted (on the associated entity) when " +
                    "we consistently wanted to resize the pool above the max allowed size, for " +
                    "maxReachedNotificationDelay milliseconds")
            .build();
    
    @SetFromFlag("maxReachedNotificationDelay")
    public static final ConfigKey<Duration> MAX_REACHED_NOTIFICATION_DELAY = BasicConfigKey.builder(Duration.class)
            .name("autoscaler.maxReachedNotificationDelay")
            .description("Time that we consistently wanted to go above the maxPoolSize for, after which the " +
                    "maxSizeReachedSensor (if any) will be emitted")
            .defaultValue(Duration.ZERO)
            .build();
    
    private Entity poolEntity;
    
    private final AtomicBoolean executorQueued = new AtomicBoolean(false);
    private volatile long executorTime = 0;
    private volatile ScheduledExecutorService executor;

    private SizeHistory recentUnboundedResizes;

    private SizeHistory recentDesiredResizes;
    
    private long maxReachedLastNotifiedTime;
    
    private final SensorEventListener<Map> utilizationEventHandler = new SensorEventListener<Map>() {
        @Override
        public void onEvent(SensorEvent<Map> event) {
            Map<String, ?> properties = event.getValue();
            Sensor<?> sensor = event.getSensor();
            
            if (sensor.equals(getPoolColdSensor())) {
                onPoolCold(properties);
            } else if (sensor.equals(getPoolHotSensor())) {
                onPoolHot(properties);
            } else if (sensor.equals(getPoolOkSensor())) {
                onPoolOk(properties);
            } else {
                throw new IllegalStateException("Unexpected sensor type: "+sensor+"; event="+event);
            }
        }
    };

    private final SensorEventListener<Number> metricEventHandler = new SensorEventListener<Number>() {
        @Override
        public void onEvent(SensorEvent<Number> event) {
            assert event.getSensor().equals(getMetric());
            onMetricChanged(event.getValue());
        }
    };

    /**
     * This should usually be a no-op as the min and max pool sizes are taken into account when
     * an automatic resize event occurs, however this covers the special case where the user
     * has manually resized the pool
     */
    private SensorEventListener<? super Integer> poolEventHandler = new SensorEventListener<Integer>() {
        @Override
        public void onEvent(SensorEvent<Integer> event) {
            Sensor<Integer> sensor = event.getSensor();
            Preconditions.checkArgument(sensor.equals(DynamicCluster.GROUP_SIZE), "Unexpected sensor " + sensor);
            Integer size = event.getValue();
            String targetRange = getMinPoolSize()+" - "+getMaxPoolSize();
            if (size > getMaxPoolSize()) {
                highlightViolation("Size "+size+" too large (target "+targetRange+")");
                scheduleResize(getMaxPoolSize(), decap(getHighlights().get(HIGHLIGHT_NAME_LAST_VIOLATION)));
            } else if (size < getMinPoolSize()) {
                highlightViolation("Size "+size+" too small (target "+targetRange+")");
                scheduleResize(getMinPoolSize(), decap(getHighlights().get(HIGHLIGHT_NAME_LAST_VIOLATION)));
            } else {
                highlightConfirmation("Size "+size+" in target range "+targetRange);
            }
        }
    };
    
    private String decap(HighlightTuple t) {
        String msg = t==null ? null : t.getDescription();
        if (Strings.isBlank(msg)) return null;
        return Character.toLowerCase(msg.charAt(0))+msg.substring(1);
    }

    public AutoScalerPolicy() {
    }
    
    /**
     * @deprecated since 0.12.0; use {@link PolicySpec}
     */
    @Deprecated
    public AutoScalerPolicy(Map<String,?> props) {
        super(props);
    }
    
    @Override
    public void init() {
        doInit();
    }

    @Override
    public void rebind() {
        doInit();
    }
    
    protected void doInit() {
        long maxReachedNotificationDelay = getMaxReachedNotificationDelay().toMilliseconds();
        recentUnboundedResizes = new SizeHistory(maxReachedNotificationDelay);
        
        long maxResizeStabilizationDelay = Math.max(getResizeUpStabilizationDelay().toMilliseconds(), getResizeDownStabilizationDelay().toMilliseconds());
        recentDesiredResizes = new SizeHistory(maxResizeStabilizationDelay);
        
        // TODO Should re-use the execution manager's thread pool, somehow
        executor = Executors.newSingleThreadScheduledExecutor(newThreadFactory());
    }

    public void setMetricLowerBound(Number val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing metricLowerBound from {} to {}", new Object[] {this, getMetricLowerBound(), val});
        config().set(METRIC_LOWER_BOUND, checkNotNull(val));
    }
    
    public void setMetricUpperBound(Number val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing metricUpperBound from {} to {}", new Object[] {this, getMetricUpperBound(), val});
        config().set(METRIC_UPPER_BOUND, checkNotNull(val));
    }
    
    private <T> void setOrDefault(ConfigKey<T> key, T val) {
        if (val==null) val = key.getDefaultValue();
        config().set(key, val);
    }
    public int getResizeUpIterationIncrement() { return getConfig(RESIZE_UP_ITERATION_INCREMENT); }
    public void setResizeUpIterationIncrement(Integer val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing resizeUpIterationIncrement from {} to {}", new Object[] {this, getResizeUpIterationIncrement(), val});
        setOrDefault(RESIZE_UP_ITERATION_INCREMENT, val);
    }
    public int getResizeDownIterationIncrement() { return getConfig(RESIZE_DOWN_ITERATION_INCREMENT); }
    public void setResizeDownIterationIncrement(Integer val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing resizeDownIterationIncrement from {} to {}", new Object[] {this, getResizeDownIterationIncrement(), val});
        setOrDefault(RESIZE_DOWN_ITERATION_INCREMENT, val);
    }
    public int getResizeUpIterationMax() { return getConfig(RESIZE_UP_ITERATION_MAX); }
    public void setResizeUpIterationMax(Integer val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing resizeUpIterationMax from {} to {}", new Object[] {this, getResizeUpIterationMax(), val});
        setOrDefault(RESIZE_UP_ITERATION_MAX, val);
    }
    public int getResizeDownIterationMax() { return getConfig(RESIZE_DOWN_ITERATION_MAX); }
    public void setResizeDownIterationMax(Integer val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing resizeDownIterationMax from {} to {}", new Object[] {this, getResizeDownIterationMax(), val});
        setOrDefault(RESIZE_DOWN_ITERATION_MAX, val);
    }

    public void setMinPeriodBetweenExecs(Duration val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing minPeriodBetweenExecs from {} to {}", new Object[] {this, getMinPeriodBetweenExecs(), val});
        config().set(MIN_PERIOD_BETWEEN_EXECS, val);
    }

    public void setResizeUpStabilizationDelay(Duration val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing resizeUpStabilizationDelay from {} to {}", new Object[] {this, getResizeUpStabilizationDelay(), val});
        config().set(RESIZE_UP_STABILIZATION_DELAY, val);
    }
    
    public void setResizeDownStabilizationDelay(Duration val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing resizeDownStabilizationDelay from {} to {}", new Object[] {this, getResizeDownStabilizationDelay(), val});
        config().set(RESIZE_DOWN_STABILIZATION_DELAY, val);
    }
    
    public void setMinPoolSize(int val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing minPoolSize from {} to {}", new Object[] {this, getMinPoolSize(), val});
        config().set(MIN_POOL_SIZE, val);
    }
    
    public void setMaxPoolSize(int val) {
        if (LOG.isInfoEnabled()) LOG.info("{} changing maxPoolSize from {} to {}", new Object[] {this, getMaxPoolSize(), val});
        config().set(MAX_POOL_SIZE, val);
    }
    
    private AttributeSensor<? extends Number> getMetric() {
        return getConfig(METRIC);
    }

    private Entity getEntityWithMetric() {
        return getConfig(ENTITY_WITH_METRIC);
    }
    
    private Number getMetricLowerBound() {
        return getConfig(METRIC_LOWER_BOUND);
    }
    
    private Number getMetricUpperBound() {
        return getConfig(METRIC_UPPER_BOUND);
    }
    
    private Duration getMinPeriodBetweenExecs() {
        return getConfig(MIN_PERIOD_BETWEEN_EXECS);
    }
    
    private Duration getResizeUpStabilizationDelay() {
        return getConfig(RESIZE_UP_STABILIZATION_DELAY);
    }
    
    private Duration getResizeDownStabilizationDelay() {
        return getConfig(RESIZE_DOWN_STABILIZATION_DELAY);
    }
    
    private int getMinPoolSize() {
        return getConfig(MIN_POOL_SIZE);
    }
    
    private int getMaxPoolSize() {
        return getConfig(MAX_POOL_SIZE);
    }
    
    private Integer getInsufficientCapacityHighWaterMark() {
        return getConfig(INSUFFICIENT_CAPACITY_HIGH_WATER_MARK);
    }
    
    private ResizeOperator getResizeOperator() {
        return getConfig(RESIZE_OPERATOR);
    }
    
    private Function<Entity,Integer> getCurrentSizeOperator() {
        return getConfig(CURRENT_SIZE_OPERATOR);
    }
    
    private BasicNotificationSensor<? extends Map> getPoolHotSensor() {
        return getConfig(POOL_HOT_SENSOR);
    }
    
    private BasicNotificationSensor<? extends Map> getPoolColdSensor() {
        return getConfig(POOL_COLD_SENSOR);
    }
    
    private BasicNotificationSensor<? extends Map> getPoolOkSensor() {
        return getConfig(POOL_OK_SENSOR);
    }
    
    private BasicNotificationSensor<? super MaxPoolSizeReachedEvent> getMaxSizeReachedSensor() {
        return getConfig(MAX_SIZE_REACHED_SENSOR);
    }
    
    private Duration getMaxReachedNotificationDelay() {
        return getConfig(MAX_REACHED_NOTIFICATION_DELAY);
    }

    @Override
    protected <T> void doReconfigureConfig(ConfigKey<T> key, T val) {
        if (key.equals(RESIZE_UP_STABILIZATION_DELAY)) {
            Duration maxResizeStabilizationDelay = Duration.max((Duration)val, getResizeDownStabilizationDelay());
            recentDesiredResizes.setWindowSize(maxResizeStabilizationDelay);
        } else if (key.equals(RESIZE_DOWN_STABILIZATION_DELAY)) {
            Duration maxResizeStabilizationDelay = Duration.max((Duration)val, getResizeUpStabilizationDelay());
            recentDesiredResizes.setWindowSize(maxResizeStabilizationDelay);
        } else if (key.equals(METRIC_LOWER_BOUND)) {
            // TODO If recorded what last metric value was then we could recalculate immediately
            // Rely on next metric-change to trigger recalculation; 
            // and same for those below...
        } else if (key.equals(METRIC_UPPER_BOUND)) {
            // see above
        } else if (key.equals(RESIZE_UP_ITERATION_INCREMENT) || key.equals(RESIZE_UP_ITERATION_MAX) || key.equals(RESIZE_DOWN_ITERATION_INCREMENT) || key.equals(RESIZE_DOWN_ITERATION_MAX)) {
            // no special actions needed
        } else if (key.equals(MIN_POOL_SIZE)) {
            int newMin = (Integer) val;
            if (newMin > getConfig(MAX_POOL_SIZE)) {
                throw new IllegalArgumentException("Min pool size "+val+" must not be greater than max pool size "+getConfig(MAX_POOL_SIZE));
            }
            onPoolSizeLimitsChanged(newMin, getConfig(MAX_POOL_SIZE));
        } else if (key.equals(MAX_POOL_SIZE)) {
            int newMax = (Integer) val;
            if (newMax < getConfig(MIN_POOL_SIZE)) {
                throw new IllegalArgumentException("Min pool size "+val+" must not be greater than max pool size "+getConfig(MAX_POOL_SIZE));
            }
            onPoolSizeLimitsChanged(getConfig(MIN_POOL_SIZE), newMax);
        } else if (key.equals(INSUFFICIENT_CAPACITY_HIGH_WATER_MARK)) {
            Integer newVal = (Integer) val;
            Integer oldVal = config().get(INSUFFICIENT_CAPACITY_HIGH_WATER_MARK);
            if (oldVal != null && (newVal == null || newVal > oldVal)) {
                LOG.info("{} resetting {} to {}, which will enable resizing above previous level of {}",
                        new Object[] {AutoScalerPolicy.this, INSUFFICIENT_CAPACITY_HIGH_WATER_MARK.getName(), newVal, oldVal});
                // TODO see above about changing metricLowerBound; not triggering resize now
            }
        } else {
            throw new UnsupportedOperationException("reconfiguring "+key+" unsupported for "+this);
        }
    }

    @Override
    public void suspend() {
        super.suspend();
        // TODO unsubscribe from everything? And resubscribe on resume?
        if (executor != null) executor.shutdownNow();
    }
    
    @Override
    public void resume() {
        super.resume();
        executor = Executors.newSingleThreadScheduledExecutor(newThreadFactory());
    }
    
    @Override
    public void setEntity(EntityLocal entity) {
        if (!config().getRaw(RESIZE_OPERATOR).isPresentAndNonNull()) {
            Preconditions.checkArgument(entity instanceof Resizable, "Provided entity "+entity+" must be an instance of Resizable, because no custom-resizer operator supplied");
        }
        super.setEntity(entity);
        this.poolEntity = entity;
        
        if (getMetric() != null) {
            Entity entityToSubscribeTo = (getEntityWithMetric() != null) ? getEntityWithMetric() : entity;
            subscriptions().subscribe(entityToSubscribeTo, getMetric(), metricEventHandler);
            highlightTriggers(getMetric(), entityToSubscribeTo);
        } else {
            highlightTriggers("Listening for standard size and pool hot/cold sensors (no specific metric)");
        }
        subscriptions().subscribe(poolEntity, getPoolColdSensor(), utilizationEventHandler);
        subscriptions().subscribe(poolEntity, getPoolHotSensor(), utilizationEventHandler);
        subscriptions().subscribe(poolEntity, getPoolOkSensor(), utilizationEventHandler);
        subscriptions().subscribe(poolEntity, DynamicCluster.GROUP_SIZE, poolEventHandler);
    }
    
    private ThreadFactory newThreadFactory() {
        return new ThreadFactoryBuilder()
                .setNameFormat("brooklyn-autoscalerpolicy-%d")
                .build();
    }

    /**
     * Forces an immediate resize (without waiting for stabilization etc) if the current size is 
     * not within the min and max limits. We schedule this so that all resize operations are done
     * by the same thread.
     */
    private void onPoolSizeLimitsChanged(final int min, final int max) {
        if (LOG.isTraceEnabled()) LOG.trace("{} checking pool size on limits changed for {} (between {} and {})", new Object[] {this, poolEntity, min, max});
        
        if (isRunning() && isEntityUp()) {
            executor.submit(new Runnable() {
                @Override public void run() {
                    try {
                        int currentSize = getCurrentSizeOperator().apply(entity);
                        int desiredSize = Math.min(max, Math.max(min, currentSize));

                        if (currentSize != desiredSize) {
                            if (LOG.isInfoEnabled()) LOG.info("{} resizing pool {} immediateley from {} to {} (due to new pool size limits)", new Object[] {this, poolEntity, currentSize, desiredSize});
                            getResizeOperator().resize(poolEntity, desiredSize);
                        }
                        
                    } catch (Exception e) {
                        if (isRunning()) {
                            LOG.error("Error resizing: "+e, e);
                        } else {
                            if (LOG.isDebugEnabled()) LOG.debug("Error resizing, but no longer running: "+e, e);
                        }
                    } catch (Throwable t) {
                        LOG.error("Error resizing: "+t, t);
                        throw Throwables.propagate(t);
                    }
                }});
        }
    }
    
    private enum ScalingType { HOT, COLD }
    private static class ScalingData {
        ScalingType scalingMode;
        int currentSize;
        double currentMetricValue;
        Double metricUpperBound;
        Double metricLowerBound;
        
        public double getCurrentTotalActivity() {
            return currentMetricValue * currentSize;
        }
        
        public boolean isHot() {
            return ((scalingMode==null || scalingMode==ScalingType.HOT) && isValid(metricUpperBound) && currentMetricValue > metricUpperBound);
        }
        public boolean isCold() {
            return ((scalingMode==null || scalingMode==ScalingType.COLD) && isValid(metricLowerBound) && currentMetricValue < metricLowerBound);
        }
        private boolean isValid(Double bound) {
            return (bound!=null && bound>0);
        }
    }

    private void onMetricChanged(Number val) {
        if (LOG.isTraceEnabled()) LOG.trace("{} recording pool-metric for {}: {}", new Object[] {this, poolEntity, val});

        if (val==null) {
            // occurs e.g. if using an aggregating enricher who returns null when empty, the sensor has gone away
            if (LOG.isTraceEnabled()) LOG.trace("{} not resizing pool {}, inbound metric is null", new Object[] {this, poolEntity});
            return;
        }
        
        ScalingData data = new ScalingData();
        data.currentMetricValue = val.doubleValue();
        data.currentSize = getCurrentSizeOperator().apply(entity);
        data.metricUpperBound = getMetricUpperBound().doubleValue();
        data.metricLowerBound = getMetricLowerBound().doubleValue();
        
        analyze(data, "pool");
    }
    
    private void onPoolCold(Map<String, ?> properties) {
        if (LOG.isTraceEnabled()) LOG.trace("{} recording pool-cold for {}: {}", new Object[] {this, poolEntity, properties});
        analyzeOnHotOrColdSensor(ScalingType.COLD, "cold pool", properties);
    }
    
    private void onPoolHot(Map<String, ?> properties) {
        if (LOG.isTraceEnabled()) LOG.trace("{} recording pool-hot for {}: {}", new Object[] {this, poolEntity, properties});
        analyzeOnHotOrColdSensor(ScalingType.HOT, "hot pool", properties);
    }
    
    private void analyzeOnHotOrColdSensor(ScalingType scalingMode, String description, Map<String, ?> properties) {
        ScalingData data = new ScalingData();
        data.scalingMode = scalingMode;
        data.currentMetricValue = (Double) properties.get(POOL_CURRENT_WORKRATE_KEY);
        data.currentSize = (Integer) properties.get(POOL_CURRENT_SIZE_KEY);
        data.metricUpperBound = (Double) properties.get(POOL_HIGH_THRESHOLD_KEY);
        data.metricLowerBound = (Double) properties.get(POOL_LOW_THRESHOLD_KEY);
        
        analyze(data, description);   
    }
    
    private void analyze(ScalingData data, String description) {
        int desiredSizeUnconstrained;
        
        /* We always scale out (modulo stabilization delay) if:
         *   currentTotalActivity > currentSize*metricUpperBound
         * With newDesiredSize the smallest n such that   n*metricUpperBound >= currentTotalActivity
         * ie  n >= currentTotalActiviy/metricUpperBound, thus n := Math.ceil(currentTotalActivity/metricUpperBound)
         * 
         * Else consider scale back if:
         *   currentTotalActivity < currentSize*metricLowerBound
         * With newDesiredSize normally the largest n such that:  
         *   n*metricLowerBound <= currentTotalActivity
         * BUT with an absolute requirement which trumps the above computation
         * that the newDesiredSize doesn't cause immediate scale out:
         *   n*metricUpperBound >= currentTotalActivity
         * thus n := Math.max ( floor(currentTotalActiviy/metricLowerBound), ceil(currentTotal/metricUpperBound) )
         */
        if (data.isHot()) {
            // scale out
            highlightViolation("Metric "+String.format("%.02f", data.currentMetricValue)+" too hot "
                + "(target range "+String.format("%.02f", data.metricLowerBound)+"-"+String.format("%.02f", data.metricUpperBound)+")");
            desiredSizeUnconstrained = (int)Math.ceil(data.getCurrentTotalActivity() / data.metricUpperBound);
            data.scalingMode = ScalingType.HOT;
            
        } else if (data.isCold()) {
            // scale back
            highlightViolation("Metric "+String.format("%.02f", data.currentMetricValue)+" too cold "
                + "(target range "+String.format("%.02f", data.metricLowerBound)+"-"+String.format("%.02f", data.metricUpperBound)+")");
            desiredSizeUnconstrained = (int)Math.floor(data.getCurrentTotalActivity() / data.metricLowerBound);
            data.scalingMode = ScalingType.COLD;
            
        } else {
            highlightConfirmation("Metric "+String.format("%.02f", data.currentMetricValue)+" in "
                + "target range "+String.format("%.02f", data.metricLowerBound)+"-"+String.format("%.02f", data.metricUpperBound));
            if (LOG.isTraceEnabled()) LOG.trace("{} not resizing pool {} from {} ({} within range {}..{})", new Object[] {this, poolEntity, data.currentSize, data.currentMetricValue, data.metricLowerBound, data.metricUpperBound});
            abortResize(data.currentSize);
            return; // within the healthy range; no-op
        }
        
        if (LOG.isTraceEnabled()) LOG.debug("{} detected unconstrained desired size {}", new Object[] {this, desiredSizeUnconstrained});
        int desiredSize = applyMinMaxConstraints(desiredSizeUnconstrained);

        if ((data.scalingMode==ScalingType.COLD) && (desiredSize < data.currentSize)) {

            int delta = data.currentSize - desiredSize;
            int scaleIncrement = getResizeDownIterationIncrement();
            int scaleMax = getResizeDownIterationMax();
            if (delta>scaleMax) {
                delta=scaleMax;
            } else if (delta % scaleIncrement != 0) {
                // keep scaling to the increment
                delta += scaleIncrement - (delta % scaleIncrement);
            }
            desiredSize = data.currentSize - delta;
            
            if (data.metricUpperBound!=null) {
                // if upper bound supplied, check that this desired scale-back size 
                // is not going to cause scale-out on next run; i.e. anti-thrashing
                while (desiredSize < data.currentSize && data.getCurrentTotalActivity() > data.metricUpperBound * desiredSize) {
                    if (LOG.isTraceEnabled()) LOG.trace("{} when resizing back pool {} from {}, tweaking from {} to prevent thrashing", new Object[] {this, poolEntity, data.currentSize, desiredSize });
                    desiredSize += scaleIncrement;
                }
            }
            desiredSize = applyMinMaxConstraints(desiredSize);
            if (desiredSize >= data.currentSize) data.scalingMode = null;
            
        } else if ((data.scalingMode==ScalingType.HOT) && (desiredSize > data.currentSize)) {

            int delta = desiredSize - data.currentSize;
            int scaleIncrement = getResizeUpIterationIncrement();
            int scaleMax = getResizeUpIterationMax();
            if (delta>scaleMax) {
                delta=scaleMax;
            } else if (delta % scaleIncrement != 0) {
                // keep scaling to the increment
                delta += scaleIncrement - (delta % scaleIncrement);
            }
            desiredSize = data.currentSize + delta;
            desiredSize = applyMinMaxConstraints(desiredSize);
            if (desiredSize <= data.currentSize) data.scalingMode = null;

        } else {
            data.scalingMode = null;
        }
    
        if (data.scalingMode!=null) {
            if (LOG.isDebugEnabled()) LOG.debug("{} provisionally resizing {} {} from {} to {} ({} < {}; ideal size {})", new Object[] {this, description, poolEntity, data.currentSize, desiredSize, data.currentMetricValue, data.metricLowerBound, desiredSizeUnconstrained});
            scheduleResize(desiredSize, "metric "+data.currentMetricValue+" out of range");
        } else {
            if (LOG.isTraceEnabled()) LOG.trace("{} not resizing {} {} from {} to {}, {} out of healthy range {}..{} but unconstrained size {} blocked by bounds/check", new Object[] {this, description, poolEntity, data.currentSize, desiredSize, data.currentMetricValue, data.metricLowerBound, data.metricUpperBound, desiredSizeUnconstrained});
            abortResize(data.currentSize);
            // but add to the unbounded record for future consideration
        }
        
        onNewUnboundedPoolSize(desiredSizeUnconstrained, "ideal unconstrained size is "+desiredSizeUnconstrained);
    }

    private int applyMinMaxConstraints(long desiredSize) {
        return applyMinMaxConstraints(desiredSize > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int)desiredSize);
    }
    
    private int applyMinMaxConstraints(int desiredSize) {
        int minSize = getMinPoolSize();
        int maxSize = getMaxPoolSize();
        Integer insufficientCapacityHighWaterMark = getInsufficientCapacityHighWaterMark();
        desiredSize = Math.max(minSize, desiredSize);
        desiredSize = Math.min(maxSize, desiredSize);
        if (insufficientCapacityHighWaterMark != null) desiredSize = Math.min(insufficientCapacityHighWaterMark, desiredSize);
        return desiredSize;
    }

    private void onPoolOk(Map<String, ?> properties) {
        if (LOG.isTraceEnabled()) LOG.trace("{} recording pool-ok for {}: {}", new Object[] {this, poolEntity, properties});
        
        int poolCurrentSize = (Integer) properties.get(POOL_CURRENT_SIZE_KEY);
        
        if (LOG.isTraceEnabled()) LOG.trace("{} not resizing ok pool {} from {}", new Object[] {this, poolEntity, poolCurrentSize});
        abortResize(poolCurrentSize);
    }

    /**
     * Schedules a resize, if there is not already a resize operation queued up. When that resize
     * executes, it will resize to whatever the latest value is to be (rather than what it was told
     * to do at the point the job was queued).
     */
    private void scheduleResize(final int newSize, String reason) {
        recentDesiredResizes.add(newSize);
        
        scheduleResize(reason);
    }

    /**
     * If a listener is registered to be notified of the max-pool-size cap being reached, then record
     * what our unbounded size would be and schedule a check to see if this unbounded size is sustained.
     * 
     * Piggy-backs off the existing scheduleResize execution, which now also checks if the listener
     * needs to be called.
     */
    private void onNewUnboundedPoolSize(final int val, String reason) {
        if (getMaxSizeReachedSensor() != null) {
            recentUnboundedResizes.add(val);
            scheduleResize(reason);
        }
    }
    
    private void abortResize(final int currentSize) {
        recentDesiredResizes.add(currentSize);
        recentUnboundedResizes.add(currentSize);
    }

    private boolean isEntityUp() {
        if (entity == null) {
            return false;
        } else if (entity.getEntityType().getSensors().contains(Startable.SERVICE_UP)) {
            return Boolean.TRUE.equals(entity.getAttribute(Startable.SERVICE_UP));
        } else {
            return true;
        }
    }

    private void scheduleResize(String reason) {
        // TODO Make scale-out calls concurrent, rather than waiting for first resize to entirely 
        // finish. On ec2 for example, this can cause us to grow very slowly if first request is for
        // just one new VM to be provisioned.
        
        if (isRunning() && isEntityUp() && executorQueued.compareAndSet(false, true)) {
            long now = System.currentTimeMillis();
            long delay = Math.max(0, (executorTime + getMinPeriodBetweenExecs().toMilliseconds()) - now);
            if (LOG.isTraceEnabled()) LOG.trace("{} scheduling resize in {}ms", this, delay);
            
            executor.schedule(new Runnable() {
                @Override public void run() {
                    try {
                        executorTime = System.currentTimeMillis();
                        executorQueued.set(false);

                        resizeNow(reason);
                        notifyMaxReachedIfRequiredNow(reason);
                        
                    } catch (Exception e) {
                        if (isRunning()) {
                            LOG.error("Error resizing: "+e, e);
                        } else {
                            if (LOG.isDebugEnabled()) LOG.debug("Error resizing, but no longer running: "+e, e);
                        }
                    } catch (Throwable t) {
                        LOG.error("Error resizing: "+t, t);
                        throw Throwables.propagate(t);
                    }
                }},
                delay,
                TimeUnit.MILLISECONDS);
        }
    }

    /**
     * Looks at the values for "unbounded pool size" (i.e. if we ignore caps of minSize and maxSize) to report what
     * those values have been within a time window. The time window used is the "maxReachedNotificationDelay",
     * which determines how many milliseconds after being consistently above the max-size will it take before
     * we emit the sensor event (if any).
     * @param reason 
     */
    private void notifyMaxReachedIfRequiredNow(String reason) {
        BasicNotificationSensor<? super MaxPoolSizeReachedEvent> maxSizeReachedSensor = getMaxSizeReachedSensor();
        if (maxSizeReachedSensor == null) {
            return;
        }
        
        WindowSummary valsSummary = recentUnboundedResizes.summarizeWindow(getMaxReachedNotificationDelay());
        long timeWindowSize = getMaxReachedNotificationDelay().toMilliseconds();
        long currentPoolSize = getCurrentSizeOperator().apply(poolEntity);
        int maxAllowedPoolSize = getMaxPoolSize();
        long unboundedSustainedMaxPoolSize = valsSummary.min; // The sustained maximum (i.e. the smallest it's dropped down to)
        long unboundedCurrentPoolSize = valsSummary.latest;
        
        if (maxReachedLastNotifiedTime > 0) {
            // already notified the listener; don't do it again
            // TODO Could have max period for notifications, or a step increment to warn when exceeded by ever bigger amounts
            
        } else if (unboundedSustainedMaxPoolSize > maxAllowedPoolSize) {
            // We have consistently wanted to be bigger than the max allowed; tell the listener
            if (LOG.isDebugEnabled()) LOG.debug("{} notifying listener of max pool size reached; current {}, max {}, unbounded current {}, unbounded max {}", 
                    new Object[] {this, currentPoolSize, maxAllowedPoolSize, unboundedCurrentPoolSize, unboundedSustainedMaxPoolSize});
            
            maxReachedLastNotifiedTime = System.currentTimeMillis();
            MaxPoolSizeReachedEvent event = MaxPoolSizeReachedEvent.builder()
                    .currentPoolSize(currentPoolSize)
                    .maxAllowed(maxAllowedPoolSize)
                    .currentUnbounded(unboundedCurrentPoolSize)
                    .maxUnbounded(unboundedSustainedMaxPoolSize)
                    .timeWindow(timeWindowSize)
                    .build();
            entity.sensors().emit(maxSizeReachedSensor, event);
            
        } else if (valsSummary.max > maxAllowedPoolSize) {
            // We temporarily wanted to be bigger than the max allowed; check back later to see if consistent
            // TODO Could check if there has been anything bigger than "min" since min happened (would be more efficient)
            if (LOG.isTraceEnabled()) LOG.trace("{} re-scheduling max-reached check for {}, as unbounded size not stable (min {}, max {}, latest {})", 
                    new Object[] {this, poolEntity, valsSummary.min, valsSummary.max, valsSummary.latest});
            scheduleResize(reason);
            
        } else {
            // nothing to write home about; continually below maxAllowed
        }
    }

    private void resizeNow(String reason) {
        final int currentPoolSize = getCurrentSizeOperator().apply(poolEntity);
        CalculatedDesiredPoolSize calculatedDesiredPoolSize = calculateDesiredPoolSize(currentPoolSize);
        long desiredPoolSize = calculatedDesiredPoolSize.size;
        boolean stable = calculatedDesiredPoolSize.stable;
        
        final int targetPoolSize = applyMinMaxConstraints(desiredPoolSize);
        
        if (!stable) {
            // the desired size fluctuations are not stable; ensure we check again later (due to time-window)
            // even if no additional events have been received
            // (note we continue now with as "good" a resize as we can given the instability)
            if (LOG.isTraceEnabled()) LOG.trace("{} re-scheduling resize check for {}, as desired size not stable (current {}, desired {}); continuing with resize...", 
                    new Object[] {this, poolEntity, currentPoolSize, targetPoolSize});
            scheduleResize(reason);
        }
        if (currentPoolSize == targetPoolSize) {
            if (LOG.isTraceEnabled()) LOG.trace("{} not resizing pool {} from {} to {}", 
                    new Object[] {this, poolEntity, currentPoolSize, targetPoolSize});
            return;
        }
        
        if (LOG.isDebugEnabled()) LOG.debug("{} requesting resize to {}; current {}, min {}, max {}", 
                new Object[] {this, targetPoolSize, currentPoolSize, getMinPoolSize(), getMaxPoolSize()});
        
        Task<Void> t = Entities.submit(entity, Tasks.<Void>builder().displayName("Auto-scaler")
            .description("Auto-scaler recommending resize from "+currentPoolSize+" to "+targetPoolSize)
            .tag(BrooklynTaskTags.NON_TRANSIENT_TASK_TAG)
            .body(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    // TODO Should we use int throughout, rather than casting here?
                    try {
                        getResizeOperator().resize(poolEntity, targetPoolSize);
                    } catch (Resizable.InsufficientCapacityException e) {
                        // cannot resize beyond this; set the high-water mark
                        int insufficientCapacityHighWaterMark = getCurrentSizeOperator().apply(poolEntity);
                        LOG.warn("{} failed to resize {} due to insufficient capacity; setting high-water mark to {}, "
                                + "and will not attempt to resize above that level again", 
                                new Object[] {AutoScalerPolicy.this, poolEntity, insufficientCapacityHighWaterMark});
                        config().set(INSUFFICIENT_CAPACITY_HIGH_WATER_MARK, insufficientCapacityHighWaterMark);
                    }
                    return null;
                }
            }).build());
        highlightAction("Resize from "+currentPoolSize+" to "+targetPoolSize+
            (reason!=null ? " because "+reason : ""), t);
        t.blockUntilEnded();
    }
    
    /**
     * Complicated logic for stabilization-delay...
     * Only grow if we have consistently been asked to grow for the resizeUpStabilizationDelay period;
     * Only shrink if we have consistently been asked to shrink for the resizeDownStabilizationDelay period.
     * 
     * @return tuple of desired pool size, and whether this is "stable" (i.e. if we receive no more events 
     *         will this continue to be the desired pool size)
     */
    private CalculatedDesiredPoolSize calculateDesiredPoolSize(long currentPoolSize) {
        long now = System.currentTimeMillis();
        WindowSummary downsizeSummary = recentDesiredResizes.summarizeWindow(getResizeDownStabilizationDelay());
        WindowSummary upsizeSummary = recentDesiredResizes.summarizeWindow(getResizeUpStabilizationDelay());
        
        // this is the _sustained_ growth value; the smallest size that has been requested in the "stable-for-growing" period
        long maxDesiredPoolSize = upsizeSummary.min;
        boolean stableForGrowing = upsizeSummary.stableForGrowth;
        
        // this is the _sustained_ shrink value; largest size that has been requested in the "stable-for-shrinking" period:
        long minDesiredPoolSize = downsizeSummary.max;
        boolean stableForShrinking = downsizeSummary.stableForShrinking;
        
        // (it is a logical consequence of the above that minDesired >= maxDesired -- this is correct, if confusing:
        // think of minDesired as the minimum size we are allowed to resize to, and similarly for maxDesired; 
        // if min > max we can scale to max if current < max, or scale to min if current > min)

        long desiredPoolSize;
        
        boolean stable;
        
        if (currentPoolSize < maxDesiredPoolSize) {
            // we have valid request to grow 
            // (we'll never have a valid request to grow and a valid to shrink simultaneously, btw)
            desiredPoolSize = maxDesiredPoolSize;
            stable = stableForGrowing;
        } else if (currentPoolSize > minDesiredPoolSize) {
            // we have valid request to shrink
            desiredPoolSize = minDesiredPoolSize;
            stable = stableForShrinking;
        } else {
            desiredPoolSize = currentPoolSize;
            stable = stableForGrowing && stableForShrinking;
        }

        if (LOG.isTraceEnabled()) LOG.trace("{} calculated desired pool size: from {} to {}; minDesired {}, maxDesired {}; " +
                "stable {}; now {}; downsizeHistory {}; upsizeHistory {}", 
                new Object[] {this, currentPoolSize, desiredPoolSize, minDesiredPoolSize, maxDesiredPoolSize, stable, now, downsizeSummary, upsizeSummary});
        
        return new CalculatedDesiredPoolSize(desiredPoolSize, stable);
    }
    
    private static class CalculatedDesiredPoolSize {
        final long size;
        final boolean stable;
        
        CalculatedDesiredPoolSize(long size, boolean stable) {
            this.size = size;
            this.stable = stable;
        }
    }
    
    @Override
    public String toString() {
        return getClass().getSimpleName() + (groovyTruth(name) ? "("+name+")" : "");
    }
}
