/*
 * 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.entity.nosql.couchbase;

import static org.apache.brooklyn.util.JavaGroovyEquivalents.groovyTruth;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import javax.annotation.Nonnull;

import org.apache.brooklyn.api.entity.Entity;
import org.apache.brooklyn.api.entity.EntitySpec;
import org.apache.brooklyn.api.policy.PolicySpec;
import org.apache.brooklyn.api.sensor.AttributeSensor;
import org.apache.brooklyn.core.config.render.RendererHints;
import org.apache.brooklyn.core.effector.Effectors;
import org.apache.brooklyn.core.entity.Attributes;
import org.apache.brooklyn.core.entity.Entities;
import org.apache.brooklyn.core.entity.EntityInternal;
import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
import org.apache.brooklyn.core.entity.trait.Startable;
import org.apache.brooklyn.core.location.access.BrooklynAccessUtils;
import org.apache.brooklyn.core.sensor.DependentConfiguration;
import org.apache.brooklyn.enricher.stock.Enrichers;
import org.apache.brooklyn.entity.group.AbstractMembershipTrackingPolicy;
import org.apache.brooklyn.entity.group.DynamicClusterImpl;
import org.apache.brooklyn.entity.software.base.SoftwareProcess;
import org.apache.brooklyn.feed.http.HttpFeed;
import org.apache.brooklyn.feed.http.HttpPollConfig;
import org.apache.brooklyn.feed.http.HttpValueFunctions;
import org.apache.brooklyn.feed.http.JsonFunctions;
import org.apache.brooklyn.util.http.HttpToolResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.brooklyn.util.collections.CollectionFunctionals;
import org.apache.brooklyn.util.collections.MutableSet;
import org.apache.brooklyn.util.collections.QuorumCheck;
import org.apache.brooklyn.util.core.task.DynamicTasks;
import org.apache.brooklyn.util.core.task.TaskBuilder;
import org.apache.brooklyn.util.core.task.Tasks;
import org.apache.brooklyn.util.exceptions.Exceptions;
import org.apache.brooklyn.util.guava.Functionals;
import org.apache.brooklyn.util.guava.IfFunctions;
import org.apache.brooklyn.util.math.MathPredicates;
import org.apache.brooklyn.util.text.ByteSizeStrings;
import org.apache.brooklyn.util.text.StringFunctions;
import org.apache.brooklyn.util.text.Strings;
import org.apache.brooklyn.util.time.Duration;
import org.apache.brooklyn.util.time.Time;

import com.google.common.base.Function;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.net.HostAndPort;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;

public class CouchbaseClusterImpl extends DynamicClusterImpl implements CouchbaseCluster {
    
    /*
     * Refactoring required:
     * 
     * Currently, on start() the cluster waits for an arbitrary SERVICE_UP_TIME_OUT (3 minutes) before assuming that a quorate 
     * number of servers are available. The servers are then added to the cluster, and a further wait period of  
     * DELAY_BEFORE_ADVERTISING_CLUSTER (30 seconds) is used before advertising the cluster
     * 
     * DELAY_BEFORE_ADVERTISING_CLUSTER: It should be possible to refactor this away by adding a repeater that will poll
     * the REST API of the primary node (once established) until the API indicates that the cluster is available
     * 
     * SERVICE_UP_TIME_OUT: The refactoring of this would be more substantial. One method would be to remove the bulk of the 
     * logic from the start() method, and rely entirely on the membership tracking policy and the onServerPoolMemberChanged()
     * method. The addition of a RUNNING sensor on the nodes would allow the cluster to determine that a node is up and
     * running but has not yet been added to the cluster. The IS_CLUSTER_INITIALIZED key could be used to determine whether
     * or not the cluster should be initialized, or a node simply added to an existing cluster. A repeater could be used
     * in the driver's to ensure that the method does not return until the node has been fully added
     * 
     * There is an (incomplete) first-pass at this here: https://github.com/Nakomis/incubator-brooklyn/compare/couchbase-running-sensor
     * however, there have been significant changes to the cluster initialization since that work was done so it will probably
     * need to be re-done
     * 
     * Additionally, during bucket creation, a HttpPoll is used to check that the bucket has been created. This should be 
     * refactored to use a Repeater in CouchbaseNodeSshDriver.bucketCreate() in a similar way to the one employed in
     * CouchbaseNodeSshDriver.rebalance(). Were this done, this class could simply queue the bucket creation tasks
     * 
     */
    
    private static final Logger log = LoggerFactory.getLogger(CouchbaseClusterImpl.class);
    private final Object mutex = new Object[0];
    // Used to serialize bucket creation as only one bucket can be created at a time,
    // so a feed is used to determine when a bucket has finished being created
    private final AtomicReference<HttpFeed> resetBucketCreation = new AtomicReference<HttpFeed>();

    @Override
    public void init() {
        log.info("Initializing the Couchbase cluster...");
        super.init();
        
        enrichers().add(
            Enrichers.builder()
                .transforming(COUCHBASE_CLUSTER_UP_NODES)
                .from(this)
                .publishing(COUCHBASE_CLUSTER_UP_NODE_ADDRESSES)
                .computing(new ListOfHostAndPort()).build() );
        enrichers().add(
            Enrichers.builder()
                .transforming(COUCHBASE_CLUSTER_UP_NODE_ADDRESSES)
                .from(this)
                .publishing(COUCHBASE_CLUSTER_CONNECTION_URL)
                .computing(
                    IfFunctions.<List<String>>ifPredicate(
                        Predicates.compose(MathPredicates.lessThan(getConfig(CouchbaseCluster.INITIAL_QUORUM_SIZE)), 
                            CollectionFunctionals.sizeFunction(0)) )
                    .value((String)null)
                    .defaultApply(
                        Functionals.chain(
                            CollectionFunctionals.<String,List<String>>limit(4), 
                            StringFunctions.joiner(","),
                            StringFunctions.formatter("http://%s/"))) )
                .build() );
        
        Map<? extends AttributeSensor<? extends Number>, ? extends AttributeSensor<? extends Number>> enricherSetup = 
            ImmutableMap.<AttributeSensor<? extends Number>, AttributeSensor<? extends Number>>builder()
                .put(CouchbaseNode.OPS, CouchbaseCluster.OPS_PER_NODE)
                .put(CouchbaseNode.COUCH_DOCS_DATA_SIZE, CouchbaseCluster.COUCH_DOCS_DATA_SIZE_PER_NODE)
                .put(CouchbaseNode.COUCH_DOCS_ACTUAL_DISK_SIZE, CouchbaseCluster.COUCH_DOCS_ACTUAL_DISK_SIZE_PER_NODE)
                .put(CouchbaseNode.EP_BG_FETCHED, CouchbaseCluster.EP_BG_FETCHED_PER_NODE)
                .put(CouchbaseNode.MEM_USED, CouchbaseCluster.MEM_USED_PER_NODE)
                .put(CouchbaseNode.COUCH_VIEWS_ACTUAL_DISK_SIZE, CouchbaseCluster.COUCH_VIEWS_ACTUAL_DISK_SIZE_PER_NODE)
                .put(CouchbaseNode.CURR_ITEMS, CouchbaseCluster.CURR_ITEMS_PER_NODE)
                .put(CouchbaseNode.VB_REPLICA_CURR_ITEMS, CouchbaseCluster.VB_REPLICA_CURR_ITEMS_PER_NODE)
                .put(CouchbaseNode.COUCH_VIEWS_DATA_SIZE, CouchbaseCluster.COUCH_VIEWS_DATA_SIZE_PER_NODE)
                .put(CouchbaseNode.GET_HITS, CouchbaseCluster.GET_HITS_PER_NODE)
                .put(CouchbaseNode.CMD_GET, CouchbaseCluster.CMD_GET_PER_NODE)
                .put(CouchbaseNode.CURR_ITEMS_TOT, CouchbaseCluster.CURR_ITEMS_TOT_PER_NODE)
            .build();
        
        for (AttributeSensor<? extends Number> nodeSensor : enricherSetup.keySet()) {
            addSummingMemberEnricher(nodeSensor);
            addAveragingMemberEnricher(nodeSensor, enricherSetup.get(nodeSensor));
        }
        
        enrichers().add(Enrichers.builder().updatingMap(Attributes.SERVICE_NOT_UP_INDICATORS)
            .from(IS_CLUSTER_INITIALIZED).computing(
                IfFunctions.ifNotEquals(true).value("The cluster is not yet completely initialized")
                    .defaultValue(null).build()).build() );
    }
    
    private void addAveragingMemberEnricher(AttributeSensor<? extends Number> fromSensor, AttributeSensor<? extends Number> toSensor) {
        enrichers().add(Enrichers.builder()
            .aggregating(fromSensor)
            .publishing(toSensor)
            .fromMembers()
            .computingAverage()
            .build()
        );
    }

    private void addSummingMemberEnricher(AttributeSensor<? extends Number> source) {
        enrichers().add(Enrichers.builder()
            .aggregating(source)
            .publishing(source)
            .fromMembers()
            .computingSum()
            .build()
        );
    }

    @Override
    protected void doStart() {
        sensors().set(IS_CLUSTER_INITIALIZED, false);
        
        super.doStart();

        connectSensors();
        
        sensors().set(BUCKET_CREATION_IN_PROGRESS, false);

        //start timeout before adding the servers
        Tasks.setBlockingDetails("Pausing while Couchbase stabilizes");
        Time.sleep(getConfig(NODES_STARTED_STABILIZATION_DELAY));

        Optional<Set<Entity>> upNodes = Optional.<Set<Entity>>fromNullable(getAttribute(COUCHBASE_CLUSTER_UP_NODES));
        if (upNodes.isPresent() && !upNodes.get().isEmpty()) {

            Tasks.setBlockingDetails("Adding servers to Couchbase");
            
            //TODO: select a new primary node if this one fails
            Entity primaryNode = upNodes.get().iterator().next();
            ((EntityInternal) primaryNode).sensors().set(CouchbaseNode.IS_PRIMARY_NODE, true);
            sensors().set(COUCHBASE_PRIMARY_NODE, primaryNode);

            Set<Entity> serversToAdd = MutableSet.<Entity>copyOf(getUpNodes());

            if (serversToAdd.size() >= getQuorumSize() && serversToAdd.size() > 1) {
                log.info("Number of SERVICE_UP nodes:{} in cluster:{} reached Quorum:{}, adding the servers", new Object[]{serversToAdd.size(), getId(), getQuorumSize()});
                addServers(serversToAdd);

                //wait for servers to be added to the couchbase server
                try {
                    Tasks.setBlockingDetails("Delaying before advertising cluster up");
                    Time.sleep(getConfig(DELAY_BEFORE_ADVERTISING_CLUSTER));
                } finally {
                    Tasks.resetBlockingDetails();
                }
                
                getPrimaryNode().rebalance();
            } else {
                if (getQuorumSize()>1) {
                    log.warn(this+" is not quorate; will likely fail later, but proceeding for now");
                }
                for (Entity server: serversToAdd) {
                    ((EntityInternal) server).sensors().set(CouchbaseNode.IS_IN_CLUSTER, true);
                }
            }
                
            if (getConfig(CREATE_BUCKETS)!=null) {
                try {
                    Tasks.setBlockingDetails("Creating buckets in Couchbase");

                    createBuckets();
                    DependentConfiguration.waitInTaskForAttributeReady(this, CouchbaseCluster.BUCKET_CREATION_IN_PROGRESS, Predicates.equalTo(false));

                } finally {
                    Tasks.resetBlockingDetails();
                }
            }

            if (getConfig(REPLICATION)!=null) {
                try {
                    Tasks.setBlockingDetails("Configuring replication rules");

                    List<Map<String, Object>> replRules = getConfig(REPLICATION);
                    for (Map<String, Object> replRule: replRules) {
                        DynamicTasks.queue(Effectors.invocation(getPrimaryNode(), CouchbaseNode.ADD_REPLICATION_RULE, replRule));
                    }
                    DynamicTasks.waitForLast();

                } finally {
                    Tasks.resetBlockingDetails();
                }
            }

            sensors().set(IS_CLUSTER_INITIALIZED, true);
            
        } else {
            throw new IllegalStateException("No up nodes available after starting");
        }
    }

    @Override
    public void stop() {
        if (resetBucketCreation.get() != null) {
            resetBucketCreation.get().stop();
        }
        super.stop();
    }

    protected void connectSensors() {
        policies().add(PolicySpec.create(MemberTrackingPolicy.class)
                .displayName("Controller targets tracker")
                .configure("group", this));
    }
    
    private final static class ListOfHostAndPort implements Function<Set<Entity>, List<String>> {
        @Override public List<String> apply(Set<Entity> input) {
            List<String> addresses = Lists.newArrayList();
            if (input == null) return addresses;
            for (Entity entity : input) {
                addresses.add(String.format("%s",
                        BrooklynAccessUtils.getBrooklynAccessibleAddress(entity, entity.getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT))));
            }
            return addresses;
        }
    }

    public static class MemberTrackingPolicy extends AbstractMembershipTrackingPolicy {
        @Override protected void onEntityChange(Entity member) {
            ((CouchbaseClusterImpl)entity).onServerPoolMemberChanged(member);
        }

        @Override protected void onEntityAdded(Entity member) {
            ((CouchbaseClusterImpl)entity).onServerPoolMemberChanged(member);
        }

        @Override protected void onEntityRemoved(Entity member) {
            ((CouchbaseClusterImpl)entity).onServerPoolMemberChanged(member);
        }
    };

    protected synchronized void onServerPoolMemberChanged(Entity member) {
        if (log.isTraceEnabled()) log.trace("For {}, considering membership of {} which is in locations {}",
                new Object[]{this, member, member.getLocations()});

        //FIXME: make use of servers to be added after cluster initialization.
        synchronized (mutex) {
            if (belongsInServerPool(member)) {

                Optional<Set<Entity>> upNodes = Optional.fromNullable(getUpNodes());
                if (upNodes.isPresent()) {

                    if (!upNodes.get().contains(member)) {
                        Set<Entity> newNodes = Sets.newHashSet(getUpNodes());
                        newNodes.add(member);
                        sensors().set(COUCHBASE_CLUSTER_UP_NODES, newNodes);

                        //add to set of servers to be added.
                        if (isClusterInitialized()) {
                            addServer(member);
                        }
                    }
                } else {
                    Set<Entity> newNodes = Sets.newHashSet();
                    newNodes.add(member);
                    sensors().set(COUCHBASE_CLUSTER_UP_NODES, newNodes);

                    if (isClusterInitialized()) {
                        addServer(member);
                    }
                }
            } else {
                Set<Entity> upNodes = getUpNodes();
                if (upNodes != null && upNodes.contains(member)) {
                    upNodes.remove(member);
                    sensors().set(COUCHBASE_CLUSTER_UP_NODES, upNodes);
                    log.info("Removing couchbase node {}: {}; from cluster", new Object[]{this, member});
                }
            }
            if (log.isTraceEnabled()) log.trace("Done {} checkEntity {}", this, member);
        }
    }

    protected boolean belongsInServerPool(Entity member) {
        if (!groovyTruth(member.getAttribute(Startable.SERVICE_UP))) {
            if (log.isTraceEnabled()) log.trace("Members of {}, checking {}, eliminating because not up", this, member);
            return false;
        }
        if (!getMembers().contains(member)) {
            if (log.isTraceEnabled())
                log.trace("Members of {}, checking {}, eliminating because not member", this, member);

            return false;
        }
        if (log.isTraceEnabled()) log.trace("Members of {}, checking {}, approving", this, member);

        return true;
    }


    @Override
    protected EntitySpec<?> getMemberSpec() {
        EntitySpec<?> result = super.getMemberSpec();
        if (result != null) return result;
        return EntitySpec.create(CouchbaseNode.class);
    }

    @Override
    public int getQuorumSize() {
        Integer quorumSize = getConfig(CouchbaseCluster.INITIAL_QUORUM_SIZE);
        if (quorumSize != null && quorumSize > 0)
            return quorumSize;
        // by default the quorum would be floor(initial_cluster_size/2) + 1
        return (int) Math.floor(getConfig(INITIAL_SIZE) / 2) + 1;
    }

    protected int getActualSize() {
        return Optional.fromNullable(getAttribute(CouchbaseCluster.ACTUAL_CLUSTER_SIZE)).or(-1);
    }

    private Set<Entity> getUpNodes() {
        return getAttribute(COUCHBASE_CLUSTER_UP_NODES);
    }

    private CouchbaseNode getPrimaryNode() {
        return (CouchbaseNode) getAttribute(COUCHBASE_PRIMARY_NODE);
    }

    @Override
    protected void initEnrichers() {
        enrichers().add(Enrichers.builder().updatingMap(ServiceStateLogic.SERVICE_NOT_UP_INDICATORS)
            .from(COUCHBASE_CLUSTER_UP_NODES)
            .computing(new Function<Set<Entity>, Object>() {
                @Override
                public Object apply(Set<Entity> input) {
                    if (input==null) return "Couchbase up nodes not set";
                    if (input.isEmpty()) return "No Couchbase up nodes";
                    if (input.size() < getQuorumSize()) return "Couchbase up nodes not quorate";
                    return null;
                }
            }).build());
        
        if (config().getLocalRaw(UP_QUORUM_CHECK).isAbsent()) {
            // TODO Only leaving CouchbaseQuorumCheck here in case it is contained in persisted state.
            // If so, need a transformer and then to delete it
            @SuppressWarnings({ "unused", "hiding" })
            @Deprecated
            class CouchbaseQuorumCheck implements QuorumCheck {
                @Override
                public boolean isQuorate(int sizeHealthy, int totalSize) {
                    // check members count passed in AND the sensor  
                    if (sizeHealthy < getQuorumSize()) return false;
                    return true;
                }
            }
            config().set(UP_QUORUM_CHECK, new CouchbaseClusterImpl.CouchbaseQuorumCheck(this));
        }
        super.initEnrichers();
    }
    
    static class CouchbaseQuorumCheck implements QuorumCheck {
        private final CouchbaseCluster cluster;
        CouchbaseQuorumCheck(CouchbaseCluster cluster) {
            this.cluster = cluster;
        }
        @Override
        public boolean isQuorate(int sizeHealthy, int totalSize) {
            // check members count passed in AND the sensor  
            if (sizeHealthy < cluster.getQuorumSize()) return false;
            return true;
        }
    }
    protected void addServers(Set<Entity> serversToAdd) {
        Preconditions.checkNotNull(serversToAdd);
        for (Entity s : serversToAdd) {
            addServerSeveralTimes(s, 12, Duration.TEN_SECONDS);
        }
    }

    /** try adding in a loop because we are seeing spurious port failures in AWS */
    protected void addServerSeveralTimes(Entity s, int numRetries, Duration delayOnFailure) {
        try {
            addServer(s);
        } catch (Exception e) {
            Exceptions.propagateIfFatal(e);
            if (numRetries<=0) throw Exceptions.propagate(e);
            // retry once after sleep because we are getting some odd primary-change events
            log.warn("Error adding "+s+" to "+this+", "+numRetries+" retries remaining, will retry after delay ("+e+")");
            Time.sleep(delayOnFailure);
            addServerSeveralTimes(s, numRetries-1, delayOnFailure);
        }
    }

    protected void addServer(Entity serverToAdd) {
        Preconditions.checkNotNull(serverToAdd);
        if (serverToAdd.equals(getPrimaryNode())) {
            // no need to add; but we pass it in anyway because it makes the calling logic easier
            return;
        }
        if (!isMemberInCluster(serverToAdd)) {
            HostAndPort webAdmin = HostAndPort.fromParts(serverToAdd.getAttribute(SoftwareProcess.SUBNET_HOSTNAME),
                    serverToAdd.getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT));
            String username = serverToAdd.getConfig(CouchbaseNode.COUCHBASE_ADMIN_USERNAME);
            String password = serverToAdd.getConfig(CouchbaseNode.COUCHBASE_ADMIN_PASSWORD);

            if (isClusterInitialized()) {
                Entities.invokeEffectorWithArgs(this, getPrimaryNode(), CouchbaseNode.SERVER_ADD_AND_REBALANCE, webAdmin.toString(), username, password).getUnchecked();
            } else {
                Entities.invokeEffectorWithArgs(this, getPrimaryNode(), CouchbaseNode.SERVER_ADD, webAdmin.toString(), username, password).getUnchecked();
            }
            //FIXME check feedback of whether the server was added.
            ((EntityInternal) serverToAdd).sensors().set(CouchbaseNode.IS_IN_CLUSTER, true);
        }
    }

    /** finds the cluster name specified for a node or a cluster, 
     * using {@link CouchbaseCluster#CLUSTER_NAME} or falling back to the cluster (or node) ID. */
    public static String getClusterName(Entity node) {
        String name = node.getConfig(CLUSTER_NAME);
        if (!Strings.isBlank(name)) return Strings.makeValidFilename(name);
        return getClusterOrNode(node).getId();
    }
    
    /** returns Couchbase cluster in ancestry, defaulting to the given node if none */
    @Nonnull public static Entity getClusterOrNode(Entity node) {
        Iterable<CouchbaseCluster> clusterNodes = Iterables.filter(Entities.ancestors(node), CouchbaseCluster.class);
        return Iterables.getFirst(clusterNodes, node);
    }
    
    public boolean isClusterInitialized() {
        return Optional.fromNullable(getAttribute(IS_CLUSTER_INITIALIZED)).or(false);
    }

    public boolean isMemberInCluster(Entity e) {
        return Optional.fromNullable(e.getAttribute(CouchbaseNode.IS_IN_CLUSTER)).or(false);
    }
    
    public void createBuckets() {
        //TODO: check for port conflicts if buckets are being created with a port
        List<Map<String, Object>> bucketsToCreate = getConfig(CREATE_BUCKETS);
        if (bucketsToCreate==null) return;
        
        Entity primaryNode = getPrimaryNode();

        for (Map<String, Object> bucketMap : bucketsToCreate) {
            String bucketName = bucketMap.containsKey("bucket") ? (String) bucketMap.get("bucket") : "default";
            String bucketType = bucketMap.containsKey("bucket-type") ? (String) bucketMap.get("bucket-type") : "couchbase";
            // default bucket must be on this port; other buckets can (must) specify their own (unique) port
            Integer bucketPort = bucketMap.containsKey("bucket-port") ? (Integer) bucketMap.get("bucket-port") : 11211;
            Integer bucketRamSize = bucketMap.containsKey("bucket-ramsize") ? (Integer) bucketMap.get("bucket-ramsize") : 100;
            Integer bucketReplica = bucketMap.containsKey("bucket-replica") ? (Integer) bucketMap.get("bucket-replica") : 1;

            createBucket(primaryNode, bucketName, bucketType, bucketPort, bucketRamSize, bucketReplica);
        }
    }

    public void createBucket(final Entity primaryNode, final String bucketName, final String bucketType, final Integer bucketPort, final Integer bucketRamSize, final Integer bucketReplica) {
        DynamicTasks.queueIfPossible(TaskBuilder.<Void>builder().displayName("Creating bucket " + bucketName).body(
                new Callable<Void>() {
                    @Override
                    public Void call() throws Exception {
                        DependentConfiguration.waitInTaskForAttributeReady(CouchbaseClusterImpl.this, CouchbaseCluster.BUCKET_CREATION_IN_PROGRESS, Predicates.equalTo(false));
                        if (CouchbaseClusterImpl.this.resetBucketCreation.get() != null) {
                            CouchbaseClusterImpl.this.resetBucketCreation.get().stop();
                        }
                        sensors().set(CouchbaseCluster.BUCKET_CREATION_IN_PROGRESS, true);
                        HostAndPort hostAndPort = BrooklynAccessUtils.getBrooklynAccessibleAddress(primaryNode, primaryNode.getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT));

                        CouchbaseClusterImpl.this.resetBucketCreation.set(HttpFeed.builder()
                                .entity(CouchbaseClusterImpl.this)
                                .period(500, TimeUnit.MILLISECONDS)
                                .baseUri(String.format("http://%s/pools/default/buckets/%s", hostAndPort, bucketName))
                                .credentials(primaryNode.getConfig(CouchbaseNode.COUCHBASE_ADMIN_USERNAME), primaryNode.getConfig(CouchbaseNode.COUCHBASE_ADMIN_PASSWORD))
                                .poll(new HttpPollConfig<Boolean>(BUCKET_CREATION_IN_PROGRESS)
                                        .onSuccess(Functionals.chain(HttpValueFunctions.jsonContents(), JsonFunctions.walkN("nodes"), new Function<JsonElement, Boolean>() {
                                            @Override
                                            public Boolean apply(JsonElement input) {
                                                // Wait until bucket has been created on all nodes and the couchApiBase element has been published (indicating that the bucket is useable)
                                                JsonArray servers = input.getAsJsonArray();
                                                if (servers.size() != CouchbaseClusterImpl.this.getMembers().size()) {
                                                    return true;
                                                }
                                                for (JsonElement server : servers) {
                                                    Object api = server.getAsJsonObject().get("couchApiBase");
                                                    if (api == null || Strings.isEmpty(String.valueOf(api))) {
                                                        return true;
                                                    }
                                                }
                                                return false;
                                            }
                                        }))
                                        .onFailureOrException(new Function<Object, Boolean>() {
                                            @Override
                                            public Boolean apply(Object input) {
                                                if (input instanceof HttpToolResponse) {
                                                    if (((HttpToolResponse) input).getResponseCode() == 404) {
                                                        return true;
                                                    }
                                                }
                                                if (input instanceof Throwable)
                                                    Exceptions.propagate((Throwable) input);
                                                throw new IllegalStateException("Unexpected response when creating bucket:" + input);
                                            }
                                        }))
                                .build());

                        // TODO: Bail out if bucket creation fails, to allow next bucket to proceed
                        Entities.invokeEffectorWithArgs(CouchbaseClusterImpl.this, primaryNode, CouchbaseNode.BUCKET_CREATE, bucketName, bucketType, bucketPort, bucketRamSize, bucketReplica);
                        DependentConfiguration.waitInTaskForAttributeReady(CouchbaseClusterImpl.this, CouchbaseCluster.BUCKET_CREATION_IN_PROGRESS, Predicates.equalTo(false));
                        if (CouchbaseClusterImpl.this.resetBucketCreation.get() != null) {
                            CouchbaseClusterImpl.this.resetBucketCreation.get().stop();
                        }
                        return null;
                    }
                }
        ).build()).orSubmitAndBlock();
    }
    
    static {
        RendererHints.register(COUCH_DOCS_DATA_SIZE_PER_NODE, RendererHints.displayValue(ByteSizeStrings.metric()));
        RendererHints.register(COUCH_DOCS_ACTUAL_DISK_SIZE_PER_NODE, RendererHints.displayValue(ByteSizeStrings.metric()));
        RendererHints.register(MEM_USED_PER_NODE, RendererHints.displayValue(ByteSizeStrings.metric()));
        RendererHints.register(COUCH_VIEWS_ACTUAL_DISK_SIZE_PER_NODE, RendererHints.displayValue(ByteSizeStrings.metric()));
        RendererHints.register(COUCH_VIEWS_DATA_SIZE_PER_NODE, RendererHints.displayValue(ByteSizeStrings.metric()));
    }
}
