http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java new file mode 100644 index 0000000..49bd515 --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakClusterImpl.java @@ -0,0 +1,264 @@ +/* + * 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 brooklyn.util.JavaGroovyEquivalents.groovyTruth; + +import java.net.URI; +import java.util.Collection; +import java.util.List; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +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.EntityPredicates; +import brooklyn.entity.basic.Lifecycle; +import brooklyn.entity.basic.ServiceStateLogic; +import brooklyn.entity.basic.ServiceStateLogic.ServiceNotUpLogic; +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.policy.EnricherSpec; +import brooklyn.policy.PolicySpec; +import brooklyn.util.task.Tasks; +import brooklyn.util.time.Duration; +import brooklyn.util.time.Time; + +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..."); + setAttribute(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()) { + setAttribute(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() { + addPolicy(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(); + addEnricher(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) { + 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() + ); + } + + 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)) { + setAttribute(IS_FIRST_NODE_SET, Boolean.TRUE); + + nodes.put(member, riakName); + setAttribute(RIAK_CLUSTER_NODES, nodes); + + ((EntityInternal) member).setAttribute(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); + setAttribute(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); + setAttribute(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)); + } + } + setAttribute(RiakCluster.NODE_LIST, Joiner.on(",").join(addresses)); + setAttribute(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); + } + } +}
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNode.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNode.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNode.java new file mode 100644 index 0000000..9542840 --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNode.java @@ -0,0 +1,238 @@ +/* + * 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.List; + +import org.apache.brooklyn.catalog.Catalog; +import brooklyn.config.ConfigKey; +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.java.UsesJava; +import brooklyn.entity.proxying.ImplementedBy; +import brooklyn.event.AttributeSensor; +import brooklyn.event.basic.AttributeSensorAndConfigKey; +import brooklyn.event.basic.PortAttributeSensorAndConfigKey; +import brooklyn.event.basic.Sensors; +import brooklyn.util.flags.SetFromFlag; + +import com.google.common.collect.ImmutableList; +import com.google.common.reflect.TypeToken; + +@Catalog(name="Riak Node", description="Riak is a distributed NoSQL key-value data store that offers " + + "extremely high availability, fault tolerance, operational simplicity and scalability.") +@ImplementedBy(RiakNodeImpl.class) +public interface RiakNode extends SoftwareProcess, UsesJava { + + @SetFromFlag("version") + ConfigKey<String> SUGGESTED_VERSION = ConfigKeys.newConfigKeyWithDefault(SoftwareProcess.SUGGESTED_VERSION, + "Version to install (Default 2.0.5)", "2.0.5"); + + @SetFromFlag("optimizeNetworking") + ConfigKey<Boolean> OPTIMIZE_HOST_NETWORKING = ConfigKeys.newBooleanConfigKey("riak.networking.optimize", "Optimize host networking when running in a VM", Boolean.TRUE); + + // vm.args and app.config are used for pre-version 2.0.0. Later versions use the (simplified) riak.conf + // see https://github.com/joedevivo/ricon/blob/master/cuttlefish.md + @SetFromFlag("vmArgsTemplateUrl") + ConfigKey<String> RIAK_VM_ARGS_TEMPLATE_URL = ConfigKeys.newStringConfigKey( + "riak.vmArgs.templateUrl", "Template file (in freemarker format) for the vm.args config file", + "classpath://org/apache/brooklyn/entity/nosql/riak/vm.args"); + @SetFromFlag("appConfigTemplateUrl") + ConfigKey<String> RIAK_APP_CONFIG_TEMPLATE_URL = ConfigKeys.newStringConfigKey( + "riak.appConfig.templateUrl", "Template file (in freemarker format) for the app.config config file", + "classpath://org/apache/brooklyn/entity/nosql/riak/app.config"); + @SetFromFlag("appConfigTemplateUrlLinux") + ConfigKey<String> RIAK_CONF_TEMPLATE_URL_LINUX = ConfigKeys.newStringConfigKey( + "riak.riakConf.templateUrl.linux", "Template file (in freemarker format) for the app.config config file", + "classpath://org/apache/brooklyn/entity/nosql/riak/riak.conf"); + @SetFromFlag("appConfigTemplateUrlMac") + ConfigKey<String> RIAK_CONF_TEMPLATE_URL_MAC = ConfigKeys.newStringConfigKey( + "riak.riakConf.templateUrl.mac", "Template file (in freemarker format) for the app.config config file", + "classpath://org/apache/brooklyn/entity/nosql/riak/riak-mac.conf"); + + ConfigKey<String> RIAK_CONF_ADDITIONAL_CONTENT = ConfigKeys.newStringConfigKey( + "riak.riakConf.additionalContent", "Template file (in freemarker format) for setting up additional settings in the riak.conf file", ""); + + // maxOpenFiles' default value (65536) is based on the Basho's recommendation - http://docs.basho.com/riak/latest/ops/tuning/open-files-limit/ + @SetFromFlag("maxOpenFiles") + ConfigKey<Integer> RIAK_MAX_OPEN_FILES = ConfigKeys.newIntegerConfigKey( + "riak.max.open.files", "Number of the open files required by Riak", 65536); + + @SetFromFlag("downloadUrlRhelCentos") + AttributeSensorAndConfigKey<String, String> DOWNLOAD_URL_RHEL_CENTOS = ConfigKeys.newTemplateSensorAndConfigKey("download.url.rhelcentos", + "URL pattern for downloading the linux RPM installer (will substitute things like ${version} automatically)", + "http://s3.amazonaws.com/downloads.basho.com/riak/${entity.majorVersion}/${entity.fullVersion}/rhel/"; + + "${entity.osMajorVersion}/riak-${entity.fullVersion}-1.el${entity.osMajorVersion}.x86_64.rpm"); + + @SetFromFlag("downloadUrlUbuntu") + AttributeSensorAndConfigKey<String, String> DOWNLOAD_URL_UBUNTU = ConfigKeys.newTemplateSensorAndConfigKey("download.url.ubuntu", + "URL pattern for downloading the linux Ubuntu installer (will substitute things like ${version} automatically)", + "http://s3.amazonaws.com/downloads.basho.com/riak/${entity.majorVersion}/${entity.fullVersion}/ubuntu/"; + + "$OS_RELEASE/riak_${entity.fullVersion}-1_amd64.deb"); + + @SetFromFlag("downloadUrlDebian") + AttributeSensorAndConfigKey<String, String> DOWNLOAD_URL_DEBIAN = ConfigKeys.newTemplateSensorAndConfigKey("download.url.debian", + "URL pattern for downloading the linux Debian installer (will substitute things like ${version} automatically)", + "http://s3.amazonaws.com/downloads.basho.com/riak/${entity.majorVersion}/${entity.fullVersion}/debian/"; + + "$OS_RELEASE/riak_${entity.fullVersion}-1_amd64.deb"); + + @SetFromFlag("downloadUrlMac") + AttributeSensorAndConfigKey<String, String> DOWNLOAD_URL_MAC = ConfigKeys.newTemplateSensorAndConfigKey("download.url.mac", + "URL pattern for downloading the MAC binaries tarball (will substitute things like ${version} automatically)", + "http://s3.amazonaws.com/downloads.basho.com/riak/${entity.majorVersion}/${entity.fullVersion}/osx/10.8/riak-${entity.fullVersion}-OSX-x86_64.tar.gz";); + + // NB these two needed for clients to access + @SetFromFlag("riakWebPort") + PortAttributeSensorAndConfigKey RIAK_WEB_PORT = new PortAttributeSensorAndConfigKey("riak.webPort", "Riak Web Port", "8098+"); + + @SetFromFlag("riakPbPort") + PortAttributeSensorAndConfigKey RIAK_PB_PORT = new PortAttributeSensorAndConfigKey("riak.pbPort", "Riak Protocol Buffers Port", "8087+"); + + AttributeSensor<Boolean> RIAK_PACKAGE_INSTALL = Sensors.newBooleanSensor( + "riak.install.package", "Flag to indicate whether Riak was installed using an OS package"); + AttributeSensor<Boolean> RIAK_ON_PATH = Sensors.newBooleanSensor( + "riak.install.onPath", "Flag to indicate whether Riak is available on the PATH"); + + AttributeSensor<Boolean> RIAK_NODE_HAS_JOINED_CLUSTER = Sensors.newBooleanSensor( + "riak.node.riakNodeHasJoinedCluster", "Flag to indicate whether the Riak node has joined a cluster member"); + + AttributeSensor<String> RIAK_NODE_NAME = Sensors.newStringSensor("riak.node", "Returns the riak node name as defined in vm.args"); + + // these needed for nodes to talk to each other, but not clients (so ideally set up in the security group for internal access) + PortAttributeSensorAndConfigKey HANDOFF_LISTENER_PORT = new PortAttributeSensorAndConfigKey("handoffListenerPort", "Handoff Listener Port", "8099+"); + PortAttributeSensorAndConfigKey EPMD_LISTENER_PORT = new PortAttributeSensorAndConfigKey("epmdListenerPort", "Erlang Port Mapper Daemon Listener Port", "4369"); + PortAttributeSensorAndConfigKey ERLANG_PORT_RANGE_START = new PortAttributeSensorAndConfigKey("erlangPortRangeStart", "Erlang Port Range Start", "6000+"); + PortAttributeSensorAndConfigKey ERLANG_PORT_RANGE_END = new PortAttributeSensorAndConfigKey("erlangPortRangeEnd", "Erlang Port Range End", "7999+"); + + @SetFromFlag("searchEnabled") + ConfigKey<Boolean> SEARCH_ENABLED = ConfigKeys.newBooleanConfigKey("riak.search", "Deploy Solr and configure Riak to use it", false); + + /** + * http://docs.basho.com/riak/latest/dev/using/search/ + * Solr is powered by Riak's Yokozuna engine and it is used through the riak webport + * So SEARCH_SOLR_PORT shouldn't be exposed + */ + ConfigKey<Integer> SEARCH_SOLR_PORT = ConfigKeys.newIntegerConfigKey("search.solr.port", "Solr port", 8983); + ConfigKey<Integer> SEARCH_SOLR_JMX_PORT = ConfigKeys.newIntegerConfigKey("search.solr.jmx_port", "Solr port", 8985); + + AttributeSensor<Integer> NODE_GETS = Sensors.newIntegerSensor("riak.node.gets", "Gets in the last minute"); + AttributeSensor<Integer> NODE_GETS_TOTAL = Sensors.newIntegerSensor("riak.node.gets.total", "Total gets since node started"); + AttributeSensor<Integer> NODE_PUTS = Sensors.newIntegerSensor("riak.node.puts", "Puts in the last minute"); + AttributeSensor<Integer> NODE_PUTS_TOTAL = Sensors.newIntegerSensor("riak.node.puts.total", "Total puts since node started"); + AttributeSensor<Integer> VNODE_GETS = Sensors.newIntegerSensor("riak.vnode.gets"); + AttributeSensor<Integer> VNODE_GETS_TOTAL = Sensors.newIntegerSensor("riak.vnode.gets.total"); + + //Sensors for Riak Node Counters (within 1 minute window or lifetime of node. + //http://docs.basho.com/riak/latest/ops/running/stats-and-monitoring/#Statistics-from-Riak + AttributeSensor<Integer> VNODE_PUTS = Sensors.newIntegerSensor("riak.vnode.puts"); + AttributeSensor<Integer> VNODE_PUTS_TOTAL = Sensors.newIntegerSensor("riak.vnode.puts.total"); + AttributeSensor<Integer> READ_REPAIRS_TOTAL = Sensors.newIntegerSensor("riak.read.repairs.total"); + AttributeSensor<Integer> COORD_REDIRS_TOTAL = Sensors.newIntegerSensor("riak.coord.redirs.total"); + //Additional Riak node counters + AttributeSensor<Integer> MEMORY_PROCESSES_USED = Sensors.newIntegerSensor("riak.memory.processes.used"); + AttributeSensor<Integer> SYS_PROCESS_COUNT = Sensors.newIntegerSensor("riak.sys.process.count"); + AttributeSensor<Integer> PBC_CONNECTS = Sensors.newIntegerSensor("riak.pbc.connects"); + AttributeSensor<Integer> PBC_ACTIVE = Sensors.newIntegerSensor("riak.pbc.active"); + @SuppressWarnings("serial") + AttributeSensor<List<String>> RING_MEMBERS = Sensors.newSensor(new TypeToken<List<String>>() {}, + "ring.members", "all the riak nodes in the ring"); + + AttributeSensor<Integer> NODE_OPS = Sensors.newIntegerSensor("riak.node.ops", "Sum of node gets and puts in the last minute"); + AttributeSensor<Integer> NODE_OPS_TOTAL = Sensors.newIntegerSensor("riak.node.ops.total", "Sum of node gets and puts since the node started"); + + MethodEffector<Void> JOIN_RIAK_CLUSTER = new MethodEffector<Void>(RiakNode.class, "joinCluster"); + MethodEffector<Void> LEAVE_RIAK_CLUSTER = new MethodEffector<Void>(RiakNode.class, "leaveCluster"); + MethodEffector<Void> REMOVE_FROM_CLUSTER = new MethodEffector<Void>(RiakNode.class, "removeNode"); + + AttributeSensor<Integer> RIAK_NODE_GET_FSM_TIME_MEAN = Sensors.newIntegerSensor("riak.node_get_fsm_time_mean", "Time between reception of client read request and subsequent response to client"); + AttributeSensor<Integer> RIAK_NODE_PUT_FSM_TIME_MEAN = Sensors.newIntegerSensor("riak.node_put_fsm_time_mean", "Time between reception of client write request and subsequent response to client"); + AttributeSensor<Integer> RIAK_OBJECT_COUNTER_MERGE_TIME_MEAN = Sensors.newIntegerSensor("riak.object_counter_merge_time_mean", "Time it takes to perform an Update Counter operation"); + AttributeSensor<Integer> RIAK_OBJECT_SET_MERGE_TIME_MEAN = Sensors.newIntegerSensor("riak.object_set_merge_time_mean", "Time it takes to perform an Update Set operation"); + AttributeSensor<Integer> RIAK_OBJECT_MAP_MERGE_TIME_MEAN = Sensors.newIntegerSensor("riak.object_map_merge_time_mean", "Time it takes to perform an Update Map operation"); + AttributeSensor<Integer> RIAK_CONSISTENT_GET_TIME_MEAN = Sensors.newIntegerSensor("riak.consistent_get_time_mean", "Strongly consistent read latency"); + AttributeSensor<Integer> RIAK_CONSISTENT_PUT_TIME_MEAN = Sensors.newIntegerSensor("riak.consistent_put_time_mean", "Strongly consistent write latency"); + + List<AttributeSensor<Integer>> ONE_MINUTE_SENSORS = ImmutableList.of(RIAK_NODE_GET_FSM_TIME_MEAN, RIAK_NODE_PUT_FSM_TIME_MEAN, + RIAK_OBJECT_COUNTER_MERGE_TIME_MEAN, RIAK_OBJECT_SET_MERGE_TIME_MEAN, RIAK_OBJECT_MAP_MERGE_TIME_MEAN, + RIAK_CONSISTENT_GET_TIME_MEAN, RIAK_CONSISTENT_PUT_TIME_MEAN); + + AttributeSensor<URI> RIAK_CONSOLE_URI = Attributes.MAIN_URI; + + // accessors, for use from template file + Integer getRiakWebPort(); + + Integer getRiakPbPort(); + + Integer getHandoffListenerPort(); + + Integer getEpmdListenerPort(); + + Integer getErlangPortRangeStart(); + + Integer getErlangPortRangeEnd(); + + Boolean isSearchEnabled(); + + Integer getSearchSolrPort(); + + Integer getSearchSolrJmxPort(); + + String getFullVersion(); + + String getMajorVersion(); + + String getOsMajorVersion(); + + // TODO add commitCluster() effector and add effectors joinCluster, leaveCluster, removeNode, recoverFailedNode which do not execute commitCluster() + // the commit where the commitCluster effector was available is adbf2dc1cb5df98b1e52d3ab35fa6bb4983b722f + + @Effector(description = "Join the Riak cluster on the given node") + void joinCluster(@EffectorParam(name = "nodeName") String nodeName); + + @Effector(description = "Leave the Riak cluster") + void leaveCluster(); + + @Effector(description = "Remove the given node from the Riak cluster") + void removeNode(@EffectorParam(name = "nodeName") String nodeName); + + @Effector(description = "Recover and join the Riak cluster on the given node") + void recoverFailedNode(@EffectorParam(name = "nodeName") String nodeName); + + @Effector(description = "Create or modify a bucket type before activation") + void bucketTypeCreate(@EffectorParam(name = "bucketTypeName") String bucketTypeName, + @EffectorParam(name = "bucketTypeProperties") String bucketTypeProperties); + + @Effector(description = "List all currently available bucket types and their activation status") + List<String> bucketTypeList(); + + @Effector(description = "Display the status and properties of a specific bucket type") + List<String> bucketTypeStatus(@EffectorParam(name = "bucketTypeName") String bucketTypeName); + + @Effector(description = "Update a bucket type after activation") + void bucketTypeUpdate(@EffectorParam(name = "bucketTypeName") String bucketTypeName, + @EffectorParam(name = "bucketTypeProperties") String bucketTypeProperties); + + @Effector(description = "Activate a bucket type") + void bucketTypeActivate(@EffectorParam(name = "bucketTypeName") String bucketTypeName); +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeDriver.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeDriver.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeDriver.java new file mode 100644 index 0000000..b9339cf --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeDriver.java @@ -0,0 +1,48 @@ +/* + * 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 brooklyn.entity.basic.SoftwareProcessDriver; + +import java.util.List; + +public interface RiakNodeDriver extends SoftwareProcessDriver { + + String getRiakEtcDir(); + + void joinCluster(String nodeName); + + void leaveCluster(); + + void removeNode(String nodeName); + + void recoverFailedNode(String nodeName); + + String getOsMajorVersion(); + + void bucketTypeCreate(String bucketTypeName, String bucketTypeProperties); + + List<String> bucketTypeList(); + + List<String> bucketTypeStatus(String bucketTypeName); + + void bucketTypeUpdate(String bucketTypeName, String bucketTypeProperties); + + void bucketTypeActivate(String bucketTypeName); +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeImpl.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeImpl.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeImpl.java new file mode 100644 index 0000000..d631516 --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeImpl.java @@ -0,0 +1,306 @@ +/* + * 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.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import javax.annotation.Nullable; + +import brooklyn.enricher.Enrichers; +import brooklyn.entity.basic.Entities; +import brooklyn.entity.basic.SoftwareProcessImpl; +import brooklyn.entity.webapp.WebAppServiceMethods; +import brooklyn.event.AttributeSensor; +import brooklyn.event.basic.AttributeSensorAndConfigKey; +import brooklyn.event.feed.http.HttpFeed; +import brooklyn.event.feed.http.HttpPollConfig; +import brooklyn.event.feed.http.HttpValueFunctions; +import brooklyn.location.MachineProvisioningLocation; +import brooklyn.location.access.BrooklynAccessUtils; +import brooklyn.location.cloud.CloudLocationConfig; +import brooklyn.util.collections.MutableSet; +import brooklyn.util.config.ConfigBag; +import brooklyn.util.guava.Functionals; +import brooklyn.util.time.Duration; + +import com.google.common.base.Preconditions; +import com.google.common.base.Function; +import com.google.common.base.Functions; +import com.google.common.collect.ContiguousSet; +import com.google.common.collect.DiscreteDomain; +import com.google.common.collect.Range; +import com.google.common.net.HostAndPort; + +public class RiakNodeImpl extends SoftwareProcessImpl implements RiakNode { + + private volatile HttpFeed httpFeed; + + @Override + public RiakNodeDriver getDriver() { + return (RiakNodeDriver) super.getDriver(); + } + + @Override + public Class<RiakNodeDriver> getDriverInterface() { + return RiakNodeDriver.class; + } + + @Override + public void init() { + super.init(); + // fail fast if config files not avail + Entities.getRequiredUrlConfig(this, RIAK_VM_ARGS_TEMPLATE_URL); + Entities.getRequiredUrlConfig(this, RIAK_APP_CONFIG_TEMPLATE_URL); + + Integer defaultMaxOpenFiles = RIAK_MAX_OPEN_FILES.getDefaultValue(); + Integer maxOpenFiles = getConfig(RiakNode.RIAK_MAX_OPEN_FILES); + Preconditions.checkArgument(maxOpenFiles >= defaultMaxOpenFiles , "Specified number of open files : %s : is less than the required minimum", + maxOpenFiles, defaultMaxOpenFiles); + } + + @SuppressWarnings("rawtypes") + public boolean isPackageDownloadUrlProvided() { + AttributeSensorAndConfigKey[] downloadProperties = { DOWNLOAD_URL_RHEL_CENTOS, DOWNLOAD_URL_UBUNTU, DOWNLOAD_URL_DEBIAN }; + for (AttributeSensorAndConfigKey property : downloadProperties) { + if (!((ConfigurationSupportInternal) config()).getRaw(property).isAbsent()) { + return true; + } + } + return false; + } + + @Override + 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(ERLANG_PORT_RANGE_START).iterator().next(); + int erlangRangeEnd = getConfig(ERLANG_PORT_RANGE_END).iterator().next(); + + Set<Integer> ports = MutableSet.copyOf(super.getRequiredOpenPorts()); + Set<Integer> erlangPorts = ContiguousSet.create(Range.open(erlangRangeStart, erlangRangeEnd), DiscreteDomain.integers()); + ports.addAll(erlangPorts); + + return ports; + } + + @Override + public void connectSensors() { + super.connectSensors(); + connectServiceUpIsRunning(); + HostAndPort accessible = BrooklynAccessUtils.getBrooklynAccessibleAddress(this, getRiakWebPort()); + + HttpFeed.Builder httpFeedBuilder = HttpFeed.builder() + .entity(this) + .period(500, TimeUnit.MILLISECONDS) + .baseUri(String.format("http://%s/stats";, accessible.toString())) + .poll(new HttpPollConfig<Integer>(NODE_GETS) + .onSuccess(HttpValueFunctions.jsonContents("node_gets", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(NODE_GETS_TOTAL) + .onSuccess(HttpValueFunctions.jsonContents("node_gets_total", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(NODE_PUTS) + .onSuccess(HttpValueFunctions.jsonContents("node_puts", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(NODE_PUTS_TOTAL) + .onSuccess(HttpValueFunctions.jsonContents("node_puts_total", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(VNODE_GETS) + .onSuccess(HttpValueFunctions.jsonContents("vnode_gets", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(VNODE_GETS_TOTAL) + .onSuccess(HttpValueFunctions.jsonContents("vnode_gets_total", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(VNODE_PUTS) + .onSuccess(HttpValueFunctions.jsonContents("vnode_puts", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(VNODE_PUTS_TOTAL) + .onSuccess(HttpValueFunctions.jsonContents("vnode_puts_total", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(READ_REPAIRS_TOTAL) + .onSuccess(HttpValueFunctions.jsonContents("read_repairs_total", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(COORD_REDIRS_TOTAL) + .onSuccess(HttpValueFunctions.jsonContents("coord_redirs_total", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(MEMORY_PROCESSES_USED) + .onSuccess(HttpValueFunctions.jsonContents("memory_processes_used", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(SYS_PROCESS_COUNT) + .onSuccess(HttpValueFunctions.jsonContents("sys_process_count", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(PBC_CONNECTS) + .onSuccess(HttpValueFunctions.jsonContents("pbc_connects", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<Integer>(PBC_ACTIVE) + .onSuccess(HttpValueFunctions.jsonContents("pbc_active", Integer.class)) + .onFailureOrException(Functions.constant(-1))) + .poll(new HttpPollConfig<List<String>>(RING_MEMBERS) + .onSuccess(Functionals.chain( + HttpValueFunctions.jsonContents("ring_members", String[].class), + new Function<String[], List<String>>() { + @Nullable + @Override + public List<String> apply(@Nullable String[] strings) { + return Arrays.asList(strings); + } + } + )) + .onFailureOrException(Functions.constant(Arrays.asList(new String[0])))); + + for (AttributeSensor<Integer> sensor : ONE_MINUTE_SENSORS) { + httpFeedBuilder.poll(new HttpPollConfig<Integer>(sensor) + .period(Duration.ONE_MINUTE) + .onSuccess(HttpValueFunctions.jsonContents(sensor.getName().substring(5), Integer.class)) + .onFailureOrException(Functions.constant(-1))); + } + + httpFeed = httpFeedBuilder.build(); + + addEnricher(Enrichers.builder().combining(NODE_GETS, NODE_PUTS).computingSum().publishing(NODE_OPS).build()); + addEnricher(Enrichers.builder().combining(NODE_GETS_TOTAL, NODE_PUTS_TOTAL).computingSum().publishing(NODE_OPS_TOTAL).build()); + WebAppServiceMethods.connectWebAppServerPolicies(this); + } + + @Override + public void disconnectSensors() { + super.disconnectSensors(); + if (httpFeed != null) { + httpFeed.stop(); + } + disconnectServiceUpIsRunning(); + } + + @Override + public void joinCluster(String nodeName) { + getDriver().joinCluster(nodeName); + } + + @Override + public void leaveCluster() { + getDriver().leaveCluster(); + } + + @Override + public void removeNode(String nodeName) { + getDriver().removeNode(nodeName); + } + + @Override + public void bucketTypeCreate(String bucketTypeName, String bucketTypeProperties) { + getDriver().bucketTypeCreate(bucketTypeName, bucketTypeProperties); + } + + @Override + public List<String> bucketTypeList() { + return getDriver().bucketTypeList(); + } + + @Override + public List<String> bucketTypeStatus(String bucketTypeName) { + return getDriver().bucketTypeStatus(bucketTypeName); + } + + @Override + public void bucketTypeUpdate(String bucketTypeName, String bucketTypeProperties) { + getDriver().bucketTypeUpdate(bucketTypeName, bucketTypeProperties); + } + + @Override + public void bucketTypeActivate(String bucketTypeName) { + getDriver().bucketTypeActivate(bucketTypeName); + } + + @Override + public void recoverFailedNode(String nodeName) { + getDriver().recoverFailedNode(nodeName); + } + + @Override + public Integer getRiakWebPort() { + return getAttribute(RiakNode.RIAK_WEB_PORT); + } + + @Override + public Integer getRiakPbPort() { + return getAttribute(RiakNode.RIAK_PB_PORT); + } + + @Override + public Integer getHandoffListenerPort() { + return getAttribute(RiakNode.HANDOFF_LISTENER_PORT); + } + + @Override + public Integer getEpmdListenerPort() { + return getAttribute(RiakNode.EPMD_LISTENER_PORT); + } + + @Override + public Integer getErlangPortRangeStart() { + return getAttribute(RiakNode.ERLANG_PORT_RANGE_START); + } + + @Override + public Integer getErlangPortRangeEnd() { + return getAttribute(RiakNode.ERLANG_PORT_RANGE_END); + } + + @Override + public Boolean isSearchEnabled() { + return getConfig(RiakNode.SEARCH_ENABLED); + } + + @Override + public Integer getSearchSolrPort() { + return getConfig(RiakNode.SEARCH_SOLR_PORT); + } + + @Override + public Integer getSearchSolrJmxPort() { + return getConfig(RiakNode.SEARCH_SOLR_JMX_PORT); + } + + @Override + public String getMajorVersion() { + return getFullVersion().substring(0, 3); + } + + @Override + public String getFullVersion() { + return getConfig(RiakNode.SUGGESTED_VERSION); + } + + @Override + public String getOsMajorVersion() { + return getDriver().getOsMajorVersion(); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java new file mode 100644 index 0000000..1815673 --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/riak/RiakNodeSshDriver.java @@ -0,0 +1,614 @@ +/* + * 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 brooklyn.util.ssh.BashCommands.INSTALL_CURL; +import static brooklyn.util.ssh.BashCommands.INSTALL_TAR; +import static brooklyn.util.ssh.BashCommands.addSbinPathCommand; +import static brooklyn.util.ssh.BashCommands.alternatives; +import static brooklyn.util.ssh.BashCommands.chainGroup; +import static brooklyn.util.ssh.BashCommands.commandToDownloadUrlAs; +import static brooklyn.util.ssh.BashCommands.ifExecutableElse; +import static brooklyn.util.ssh.BashCommands.ifNotExecutable; +import static brooklyn.util.ssh.BashCommands.ok; +import static brooklyn.util.ssh.BashCommands.sudo; +import static brooklyn.util.text.StringEscapes.BashStringEscapes.escapeLiteralForDoubleQuotedBash; +import static java.lang.String.format; + +import java.net.URI; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import brooklyn.entity.java.JavaSoftwareProcessSshDriver; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import brooklyn.entity.basic.Attributes; +import brooklyn.entity.basic.Entities; +import brooklyn.entity.basic.lifecycle.ScriptHelper; +import brooklyn.entity.software.SshEffectorTasks; +import brooklyn.location.OsDetails; +import brooklyn.location.basic.SshMachineLocation; +import brooklyn.util.collections.MutableMap; +import brooklyn.util.net.Urls; +import brooklyn.util.os.Os; +import brooklyn.util.ssh.BashCommands; +import brooklyn.util.task.DynamicTasks; +import brooklyn.util.task.ssh.SshTasks; +import brooklyn.util.text.Strings; + +import com.google.common.base.Joiner; +import com.google.common.base.Optional; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Lists; + +// TODO: Alter -env ERL_CRASH_DUMP path in vm.args +public class RiakNodeSshDriver extends JavaSoftwareProcessSshDriver implements RiakNodeDriver { + + private static final Logger LOG = LoggerFactory.getLogger(RiakNodeSshDriver.class); + private static final String sbinPath = "$PATH:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"; + private static final String INSTALLING_FALLBACK = INSTALLING + "_fallback"; + + public RiakNodeSshDriver(final RiakNodeImpl entity, final SshMachineLocation machine) { + super(entity, machine); + } + + @Override + protected String getLogFileLocation() { + return "/var/log/riak/solr.log"; + } + + @Override + public RiakNodeImpl getEntity() { + return RiakNodeImpl.class.cast(super.getEntity()); + } + + @Override + public Map<String, String> getShellEnvironment() { + MutableMap<String, String> result = MutableMap.copyOf(super.getShellEnvironment()); + // how to change epmd port, according to + // http://serverfault.com/questions/582787/how-to-change-listening-interface-of-rabbitmqs-epmd-port-4369 + if (getEntity().getEpmdListenerPort() != null) { + result.put("ERL_EPMD_PORT", Integer.toString(getEntity().getEpmdListenerPort())); + } + result.put("WAIT_FOR_ERLANG", "60"); + return result; + } + + @Override + public void preInstall() { + resolver = Entities.newDownloader(this); + setExpandedInstallDir(Os.mergePaths(getInstallDir(), resolver.getUnpackedDirectoryName(format("riak-%s", getVersion())))); + + // Set package install attribute + OsDetails osDetails = getMachine().getMachineDetails().getOsDetails(); + if (osDetails.isLinux()) { + entity.setAttribute(RiakNode.RIAK_PACKAGE_INSTALL, true); + } else if (osDetails.isMac()) { + entity.setAttribute(RiakNode.RIAK_PACKAGE_INSTALL, false); + } + } + + @Override + public void install() { + if (entity.getConfig(Attributes.DOWNLOAD_URL) != null) { + LOG.warn("Ignoring download.url {}, use download.url.rhelcentos or download.url.mac", entity.getConfig(Attributes.DOWNLOAD_URL)); + } + + OsDetails osDetails = getMachine().getMachineDetails().getOsDetails(); + List<String> commands = Lists.newLinkedList(); + if (osDetails.isLinux()) { + if (getEntity().isPackageDownloadUrlProvided()) { + commands.addAll(installLinuxFromPackageUrl()); + } else { + commands.addAll(installFromPackageCloud()); + } + } else if (osDetails.isMac()) { + commands.addAll(installMac()); + } else if (osDetails.isWindows()) { + throw new UnsupportedOperationException("RiakNode not supported on Windows instances"); + } else { + throw new IllegalStateException("Machine was not detected as linux, mac or windows! Installation does not know how to proceed with " + + getMachine() + ". Details: " + getMachine().getMachineDetails().getOsDetails()); + } + + int result = newScript(INSTALLING) + .body.append(commands) + .failIfBodyEmpty() + .execute(); + + if (result != 0 && osDetails.isLinux()) { + result = newScript(INSTALLING_FALLBACK) + .body.append(installLinuxFromPackageUrl()) + .execute(); + } + + if (result != 0) { + throw new IllegalStateException(String.format("Install failed with result %d", result)); + } + } + + private List<String> installLinuxFromPackageUrl() { + DynamicTasks.queueIfPossible(SshTasks.dontRequireTtyForSudo(getMachine(), SshTasks.OnFailingTask.WARN_OR_IF_DYNAMIC_FAIL_MARKING_INESSENTIAL)).orSubmitAndBlock(); + + String expandedInstallDir = getExpandedInstallDir(); + String installBin = Urls.mergePaths(expandedInstallDir, "bin"); + String saveAsYum = "riak.rpm"; + String saveAsApt = "riak.deb"; + OsDetails osDetails = getMachine().getOsDetails(); + + String downloadUrl; + String osReleaseCmd; + if ("debian".equalsIgnoreCase(osDetails.getName())) { + // TODO osDetails.getName() is returning "linux", instead of debian/ubuntu on AWS with jenkins image, + // running as integration test targetting localhost. + // TODO Debian support (default debian image fails with 'sudo: command not found') + downloadUrl = (String)entity.getAttribute(RiakNode.DOWNLOAD_URL_DEBIAN); + osReleaseCmd = osDetails.getVersion().substring(0, osDetails.getVersion().indexOf(".")); + } else { + // assume Ubuntu + downloadUrl = (String)entity.getAttribute(RiakNode.DOWNLOAD_URL_UBUNTU); + osReleaseCmd = "`lsb_release -sc` && " + + "export OS_RELEASE=`([[ \"lucid natty precise\" =~ (^| )\\$OS_RELEASE($| ) ]] && echo $OS_RELEASE || echo precise)`"; + } + String apt = chainGroup( + //debian fix + "export PATH=" + sbinPath, + "which apt-get", + ok(sudo("apt-get -y --allow-unauthenticated install logrotate libpam0g-dev libssl0.9.8")), + "export OS_NAME=" + Strings.toLowerCase(osDetails.getName()), + "export OS_RELEASE=" + osReleaseCmd, + String.format("wget -O %s %s", saveAsApt, downloadUrl), + sudo(String.format("dpkg -i %s", saveAsApt))); + String yum = chainGroup( + "which yum", + ok(sudo("yum -y install openssl")), + String.format("wget -O %s %s", saveAsYum, entity.getAttribute(RiakNode.DOWNLOAD_URL_RHEL_CENTOS)), + sudo(String.format("yum localinstall -y %s", saveAsYum))); + return ImmutableList.<String>builder() + .add("mkdir -p " + installBin) + .add(INSTALL_CURL) + .add(alternatives(apt, yum)) + .add("ln -s `which riak` " + Urls.mergePaths(installBin, "riak")) + .add("ln -s `which riak-admin` " + Urls.mergePaths(installBin, "riak-admin")) + .build(); + } + + private List<String> installFromPackageCloud() { + OsDetails osDetails = getMachine().getMachineDetails().getOsDetails(); + return ImmutableList.<String>builder() + .add(osDetails.getName().toLowerCase().contains("debian") ? addSbinPathCommand() : "") + .add(ifNotExecutable("curl", INSTALL_CURL)) + .addAll(ifExecutableElse("yum", installDebianBased(), installRpmBased())) + .build(); + } + + private ImmutableList<String> installDebianBased() { + return ImmutableList.<String>builder() + .add("curl https://packagecloud.io/install/repositories/basho/riak/script.deb.sh | " + BashCommands.sudo("bash")) + .add(BashCommands.sudo("apt-get install --assume-yes riak=" + getEntity().getFullVersion() + "-1")) + .build(); + } + + private ImmutableList<String> installRpmBased() { + return ImmutableList.<String>builder() + .add("curl https://packagecloud.io/install/repositories/basho/riak/script.rpm.sh | " + BashCommands.sudo("bash")) + .add(BashCommands.sudo("yum install -y riak-" + getEntity().getFullVersion() + "*")) + .build(); + } + + protected List<String> installMac() { + String saveAs = resolver.getFilename(); + String url = entity.getAttribute(RiakNode.DOWNLOAD_URL_MAC); + return ImmutableList.<String>builder() + .add(INSTALL_TAR) + .add(INSTALL_CURL) + .add(commandToDownloadUrlAs(url, saveAs)) + .add("tar xzvf " + saveAs) + .build(); + } + + @Override + public void customize() { + checkRiakOnPath(); + + //create entity's runDir + newScript(CUSTOMIZING).execute(); + + OsDetails osDetails = getMachine().getMachineDetails().getOsDetails(); + + List<String> commands = Lists.newLinkedList(); + commands.add(sudo("mkdir -p " + getRiakEtcDir())); + + if (isVersion1()) { + String vmArgsTemplate = processTemplate(entity.getConfig(RiakNode.RIAK_VM_ARGS_TEMPLATE_URL)); + String saveAsVmArgs = Urls.mergePaths(getRunDir(), "vm.args"); + DynamicTasks.queue(SshEffectorTasks.put(saveAsVmArgs).contents(vmArgsTemplate)); + commands.add(sudo("mv " + saveAsVmArgs + " " + getRiakEtcDir())); + + String appConfigTemplate = processTemplate(entity.getConfig(RiakNode.RIAK_APP_CONFIG_TEMPLATE_URL)); + String saveAsAppConfig = Urls.mergePaths(getRunDir(), "app.config"); + DynamicTasks.queue(SshEffectorTasks.put(saveAsAppConfig).contents(appConfigTemplate)); + commands.add(sudo("mv " + saveAsAppConfig + " " + getRiakEtcDir())); + } else { + String templateUrl = osDetails.isMac() ? entity.getConfig(RiakNode.RIAK_CONF_TEMPLATE_URL_MAC) : + entity.getConfig(RiakNode.RIAK_CONF_TEMPLATE_URL_LINUX); + String riakConfContent = processTemplate(templateUrl); + String saveAsRiakConf = Urls.mergePaths(getRunDir(), "riak.conf"); + + if(Strings.isNonBlank(entity.getConfig(RiakNode.RIAK_CONF_ADDITIONAL_CONTENT))) { + String additionalConfigContent = processTemplateContents(entity.getConfig(RiakNode.RIAK_CONF_ADDITIONAL_CONTENT)); + riakConfContent += "\n## Brooklyn note: additional config\n"; + riakConfContent += additionalConfigContent; + } + + DynamicTasks.queue(SshEffectorTasks.put(saveAsRiakConf).contents(riakConfContent)); + commands.add(sudo("mv " + saveAsRiakConf + " " + getRiakEtcDir())); + } + + //increase open file limit (default min for riak is: 4096) + //TODO: detect the actual limit then do the modification. + //TODO: modify ulimit for linux distros + // commands.add(sudo("launchctl limit maxfiles 4096 32768")); + if (osDetails.isMac()) { + commands.add("ulimit -n 4096"); + } + + if (osDetails.isLinux() && isVersion1()) { + commands.add(sudo("chown -R riak:riak " + getRiakEtcDir())); + } + + // TODO platform_*_dir + // TODO riak config log + + ScriptHelper customizeScript = newScript(CUSTOMIZING) + .failOnNonZeroResultCode() + .body.append(commands); + + if (!isRiakOnPath()) { + addRiakOnPath(customizeScript); + } + customizeScript.failOnNonZeroResultCode().execute(); + + if (osDetails.isLinux()) { + ImmutableMap<String, String> sysctl = ImmutableMap.<String, String>builder() + .put("vm.swappiness", "0") + .put("net.core.somaxconn", "40000") + .put("net.ipv4.tcp_max_syn_backlog", "40000") + .put("net.ipv4.tcp_sack", "1") + .put("net.ipv4.tcp_window_scaling", "15") + .put("net.ipv4.tcp_fin_timeout", "1") + .put("net.ipv4.tcp_keepalive_intvl", "30") + .put("net.ipv4.tcp_tw_reuse", "1") + .put("net.ipv4.tcp_moderate_rcvbuf", "1") + .build(); + + ScriptHelper optimize = newScript(CUSTOMIZING + "network") + .body.append(sudo("sysctl " + Joiner.on(' ').withKeyValueSeparator("=").join(sysctl))); + + Optional<Boolean> enable = Optional.fromNullable(entity.getConfig(RiakNode.OPTIMIZE_HOST_NETWORKING)); + if (!enable.isPresent()) optimize.inessential(); + if (enable.or(true)) optimize.execute(); + } + + //set the riak node name + entity.setAttribute(RiakNode.RIAK_NODE_NAME, format("riak@%s", getSubnetHostname())); + } + + @Override + public void launch() { + List<String> commands = Lists.newLinkedList(); + + if (isPackageInstall()) { + commands.add(addSbinPathCommand()); + commands.add(sudo(format("sh -c \"ulimit -n %s && service riak start\"", maxOpenFiles()))); + } else { + // NOTE: See instructions at http://superuser.com/questions/433746/is-there-a-fix-for-the-too-many-open-files-in-system-error-on-os-x-10-7-1 + // for increasing the system limit for number of open files + commands.add("ulimit -n 65536 || true"); // `BashCommands.ok` will put this in parentheses, which will set ulimit -n in the subshell + commands.add(format("%s start >/dev/null 2>&1 < /dev/null &", getRiakCmd())); + } + + ScriptHelper launchScript = newScript(LAUNCHING) + .body.append(commands); + + if (!isRiakOnPath()) { + addRiakOnPath(launchScript); + } + launchScript.failOnNonZeroResultCode().execute(); + + String mainUri = String.format("http://%s:%s/admin";, entity.getAttribute(Attributes.HOSTNAME), entity.getAttribute(RiakNode.RIAK_WEB_PORT)); + entity.setAttribute(Attributes.MAIN_URI, URI.create(mainUri)); + } + + @Override + public void stop() { + leaveCluster(); + + String command = format("%s stop", getRiakCmd()); + command = isPackageInstall() ? sudo(command) : command; + + ScriptHelper stopScript = newScript(ImmutableMap.of(USE_PID_FILE, false), STOPPING) + .body.append(command); + + if (!isRiakOnPath()) { + addRiakOnPath(stopScript); + } + + int result = stopScript.failOnNonZeroResultCode().execute(); + if (result != 0) { + newScript(ImmutableMap.of(USE_PID_FILE, false), STOPPING).execute(); + } + } + + @Override + public boolean isRunning() { + // Version 2.0.0 requires sudo for `riak ping` + ScriptHelper checkRunningScript = newScript(CHECK_RUNNING) + .body.append(sudo(format("%s ping", getRiakCmd()))); + + if (!isRiakOnPath()) { + addRiakOnPath(checkRunningScript); + } + return (checkRunningScript.execute() == 0); + } + + public boolean isPackageInstall() { + return entity.getAttribute(RiakNode.RIAK_PACKAGE_INSTALL); + } + + public boolean isRiakOnPath() { + return entity.getAttribute(RiakNode.RIAK_ON_PATH); + } + + public String getRiakEtcDir() { + return isPackageInstall() ? "/etc/riak" : Urls.mergePaths(getExpandedInstallDir(), "etc"); + } + + protected String getRiakCmd() { + return isPackageInstall() ? "riak" : Urls.mergePaths(getExpandedInstallDir(), "bin/riak"); + } + + protected String getRiakAdminCmd() { + return isPackageInstall() ? "riak-admin" : Urls.mergePaths(getExpandedInstallDir(), "bin/riak-admin"); + } + + // TODO find a way to batch commit the changes, instead of committing for every operation. + + @Override + public void joinCluster(String nodeName) { + if (getRiakName().equals(nodeName)) { + log.warn("Cannot join Riak node: {} to itself", nodeName); + } else { + if (!hasJoinedCluster()) { + ScriptHelper joinClusterScript = newScript("joinCluster") + .body.append(sudo(format("%s cluster join %s", getRiakAdminCmd(), nodeName))) + .body.append(sudo(format("%s cluster plan", getRiakAdminCmd()))) + .body.append(sudo(format("%s cluster commit", getRiakAdminCmd()))) + .failOnNonZeroResultCode(); + + if (!isRiakOnPath()) { + addRiakOnPath(joinClusterScript); + } + + joinClusterScript.execute(); + + entity.setAttribute(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, Boolean.TRUE); + } else { + log.warn("entity {}: is already in the riak cluster", entity.getId()); + } + } + } + + @Override + public void leaveCluster() { + if (hasJoinedCluster()) { + ScriptHelper leaveClusterScript = newScript("leaveCluster") + .body.append(sudo(format("%s cluster leave", getRiakAdminCmd()))) + .body.append(sudo(format("%s cluster plan", getRiakAdminCmd()))) + .body.append(sudo(format("%s cluster commit", getRiakAdminCmd()))) + .failOnNonZeroResultCode(); + + if (!isRiakOnPath()) { + addRiakOnPath(leaveClusterScript); + } + + leaveClusterScript.execute(); + + entity.setAttribute(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER, Boolean.FALSE); + } else { + log.warn("entity {}: has already left the riak cluster", entity.getId()); + } + } + + @Override + public void removeNode(String nodeName) { + ScriptHelper removeNodeScript = newScript("removeNode") + .body.append(sudo(format("%s cluster force-remove %s", getRiakAdminCmd(), nodeName))) + .body.append(sudo(format("%s down %s", getRiakAdminCmd(), nodeName))) + .body.append(sudo(format("%s cluster plan", getRiakAdminCmd()))) + .body.append(sudo(format("%s cluster commit", getRiakAdminCmd()))) + .failOnNonZeroResultCode(); + + if (!isRiakOnPath()) { + addRiakOnPath(removeNodeScript); + } + + removeNodeScript.execute(); + } + + @Override + public void bucketTypeCreate(String bucketTypeName, String bucketTypeProperties) { + ScriptHelper bucketTypeCreateScript = newScript("bucket-type_create " + bucketTypeName) + .body.append(sudo(format("%s bucket-type create %s %s", + getRiakAdminCmd(), + bucketTypeName, + escapeLiteralForDoubleQuotedBash(bucketTypeProperties)))); + if(!isRiakOnPath()) { + addRiakOnPath(bucketTypeCreateScript); + } + bucketTypeCreateScript.body.append(sudo(format("%s bucket-type activate %s", getRiakAdminCmd(), bucketTypeName))) + .failOnNonZeroResultCode(); + + bucketTypeCreateScript.execute(); + } + + @Override + public List<String> bucketTypeList() { + ScriptHelper bucketTypeListScript = newScript("bucket-types_list") + .body.append(sudo(format("%s bucket-type list", getRiakAdminCmd()))) + .gatherOutput() + .noExtraOutput() + .failOnNonZeroResultCode(); + if (!isRiakOnPath()) { + addRiakOnPath(bucketTypeListScript); + } + bucketTypeListScript.execute(); + String stdout = bucketTypeListScript.getResultStdout(); + return Arrays.asList(stdout.split("[\\r\\n]+")); + } + + @Override + public List<String> bucketTypeStatus(String bucketTypeName) { + ScriptHelper bucketTypeStatusScript = newScript("bucket-type_status") + .body.append(sudo(format("%s bucket-type status %s", getRiakAdminCmd(), bucketTypeName))) + .gatherOutput() + .noExtraOutput() + .failOnNonZeroResultCode(); + if (!isRiakOnPath()) { + addRiakOnPath(bucketTypeStatusScript); + } + bucketTypeStatusScript.execute(); + String stdout = bucketTypeStatusScript.getResultStdout(); + return Arrays.asList(stdout.split("[\\r\\n]+")); + } + + @Override + public void bucketTypeUpdate(String bucketTypeName, String bucketTypeProperties) { + ScriptHelper bucketTypeStatusScript = newScript("bucket-type_update") + .body.append(sudo(format("%s bucket-type update %s %s", + getRiakAdminCmd(), + bucketTypeName, + escapeLiteralForDoubleQuotedBash(bucketTypeProperties)))) + .failOnNonZeroResultCode(); + if (!isRiakOnPath()) { + addRiakOnPath(bucketTypeStatusScript); + } + bucketTypeStatusScript.execute(); + } + + @Override + public void bucketTypeActivate(String bucketTypeName) { + ScriptHelper bucketTypeStatusScript = newScript("bucket-type_activate") + .body.append(sudo(format("%s bucket-type activate %s", getRiakAdminCmd(), bucketTypeName))) + .failOnNonZeroResultCode(); + if (!isRiakOnPath()) { + addRiakOnPath(bucketTypeStatusScript); + } + bucketTypeStatusScript.execute(); + } + + @Override + public void recoverFailedNode(String nodeName) { + //TODO find ways to detect a faulty/failed node + //argument passed 'node' is any working node in the riak cluster + //following the instruction from: http://docs.basho.com/riak/latest/ops/running/recovery/failed-node/ + + if (hasJoinedCluster()) { + String failedNodeName = getRiakName(); + + + String stopCommand = format("%s stop", getRiakCmd()); + stopCommand = isPackageInstall() ? sudo(stopCommand) : stopCommand; + + String startCommand = format("%s start > /dev/null 2>&1 < /dev/null &", getRiakCmd()); + startCommand = isPackageInstall() ? sudo(startCommand) : startCommand; + + ScriptHelper recoverNodeScript = newScript("recoverNode") + .body.append(stopCommand) + .body.append(format("%s down %s", getRiakAdminCmd(), failedNodeName)) + .body.append(sudo(format("rm -rf %s", getRingStateDir()))) + .body.append(startCommand) + .body.append(sudo(format("%s cluster join %s", getRiakAdminCmd(), nodeName))) + .body.append(sudo(format("%s cluster plan", getRiakAdminCmd()))) + .body.append(sudo(format("%s cluster commit", getRiakAdminCmd()))) + .failOnNonZeroResultCode(); + + if (!isRiakOnPath()) { + addRiakOnPath(recoverNodeScript); + } + + recoverNodeScript.execute(); + + } else { + log.warn("entity {}: is not in the riak cluster", entity.getId()); + } + } + + @Override + public void setup() { + if(entity.getConfig(RiakNode.SEARCH_ENABLED)) { + // JavaSoftwareProcessSshDriver.setup() is called in order to install java + super.setup(); + } + } + + private Boolean hasJoinedCluster() { + return Boolean.TRUE.equals(entity.getAttribute(RiakNode.RIAK_NODE_HAS_JOINED_CLUSTER)); + } + + protected void checkRiakOnPath() { + boolean riakOnPath = newScript("riakOnPath") + .body.append("which riak") + .execute() == 0; + entity.setAttribute(RiakNode.RIAK_ON_PATH, riakOnPath); + } + + private String getRiakName() { + return entity.getAttribute(RiakNode.RIAK_NODE_NAME); + } + + private String getRingStateDir() { + //TODO: check for non-package install. + return isPackageInstall() ? "/var/lib/riak/ring" : Urls.mergePaths(getExpandedInstallDir(), "lib/ring"); + } + + protected boolean isVersion1() { + return getVersion().startsWith("1."); + } + + @Override + public String getOsMajorVersion() { + OsDetails osDetails = getMachine().getMachineDetails().getOsDetails(); + String osVersion = osDetails.getVersion(); + return osVersion.contains(".") ? osVersion.substring(0, osVersion.indexOf(".")) : osVersion; + } + + private void addRiakOnPath(ScriptHelper scriptHelper) { + Map<String, String> newPathVariable = ImmutableMap.of("PATH", sbinPath); +// log.warn("riak command not found on PATH. Altering future commands' environment variables from {} to {}", getShellEnvironment(), newPathVariable); + scriptHelper.environmentVariablesReset(newPathVariable); + } + + public Integer maxOpenFiles() { + return entity.getConfig(RiakNode.RIAK_MAX_OPEN_FILES); + } +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServer.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServer.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServer.java new file mode 100644 index 0000000..f04231a --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServer.java @@ -0,0 +1,81 @@ +/* + * 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.solr; + +import java.util.Map; + +import org.apache.brooklyn.catalog.Catalog; +import brooklyn.config.ConfigKey; +import brooklyn.entity.basic.BrooklynConfigKeys; +import brooklyn.entity.basic.ConfigKeys; +import brooklyn.entity.basic.SoftwareProcess; +import brooklyn.entity.java.UsesJava; +import brooklyn.entity.java.UsesJavaMXBeans; +import brooklyn.entity.java.UsesJmx; +import brooklyn.entity.proxying.ImplementedBy; +import brooklyn.event.basic.BasicAttributeSensorAndConfigKey; +import brooklyn.event.basic.PortAttributeSensorAndConfigKey; +import brooklyn.location.basic.PortRanges; +import brooklyn.util.flags.SetFromFlag; +import brooklyn.util.time.Duration; + +import com.google.common.collect.Maps; +import com.google.common.reflect.TypeToken; + +/** + * An {@link brooklyn.entity.Entity} that represents a Solr node. + */ +@Catalog(name="Apache Solr Node", description="Solr is the popular, blazing fast open source enterprise search " + + "platform from the Apache Lucene project.", iconUrl="classpath:///solr-logo.jpeg") +@ImplementedBy(SolrServerImpl.class) +public interface SolrServer extends SoftwareProcess, UsesJava, UsesJmx, UsesJavaMXBeans { + + @SetFromFlag("version") + ConfigKey<String> SUGGESTED_VERSION = ConfigKeys.newConfigKeyWithDefault(SoftwareProcess.SUGGESTED_VERSION, "4.7.2"); + + @SetFromFlag("downloadUrl") + BasicAttributeSensorAndConfigKey<String> DOWNLOAD_URL = new BasicAttributeSensorAndConfigKey<String>( + SoftwareProcess.DOWNLOAD_URL, "${driver.mirrorUrl}/${version}/solr-${version}.tgz"); + + /** download mirror, if desired */ + @SetFromFlag("mirrorUrl") + ConfigKey<String> MIRROR_URL = ConfigKeys.newStringConfigKey("solr.install.mirror.url", "URL of mirror", + "http://mirrors.ukfast.co.uk/sites/ftp.apache.org/lucene/solr/";); + + @SetFromFlag("solrPort") + PortAttributeSensorAndConfigKey SOLR_PORT = new PortAttributeSensorAndConfigKey("solr.http.port", "Solr HTTP communications port", + PortRanges.fromString("8983+")); + + @SetFromFlag("solrConfigTemplateUrl") + ConfigKey<String> SOLR_CONFIG_TEMPLATE_URL = ConfigKeys.newStringConfigKey( + "solr.config.templateUrl", "Template file (in freemarker format) for the solr.xml config file", + "classpath://org/apache/brooklyn/entity/nosql/solr/solr.xml"); + + @SetFromFlag("coreConfigMap") + ConfigKey<Map<String, String>> SOLR_CORE_CONFIG = ConfigKeys.newConfigKey(new TypeToken<Map<String, String>>() { }, + "solr.core.config", "Map of core names to core configuration archive URL", + Maps.<String, String>newHashMap()); + + ConfigKey<Duration> START_TIMEOUT = ConfigKeys.newConfigKeyWithDefault(BrooklynConfigKeys.START_TIMEOUT, Duration.FIVE_MINUTES); + + /* Accessors used from template */ + + Integer getSolrPort(); + +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerDriver.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerDriver.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerDriver.java new file mode 100644 index 0000000..72e1049 --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerDriver.java @@ -0,0 +1,30 @@ +/* + * 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.solr; + +import brooklyn.entity.basic.SoftwareProcessDriver; +import brooklyn.entity.java.JavaSoftwareProcessDriver; + +public interface SolrServerDriver extends JavaSoftwareProcessDriver { + + Integer getSolrPort(); + + String getSolrConfigTemplateUrl(); + +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerImpl.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerImpl.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerImpl.java new file mode 100644 index 0000000..a42d15b --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerImpl.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.solr; + +import brooklyn.entity.basic.Attributes; +import brooklyn.entity.basic.SoftwareProcessImpl; +import brooklyn.event.feed.http.HttpFeed; +import brooklyn.event.feed.http.HttpPollConfig; +import brooklyn.event.feed.http.HttpValueFunctions; +import brooklyn.location.access.BrooklynAccessUtils; +import com.google.common.base.Functions; +import com.google.common.net.HostAndPort; + +import java.net.URI; +import java.util.concurrent.TimeUnit; + +/** + * Implementation of {@link SolrServer}. + */ +public class SolrServerImpl extends SoftwareProcessImpl implements SolrServer { + + @Override + public Integer getSolrPort() { + return getAttribute(SolrServer.SOLR_PORT); + } + + @Override + public Class<SolrServerDriver> getDriverInterface() { + return SolrServerDriver.class; + } + + private volatile HttpFeed httpFeed; + + @Override + protected void connectSensors() { + super.connectSensors(); + + HostAndPort hp = BrooklynAccessUtils.getBrooklynAccessibleAddress(this, getSolrPort()); + + String solrUri = String.format("http://%s:%d/solr";, hp.getHostText(), hp.getPort()); + setAttribute(Attributes.MAIN_URI, URI.create(solrUri)); + + httpFeed = HttpFeed.builder() + .entity(this) + .period(500, TimeUnit.MILLISECONDS) + .baseUri(solrUri) + .poll(new HttpPollConfig<Boolean>(SERVICE_UP) + .onSuccess(HttpValueFunctions.responseCodeEquals(200)) + .onFailureOrException(Functions.constant(false))) + .build(); + } + + @Override + public void disconnectSensors() { + super.disconnectSensors(); + + if (httpFeed != null) httpFeed.stop(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/d5cf5285/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerSshDriver.java ---------------------------------------------------------------------- diff --git a/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerSshDriver.java b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerSshDriver.java new file mode 100644 index 0000000..f05624e --- /dev/null +++ b/software/nosql/src/main/java/org/apache/brooklyn/entity/nosql/solr/SolrServerSshDriver.java @@ -0,0 +1,158 @@ +/* + * 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.solr; + +import static java.lang.String.format; + +import java.util.List; +import java.util.Map; +import java.util.Set; + +import brooklyn.entity.java.JavaSoftwareProcessSshDriver; +import brooklyn.entity.java.UsesJmx; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import brooklyn.entity.basic.AbstractSoftwareProcessSshDriver; +import brooklyn.entity.basic.Entities; +import brooklyn.location.Location; +import brooklyn.location.basic.SshMachineLocation; +import brooklyn.util.collections.MutableMap; +import brooklyn.util.file.ArchiveUtils; +import brooklyn.util.net.Networking; +import brooklyn.util.net.Urls; +import brooklyn.util.os.Os; +import brooklyn.util.ssh.BashCommands; +import brooklyn.util.stream.Streams; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Sets; + +/** + * Start a {@link SolrServer} in a {@link Location} accessible over ssh. + */ +public class SolrServerSshDriver extends JavaSoftwareProcessSshDriver implements SolrServerDriver { + + private static final Logger log = LoggerFactory.getLogger(SolrServerSshDriver.class); + + public SolrServerSshDriver(SolrServerImpl entity, SshMachineLocation machine) { + super(entity, machine); + } + + @Override + public Integer getSolrPort() { return entity.getAttribute(SolrServer.SOLR_PORT); } + + @Override + public String getSolrConfigTemplateUrl() { return entity.getConfig(SolrServer.SOLR_CONFIG_TEMPLATE_URL); } + + public String getMirrorUrl() { return entity.getConfig(SolrServer.MIRROR_URL); } + + public String getPidFile() { return Os.mergePaths(getRunDir(), "solr.pid"); } + + @Override + public void preInstall() { + resolver = Entities.newDownloader(this); + setExpandedInstallDir(Os.mergePaths(getInstallDir(), resolver.getUnpackedDirectoryName(format("solr-%s", getVersion())))); + } + + @Override + public void install() { + List<String> urls = resolver.getTargets(); + String saveAs = resolver.getFilename(); + + List<String> commands = ImmutableList.<String>builder() + .addAll(BashCommands.commandsToDownloadUrlsAs(urls, saveAs)) + .add(BashCommands.INSTALL_TAR) + .add("tar xzfv " + saveAs) + .build(); + + newScript(INSTALLING) + .failOnNonZeroResultCode() + .body.append(commands) + .execute(); + } + + public Set<Integer> getPortsUsed() { + Set<Integer> result = Sets.newLinkedHashSet(super.getPortsUsed()); + result.addAll(getPortMap().values()); + return result; + } + + private Map<String, Integer> getPortMap() { + return ImmutableMap.<String, Integer>builder() + .put("solrPort", getSolrPort()) + .put("jmxPort", entity.getAttribute(UsesJmx.JMX_PORT)) + .put("rmiPort", entity.getAttribute(UsesJmx.RMI_REGISTRY_PORT)) + .build(); + } + + @Override + public void customize() { + log.debug("Customizing {}", entity); + Networking.checkPortsValid(getPortMap()); + + ImmutableList.Builder<String> commands = new ImmutableList.Builder<String>() + .add("mkdir contrib") + .add("mkdir solr") + .add(String.format("cp -R %s/example/{etc,contexts,lib,logs,resources,webapps} .", getExpandedInstallDir())) + .add(String.format("cp %s/example/start.jar .", getExpandedInstallDir())) + .add(String.format("cp %s/dist/*.jar lib/", getExpandedInstallDir())) + .add(String.format("cp %s/contrib/*/lib/*.jar contrib/", getExpandedInstallDir())); + + newScript(CUSTOMIZING) + .body.append(commands.build()) + .execute(); + + // Copy the solr.xml configuration file across + String configFileContents = processTemplate(getSolrConfigTemplateUrl()); + String destinationConfigFile = String.format("%s/solr/solr.xml", getRunDir()); + getMachine().copyTo(Streams.newInputStreamWithContents(configFileContents), destinationConfigFile); + + // Copy the core definitions across + Map<String, String> coreConfig = entity.getConfig(SolrServer.SOLR_CORE_CONFIG); + for (String core : coreConfig.keySet()) { + String url = coreConfig.get(core); + String solr = Urls.mergePaths(getRunDir(), "solr"); + ArchiveUtils.deploy(url, getMachine(), solr); + } + } + + @Override + public void launch() { + newScript(MutableMap.of(USE_PID_FILE, getPidFile()), LAUNCHING) + .body.append("nohup java $JAVA_OPTS -jar start.jar > ./logs/console.log 2>&1 &") + .execute(); + } + + @Override + public boolean isRunning() { + return newScript(MutableMap.of(USE_PID_FILE, getPidFile()), CHECK_RUNNING).execute() == 0; + } + + @Override + public void stop() { + newScript(MutableMap.of(USE_PID_FILE, getPidFile()), STOPPING).execute(); + } + + @Override + protected String getLogFileLocation() { + return Urls.mergePaths(getRunDir(), "solr", "logs", "solr.log"); + } +}
