You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by he...@apache.org on 2016/02/01 18:47:47 UTC
[11/51] [abbrv] [partial] brooklyn-library git commit: move subdir
from incubator up a level as it is promoted to its own repo (first
non-incubator commit!)
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardClusterImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardClusterImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardClusterImpl.java
deleted file mode 100644
index e0c0f35..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardClusterImpl.java
+++ /dev/null
@@ -1,182 +0,0 @@
-/*
- * 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.mongodb.sharding;
-
-import java.net.UnknownHostException;
-import java.util.Collection;
-import java.util.Set;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.EntitySpec;
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.api.sensor.SensorEvent;
-import org.apache.brooklyn.api.sensor.SensorEventListener;
-import org.apache.brooklyn.core.entity.trait.Startable;
-import org.apache.brooklyn.entity.group.DynamicClusterImpl;
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBClientSupport;
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBReplicaSet;
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBServer;
-import org.apache.brooklyn.util.exceptions.Exceptions;
-import org.apache.brooklyn.util.time.Duration;
-import org.apache.brooklyn.util.time.Time;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Stopwatch;
-import com.google.common.collect.Sets;
-
-public class MongoDBShardClusterImpl extends DynamicClusterImpl implements MongoDBShardCluster {
-
- private static final Logger LOG = LoggerFactory.getLogger(MongoDBShardClusterImpl.class);
-
- // TODO: Need to use attributes for this in order to support brooklyn restart
- private Set<Entity> addedMembers = Sets.newConcurrentHashSet();
-
- // TODO: Need to use attributes for this in order to support brooklyn restart
- private Set<Entity> addingMembers = Sets.newConcurrentHashSet();
-
- /**
- * For shard addition and removal.
- * Used for retrying.
- *
- * TODO Should use ExecutionManager.
- */
- private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-
- @Override
- protected EntitySpec<?> getMemberSpec() {
- EntitySpec<?> result = super.getMemberSpec();
- if (result == null)
- result = EntitySpec.create(MongoDBReplicaSet.class);
- result.configure(DynamicClusterImpl.INITIAL_SIZE, getConfig(MongoDBShardedDeployment.SHARD_REPLICASET_SIZE));
- return result;
- }
-
- @Override
- public void start(Collection<? extends Location> locations) {
- subscriptions().subscribeToMembers(this, Startable.SERVICE_UP, new SensorEventListener<Boolean>() {
- public void onEvent(SensorEvent<Boolean> event) {
- addShards();
- }
- });
-
- super.start(locations);
-
- MongoDBRouterCluster routers = getParent().getAttribute(MongoDBShardedDeployment.ROUTER_CLUSTER);
- subscriptions().subscribe(routers, MongoDBRouterCluster.ANY_RUNNING_ROUTER, new SensorEventListener<MongoDBRouter>() {
- public void onEvent(SensorEvent<MongoDBRouter> event) {
- if (event.getValue() != null)
- addShards();
- }
- });
- }
-
- @Override
- public void stop() {
- // TODO Note that after this the executor will not run if the set is restarted.
- executor.shutdownNow();
- super.stop();
- }
-
- @Override
- public void onManagementStopped() {
- super.onManagementStopped();
- executor.shutdownNow();
- }
-
- protected void addShards() {
- MongoDBRouter router = getParent().getAttribute(MongoDBShardedDeployment.ROUTER_CLUSTER).getAttribute(MongoDBRouterCluster.ANY_RUNNING_ROUTER);
- if (router == null) {
- if (LOG.isTraceEnabled()) LOG.trace("Not adding shards because no running router in {}", this);
- return;
- }
-
- for (Entity member : this.getMembers()) {
- if (member.getAttribute(Startable.SERVICE_UP) && !addingMembers.contains(member)) {
- LOG.info("{} adding shard {}", new Object[] {MongoDBShardClusterImpl.this, member});
- addingMembers.add(member);
- addShardAsync(member);
- }
- }
- }
-
- protected void addShardAsync(final Entity replicaSet) {
- final Duration timeout = Duration.minutes(20);
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final AtomicInteger attempts = new AtomicInteger();
-
- // TODO Don't use executor, use ExecutionManager; but following pattern in MongoDBReplicaSetImpl for now.
- executor.submit(new Runnable() {
- @Override
- public void run() {
- boolean reschedule;
- MongoDBRouter router = getParent().getAttribute(MongoDBShardedDeployment.ROUTER_CLUSTER).getAttribute(MongoDBRouterCluster.ANY_RUNNING_ROUTER);
- if (router == null) {
- LOG.debug("Rescheduling adding shard {} because no running router for cluster {}", replicaSet, this);
- reschedule = true;
- } else {
- MongoDBClientSupport client;
- try {
- client = MongoDBClientSupport.forServer(router);
- } catch (UnknownHostException e) {
- throw Exceptions.propagate(e);
- }
-
- try {
- MongoDBServer primary = replicaSet.getAttribute(MongoDBReplicaSet.PRIMARY_ENTITY);
- if (primary != null) {
- String addr = String.format("%s:%d", primary.getAttribute(MongoDBServer.SUBNET_HOSTNAME), primary.getAttribute(MongoDBServer.PORT));
- String replicaSetURL = ((MongoDBReplicaSet) replicaSet).getName() + "/" + addr;
- boolean added = client.addShardToRouter(replicaSetURL);
- if (added) {
- LOG.info("{} added shard {} via {}", new Object[]{MongoDBShardClusterImpl.this, replicaSetURL, router});
- addedMembers.add(replicaSet);
- reschedule = false;
- } else {
- LOG.debug("Rescheduling addition of shard {} because add failed via router {}", replicaSetURL, router);
- reschedule = true;
- }
- } else {
- LOG.debug("Rescheduling addition of shard {} because primary is null", replicaSet);
- reschedule = true;
- }
- } catch (Exception e) {
- LOG.error("Failed to add shard to router {}: ", router, e);
- throw Exceptions.propagate(e);
- }
- }
-
- if (reschedule) {
- int numAttempts = attempts.incrementAndGet();
- if (numAttempts > 1 && timeout.toMilliseconds() > stopwatch.elapsed(TimeUnit.MILLISECONDS)) {
- executor.schedule(this, 3, TimeUnit.SECONDS);
- } else {
- LOG.warn("Timeout after {} attempts ({}) adding shard {}; aborting",
- new Object[] {numAttempts, Time.makeTimeStringRounded(stopwatch), replicaSet});
- addingMembers.remove(replicaSet);
- }
- }
- }
- });
- }
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeployment.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeployment.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeployment.java
deleted file mode 100644
index d48313c..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeployment.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.mongodb.sharding;
-
-import org.apache.brooklyn.api.catalog.Catalog;
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.EntitySpec;
-import org.apache.brooklyn.api.entity.Group;
-import org.apache.brooklyn.api.entity.ImplementedBy;
-import org.apache.brooklyn.api.sensor.AttributeSensor;
-import org.apache.brooklyn.config.ConfigKey;
-import org.apache.brooklyn.core.config.ConfigKeys;
-import org.apache.brooklyn.core.entity.trait.Startable;
-import org.apache.brooklyn.core.sensor.Sensors;
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBReplicaSet;
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBServer;
-import org.apache.brooklyn.util.core.flags.SetFromFlag;
-import org.apache.brooklyn.util.time.Duration;
-
-import com.google.common.reflect.TypeToken;
-
-@Catalog(name="MongoDB Sharded Deployment",
- description="MongoDB (from \"humongous\") is a scalable, high-performance, open source NoSQL database",
- iconUrl="classpath:///mongodb-logo.png")
-@ImplementedBy(MongoDBShardedDeploymentImpl.class)
-public interface MongoDBShardedDeployment extends Entity, Startable {
- @SetFromFlag("configClusterSize")
- ConfigKey<Integer> CONFIG_CLUSTER_SIZE = ConfigKeys.newIntegerConfigKey("mongodb.config.cluster.size",
- "Number of config servers", 3);
-
- @SetFromFlag("initialRouterClusterSize")
- ConfigKey<Integer> INITIAL_ROUTER_CLUSTER_SIZE = ConfigKeys.newIntegerConfigKey("mongodb.router.cluster.initial.size",
- "Initial number of routers (mongos)", 0);
-
- @SetFromFlag("initialShardClusterSize")
- ConfigKey<Integer> INITIAL_SHARD_CLUSTER_SIZE = ConfigKeys.newIntegerConfigKey("mongodb.shard.cluster.initial.size",
- "Initial number of shards (replicasets)", 2);
-
- @SetFromFlag("shardReplicaSetSize")
- ConfigKey<Integer> SHARD_REPLICASET_SIZE = ConfigKeys.newIntegerConfigKey("mongodb.shard.replicaset.size",
- "Number of servers (mongod) in each shard (replicaset)", 3);
-
- @SetFromFlag("routerUpTimeout")
- ConfigKey<Duration> ROUTER_UP_TIMEOUT = ConfigKeys.newConfigKey(Duration.class, "mongodb.router.up.timeout",
- "Maximum time to wait for the routers to become available before adding the shards", Duration.FIVE_MINUTES);
-
- @SetFromFlag("coLocatedRouterGroup")
- ConfigKey<Group> CO_LOCATED_ROUTER_GROUP = ConfigKeys.newConfigKey(Group.class, "mongodb.colocated.router.group",
- "Group to be monitored for the addition of new CoLocatedMongoDBRouter entities");
-
- @SuppressWarnings("serial")
- ConfigKey<EntitySpec<?>> MONGODB_ROUTER_SPEC = ConfigKeys.newConfigKey(
- new TypeToken<EntitySpec<?>>() {},
- "mongodb.router.spec",
- "Spec for Router instances",
- EntitySpec.create(MongoDBRouter.class));
-
- @SuppressWarnings("serial")
- ConfigKey<EntitySpec<?>> MONGODB_REPLICA_SET_SPEC = ConfigKeys.newConfigKey(
- new TypeToken<EntitySpec<?>>() {},
- "mongodb.replicaset.spec",
- "Spec for Replica Set",
- EntitySpec.create(MongoDBReplicaSet.class)
- .configure(MongoDBReplicaSet.MEMBER_SPEC, EntitySpec.create(MongoDBServer.class)));
-
- @SuppressWarnings("serial")
- ConfigKey<EntitySpec<?>> MONGODB_CONFIG_SERVER_SPEC = ConfigKeys.newConfigKey(
- new TypeToken<EntitySpec<?>>() {},
- "mongodb.configserver.spec",
- "Spec for Config Server instances",
- EntitySpec.create(MongoDBConfigServer.class));
-
- public static AttributeSensor<MongoDBConfigServerCluster> CONFIG_SERVER_CLUSTER = Sensors.newSensor(
- MongoDBConfigServerCluster.class, "mongodbshardeddeployment.configservers", "Config servers");
- public static AttributeSensor<MongoDBRouterCluster> ROUTER_CLUSTER = Sensors.newSensor(
- MongoDBRouterCluster.class, "mongodbshardeddeployment.routers", "Routers");
-
- public static AttributeSensor<MongoDBShardCluster> SHARD_CLUSTER = Sensors.newSensor(
- MongoDBShardCluster.class, "mongodbshardeddeployment.shards", "Shards");
-
- public MongoDBConfigServerCluster getConfigCluster();
-
- public MongoDBRouterCluster getRouterCluster();
-
- public MongoDBShardCluster getShardCluster();
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeploymentImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeploymentImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeploymentImpl.java
deleted file mode 100644
index 05ab66c..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/mongodb/sharding/MongoDBShardedDeploymentImpl.java
+++ /dev/null
@@ -1,162 +0,0 @@
-/*
- * 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.mongodb.sharding;
-
-import static org.apache.brooklyn.core.sensor.DependentConfiguration.attributeWhenReady;
-
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.EntitySpec;
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.api.policy.PolicySpec;
-import org.apache.brooklyn.core.entity.AbstractEntity;
-import org.apache.brooklyn.core.entity.Attributes;
-import org.apache.brooklyn.core.entity.Entities;
-import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceNotUpLogic;
-import org.apache.brooklyn.core.entity.trait.Startable;
-import org.apache.brooklyn.enricher.stock.Enrichers;
-import org.apache.brooklyn.entity.group.AbstractMembershipTrackingPolicy;
-import org.apache.brooklyn.entity.group.DynamicCluster;
-import org.apache.brooklyn.util.exceptions.Exceptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBAuthenticationMixins;
-import org.apache.brooklyn.entity.nosql.mongodb.MongoDBAuthenticationUtils;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-
-public class MongoDBShardedDeploymentImpl extends AbstractEntity implements MongoDBShardedDeployment, MongoDBAuthenticationMixins {
-
- @SuppressWarnings("unused")
- private static final Logger LOG = LoggerFactory.getLogger(MongoDBShardedDeploymentImpl.class);
-
- @Override
- public void init() {
- super.init();
-
- EntitySpec<MongoDBConfigServerCluster> configServerClusterSpec = EntitySpec.create(MongoDBConfigServerCluster.class)
- .configure(MongoDBConfigServerCluster.MEMBER_SPEC, getConfig(MONGODB_CONFIG_SERVER_SPEC))
- .configure(DynamicCluster.INITIAL_SIZE, getConfig(CONFIG_CLUSTER_SIZE));
- MongoDBAuthenticationUtils.setAuthenticationConfig(configServerClusterSpec, this);
- sensors().set(CONFIG_SERVER_CLUSTER, addChild(configServerClusterSpec));
-
- EntitySpec<MongoDBRouterCluster> routerClusterSpec = EntitySpec.create(MongoDBRouterCluster.class)
- .configure(MongoDBRouterCluster.MEMBER_SPEC, getConfig(MONGODB_ROUTER_SPEC))
- .configure(DynamicCluster.INITIAL_SIZE, getConfig(INITIAL_ROUTER_CLUSTER_SIZE))
- .configure(MongoDBRouter.CONFIG_SERVERS, attributeWhenReady(getAttribute(CONFIG_SERVER_CLUSTER), MongoDBConfigServerCluster.CONFIG_SERVER_ADDRESSES));
- MongoDBAuthenticationUtils.setAuthenticationConfig(routerClusterSpec, this);
- sensors().set(ROUTER_CLUSTER, addChild(routerClusterSpec));
-
- EntitySpec<MongoDBShardCluster> shardClusterSpec = EntitySpec.create(MongoDBShardCluster.class)
- .configure(MongoDBShardCluster.MEMBER_SPEC, getConfig(MONGODB_REPLICA_SET_SPEC))
- .configure(DynamicCluster.INITIAL_SIZE, getConfig(INITIAL_SHARD_CLUSTER_SIZE));
- MongoDBAuthenticationUtils.setAuthenticationConfig(shardClusterSpec, this);
- sensors().set(SHARD_CLUSTER, addChild(shardClusterSpec));
-
- enrichers().add(Enrichers.builder()
- .propagating(MongoDBConfigServerCluster.CONFIG_SERVER_ADDRESSES)
- .from(getAttribute(CONFIG_SERVER_CLUSTER))
- .build());
-
- // Advertise even if default are used (root password is set in MongoDBAuthenticationUtils)
- sensors().set(MongoDBAuthenticationMixins.AUTHENTICATION_DATABASE, config().get(MongoDBAuthenticationMixins.AUTHENTICATION_DATABASE));
- sensors().set(MongoDBAuthenticationMixins.ROOT_USERNAME, config().get(MongoDBAuthenticationMixins.ROOT_USERNAME));
-
- ServiceNotUpLogic.updateNotUpIndicator(this, Attributes.SERVICE_STATE_ACTUAL, "stopped");
- }
-
- @Override
- public void start(Collection<? extends Location> locations) {
- ServiceStateLogic.setExpectedState(this, Lifecycle.STARTING);
- try {
- final MongoDBRouterCluster routers = getAttribute(ROUTER_CLUSTER);
- final MongoDBShardCluster shards = getAttribute(SHARD_CLUSTER);
- List<DynamicCluster> clusters = ImmutableList.of(getAttribute(CONFIG_SERVER_CLUSTER), routers, shards);
- Entities.invokeEffectorList(this, clusters, Startable.START, ImmutableMap.of("locations", locations))
- .get();
-
- if (getConfigRaw(MongoDBShardedDeployment.CO_LOCATED_ROUTER_GROUP, true).isPresent()) {
- policies().add(PolicySpec.create(ColocatedRouterTrackingPolicy.class)
- .displayName("Co-located router tracker")
- .configure("group", getConfig(MongoDBShardedDeployment.CO_LOCATED_ROUTER_GROUP)));
- }
-
- ServiceNotUpLogic.clearNotUpIndicator(this, Attributes.SERVICE_STATE_ACTUAL);
- ServiceStateLogic.setExpectedState(this, Lifecycle.RUNNING);
- } catch (Exception e) {
- ServiceStateLogic.setExpectedState(this, Lifecycle.ON_FIRE);
- // no need to log here; the effector invocation should do that
- throw Exceptions.propagate(e);
- }
- }
-
- public static class ColocatedRouterTrackingPolicy extends AbstractMembershipTrackingPolicy {
- @Override
- protected void onEntityAdded(Entity member) {
- MongoDBAuthenticationUtils.setAuthenticationConfig(member, entity);
- MongoDBRouterCluster cluster = entity.getAttribute(ROUTER_CLUSTER);
- cluster.addMember(member.getAttribute(CoLocatedMongoDBRouter.ROUTER));
- }
- @Override
- protected void onEntityRemoved(Entity member) {
- MongoDBRouterCluster cluster = entity.getAttribute(ROUTER_CLUSTER);
- cluster.removeMember(member.getAttribute(CoLocatedMongoDBRouter.ROUTER));
- }
- }
-
- @Override
- public void stop() {
- ServiceStateLogic.setExpectedState(this, Lifecycle.STOPPING);
- try {
- Entities.invokeEffectorList(this, ImmutableList.of(getAttribute(CONFIG_SERVER_CLUSTER), getAttribute(ROUTER_CLUSTER),
- getAttribute(SHARD_CLUSTER)), Startable.STOP).get();
- } catch (Exception e) {
- ServiceStateLogic.setExpectedState(this, Lifecycle.ON_FIRE);
- throw Exceptions.propagate(e);
- }
- ServiceStateLogic.setExpectedState(this, Lifecycle.STOPPED);
- ServiceNotUpLogic.updateNotUpIndicator(this, Attributes.SERVICE_STATE_ACTUAL, "stopped");
- }
-
- @Override
- public void restart() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public MongoDBConfigServerCluster getConfigCluster() {
- return getAttribute(CONFIG_SERVER_CLUSTER);
- }
-
- @Override
- public MongoDBRouterCluster getRouterCluster() {
- return getAttribute(ROUTER_CLUSTER);
- }
-
- @Override
- public MongoDBShardCluster getShardCluster() {
- return getAttribute(SHARD_CLUSTER);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisCluster.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisCluster.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisCluster.java
deleted file mode 100644
index bc80a96..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisCluster.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.redis;
-
-import org.apache.brooklyn.api.catalog.Catalog;
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.ImplementedBy;
-import org.apache.brooklyn.core.entity.trait.Startable;
-import org.apache.brooklyn.entity.group.DynamicCluster;
-
-/**
- * A cluster of {@link RedisStore}s with one master and a group of slaves.
- *
- * The slaves are contained in a {@link DynamicCluster} which can be resized by a policy if required.
- *
- * TODO add sensors with aggregated Redis statistics from cluster
- */
-@Catalog(name="Redis Cluster", description="Redis is an open-source, networked, in-memory, key-value data store with optional durability", iconUrl="classpath:///redis-logo.png")
-@ImplementedBy(RedisClusterImpl.class)
-public interface RedisCluster extends Entity, Startable {
-
- public RedisStore getMaster();
-
- public DynamicCluster getSlaves();
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisClusterImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisClusterImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisClusterImpl.java
deleted file mode 100644
index 8d847e7..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisClusterImpl.java
+++ /dev/null
@@ -1,158 +0,0 @@
-/*
- * 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.redis;
-
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.brooklyn.api.entity.EntitySpec;
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.api.mgmt.Task;
-import org.apache.brooklyn.api.sensor.AttributeSensor;
-import org.apache.brooklyn.core.entity.AbstractEntity;
-import org.apache.brooklyn.core.entity.Attributes;
-import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ComputeServiceIndicatorsFromChildrenAndMembers;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceProblemsLogic;
-import org.apache.brooklyn.core.sensor.DependentConfiguration;
-import org.apache.brooklyn.core.sensor.Sensors;
-import org.apache.brooklyn.enricher.stock.Enrichers;
-import org.apache.brooklyn.entity.group.DynamicCluster;
-import org.apache.brooklyn.entity.software.base.SoftwareProcess;
-import org.apache.brooklyn.util.collections.QuorumCheck.QuorumChecks;
-import org.apache.brooklyn.util.exceptions.CompoundRuntimeException;
-import org.apache.brooklyn.util.exceptions.Exceptions;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Lists;
-
-public class RedisClusterImpl extends AbstractEntity implements RedisCluster {
-
- private static final AttributeSensor<RedisStore> MASTER = Sensors.newSensor(RedisStore.class, "redis.master");
- private static final AttributeSensor<DynamicCluster> SLAVES = Sensors.newSensor(DynamicCluster.class, "redis.slaves");
-
- public RedisClusterImpl() {
- }
-
- @Override
- public RedisStore getMaster() {
- return getAttribute(MASTER);
- }
-
- @Override
- public DynamicCluster getSlaves() {
- return getAttribute(SLAVES);
- }
-
- @Override
- public void init() {
- super.init();
-
- RedisStore master = addChild(EntitySpec.create(RedisStore.class));
- sensors().set(MASTER, master);
-
- DynamicCluster slaves = addChild(EntitySpec.create(DynamicCluster.class)
- .configure(DynamicCluster.MEMBER_SPEC, EntitySpec.create(RedisSlave.class).configure(RedisSlave.MASTER, master)
- .configure(SoftwareProcess.CUSTOMIZE_LATCH, DependentConfiguration.attributeWhenReady(master, Attributes.SERVICE_UP))));
- sensors().set(SLAVES, slaves);
-
- enrichers().add(Enrichers.builder()
- .propagating(RedisStore.HOSTNAME, RedisStore.ADDRESS, RedisStore.SUBNET_HOSTNAME, RedisStore.SUBNET_ADDRESS, RedisStore.REDIS_PORT)
- .from(master)
- .build());
- }
-
- @Override
- protected void initEnrichers() {
- super.initEnrichers();
- ServiceStateLogic.newEnricherFromChildrenUp().
- checkChildrenOnly().
- requireUpChildren(QuorumChecks.all()).
- configure(ComputeServiceIndicatorsFromChildrenAndMembers.IGNORE_ENTITIES_WITH_THESE_SERVICE_STATES, ImmutableSet.<Lifecycle>of()).
- addTo(this);
- }
-
- @Override
- public void start(Collection<? extends Location> locations) {
- ServiceStateLogic.setExpectedState(this, Lifecycle.STARTING);
- ServiceProblemsLogic.clearProblemsIndicator(this, START);
- try {
- doStart(locations);
- ServiceStateLogic.setExpectedState(this, Lifecycle.RUNNING);
- } catch (Exception e) {
- ServiceProblemsLogic.updateProblemsIndicator(this, START, "Start failed with error: "+e);
- ServiceStateLogic.setExpectedState(this, Lifecycle.ON_FIRE);
- throw Exceptions.propagate(e);
- }
- }
-
- private void doStart(Collection<? extends Location> locations) {
- // Start the master and slaves asynchronously (the slave has a LAUNCH_LATCH on master to ensure it is available before the slaves are launched)
- Task<Void> masterStartTask = getMaster().invoke(RedisStore.START, ImmutableMap.<String, Object>of("locations", ImmutableList.copyOf(locations)));
- Task<Void> slaveStartTask = getSlaves().invoke(DynamicCluster.START, ImmutableMap.<String, Object>of("locations", ImmutableList.copyOf(locations)));
-
- // Wait for both master and slave to start before returning
- masterStartTask.getUnchecked();
- slaveStartTask.getUnchecked();
- }
-
- @Override
- public void stop() {
- ServiceStateLogic.setExpectedState(this, Lifecycle.STOPPING);
- try {
- doStop();
- ServiceStateLogic.setExpectedState(this, Lifecycle.STOPPED);
- } catch (Exception e) {
- ServiceProblemsLogic.updateProblemsIndicator(this, STOP, "Stop failed with error: "+e);
- ServiceStateLogic.setExpectedState(this, Lifecycle.ON_FIRE);
- throw Exceptions.propagate(e);
- }
- }
-
- private void doStop() {
- StringBuilder message = new StringBuilder();
- List<Exception> exceptions = Lists.newLinkedList();
-
- try {
- getSlaves().invoke(DynamicCluster.STOP, ImmutableMap.<String, Object>of()).getUnchecked();
- } catch (Exception e) {
- message.append("Failed to stop Redis slaves");
- exceptions.add(e);
- }
-
- try {
- getMaster().invoke(RedisStore.STOP, ImmutableMap.<String, Object>of()).getUnchecked();
- } catch (Exception e) {
- message.append((message.length() == 0) ? "Failed to stop Redis master" : " and master");
- exceptions.add(e);
- }
-
- if (!exceptions.isEmpty()) {
- throw new CompoundRuntimeException(message.toString(), exceptions);
- }
- }
-
- @Override
- public void restart() {
- throw new UnsupportedOperationException();
- }
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShard.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShard.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShard.java
deleted file mode 100644
index 4d50d90..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShard.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.redis;
-
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.ImplementedBy;
-
-@ImplementedBy(RedisShardImpl.class)
-public interface RedisShard extends Entity {
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShardImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShardImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShardImpl.java
deleted file mode 100644
index ff01869..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisShardImpl.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.redis;
-
-import org.apache.brooklyn.core.entity.AbstractEntity;
-
-public class RedisShardImpl extends AbstractEntity implements RedisShard {
- public RedisShardImpl() {
- }
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlave.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlave.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlave.java
deleted file mode 100644
index d3882c2..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlave.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.redis;
-
-import org.apache.brooklyn.api.entity.ImplementedBy;
-import org.apache.brooklyn.config.ConfigKey;
-import org.apache.brooklyn.core.config.BasicConfigKey;
-import org.apache.brooklyn.util.core.flags.SetFromFlag;
-
-/**
- * A {@link RedisStore} configured as a slave.
- */
-@ImplementedBy(RedisSlaveImpl.class)
-public interface RedisSlave extends RedisStore {
-
- @SetFromFlag("master")
- ConfigKey<RedisStore> MASTER = new BasicConfigKey<RedisStore>(RedisStore.class, "redis.master", "Redis master");
-
- @SetFromFlag("redisConfigTemplateUrl")
- ConfigKey<String> REDIS_CONFIG_TEMPLATE_URL = new BasicConfigKey<String>(
- String.class, "redis.config.templateUrl", "Template file (in freemarker format) for the redis.conf config file",
- "classpath://org/apache/brooklyn/entity/nosql/redis/slave.conf");
-
- RedisStore getMaster();
-
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlaveImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlaveImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlaveImpl.java
deleted file mode 100644
index b58ce7d..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisSlaveImpl.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.redis;
-
-
-/**
- * A {@link RedisStore} configured as a slave.
- */
-public class RedisSlaveImpl extends RedisStoreImpl implements RedisSlave {
-
- public RedisSlaveImpl() {
- }
-
- @Override
- public RedisStore getMaster() {
- return getConfig(MASTER);
- }
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStore.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStore.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStore.java
deleted file mode 100644
index 64c0344..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStore.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.redis;
-
-import org.apache.brooklyn.api.catalog.Catalog;
-import org.apache.brooklyn.api.entity.ImplementedBy;
-import org.apache.brooklyn.api.sensor.AttributeSensor;
-import org.apache.brooklyn.config.ConfigKey;
-import org.apache.brooklyn.core.config.ConfigKeys;
-import org.apache.brooklyn.core.sensor.BasicAttributeSensorAndConfigKey;
-import org.apache.brooklyn.core.sensor.PortAttributeSensorAndConfigKey;
-import org.apache.brooklyn.core.sensor.Sensors;
-import org.apache.brooklyn.entity.software.base.SoftwareProcess;
-import org.apache.brooklyn.util.core.ResourcePredicates;
-import org.apache.brooklyn.util.core.flags.SetFromFlag;
-
-/**
- * An entity that represents a Redis key-value store service.
- */
-@Catalog(name="Redis Server", description="Redis is an open-source, networked, in-memory, key-value data store with optional durability", iconUrl="classpath:///redis-logo.png")
-@ImplementedBy(RedisStoreImpl.class)
-public interface RedisStore extends SoftwareProcess {
-
- @SetFromFlag("version")
- ConfigKey<String> SUGGESTED_VERSION =
- ConfigKeys.newConfigKeyWithDefault(SoftwareProcess.SUGGESTED_VERSION, "2.8.4");
-
- @SetFromFlag("downloadUrl")
- BasicAttributeSensorAndConfigKey<String> DOWNLOAD_URL = new BasicAttributeSensorAndConfigKey<String>(
- SoftwareProcess.DOWNLOAD_URL, "http://download.redis.io/releases/redis-${version}.tar.gz");
-
- @SetFromFlag("redisPort")
- PortAttributeSensorAndConfigKey REDIS_PORT = new PortAttributeSensorAndConfigKey("redis.port", "Redis port number", "6379+");
-
- @SetFromFlag("redisConfigTemplateUrl")
- ConfigKey<String> REDIS_CONFIG_TEMPLATE_URL = ConfigKeys.builder(String.class)
- .name("redis.config.templateUrl")
- .description("Template file (in freemarker format) for the redis.conf config file")
- .defaultValue("classpath://org/apache/brooklyn/entity/nosql/redis/redis.conf")
- .constraint(ResourcePredicates.urlExists())
- .build();
-
- AttributeSensor<Integer> UPTIME = Sensors.newIntegerSensor("redis.uptime", "Redis uptime in seconds");
-
- // See http://redis.io/commands/info for details of all information available
- AttributeSensor<Integer> TOTAL_CONNECTIONS_RECEIVED = Sensors.newIntegerSensor("redis.connections.received.total", "Total number of connections accepted by the server");
- AttributeSensor<Integer> TOTAL_COMMANDS_PROCESSED = Sensors.newIntegerSensor("redis.commands.processed.total", "Total number of commands processed by the server");
- AttributeSensor<Integer> EXPIRED_KEYS = Sensors.newIntegerSensor("redis.keys.expired", "Total number of key expiration events");
- AttributeSensor<Integer> EVICTED_KEYS = Sensors.newIntegerSensor("redis.keys.evicted", "Number of evicted keys due to maxmemory limit");
- AttributeSensor<Integer> KEYSPACE_HITS = Sensors.newIntegerSensor("redis.keyspace.hits", "Number of successful lookup of keys in the main dictionary");
- AttributeSensor<Integer> KEYSPACE_MISSES = Sensors.newIntegerSensor("redis.keyspace.misses", "Number of failed lookup of keys in the main dictionary");
-
- String getAddress();
-
- Integer getRedisPort();
-
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreDriver.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreDriver.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreDriver.java
deleted file mode 100644
index 63424ba..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreDriver.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.redis;
-
-import org.apache.brooklyn.entity.software.base.SoftwareProcessDriver;
-
-public interface RedisStoreDriver extends SoftwareProcessDriver {
-
- String getRunDir();
-
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreImpl.java
deleted file mode 100644
index 2fed9b1..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreImpl.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*
- * 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.redis;
-
-import java.util.concurrent.TimeUnit;
-
-import javax.annotation.Nullable;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.api.location.MachineLocation;
-import org.apache.brooklyn.entity.software.base.SoftwareProcessImpl;
-import org.apache.brooklyn.feed.ssh.SshFeed;
-import org.apache.brooklyn.feed.ssh.SshPollConfig;
-import org.apache.brooklyn.feed.ssh.SshPollValue;
-import org.apache.brooklyn.feed.ssh.SshValueFunctions;
-import org.apache.brooklyn.location.ssh.SshMachineLocation;
-
-import com.google.common.base.Function;
-import com.google.common.base.Functions;
-import com.google.common.base.Optional;
-import com.google.common.base.Predicates;
-import com.google.common.base.Splitter;
-import com.google.common.collect.Iterables;
-
-/**
- * An entity that represents a Redis key-value store service.
- */
-public class RedisStoreImpl extends SoftwareProcessImpl implements RedisStore {
- @SuppressWarnings("unused")
- private static final Logger LOG = LoggerFactory.getLogger(RedisStoreImpl.class);
-
- private transient SshFeed sshFeed;
-
- public RedisStoreImpl() {
- }
-
- @Override
- protected void connectSensors() {
- super.connectSensors();
-
- connectServiceUpIsRunning();
-
- // Find an SshMachineLocation for the UPTIME feed
- Optional<Location> location = Iterables.tryFind(getLocations(), Predicates.instanceOf(SshMachineLocation.class));
- if (!location.isPresent()) throw new IllegalStateException("Could not find SshMachineLocation in list of locations");
- SshMachineLocation machine = (SshMachineLocation) location.get();
- String statsCommand = getDriver().getRunDir() + "/bin/redis-cli -p " + getRedisPort() + " info stats";
- boolean retrieveUsageMetrics = getConfig(RETRIEVE_USAGE_METRICS);
-
- sshFeed = SshFeed.builder()
- .entity(this)
- .machine(machine)
- .period(5, TimeUnit.SECONDS)
- .poll(new SshPollConfig<Integer>(UPTIME)
- .command(getDriver().getRunDir() + "/bin/redis-cli -p " + getRedisPort() + " info server")
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("uptime_in_seconds"))
- .enabled(retrieveUsageMetrics))
- .poll(new SshPollConfig<Integer>(TOTAL_CONNECTIONS_RECEIVED)
- .command(statsCommand)
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("total_connections_received"))
- .enabled(retrieveUsageMetrics))
- .poll(new SshPollConfig<Integer>(TOTAL_COMMANDS_PROCESSED)
- .command(statsCommand)
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("total_commands_processed"))
- .enabled(retrieveUsageMetrics))
- .poll(new SshPollConfig<Integer>(EXPIRED_KEYS)
- .command(statsCommand)
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("expired_keys"))
- .enabled(retrieveUsageMetrics))
- .poll(new SshPollConfig<Integer>(EVICTED_KEYS)
- .command(statsCommand)
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("evicted_keys"))
- .enabled(retrieveUsageMetrics))
- .poll(new SshPollConfig<Integer>(KEYSPACE_HITS)
- .command(statsCommand)
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("keyspace_hits"))
- .enabled(retrieveUsageMetrics))
- .poll(new SshPollConfig<Integer>(KEYSPACE_MISSES)
- .command(statsCommand)
- .onFailureOrException(Functions.constant(-1))
- .onSuccess(infoFunction("keyspace_misses"))
- .enabled(retrieveUsageMetrics))
- .build();
- }
-
- /**
- * Create a {@link Function} to retrieve a particular field value from a {@code redis-cli info}
- * command.
- *
- * @param field the info field to retrieve and convert
- * @return a new function that converts a {@link SshPollValue} to an {@link Integer}
- */
- private static Function<SshPollValue, Integer> infoFunction(final String field) {
- return Functions.compose(new Function<String, Integer>() {
- @Override
- public Integer apply(@Nullable String input) {
- Optional<String> line = Iterables.tryFind(Splitter.on('\n').split(input), Predicates.containsPattern(field + ":"));
- if (line.isPresent()) {
- String data = line.get().trim();
- int colon = data.indexOf(":");
- return Integer.parseInt(data.substring(colon + 1));
- } else {
- throw new IllegalStateException("Data for field "+field+" not found: "+input);
- }
- }
- }, SshValueFunctions.stdout());
- }
-
- @Override
- public void disconnectSensors() {
- disconnectServiceUpIsRunning();
- if (sshFeed != null) sshFeed.stop();
- super.disconnectSensors();
- }
-
- @Override
- public Class<?> getDriverInterface() {
- return RedisStoreDriver.class;
- }
-
- @Override
- public RedisStoreDriver getDriver() {
- return (RedisStoreDriver) super.getDriver();
- }
-
- @Override
- public String getAddress() {
- MachineLocation machine = getMachineOrNull();
- return (machine != null) ? machine.getAddress().getHostAddress() : null;
- }
-
- @Override
- public Integer getRedisPort() {
- return getAttribute(RedisStore.REDIS_PORT);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreSshDriver.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreSshDriver.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreSshDriver.java
deleted file mode 100644
index 81ebfe3..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/redis/RedisStoreSshDriver.java
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * 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.redis;
-
-import static java.lang.String.format;
-
-import java.util.List;
-import java.util.Map;
-
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.core.entity.Entities;
-import org.apache.brooklyn.entity.software.base.AbstractSoftwareProcessSshDriver;
-import org.apache.brooklyn.location.ssh.SshMachineLocation;
-import org.apache.brooklyn.util.collections.MutableMap;
-import org.apache.brooklyn.util.os.Os;
-import org.apache.brooklyn.util.ssh.BashCommands;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-
-/**
- * Start a {@link RedisStore} in a {@link Location} accessible over ssh.
- */
-public class RedisStoreSshDriver extends AbstractSoftwareProcessSshDriver implements RedisStoreDriver {
-
- private static final Logger LOG = LoggerFactory.getLogger(RedisStoreSshDriver.class);
-
- public RedisStoreSshDriver(RedisStoreImpl entity, SshMachineLocation machine) {
- super(entity, machine);
- }
-
- @Override
- public void preInstall() {
- resolver = Entities.newDownloader(this);
- setExpandedInstallDir(Os.mergePaths(getInstallDir(), resolver.getUnpackedDirectoryName(format("redis-%s", getVersion()))));
- }
-
- @Override
- public void install() {
- List<String> urls = resolver.getTargets();
- String saveAs = resolver.getFilename();
-
- MutableMap<String, String> installGccPackageFlags = MutableMap.of(
- "onlyifmissing", "gcc",
- "yum", "gcc",
- "apt", "gcc",
- "port", null);
- MutableMap<String, String> installMakePackageFlags = MutableMap.of(
- "onlyifmissing", "make",
- "yum", "make",
- "apt", "make",
- "port", null);
- //Headers still might be missing on some systems (especially if gcc & make are pre-installed)
- Map<String, String> installBuildEssentialPackageFlags = ImmutableMap.of(
- "apt", "build-essential");
-
- List<String> commands = ImmutableList.<String>builder()
- .addAll(BashCommands.commandsToDownloadUrlsAs(urls, saveAs))
- .add(BashCommands.INSTALL_TAR)
- .add(BashCommands.INSTALL_CURL)
- .add(BashCommands.installPackage(installGccPackageFlags, "redis-prerequisites-gcc"))
- .add(BashCommands.installPackage(installMakePackageFlags, "redis-prerequisites-make"))
- .add(BashCommands.installPackage(installBuildEssentialPackageFlags, null))
- .add("tar xzfv " + saveAs)
- .add(format("cd redis-%s", getVersion()))
- .add("pushd deps")
- .add("make lua hiredis linenoise")
- .add("popd")
- .add("make clean && make")
- .build();
-
- newScript(INSTALLING)
- .failOnNonZeroResultCode()
- .body.append(commands).execute();
- }
-
- @Override
- public void customize() {
- newScript(MutableMap.of("usePidFile", false), CUSTOMIZING)
- .failOnNonZeroResultCode()
- .body.append(
- format("cd %s", getExpandedInstallDir()),
- "make install PREFIX="+getRunDir())
- .execute();
-
- copyTemplate(getEntity().getConfig(RedisStore.REDIS_CONFIG_TEMPLATE_URL), "redis.conf");
- }
-
- @Override
- public void launch() {
- // TODO Should we redirect stdout/stderr: format(" >> %s/console 2>&1 </dev/null &", getRunDir())
- newScript(MutableMap.of("usePidFile", false), LAUNCHING)
- .failOnNonZeroResultCode()
- .body.append("./bin/redis-server redis.conf")
- .execute();
- }
-
- @Override
- public boolean isRunning() {
- return newScript(MutableMap.of("usePidFile", false), CHECK_RUNNING)
- .body.append("./bin/redis-cli -p " + getEntity().getAttribute(RedisStore.REDIS_PORT) + " ping > /dev/null")
- .execute() == 0;
- }
-
- /**
- * Restarts redis with the current configuration.
- */
- @Override
- public void stop() {
- int exitCode = newScript(MutableMap.of("usePidFile", false), STOPPING)
- .body.append("./bin/redis-cli -p " + getEntity().getAttribute(RedisStore.REDIS_PORT) + " shutdown")
- .execute();
- // TODO: Good enough? Will cause warnings when trying to stop a server that is already not running.
- if (exitCode != 0) {
- LOG.warn("Unexpected exit code when stopping {}: {}", entity, exitCode);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakCluster.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakCluster.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakCluster.java
deleted file mode 100644
index e6de705..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakCluster.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.riak;
-
-import java.net.URI;
-import java.util.Map;
-
-import org.apache.brooklyn.api.catalog.Catalog;
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.ImplementedBy;
-import org.apache.brooklyn.api.sensor.AttributeSensor;
-import org.apache.brooklyn.config.ConfigKey;
-import org.apache.brooklyn.core.config.ConfigKeys;
-import org.apache.brooklyn.core.entity.Attributes;
-import org.apache.brooklyn.core.sensor.Sensors;
-import org.apache.brooklyn.entity.group.DynamicCluster;
-import org.apache.brooklyn.util.core.flags.SetFromFlag;
-import org.apache.brooklyn.util.time.Duration;
-
-import com.google.common.reflect.TypeToken;
-
-@Catalog(name="Riak Cluster", description="Riak is a distributed NoSQL key-value data store that offers "
- + "extremely high availability, fault tolerance, operational simplicity and scalability.")
-@ImplementedBy(RiakClusterImpl.class)
-public interface RiakCluster extends DynamicCluster {
-
- @SuppressWarnings("serial")
- AttributeSensor<Map<Entity, String>> RIAK_CLUSTER_NODES = Sensors.newSensor(
- new TypeToken<Map<Entity, String>>() {},
- "riak.cluster.nodes", "Names of all active Riak nodes in the cluster <Entity,Riak Name>");
-
- @SetFromFlag("delayBeforeAdvertisingCluster")
- ConfigKey<Duration> DELAY_BEFORE_ADVERTISING_CLUSTER = ConfigKeys.newConfigKey(Duration.class, "riak.cluster.delayBeforeAdvertisingCluster", "Delay after cluster is started before checking and advertising its availability", Duration.seconds(2 * 60));
-
- AttributeSensor<Boolean> IS_CLUSTER_INIT = Sensors.newBooleanSensor("riak.cluster.isClusterInit", "Flag to determine if the cluster was already initialized");
-
- AttributeSensor<Boolean> IS_FIRST_NODE_SET = Sensors.newBooleanSensor("riak.cluster.isFirstNodeSet", "Flag to determine if the first node has been set");
-
- AttributeSensor<String> NODE_LIST = Sensors.newStringSensor("riak.cluster.nodeList", "List of nodes (including ports), comma separated");
-
- AttributeSensor<String> NODE_LIST_PB_PORT = Sensors.newStringSensor("riak.cluster.nodeListPbPort", "List of nodes (including ports for riak db clients), comma separated");
-
- AttributeSensor<URI> RIAK_CONSOLE_URI = Attributes.MAIN_URI;
-
- AttributeSensor<Integer> NODE_GETS_1MIN_PER_NODE = Sensors.newIntegerSensor("riak.node.gets.1m.perNode", "Gets in the last minute, averaged across cluster");
- AttributeSensor<Integer> NODE_PUTS_1MIN_PER_NODE = Sensors.newIntegerSensor("riak.node.puts.1m.perNode", "Puts in the last minute, averaged across cluster");
- AttributeSensor<Integer> NODE_OPS_1MIN_PER_NODE = Sensors.newIntegerSensor("riak.node.ops.1m.perNode", "Sum of node gets and puts in the last minute, averaged across cluster");
-
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-library/blob/02abbab0/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java
----------------------------------------------------------------------
diff --git a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java b/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java
deleted file mode 100644
index 8272815..0000000
--- a/brooklyn-library/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java
+++ /dev/null
@@ -1,263 +0,0 @@
-/*
- * 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.riak;
-
-import static org.apache.brooklyn.util.JavaGroovyEquivalents.groovyTruth;
-
-import java.net.URI;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
-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.api.sensor.EnricherSpec;
-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.EntityPredicates;
-import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
-import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceNotUpLogic;
-import org.apache.brooklyn.core.entity.trait.Startable;
-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.util.core.task.Tasks;
-import org.apache.brooklyn.util.time.Duration;
-import org.apache.brooklyn.util.time.Time;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-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.ImmutableSet;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-
-public class RiakClusterImpl extends DynamicClusterImpl implements RiakCluster {
-
- private static final Logger log = LoggerFactory.getLogger(RiakClusterImpl.class);
-
- private transient Object mutex = new Object[0];
-
- public void init() {
- super.init();
- log.info("Initializing the riak cluster...");
- sensors().set(IS_CLUSTER_INIT, false);
- }
-
- @Override
- protected void doStart() {
- super.doStart();
- connectSensors();
-
- try {
- Duration delay = getConfig(DELAY_BEFORE_ADVERTISING_CLUSTER);
- Tasks.setBlockingDetails("Sleeping for "+delay+" before advertising cluster available");
- Time.sleep(delay);
- } finally {
- Tasks.resetBlockingDetails();
- }
-
- //FIXME: add a quorum to tolerate failed nodes before setting on fire.
- @SuppressWarnings("unchecked")
- Optional<Entity> anyNode = Iterables.tryFind(getMembers(), Predicates.and(
- Predicates.instanceOf(RiakNode.class),
- EntityPredicates.attributeEqualTo(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, true),
- EntityPredicates.attributeEqualTo(RiakNode.SERVICE_UP, true)));
- if (anyNode.isPresent()) {
- sensors().set(IS_CLUSTER_INIT, true);
- } else {
- log.warn("No Riak Nodes are found on the cluster: {}. Initialization Failed", getId());
- ServiceStateLogic.setExpectedState(this, Lifecycle.ON_FIRE);
- }
- }
-
- protected EntitySpec<?> getMemberSpec() {
- EntitySpec<?> result = config().get(MEMBER_SPEC);
- if (result!=null) return result;
- return EntitySpec.create(RiakNode.class);
- }
-
- protected void connectSensors() {
- policies().add(PolicySpec.create(MemberTrackingPolicy.class)
- .displayName("Controller targets tracker")
- .configure("sensorsToTrack", ImmutableSet.of(RiakNode.SERVICE_UP))
- .configure("group", this));
-
- EnricherSpec<?> first = Enrichers.builder()
- .aggregating(Attributes.MAIN_URI)
- .publishing(Attributes.MAIN_URI)
- .computing(new Function<Collection<URI>,URI>() {
- @Override
- public URI apply(Collection<URI> input) {
- return input.iterator().next();
- } })
- .fromMembers()
- .build();
- enrichers().add(first);
-
- Map<? extends AttributeSensor<? extends Number>, ? extends AttributeSensor<? extends Number>> enricherSetup =
- ImmutableMap.<AttributeSensor<? extends Number>, AttributeSensor<? extends Number>>builder()
- .put(RiakNode.NODE_PUTS, RiakCluster.NODE_PUTS_1MIN_PER_NODE)
- .put(RiakNode.NODE_GETS, RiakCluster.NODE_GETS_1MIN_PER_NODE)
- .put(RiakNode.NODE_OPS, RiakCluster.NODE_OPS_1MIN_PER_NODE)
- .build();
- // construct sum and average over cluster
- for (AttributeSensor<? extends Number> nodeSensor : enricherSetup.keySet()) {
- addSummingMemberEnricher(nodeSensor);
- addAveragingMemberEnricher(nodeSensor, enricherSetup.get(nodeSensor));
- }
- }
-
- 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()
- );
- }
-
- protected void onServerPoolMemberChanged(final Entity member) {
- synchronized (mutex) {
- log.trace("For {}, considering membership of {} which is in locations {}", new Object[]{ this, member, member.getLocations() });
-
- Map<Entity, String> nodes = getAttribute(RIAK_CLUSTER_NODES);
- if (belongsInServerPool(member)) {
- // TODO can we discover the nodes by asking the riak cluster, rather than assuming what we add will be in there?
- // TODO and can we do join as part of node starting?
-
- if (nodes == null) {
- nodes = Maps.newLinkedHashMap();
- }
- String riakName = getRiakName(member);
- Preconditions.checkNotNull(riakName);
-
- // flag a first node to be the first node in the riak cluster.
- Boolean firstNode = getAttribute(IS_FIRST_NODE_SET);
- if (!Boolean.TRUE.equals(firstNode)) {
- sensors().set(IS_FIRST_NODE_SET, Boolean.TRUE);
-
- nodes.put(member, riakName);
- sensors().set(RIAK_CLUSTER_NODES, nodes);
-
- ((EntityInternal) member).sensors().set(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, Boolean.TRUE);
-
- log.info("Added initial Riak node {}: {}; {} to new cluster", new Object[] { this, member, getRiakName(member) });
- } else {
- // TODO: be wary of erroneous nodes but are still flagged 'in cluster'
- // add the new node to be part of the riak cluster.
- Optional<Entity> anyNodeInCluster = Iterables.tryFind(nodes.keySet(), Predicates.and(
- Predicates.instanceOf(RiakNode.class),
- EntityPredicates.attributeEqualTo(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, true)));
- if (anyNodeInCluster.isPresent()) {
- if (!nodes.containsKey(member) && member.getAttribute(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER) == null) {
- String anyNodeName = anyNodeInCluster.get().getAttribute(RiakNode.RIAK_NODE_NAME);
- Entities.invokeEffectorWithArgs(this, member, RiakNode.JOIN_RIAK_CLUSTER, anyNodeName).blockUntilEnded();
- nodes.put(member, riakName);
- sensors().set(RIAK_CLUSTER_NODES, nodes);
- log.info("Added Riak node {}: {}; {} to cluster", new Object[] { this, member, getRiakName(member) });
- }
- } else {
- log.error("isFirstNodeSet, but no cluster members found to add {}", member.getId());
- }
- }
- } else {
- if (nodes != null && nodes.containsKey(member)) {
- DependentConfiguration.attributeWhenReady(member, RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, Predicates.equalTo(false)).blockUntilEnded(Duration.TWO_MINUTES);
- @SuppressWarnings("unchecked")
- Optional<Entity> anyNodeInCluster = Iterables.tryFind(nodes.keySet(), Predicates.and(
- Predicates.instanceOf(RiakNode.class),
- EntityPredicates.attributeEqualTo(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, true),
- Predicates.not(Predicates.equalTo(member))));
- if (anyNodeInCluster.isPresent()) {
- Entities.invokeEffectorWithArgs(this, anyNodeInCluster.get(), RiakNode.REMOVE_FROM_CLUSTER, getRiakName(member)).blockUntilEnded();
- }
- nodes.remove(member);
- sensors().set(RIAK_CLUSTER_NODES, nodes);
- log.info("Removed Riak node {}: {}; {} from cluster", new Object[]{ this, member, getRiakName(member) });
- }
- }
-
- ServiceNotUpLogic.updateNotUpIndicatorRequiringNonEmptyMap(this, RIAK_CLUSTER_NODES);
-
- calculateClusterAddresses();
- }
- }
-
- private void calculateClusterAddresses() {
- List<String> addresses = Lists.newArrayList();
- List<String> addressesPbPort = Lists.newArrayList();
- for (Entity entity : this.getMembers()) {
- if (entity instanceof RiakNode && entity.getAttribute(Attributes.SERVICE_UP)) {
- RiakNode riakNode = (RiakNode) entity;
- addresses.add(riakNode.getAttribute(Attributes.SUBNET_HOSTNAME) + ":" + riakNode.getAttribute(RiakNode.RIAK_WEB_PORT));
- addressesPbPort.add(riakNode.getAttribute(Attributes.SUBNET_HOSTNAME) + ":" + riakNode.getAttribute(RiakNode.RIAK_PB_PORT));
- }
- }
- sensors().set(RiakCluster.NODE_LIST, Joiner.on(",").join(addresses));
- sensors().set(RiakCluster.NODE_LIST_PB_PORT, Joiner.on(",").join(addressesPbPort));
- }
-
- protected boolean belongsInServerPool(Entity member) {
- if (!groovyTruth(member.getAttribute(Startable.SERVICE_UP))) {
- log.trace("Members of {}, checking {}, eliminating because not up", this, member);
- return false;
- }
- if (!getMembers().contains(member)) {
- log.trace("Members of {}, checking {}, eliminating because not member", this, member);
- return false;
- }
- log.trace("Members of {}, checking {}, approving", this, member);
-
- return true;
- }
-
- private String getRiakName(Entity node) {
- return node.getAttribute(RiakNode.RIAK_NODE_NAME);
- }
-
- public static class MemberTrackingPolicy extends AbstractMembershipTrackingPolicy {
- @Override
- protected void onEntityEvent(EventType type, Entity entity) {
- ((RiakClusterImpl) super.entity).onServerPoolMemberChanged(entity);
- }
- }
-}