You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by al...@apache.org on 2015/08/06 18:32:29 UTC
[15/26] incubator-brooklyn git commit: [BROOKLYN-162] Renaming of the
NoSQL packages
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseClusterImpl.java
----------------------------------------------------------------------
diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseClusterImpl.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseClusterImpl.java
new file mode 100644
index 0000000..f279987
--- /dev/null
+++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseClusterImpl.java
@@ -0,0 +1,597 @@
+/*
+ * 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 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import brooklyn.config.render.RendererHints;
+import brooklyn.enricher.Enrichers;
+import brooklyn.entity.Entity;
+import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.Entities;
+import brooklyn.entity.basic.EntityInternal;
+import brooklyn.entity.basic.ServiceStateLogic;
+import brooklyn.entity.basic.SoftwareProcess;
+import brooklyn.entity.effector.Effectors;
+import brooklyn.entity.group.AbstractMembershipTrackingPolicy;
+import brooklyn.entity.group.DynamicClusterImpl;
+import brooklyn.entity.proxying.EntitySpec;
+import brooklyn.entity.trait.Startable;
+import brooklyn.event.AttributeSensor;
+import brooklyn.event.basic.DependentConfiguration;
+import brooklyn.event.feed.http.HttpFeed;
+import brooklyn.event.feed.http.HttpPollConfig;
+import brooklyn.event.feed.http.HttpValueFunctions;
+import brooklyn.event.feed.http.JsonFunctions;
+import brooklyn.location.access.BrooklynAccessUtils;
+import brooklyn.policy.PolicySpec;
+import brooklyn.util.collections.CollectionFunctionals;
+import brooklyn.util.collections.MutableSet;
+import brooklyn.util.collections.QuorumCheck;
+import brooklyn.util.exceptions.Exceptions;
+import brooklyn.util.guava.Functionals;
+import brooklyn.util.guava.IfFunctions;
+import brooklyn.util.math.MathPredicates;
+import brooklyn.util.task.DynamicTasks;
+import brooklyn.util.task.TaskBuilder;
+import brooklyn.util.task.Tasks;
+import brooklyn.util.text.ByteSizeStrings;
+import brooklyn.util.text.StringFunctions;
+import brooklyn.util.text.Strings;
+import brooklyn.util.time.Duration;
+import 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>();
+
+ public void init() {
+ log.info("Initializing the Couchbase cluster...");
+ super.init();
+
+ addEnricher(
+ Enrichers.builder()
+ .transforming(COUCHBASE_CLUSTER_UP_NODES)
+ .from(this)
+ .publishing(COUCHBASE_CLUSTER_UP_NODE_ADDRESSES)
+ .computing(new ListOfHostAndPort()).build() );
+ addEnricher(
+ 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));
+ }
+
+ addEnricher(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) {
+ addEnricher(Enrichers.builder()
+ .aggregating(fromSensor)
+ .publishing(toSensor)
+ .fromMembers()
+ .computingAverage()
+ .build()
+ );
+ }
+
+ private void addSummingMemberEnricher(AttributeSensor<? extends Number> source) {
+ addEnricher(Enrichers.builder()
+ .aggregating(source)
+ .publishing(source)
+ .fromMembers()
+ .computingSum()
+ .build()
+ );
+ }
+
+ @Override
+ protected void doStart() {
+ setAttribute(IS_CLUSTER_INITIALIZED, false);
+
+ super.doStart();
+
+ connectSensors();
+
+ setAttribute(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).setAttribute(CouchbaseNode.IS_PRIMARY_NODE, true);
+ setAttribute(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();
+ }
+
+ ((CouchbaseNode)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).setAttribute(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();
+ }
+ }
+
+ setAttribute(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() {
+ addPolicy(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();
+ 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);
+ setAttribute(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);
+ setAttribute(COUCHBASE_CLUSTER_UP_NODES, newNodes);
+
+ if (isClusterInitialized()) {
+ addServer(member);
+ }
+ }
+ } else {
+ Set<Entity> upNodes = getUpNodes();
+ if (upNodes != null && upNodes.contains(member)) {
+ upNodes.remove(member);
+ setAttribute(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;
+ }
+
+
+ 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() {
+ addEnricher(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).setAttribute(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().name("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();
+ }
+ setAttribute(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 brooklyn.util.http.HttpToolResponse) {
+ if (((brooklyn.util.http.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()));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNode.java
----------------------------------------------------------------------
diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNode.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNode.java
new file mode 100644
index 0000000..16434fa
--- /dev/null
+++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNode.java
@@ -0,0 +1,159 @@
+/*
+ * 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 java.net.URI;
+
+import org.apache.brooklyn.catalog.Catalog;
+import brooklyn.config.ConfigKey;
+import brooklyn.config.render.RendererHints;
+import brooklyn.entity.annotation.Effector;
+import brooklyn.entity.annotation.EffectorParam;
+import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.ConfigKeys;
+import brooklyn.entity.basic.MethodEffector;
+import brooklyn.entity.basic.SoftwareProcess;
+import brooklyn.entity.effector.Effectors;
+import brooklyn.entity.proxying.ImplementedBy;
+import brooklyn.event.AttributeSensor;
+import brooklyn.event.basic.BasicAttributeSensorAndConfigKey;
+import brooklyn.event.basic.PortAttributeSensorAndConfigKey;
+import brooklyn.event.basic.Sensors;
+import brooklyn.util.flags.SetFromFlag;
+import brooklyn.util.text.ByteSizeStrings;
+
+@Catalog(name="CouchBase Node", description="Couchbase Server is an open source, distributed (shared-nothing architecture) "
+ + "NoSQL document-oriented database that is optimized for interactive applications.")
+@ImplementedBy(CouchbaseNodeImpl.class)
+public interface CouchbaseNode extends SoftwareProcess {
+
+ @SetFromFlag("adminUsername")
+ ConfigKey<String> COUCHBASE_ADMIN_USERNAME = ConfigKeys.newStringConfigKey("couchbase.adminUsername", "Username for the admin user on the node", "Administrator");
+
+ @SetFromFlag("adminPassword")
+ ConfigKey<String> COUCHBASE_ADMIN_PASSWORD = ConfigKeys.newStringConfigKey("couchbase.adminPassword", "Password for the admin user on the node", "Password");
+
+ @SetFromFlag("version")
+ ConfigKey<String> SUGGESTED_VERSION = ConfigKeys.newConfigKeyWithDefault(SoftwareProcess.SUGGESTED_VERSION,
+ "3.0.0");
+
+ @SetFromFlag("enterprise")
+ ConfigKey<Boolean> USE_ENTERPRISE = ConfigKeys.newBooleanConfigKey("couchbase.enterprise.enabled",
+ "Whether to use Couchbase Enterprise; if false uses the community version. Defaults to true.", true);
+
+ @SetFromFlag("downloadUrl")
+ BasicAttributeSensorAndConfigKey<String> DOWNLOAD_URL = new BasicAttributeSensorAndConfigKey<String>(
+ SoftwareProcess.DOWNLOAD_URL, "http://packages.couchbase.com/releases/${version}/"
+ + "couchbase-server-${driver.communityOrEnterprise}${driver.downloadLinkPreVersionSeparator}${version}${driver.downloadLinkOsTagWithPrefix}");
+
+ @SetFromFlag("clusterInitRamSize")
+ BasicAttributeSensorAndConfigKey<Integer> COUCHBASE_CLUSTER_INIT_RAM_SIZE = new BasicAttributeSensorAndConfigKey<Integer>(
+ Integer.class, "couchbase.clusterInitRamSize", "initial ram size of the cluster", 300);
+
+ PortAttributeSensorAndConfigKey COUCHBASE_WEB_ADMIN_PORT = new PortAttributeSensorAndConfigKey("couchbase.webAdminPort", "Web Administration Port", "8091+");
+ PortAttributeSensorAndConfigKey COUCHBASE_API_PORT = new PortAttributeSensorAndConfigKey("couchbase.apiPort", "Couchbase API Port", "8092+");
+ PortAttributeSensorAndConfigKey COUCHBASE_INTERNAL_BUCKET_PORT = new PortAttributeSensorAndConfigKey("couchbase.internalBucketPort", "Internal Bucket Port", "11209");
+ PortAttributeSensorAndConfigKey COUCHBASE_INTERNAL_EXTERNAL_BUCKET_PORT = new PortAttributeSensorAndConfigKey("couchbase.internalExternalBucketPort", "Internal/External Bucket Port", "11210");
+ PortAttributeSensorAndConfigKey COUCHBASE_CLIENT_INTERFACE_PROXY = new PortAttributeSensorAndConfigKey("couchbase.clientInterfaceProxy", "Client interface (proxy)", "11211");
+ PortAttributeSensorAndConfigKey COUCHBASE_INCOMING_SSL_PROXY = new PortAttributeSensorAndConfigKey("couchbase.incomingSslProxy", "Incoming SSL Proxy", "11214");
+ PortAttributeSensorAndConfigKey COUCHBASE_INTERNAL_OUTGOING_SSL_PROXY = new PortAttributeSensorAndConfigKey("couchbase.internalOutgoingSslProxy", "Internal Outgoing SSL Proxy", "11215");
+ PortAttributeSensorAndConfigKey COUCHBASE_REST_HTTPS_FOR_SSL = new PortAttributeSensorAndConfigKey("couchbase.internalRestHttpsForSsl", "Internal REST HTTPS for SSL", "18091");
+ PortAttributeSensorAndConfigKey COUCHBASE_CAPI_HTTPS_FOR_SSL = new PortAttributeSensorAndConfigKey("couchbase.internalCapiHttpsForSsl", "Internal CAPI HTTPS for SSL", "18092");
+ PortAttributeSensorAndConfigKey ERLANG_PORT_MAPPER = new PortAttributeSensorAndConfigKey("couchbase.erlangPortMapper", "Erlang Port Mapper Daemon Listener Port (epmd)", "4369");
+ PortAttributeSensorAndConfigKey NODE_DATA_EXCHANGE_PORT_RANGE_START = new PortAttributeSensorAndConfigKey("couchbase.nodeDataExchangePortRangeStart", "Node data exchange Port Range Start", "21100+");
+ PortAttributeSensorAndConfigKey NODE_DATA_EXCHANGE_PORT_RANGE_END = new PortAttributeSensorAndConfigKey("couchbase.nodeDataExchangePortRangeEnd", "Node data exchange Port Range End", "21199+");
+
+ AttributeSensor<Boolean> IS_PRIMARY_NODE = Sensors.newBooleanSensor("couchbase.isPrimaryNode", "flag to determine if the current couchbase node is the primary node for the cluster");
+ AttributeSensor<Boolean> IS_IN_CLUSTER = Sensors.newBooleanSensor("couchbase.isInCluster", "flag to determine if the current couchbase node has been added to a cluster, "
+ + "including being the first / primary node");
+ AttributeSensor<URI> COUCHBASE_WEB_ADMIN_URL = Attributes.MAIN_URI;
+
+ // Interesting stats
+ AttributeSensor<Double> OPS = Sensors.newDoubleSensor("couchbase.stats.ops",
+ "Retrieved from pools/nodes/<current node>/interestingStats/ops");
+ AttributeSensor<Long> COUCH_DOCS_DATA_SIZE = Sensors.newLongSensor("couchbase.stats.couch.docs.data.size",
+ "Retrieved from pools/nodes/<current node>/interestingStats/couch_docs_data_size");
+ AttributeSensor<Long> COUCH_DOCS_ACTUAL_DISK_SIZE = Sensors.newLongSensor("couchbase.stats.couch.docs.actual.disk.size",
+ "Retrieved from pools/nodes/<current node>/interestingStats/couch_docs_actual_disk_size");
+ AttributeSensor<Long> EP_BG_FETCHED = Sensors.newLongSensor("couchbase.stats.ep.bg.fetched",
+ "Retrieved from pools/nodes/<current node>/interestingStats/ep_bg_fetched");
+ AttributeSensor<Long> MEM_USED = Sensors.newLongSensor("couchbase.stats.mem.used",
+ "Retrieved from pools/nodes/<current node>/interestingStats/mem_used");
+ AttributeSensor<Long> COUCH_VIEWS_ACTUAL_DISK_SIZE = Sensors.newLongSensor("couchbase.stats.couch.views.actual.disk.size",
+ "Retrieved from pools/nodes/<current node>/interestingStats/couch_views_actual_disk_size");
+ AttributeSensor<Long> CURR_ITEMS = Sensors.newLongSensor("couchbase.stats.curr.items",
+ "Retrieved from pools/nodes/<current node>/interestingStats/curr_items");
+ AttributeSensor<Long> VB_REPLICA_CURR_ITEMS = Sensors.newLongSensor("couchbase.stats.vb.replica.curr.items",
+ "Retrieved from pools/nodes/<current node>/interestingStats/vb_replica_curr_items");
+ AttributeSensor<Long> COUCH_VIEWS_DATA_SIZE = Sensors.newLongSensor("couchbase.stats.couch.views.data.size",
+ "Retrieved from pools/nodes/<current node>/interestingStats/couch_views_data_size");
+ AttributeSensor<Long> GET_HITS = Sensors.newLongSensor("couchbase.stats.get.hits",
+ "Retrieved from pools/nodes/<current node>/interestingStats/get_hits");
+ AttributeSensor<Double> CMD_GET = Sensors.newDoubleSensor("couchbase.stats.cmd.get",
+ "Retrieved from pools/nodes/<current node>/interestingStats/cmd_get");
+ AttributeSensor<Long> CURR_ITEMS_TOT = Sensors.newLongSensor("couchbase.stats.curr.items.tot",
+ "Retrieved from pools/nodes/<current node>/interestingStats/curr_items_tot");
+ AttributeSensor<String> REBALANCE_STATUS = Sensors.newStringSensor("couchbase.rebalance.status",
+ "Displays the current rebalance status from pools/nodes/rebalanceStatus");
+
+ class MainUri {
+ public static final AttributeSensor<URI> MAIN_URI = Attributes.MAIN_URI;
+
+ static {
+ // ROOT_URL does not need init because it refers to something already initialized
+ RendererHints.register(COUCHBASE_WEB_ADMIN_URL, RendererHints.namedActionWithUrl());
+
+ RendererHints.register(COUCH_DOCS_DATA_SIZE, RendererHints.displayValue(ByteSizeStrings.metric()));
+ RendererHints.register(COUCH_DOCS_ACTUAL_DISK_SIZE, RendererHints.displayValue(ByteSizeStrings.metric()));
+ RendererHints.register(MEM_USED, RendererHints.displayValue(ByteSizeStrings.metric()));
+ RendererHints.register(COUCH_VIEWS_ACTUAL_DISK_SIZE, RendererHints.displayValue(ByteSizeStrings.metric()));
+ RendererHints.register(COUCH_VIEWS_DATA_SIZE, RendererHints.displayValue(ByteSizeStrings.metric()));
+ }
+ }
+
+ // this long-winded reference is done just to trigger the initialization above
+ AttributeSensor<URI> MAIN_URI = MainUri.MAIN_URI;
+
+ MethodEffector<Void> SERVER_ADD = new MethodEffector<Void>(CouchbaseNode.class, "serverAdd");
+ MethodEffector<Void> SERVER_ADD_AND_REBALANCE = new MethodEffector<Void>(CouchbaseNode.class, "serverAddAndRebalance");
+ MethodEffector<Void> REBALANCE = new MethodEffector<Void>(CouchbaseNode.class, "rebalance");
+ MethodEffector<Void> BUCKET_CREATE = new MethodEffector<Void>(CouchbaseNode.class, "bucketCreate");
+ brooklyn.entity.Effector<Void> ADD_REPLICATION_RULE = Effectors.effector(Void.class, "addReplicationRule")
+ .description("Adds a replication rule from the indicated bucket on the cluster where this node is located "
+ + "to the indicated cluster and optional destination bucket")
+ .parameter(String.class, "fromBucket", "Bucket to be replicated")
+ .parameter(Object.class, "toCluster", "Entity (or ID) of the cluster to which this should replicate")
+ .parameter(String.class, "toBucket", "Destination bucket for replication in the toCluster, defaulting to the same as the fromBucket")
+ .buildAbstract();
+
+ @Effector(description = "add a server to a cluster")
+ public void serverAdd(@EffectorParam(name = "serverHostname") String serverToAdd, @EffectorParam(name = "username") String username, @EffectorParam(name = "password") String password);
+
+ @Effector(description = "add a server to a cluster, and immediately rebalances")
+ public void serverAddAndRebalance(@EffectorParam(name = "serverHostname") String serverToAdd, @EffectorParam(name = "username") String username, @EffectorParam(name = "password") String password);
+
+ @Effector(description = "rebalance the couchbase cluster")
+ public void rebalance();
+
+ @Effector(description = "create a new bucket")
+ public void bucketCreate(@EffectorParam(name = "bucketName") String bucketName, @EffectorParam(name = "bucketType") String bucketType,
+ @EffectorParam(name = "bucketPort") Integer bucketPort, @EffectorParam(name = "bucketRamSize") Integer bucketRamSize,
+ @EffectorParam(name = "bucketReplica") Integer bucketReplica);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeDriver.java
----------------------------------------------------------------------
diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeDriver.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeDriver.java
new file mode 100644
index 0000000..1ff28fd
--- /dev/null
+++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeDriver.java
@@ -0,0 +1,41 @@
+/*
+ * 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 brooklyn.entity.Entity;
+import brooklyn.entity.basic.SoftwareProcessDriver;
+
+public interface CouchbaseNodeDriver extends SoftwareProcessDriver {
+ public String getOsTag();
+ public String getDownloadLinkPreVersionSeparator();
+ public String getDownloadLinkOsTagWithPrefix();
+
+ public String getCommunityOrEnterprise();
+
+ public void serverAdd(String serverToAdd, String username, String password);
+
+ public void rebalance();
+
+ public void bucketCreate(String bucketName, String bucketType, Integer bucketPort, Integer bucketRamSize, Integer bucketReplica);
+
+ public void serverAddAndRebalance(String serverToAdd, String username, String password);
+
+ public void addReplicationRule(Entity toCluster, String fromBucket, String toBucket);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeImpl.java
----------------------------------------------------------------------
diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeImpl.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeImpl.java
new file mode 100644
index 0000000..a0654db
--- /dev/null
+++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeImpl.java
@@ -0,0 +1,269 @@
+/*
+ * 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 java.lang.String.format;
+
+import java.net.URI;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import brooklyn.entity.Entity;
+import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.SoftwareProcessImpl;
+import brooklyn.entity.effector.EffectorBody;
+import brooklyn.event.AttributeSensor;
+import brooklyn.event.SensorEvent;
+import brooklyn.event.SensorEventListener;
+import brooklyn.event.feed.http.HttpFeed;
+import brooklyn.event.feed.http.HttpPollConfig;
+import brooklyn.event.feed.http.HttpValueFunctions;
+import brooklyn.event.feed.http.JsonFunctions;
+import brooklyn.location.MachineProvisioningLocation;
+import brooklyn.location.access.BrooklynAccessUtils;
+import brooklyn.location.cloud.CloudLocationConfig;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.collections.MutableSet;
+import brooklyn.util.config.ConfigBag;
+import brooklyn.util.exceptions.Exceptions;
+import brooklyn.util.guava.Functionals;
+import brooklyn.util.guava.MaybeFunctions;
+import brooklyn.util.guava.TypeTokens;
+import brooklyn.util.http.HttpTool;
+import brooklyn.util.http.HttpToolResponse;
+import brooklyn.util.net.Urls;
+import brooklyn.util.task.Tasks;
+import brooklyn.util.text.Strings;
+import brooklyn.util.time.Duration;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Function;
+import com.google.common.base.Functions;
+import com.google.common.base.Preconditions;
+import com.google.common.net.HostAndPort;
+import com.google.common.net.HttpHeaders;
+import com.google.common.net.MediaType;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+
+public class CouchbaseNodeImpl extends SoftwareProcessImpl implements CouchbaseNode {
+
+ private static final Logger log = LoggerFactory.getLogger(CouchbaseNodeImpl.class);
+
+ private volatile HttpFeed httpFeed;
+
+ @Override
+ public Class<CouchbaseNodeDriver> getDriverInterface() {
+ return CouchbaseNodeDriver.class;
+ }
+
+ @Override
+ public CouchbaseNodeDriver getDriver() {
+ return (CouchbaseNodeDriver) super.getDriver();
+ }
+
+ @Override
+ public void init() {
+ super.init();
+
+ subscribe(this, Attributes.SERVICE_UP, new SensorEventListener<Boolean>() {
+ @Override
+ public void onEvent(SensorEvent<Boolean> booleanSensorEvent) {
+ if (Boolean.TRUE.equals(booleanSensorEvent.getValue())) {
+ Integer webPort = getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT);
+ Preconditions.checkNotNull(webPort, CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT+" not set for %s; is an acceptable port available?", this);
+ String hostAndPort = BrooklynAccessUtils.getBrooklynAccessibleAddress(CouchbaseNodeImpl.this, webPort).toString();
+ setAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_URL, URI.create(format("http://%s", hostAndPort)));
+ }
+ }
+ });
+
+ getMutableEntityType().addEffector(ADD_REPLICATION_RULE, new EffectorBody<Void>() {
+ @Override
+ public Void call(ConfigBag parameters) {
+ addReplicationRule(parameters);
+ return null;
+ }
+ });
+ }
+
+ protected Map<String, Object> obtainProvisioningFlags(@SuppressWarnings("rawtypes") MachineProvisioningLocation location) {
+ ConfigBag result = ConfigBag.newInstance(super.obtainProvisioningFlags(location));
+ result.configure(CloudLocationConfig.OS_64_BIT, true);
+ return result.getAllConfig();
+ }
+
+ @Override
+ protected Collection<Integer> getRequiredOpenPorts() {
+ // TODO this creates a huge list of inbound ports; much better to define on a security group using range syntax!
+ int erlangRangeStart = getConfig(NODE_DATA_EXCHANGE_PORT_RANGE_START).iterator().next();
+ int erlangRangeEnd = getConfig(NODE_DATA_EXCHANGE_PORT_RANGE_END).iterator().next();
+
+ Set<Integer> newPorts = MutableSet.<Integer>copyOf(super.getRequiredOpenPorts());
+ newPorts.remove(erlangRangeStart);
+ newPorts.remove(erlangRangeEnd);
+ for (int i = erlangRangeStart; i <= erlangRangeEnd; i++)
+ newPorts.add(i);
+ return newPorts;
+ }
+
+ @Override
+ public void serverAdd(String serverToAdd, String username, String password) {
+ getDriver().serverAdd(serverToAdd, username, password);
+ }
+
+ @Override
+ public void serverAddAndRebalance(String serverToAdd, String username, String password) {
+ getDriver().serverAddAndRebalance(serverToAdd, username, password);
+ }
+
+ @Override
+ public void rebalance() {
+ getDriver().rebalance();
+ }
+
+ protected final static Function<HttpToolResponse, JsonElement> GET_THIS_NODE_STATS = Functionals.chain(
+ HttpValueFunctions.jsonContents(),
+ JsonFunctions.walk("nodes"),
+ new Function<JsonElement, JsonElement>() {
+ @Override public JsonElement apply(JsonElement input) {
+ JsonArray nodes = input.getAsJsonArray();
+ for (JsonElement element : nodes) {
+ JsonElement thisNode = element.getAsJsonObject().get("thisNode");
+ if (thisNode!=null && Boolean.TRUE.equals(thisNode.getAsBoolean())) {
+ return element.getAsJsonObject().get("interestingStats");
+ }
+ }
+ return null;
+ }}
+ );
+
+ protected final static <T> HttpPollConfig<T> getSensorFromNodeStat(AttributeSensor<T> sensor, String ...jsonPath) {
+ return new HttpPollConfig<T>(sensor)
+ .onSuccess(Functionals.chain(GET_THIS_NODE_STATS,
+ MaybeFunctions.<JsonElement>wrap(),
+ JsonFunctions.walkM(jsonPath),
+ JsonFunctions.castM(TypeTokens.getRawRawType(sensor.getTypeToken()), null)))
+ .onFailureOrException(Functions.<T>constant(null));
+ }
+
+ @Override
+ protected void postStart() {
+ super.postStart();
+ renameServerToPublicHostname();
+ }
+
+ protected void renameServerToPublicHostname() {
+ // http://docs.couchbase.com/couchbase-manual-2.5/cb-install/#couchbase-getting-started-hostnames
+ URI apiUri = null;
+ try {
+ HostAndPort accessible = BrooklynAccessUtils.getBrooklynAccessibleAddress(this, getAttribute(COUCHBASE_WEB_ADMIN_PORT));
+ apiUri = URI.create(String.format("http://%s:%d/node/controller/rename", accessible.getHostText(), accessible.getPort()));
+ UsernamePasswordCredentials credentials = new UsernamePasswordCredentials(getConfig(COUCHBASE_ADMIN_USERNAME), getConfig(COUCHBASE_ADMIN_PASSWORD));
+ HttpToolResponse response = HttpTool.httpPost(
+ // the uri is required by the HttpClientBuilder in order to set the AuthScope of the credentials
+ HttpTool.httpClientBuilder().uri(apiUri).credentials(credentials).build(),
+ apiUri,
+ MutableMap.of(
+ HttpHeaders.CONTENT_TYPE, MediaType.FORM_DATA.toString(),
+ HttpHeaders.ACCEPT, "*/*",
+ // this appears needed; without it we get org.apache.http.NoHttpResponseException !?
+ HttpHeaders.AUTHORIZATION, HttpTool.toBasicAuthorizationValue(credentials)),
+ Charsets.UTF_8.encode("hostname="+Urls.encode(accessible.getHostText())).array());
+ log.debug("Renamed Couchbase server "+this+" via "+apiUri+": "+response);
+ if (!HttpTool.isStatusCodeHealthy(response.getResponseCode())) {
+ log.warn("Invalid response code, renaming "+apiUri+": "+response);
+ }
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ log.warn("Error renaming server, using "+apiUri+": "+e, e);
+ }
+ }
+
+ public void connectSensors() {
+ super.connectSensors();
+ connectServiceUpIsRunning();
+
+ HostAndPort hostAndPort = BrooklynAccessUtils.getBrooklynAccessibleAddress(this, this.getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT));
+ httpFeed = HttpFeed.builder()
+ .entity(this)
+ .period(Duration.seconds(3))
+ .baseUri("http://" + hostAndPort + "/pools/nodes/")
+ .credentialsIfNotNull(getConfig(CouchbaseNode.COUCHBASE_ADMIN_USERNAME), getConfig(CouchbaseNode.COUCHBASE_ADMIN_PASSWORD))
+ .poll(getSensorFromNodeStat(CouchbaseNode.OPS, "ops"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.COUCH_DOCS_DATA_SIZE, "couch_docs_data_size"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.COUCH_DOCS_ACTUAL_DISK_SIZE, "couch_docs_actual_disk_size"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.EP_BG_FETCHED, "ep_bg_fetched"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.MEM_USED, "mem_used"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.COUCH_VIEWS_ACTUAL_DISK_SIZE, "couch_views_actual_disk_size"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.CURR_ITEMS, "curr_items"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.VB_REPLICA_CURR_ITEMS, "vb_replica_curr_items"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.COUCH_VIEWS_DATA_SIZE, "couch_views_data_size"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.GET_HITS, "get_hits"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.CMD_GET, "cmd_get"))
+ .poll(getSensorFromNodeStat(CouchbaseNode.CURR_ITEMS_TOT, "curr_items_tot"))
+ .poll(new HttpPollConfig<String>(CouchbaseNode.REBALANCE_STATUS)
+ .onSuccess(HttpValueFunctions.jsonContents("rebalanceStatus", String.class))
+ .onFailureOrException(Functions.constant("Could not retrieve")))
+ .build();
+ }
+
+ public void disconnectSensors() {
+ super.disconnectSensors();
+ disconnectServiceUpIsRunning();
+ if (httpFeed != null) {
+ httpFeed.stop();
+ }
+ }
+
+ @Override
+ public void bucketCreate(String bucketName, String bucketType, Integer bucketPort, Integer bucketRamSize, Integer bucketReplica) {
+ if (Strings.isBlank(bucketType)) bucketType = "couchbase";
+ if (bucketRamSize==null || bucketRamSize<=0) bucketRamSize = 200;
+ if (bucketReplica==null || bucketReplica<0) bucketReplica = 1;
+
+ getDriver().bucketCreate(bucketName, bucketType, bucketPort, bucketRamSize, bucketReplica);
+ }
+
+ /** exposed through {@link CouchbaseNode#ADD_REPLICATION_RULE} */
+ protected void addReplicationRule(ConfigBag ruleArgs) {
+ Object toClusterO = Preconditions.checkNotNull(ruleArgs.getStringKey("toCluster"), "toCluster must not be null");
+ if (toClusterO instanceof String) {
+ toClusterO = getManagementContext().lookup((String)toClusterO);
+ }
+ Entity toCluster = Tasks.resolving(toClusterO, Entity.class).context(getExecutionContext()).get();
+
+ String fromBucket = Preconditions.checkNotNull( (String)ruleArgs.getStringKey("fromBucket"), "fromBucket must be specified" );
+
+ String toBucket = (String)ruleArgs.getStringKey("toBucket");
+ if (toBucket==null) toBucket = fromBucket;
+
+ if (!ruleArgs.getUnusedConfig().isEmpty()) {
+ throw new IllegalArgumentException("Unsupported replication rule data: "+ruleArgs.getUnusedConfig());
+ }
+
+ getDriver().addReplicationRule(toCluster, fromBucket, toBucket);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeSshDriver.java
----------------------------------------------------------------------
diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeSshDriver.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeSshDriver.java
new file mode 100644
index 0000000..73ed934
--- /dev/null
+++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseNodeSshDriver.java
@@ -0,0 +1,512 @@
+/*
+ * 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 brooklyn.util.ssh.BashCommands.*;
+import static java.lang.String.format;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.http.auth.UsernamePasswordCredentials;
+
+import brooklyn.entity.Entity;
+import brooklyn.entity.Group;
+import brooklyn.entity.basic.AbstractSoftwareProcessSshDriver;
+import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.Entities;
+import brooklyn.entity.basic.ServiceStateLogic;
+import brooklyn.entity.drivers.downloads.BasicDownloadRequirement;
+import brooklyn.entity.drivers.downloads.DownloadProducerFromUrlAttribute;
+import brooklyn.entity.software.SshEffectorTasks;
+import brooklyn.event.basic.DependentConfiguration;
+import brooklyn.event.feed.http.HttpValueFunctions;
+import brooklyn.location.OsDetails;
+import brooklyn.location.access.BrooklynAccessUtils;
+import brooklyn.location.basic.SshMachineLocation;
+import brooklyn.management.Task;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.http.HttpTool;
+import brooklyn.util.http.HttpToolResponse;
+import brooklyn.util.repeat.Repeater;
+import brooklyn.util.ssh.BashCommands;
+import brooklyn.util.task.DynamicTasks;
+import brooklyn.util.task.TaskBuilder;
+import brooklyn.util.task.TaskTags;
+import brooklyn.util.task.Tasks;
+import brooklyn.util.text.NaturalOrderComparator;
+import brooklyn.util.text.StringEscapes.BashStringEscapes;
+import brooklyn.util.text.Strings;
+import brooklyn.util.time.Duration;
+
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.net.HostAndPort;
+
+public class CouchbaseNodeSshDriver extends AbstractSoftwareProcessSshDriver implements CouchbaseNodeDriver {
+
+ public CouchbaseNodeSshDriver(final CouchbaseNodeImpl entity, final SshMachineLocation machine) {
+ super(entity, machine);
+ }
+
+ public static String couchbaseCli(String cmd) {
+ return "/opt/couchbase/bin/couchbase-cli " + cmd + " ";
+ }
+
+ @Override
+ public void preInstall() {
+ resolver = Entities.newDownloader(this);
+ setExpandedInstallDir(getInstallDir());
+ }
+
+ @Override
+ public void install() {
+ //for reference https://github.com/urbandecoder/couchbase/blob/master/recipes/server.rb
+ //installation instructions (http://docs.couchbase.com/couchbase-manual-2.5/cb-install/#preparing-to-install)
+
+ List<String> urls = resolver.getTargets();
+ String saveAs = resolver.getFilename();
+
+ OsDetails osDetails = getMachine().getMachineDetails().getOsDetails();
+
+ if (osDetails.isLinux()) {
+ List<String> commands = installLinux(urls, saveAs);
+ //FIXME installation return error but the server is up and running.
+ newScript(INSTALLING)
+ .body.append(commands).execute();
+ } else {
+ Tasks.markInessential();
+ throw new IllegalStateException("Unsupported OS for installing Couchbase. Will continue but may fail later.");
+ }
+ }
+
+ private List<String> installLinux(List<String> urls, String saveAs) {
+
+ log.info("Installing " + getEntity() + " using couchbase-server-{} {}", getCommunityOrEnterprise(), getVersion());
+
+ String apt = chainGroup(
+ installPackage(MutableMap.of("apt", "python-httplib2 libssl0.9.8"), null),
+ sudo(format("dpkg -i %s", saveAs)));
+
+ String yum = chainGroup(
+ "which yum",
+ // The following prevents failure on RHEL AWS nodes:
+ // https://forums.aws.amazon.com/thread.jspa?threadID=100509
+ ok(sudo("sed -i.bk s/^enabled=1$/enabled=0/ /etc/yum/pluginconf.d/subscription-manager.conf")),
+ ok(sudo("yum check-update")),
+ sudo("yum install -y pkgconfig"),
+ // RHEL requires openssl version 098
+ sudo("[ -f /etc/redhat-release ] && (grep -i \"red hat\" /etc/redhat-release && sudo yum install -y openssl098e) || :"),
+ sudo(format("rpm --install %s", saveAs)));
+
+ String link = new DownloadProducerFromUrlAttribute().apply(new BasicDownloadRequirement(this)).getPrimaryLocations().iterator().next();
+ return ImmutableList.<String>builder()
+ .add(INSTALL_CURL)
+ .addAll(Arrays.asList(INSTALL_CURL,
+ BashCommands.require(BashCommands.alternatives(BashCommands.simpleDownloadUrlAs(urls, saveAs),
+ // Referer link is required for 3.0.0; note mis-spelling is correct, as per http://en.wikipedia.org/wiki/HTTP_referer
+ "curl -f -L -k " + BashStringEscapes.wrapBash(link)
+ + " -H 'Referer: http://www.couchbase.com/downloads'"
+ + " -o " + saveAs),
+ "Could not retrieve " + saveAs + " (from " + urls.size() + " sites)", 9)))
+ .add(alternatives(apt, yum))
+ .build();
+ }
+
+ @Override
+ public void customize() {
+ //TODO: add linux tweaks for couchbase
+ //http://blog.couchbase.com/often-overlooked-linux-os-tweaks
+ //http://blog.couchbase.com/kirk
+
+ //turn off swappiness
+ //vm.swappiness=0
+ //sudo echo 0 > /proc/sys/vm/swappiness
+
+ //os page cache = 20%
+
+ //disable THP
+ //sudo echo never > /sys/kernel/mm/transparent_hugepage/enabled
+ //sudo echo never > /sys/kernel/mm/transparent_hugepage/defrag
+
+ //turn off transparent huge pages
+ //limit page cache disty bytes
+ //control the rate page cache is flused ... vm.dirty_*
+ }
+
+ @Override
+ public void launch() {
+ String clusterPrefix = "--cluster-" + (isPreV3() ? "init-" : "");
+ // in v30, the cluster arguments were changed, and it became mandatory to supply a url + password (if there is none, these are ignored)
+ newScript(LAUNCHING)
+ .body.append(
+ sudo("/etc/init.d/couchbase-server start"),
+ "for i in {0..120}\n" +
+ "do\n" +
+ " if [ $i -eq 120 ]; then echo REST API unavailable after 120 seconds, failing; exit 1; fi;\n" +
+ " curl -s " + String.format("http://localhost:%s", getWebPort()) + " > /dev/null && echo REST API available after $i seconds && break\n" +
+ " sleep 1\n" +
+ "done\n" +
+ couchbaseCli("cluster-init") +
+ (isPreV3() ? getCouchbaseHostnameAndPort() : getCouchbaseHostnameAndCredentials()) +
+ " " + clusterPrefix + "username=" + getUsername() +
+ " " + clusterPrefix + "password=" + getPassword() +
+ " " + clusterPrefix + "port=" + getWebPort() +
+ " " + clusterPrefix + "ramsize=" + getClusterInitRamSize())
+ .execute();
+ }
+
+ @Override
+ public boolean isRunning() {
+ //TODO add a better way to check if couchbase server is running
+ return (newScript(CHECK_RUNNING)
+ .body.append(format("curl -u %s:%s http://localhost:%s/pools/nodes", getUsername(), getPassword(), getWebPort()))
+ .execute() == 0);
+ }
+
+ @Override
+ public void stop() {
+ newScript(STOPPING)
+ .body.append(sudo("/etc/init.d/couchbase-server stop"))
+ .execute();
+ }
+
+ @Override
+ public String getVersion() {
+ return entity.getConfig(CouchbaseNode.SUGGESTED_VERSION);
+ }
+
+ @Override
+ public String getOsTag() {
+ return newDownloadLinkSegmentComputer().getOsTag();
+ }
+
+ protected DownloadLinkSegmentComputer newDownloadLinkSegmentComputer() {
+ return new DownloadLinkSegmentComputer(getLocation().getOsDetails(), !isPreV3(), Strings.toString(getEntity()));
+ }
+
+ public static class DownloadLinkSegmentComputer {
+ // links are:
+ // http://packages.couchbase.com/releases/2.2.0/couchbase-server-community_2.2.0_x86_64.rpm
+ // http://packages.couchbase.com/releases/2.2.0/couchbase-server-community_2.2.0_x86_64.deb
+ // ^^^ preV3 is _ everywhere
+ // http://packages.couchbase.com/releases/3.0.0/couchbase-server-community_3.0.0-ubuntu12.04_amd64.deb
+ // ^^^ most V3 is _${version}-
+ // http://packages.couchbase.com/releases/3.0.0/couchbase-server-community-3.0.0-centos6.x86_64.rpm
+ // ^^^ but RHEL is -${version}-
+
+ @Nullable
+ private final OsDetails os;
+ @Nonnull
+ private final boolean isV3OrLater;
+ @Nonnull
+ private final String context;
+ @Nonnull
+ private final String osName;
+ @Nonnull
+ private final boolean isRpm;
+ @Nonnull
+ private final boolean is64bit;
+
+ public DownloadLinkSegmentComputer(@Nullable OsDetails os, boolean isV3OrLater, @Nonnull String context) {
+ this.os = os;
+ this.isV3OrLater = isV3OrLater;
+ this.context = context;
+ if (os == null) {
+ // guess centos as RPM is sensible default
+ log.warn("No details known for OS of " + context + "; assuming 64-bit RPM distribution of Couchbase");
+ osName = "centos";
+ isRpm = true;
+ is64bit = true;
+ return;
+ }
+ osName = os.getName().toLowerCase();
+ isRpm = !(osName.contains("deb") || osName.contains("ubuntu"));
+ is64bit = os.is64bit();
+ }
+
+ /**
+ * separator after the version number used to be _ but is - in 3.0 and later
+ */
+ public String getPreVersionSeparator() {
+ if (!isV3OrLater) return "_";
+ if (isRpm) return "-";
+ return "_";
+ }
+
+ public String getOsTag() {
+ // couchbase only provide certain versions; if on other platforms let's suck-it-and-see
+ String family;
+ if (osName.contains("debian")) family = "debian7_";
+ else if (osName.contains("ubuntu")) family = "ubuntu12.04_";
+ else if (osName.contains("centos") || osName.contains("rhel") || (osName.contains("red") && osName.contains("hat")))
+ family = "centos6.";
+ else {
+ log.warn("Unrecognised OS " + os + " of " + context + "; assuming RPM distribution of Couchbase");
+ family = "centos6.";
+ }
+
+ if (!is64bit && !isV3OrLater) {
+ // NB: 32-bit binaries aren't (yet?) available for v30
+ log.warn("32-bit binaries for Couchbase might not be available, when deploying " + context);
+ }
+ String arch = !is64bit ? "x86" : !isRpm && isV3OrLater ? "amd64" : "x86_64";
+ String fileExtension = isRpm ? ".rpm" : ".deb";
+
+ if (isV3OrLater)
+ return family + arch + fileExtension;
+ else
+ return arch + fileExtension;
+ }
+
+ public String getOsTagWithPrefix() {
+ return (!isV3OrLater ? "_" : "-") + getOsTag();
+ }
+ }
+
+ @Override
+ public String getDownloadLinkOsTagWithPrefix() {
+ return newDownloadLinkSegmentComputer().getOsTagWithPrefix();
+ }
+
+ @Override
+ public String getDownloadLinkPreVersionSeparator() {
+ return newDownloadLinkSegmentComputer().getPreVersionSeparator();
+ }
+
+ private boolean isPreV3() {
+ return NaturalOrderComparator.INSTANCE.compare(getEntity().getConfig(CouchbaseNode.SUGGESTED_VERSION), "3.0") < 0;
+ }
+
+ @Override
+ public String getCommunityOrEnterprise() {
+ Boolean isEnterprise = getEntity().getConfig(CouchbaseNode.USE_ENTERPRISE);
+ return isEnterprise ? "enterprise" : "community";
+ }
+
+ private String getUsername() {
+ return entity.getConfig(CouchbaseNode.COUCHBASE_ADMIN_USERNAME);
+ }
+
+ private String getPassword() {
+ return entity.getConfig(CouchbaseNode.COUCHBASE_ADMIN_PASSWORD);
+ }
+
+ private String getWebPort() {
+ return "" + entity.getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT);
+ }
+
+ private String getCouchbaseHostnameAndCredentials() {
+ return format("-c %s:%s -u %s -p %s", getSubnetHostname(), getWebPort(), getUsername(), getPassword());
+ }
+
+ private String getCouchbaseHostnameAndPort() {
+ return format("-c %s:%s", getSubnetHostname(), getWebPort());
+ }
+
+ private String getClusterInitRamSize() {
+ return entity.getConfig(CouchbaseNode.COUCHBASE_CLUSTER_INIT_RAM_SIZE).toString();
+ }
+
+ @Override
+ public void rebalance() {
+ entity.setAttribute(CouchbaseNode.REBALANCE_STATUS, "explicitly started");
+ newScript("rebalance")
+ .body.append(
+ couchbaseCli("rebalance") + getCouchbaseHostnameAndCredentials())
+ .failOnNonZeroResultCode()
+ .execute();
+
+ // wait until the re-balance is started
+ // (if it's quick, this might miss it, but it will only block for 30s if so)
+ Repeater.create()
+ .backoff(Repeater.DEFAULT_REAL_QUICK_PERIOD, 2, Duration.millis(500))
+ .limitTimeTo(Duration.THIRTY_SECONDS)
+ .until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ for (HostAndPort nodeHostAndPort : getNodesHostAndPort()) {
+ if (isNodeRebalancing(nodeHostAndPort.toString())) {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+ ).run();
+
+ entity.setAttribute(CouchbaseNode.REBALANCE_STATUS, "waiting for completion");
+ // Wait until the Couchbase node finishes the re-balancing
+ Task<Boolean> reBalance = TaskBuilder.<Boolean>builder()
+ .name("Waiting until node is rebalancing")
+ .body(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ return Repeater.create()
+ .backoff(Duration.ONE_SECOND, 1.2, Duration.TEN_SECONDS)
+ .limitTimeTo(Duration.FIVE_MINUTES)
+ .until(new Callable<Boolean>() {
+ @Override
+ public Boolean call() throws Exception {
+ for (HostAndPort nodeHostAndPort : getNodesHostAndPort()) {
+ if (isNodeRebalancing(nodeHostAndPort.toString())) {
+ return false;
+ }
+ }
+ return true;
+ }
+ })
+ .run();
+ }
+ })
+ .build();
+ Boolean completed = DynamicTasks.queueIfPossible(reBalance)
+ .orSubmitAndBlock()
+ .andWaitForSuccess();
+ if (completed) {
+ entity.setAttribute(CouchbaseNode.REBALANCE_STATUS, "completed");
+ ServiceStateLogic.ServiceNotUpLogic.clearNotUpIndicator(getEntity(), "rebalancing");
+ log.info("Rebalanced cluster via primary node {}", getEntity());
+ } else {
+ entity.setAttribute(CouchbaseNode.REBALANCE_STATUS, "timed out");
+ ServiceStateLogic.ServiceNotUpLogic.updateNotUpIndicator(getEntity(), "rebalancing", "rebalance did not complete within time limit");
+ log.warn("Timeout rebalancing cluster via primary node {}", getEntity());
+ }
+ }
+
+ private Iterable<HostAndPort> getNodesHostAndPort() {
+ Group group = Iterables.getFirst(getEntity().getGroups(), null);
+ if (group == null) return Lists.newArrayList();
+ return Iterables.transform(group.getAttribute(CouchbaseCluster.COUCHBASE_CLUSTER_UP_NODES),
+ new Function<Entity, HostAndPort>() {
+ @Override
+ public HostAndPort apply(Entity input) {
+ return BrooklynAccessUtils.getBrooklynAccessibleAddress(input, input.getAttribute(CouchbaseNode.COUCHBASE_WEB_ADMIN_PORT));
+ }
+ });
+ }
+
+ private boolean isNodeRebalancing(String nodeHostAndPort) {
+ HttpToolResponse response = getApiResponse("http://" + nodeHostAndPort + "/pools/default/rebalanceProgress");
+ if (response.getResponseCode() != 200) {
+ throw new IllegalStateException("failed retrieving rebalance status: " + response);
+ }
+ return !"none".equals(HttpValueFunctions.jsonContents("status", String.class).apply(response));
+ }
+
+ private HttpToolResponse getApiResponse(String uri) {
+ return HttpTool.httpGet(HttpTool.httpClientBuilder()
+ // the uri is required by the HttpClientBuilder in order to set the AuthScope of the credentials
+ .uri(uri)
+ .credentials(new UsernamePasswordCredentials(getUsername(), getPassword()))
+ .build(),
+ URI.create(uri),
+ ImmutableMap.<String, String>of());
+ }
+
+ @Override
+ public void serverAdd(String serverToAdd, String username, String password) {
+ newScript("serverAdd").body.append(couchbaseCli("server-add")
+ + getCouchbaseHostnameAndCredentials() +
+ " --server-add=" + BashStringEscapes.wrapBash(serverToAdd) +
+ " --server-add-username=" + BashStringEscapes.wrapBash(username) +
+ " --server-add-password=" + BashStringEscapes.wrapBash(password))
+ .failOnNonZeroResultCode()
+ .execute();
+ }
+
+ @Override
+ public void serverAddAndRebalance(String serverToAdd, String username, String password) {
+ newScript("serverAddAndRebalance").body.append(couchbaseCli("rebalance")
+ + getCouchbaseHostnameAndCredentials() +
+ " --server-add=" + BashStringEscapes.wrapBash(serverToAdd) +
+ " --server-add-username=" + BashStringEscapes.wrapBash(username) +
+ " --server-add-password=" + BashStringEscapes.wrapBash(password))
+ .failOnNonZeroResultCode()
+ .execute();
+ entity.setAttribute(CouchbaseNode.REBALANCE_STATUS, "triggered as part of server-add");
+ }
+
+ @Override
+ public void bucketCreate(String bucketName, String bucketType, Integer bucketPort, Integer bucketRamSize, Integer bucketReplica) {
+ log.info("Adding bucket: {} to cluster {} primary node: {}", new Object[]{bucketName, CouchbaseClusterImpl.getClusterOrNode(getEntity()), getEntity()});
+
+ newScript("bucketCreate").body.append(couchbaseCli("bucket-create")
+ + getCouchbaseHostnameAndCredentials() +
+ " --bucket=" + BashStringEscapes.wrapBash(bucketName) +
+ " --bucket-type=" + BashStringEscapes.wrapBash(bucketType) +
+ " --bucket-port=" + bucketPort +
+ " --bucket-ramsize=" + bucketRamSize +
+ " --bucket-replica=" + bucketReplica)
+ .failOnNonZeroResultCode()
+ .execute();
+ }
+
+ @Override
+ public void addReplicationRule(Entity toCluster, String fromBucket, String toBucket) {
+ DynamicTasks.queue(DependentConfiguration.attributeWhenReady(toCluster, Attributes.SERVICE_UP)).getUnchecked();
+
+ String destName = CouchbaseClusterImpl.getClusterName(toCluster);
+
+ log.info("Setting up XDCR for " + fromBucket + " from " + CouchbaseClusterImpl.getClusterName(getEntity()) + " (via " + getEntity() + ") "
+ + "to " + destName + " (" + toCluster + ")");
+
+ Entity destPrimaryNode = toCluster.getAttribute(CouchbaseCluster.COUCHBASE_PRIMARY_NODE);
+ String destHostname = destPrimaryNode.getAttribute(Attributes.HOSTNAME);
+ String destUsername = toCluster.getConfig(CouchbaseNode.COUCHBASE_ADMIN_USERNAME);
+ String destPassword = toCluster.getConfig(CouchbaseNode.COUCHBASE_ADMIN_PASSWORD);
+
+ // on the REST API there is mention of a 'type' 'continuous' but i don't see other refs to this
+
+ // PROTOCOL Select REST protocol or memcached for replication. xmem indicates memcached while capi indicates REST protocol.
+ // looks like xmem is the default; leave off for now
+// String replMode = "xmem";
+
+ DynamicTasks.queue(TaskTags.markInessential(SshEffectorTasks.ssh(
+ couchbaseCli("xdcr-setup") +
+ getCouchbaseHostnameAndCredentials() +
+ " --create" +
+ " --xdcr-cluster-name=" + BashStringEscapes.wrapBash(destName) +
+ " --xdcr-hostname=" + BashStringEscapes.wrapBash(destHostname) +
+ " --xdcr-username=" + BashStringEscapes.wrapBash(destUsername) +
+ " --xdcr-password=" + BashStringEscapes.wrapBash(destPassword)
+ ).summary("create xdcr destination " + destName).newTask()));
+
+ // would be nice to auto-create bucket, but we'll need to know the parameters; the port in particular is tedious
+// ((CouchbaseNode)destPrimaryNode).bucketCreate(toBucket, "couchbase", null, 0, 0);
+
+ DynamicTasks.queue(SshEffectorTasks.ssh(
+ couchbaseCli("xdcr-replicate") +
+ getCouchbaseHostnameAndCredentials() +
+ " --create" +
+ " --xdcr-cluster-name=" + BashStringEscapes.wrapBash(destName) +
+ " --xdcr-from-bucket=" + BashStringEscapes.wrapBash(fromBucket) +
+ " --xdcr-to-bucket=" + BashStringEscapes.wrapBash(toBucket)
+// + " --xdcr-replication-mode="+replMode
+ ).summary("configure replication for " + fromBucket + " to " + destName + ":" + toBucket).newTask());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseSyncGateway.java
----------------------------------------------------------------------
diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseSyncGateway.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseSyncGateway.java
new file mode 100644
index 0000000..384434c
--- /dev/null
+++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/couchbase/CouchbaseSyncGateway.java
@@ -0,0 +1,75 @@
+/*
+ * 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 brooklyn.config.ConfigKey;
+import brooklyn.entity.Entity;
+import brooklyn.entity.basic.ConfigKeys;
+import brooklyn.entity.basic.SoftwareProcess;
+import brooklyn.entity.proxying.ImplementedBy;
+import brooklyn.event.AttributeSensor;
+import brooklyn.event.basic.BasicAttributeSensorAndConfigKey;
+import brooklyn.event.basic.PortAttributeSensorAndConfigKey;
+import brooklyn.event.basic.Sensors;
+import brooklyn.util.flags.SetFromFlag;
+
+@ImplementedBy(CouchbaseSyncGatewayImpl.class)
+public interface CouchbaseSyncGateway extends SoftwareProcess {
+
+ @SetFromFlag("version")
+ ConfigKey<String> SUGGESTED_VERSION = ConfigKeys.newConfigKeyWithDefault(SoftwareProcess.SUGGESTED_VERSION,
+ "1.0-beta3.1");
+
+ @SetFromFlag("downloadUrl")
+ BasicAttributeSensorAndConfigKey<String> DOWNLOAD_URL = new BasicAttributeSensorAndConfigKey<String>(
+ SoftwareProcess.DOWNLOAD_URL, "http://packages.couchbase.com/releases/couchbase-sync-gateway/1.0-beta/couchbase-sync-gateway-community_${version}_${driver.osTag}");
+
+ @SetFromFlag("couchbaseServer")
+ ConfigKey<Entity> COUCHBASE_SERVER = ConfigKeys.newConfigKey(Entity.class, "couchbaseSyncGateway.couchbaseNode",
+ "Couchbase server node or cluster the sync gateway connects to");
+
+ @SetFromFlag("serverPool")
+ ConfigKey<String> COUCHBASE_SERVER_POOL = ConfigKeys.newStringConfigKey("couchbaseSyncGateway.serverPool",
+ "Couchbase Server pool name in which to find buckets", "default");
+
+ @SetFromFlag("couchbaseServerBucket")
+ ConfigKey<String> COUCHBASE_SERVER_BUCKET = ConfigKeys.newStringConfigKey("couchbaseSyncGateway.serverBucket",
+ "Name of the Couchbase bucket to use", "sync_gateway");
+
+ @SetFromFlag("pretty")
+ ConfigKey<Boolean> PRETTY = ConfigKeys.newBooleanConfigKey("couchbaseSyncGateway.pretty",
+ "Pretty-print JSON responses. This is useful for debugging, but reduces performance.", false);
+
+ @SetFromFlag("verbose")
+ ConfigKey<Boolean> VERBOSE = ConfigKeys.newBooleanConfigKey("couchbaseSyncGateway.verbose",
+ "Logs more information about requests.", false);
+
+ AttributeSensor<String> COUCHBASE_SERVER_WEB_URL = Sensors.newStringSensor("couchbaseSyncGateway.serverWebUrl",
+ "The Url and web port of the couchbase server to connect to");
+
+ AttributeSensor<String> MANAGEMENT_URL = Sensors.newStringSensor("coucbaseSyncGateway.managementUrl",
+ "Management URL for Couchbase Sycn Gateway");
+
+ PortAttributeSensorAndConfigKey SYNC_REST_API_PORT = new PortAttributeSensorAndConfigKey("couchbaseSyncGateway.syncRestPort",
+ "Port the Sync REST API listens on", "4984");
+
+ PortAttributeSensorAndConfigKey ADMIN_REST_API_PORT = new PortAttributeSensorAndConfigKey("couchbaseSyncGateway.adminRestPort",
+ "Port the Admin REST API listens on", "4985");
+
+}
\ No newline at end of file