| /* |
| * 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.cassandra; |
| |
| import java.io.IOException; |
| import java.lang.reflect.Method; |
| import java.math.BigInteger; |
| import java.net.Socket; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import javax.annotation.Nullable; |
| import javax.management.ObjectName; |
| |
| import org.apache.brooklyn.api.entity.Entity; |
| import org.apache.brooklyn.api.location.MachineLocation; |
| import org.apache.brooklyn.api.location.MachineProvisioningLocation; |
| import org.apache.brooklyn.api.sensor.EnricherSpec; |
| import org.apache.brooklyn.core.effector.EffectorBody; |
| import org.apache.brooklyn.core.entity.Attributes; |
| import org.apache.brooklyn.core.entity.Entities; |
| import org.apache.brooklyn.core.location.Machines; |
| import org.apache.brooklyn.core.location.access.BrooklynAccessUtils; |
| import org.apache.brooklyn.core.location.cloud.CloudLocationConfig; |
| 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.java.JavaAppUtils; |
| import org.apache.brooklyn.entity.java.UsesJmx; |
| import org.apache.brooklyn.entity.software.base.SoftwareProcessImpl; |
| import org.apache.brooklyn.feed.function.FunctionFeed; |
| import org.apache.brooklyn.feed.function.FunctionPollConfig; |
| import org.apache.brooklyn.feed.jmx.JmxAttributePollConfig; |
| import org.apache.brooklyn.feed.jmx.JmxFeed; |
| import org.apache.brooklyn.feed.jmx.JmxHelper; |
| import org.apache.brooklyn.feed.jmx.JmxOperationPollConfig; |
| import org.apache.brooklyn.policy.enricher.RollingTimeWindowMeanEnricher; |
| import org.apache.brooklyn.policy.enricher.TimeWeightedDeltaEnricher; |
| import org.apache.brooklyn.util.collections.MutableSet; |
| import org.apache.brooklyn.util.core.config.ConfigBag; |
| import org.apache.brooklyn.util.core.text.TemplateProcessor; |
| import org.apache.brooklyn.util.exceptions.Exceptions; |
| import org.apache.brooklyn.util.guava.Functionals; |
| import org.apache.brooklyn.util.guava.Maybe; |
| import org.apache.brooklyn.util.text.Strings; |
| import org.apache.brooklyn.util.time.Duration; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.base.Function; |
| import com.google.common.base.Functions; |
| import com.google.common.base.Joiner; |
| import com.google.common.base.Predicate; |
| import com.google.common.base.Predicates; |
| import com.google.common.base.Splitter; |
| import com.google.common.base.Throwables; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import com.google.common.net.HostAndPort; |
| |
| /** |
| * Implementation of {@link CassandraNode}. |
| */ |
| public class CassandraNodeImpl extends SoftwareProcessImpl implements CassandraNode { |
| |
| private static final Logger log = LoggerFactory.getLogger(CassandraNodeImpl.class); |
| |
| private final AtomicReference<Boolean> detectedCloudSensors = new AtomicReference<Boolean>(false); |
| |
| public CassandraNodeImpl() { |
| } |
| |
| @Override |
| public void init() { |
| super.init(); |
| |
| // TODO PERSISTENCE WORKAROUND kept anonymous class in case referenced in persisted state |
| new EffectorBody<String>() { |
| @Override |
| public String call(ConfigBag parameters) { |
| return executeScript((String)parameters.getStringKey("commands")); |
| } |
| }; |
| |
| getMutableEntityType().addEffector(EXECUTE_SCRIPT, new ExecuteScriptEffectorBody(this)); |
| |
| Entities.checkRequiredUrl(this, getCassandraConfigTemplateUrl()); |
| Entities.getRequiredUrlConfig(this, CASSANDRA_RACKDC_CONFIG_TEMPLATE_URL); |
| |
| connectEnrichers(); |
| } |
| |
| private static class ExecuteScriptEffectorBody extends EffectorBody<String> { |
| private final CassandraNode entity; |
| |
| public ExecuteScriptEffectorBody(CassandraNode entity) { |
| this.entity = entity; |
| } |
| @Override |
| public String call(ConfigBag parameters) { |
| return entity.executeScript((String)parameters.getStringKey("commands")); |
| } |
| } |
| |
| /** |
| * Some clouds (e.g. Rackspace) give us VMs that have two nics: one for private and one for public. |
| * If the private IP is used then it doesn't work, even for a cluster purely internal to Rackspace! |
| * |
| * TODO Ugly. Need to understand more and find a better fix. Perhaps in Cassandra itself if necessary. |
| * Also need to investigate further: |
| * - does it still fail if BroadcastAddress is set to private IP? |
| * - is `openIptables` opening it up for both interfaces? |
| * - for aws->rackspace comms between nodes (thus using the public IP), will it be listening on an accessible port? |
| * - ideally do a check, open a server on one port on the machine, see if it is contactable on the public address; |
| * and set that as a flag on the cloud |
| */ |
| protected void setCloudPreferredSensorNames() { |
| if (detectedCloudSensors.get()) return; |
| synchronized (detectedCloudSensors) { |
| if (detectedCloudSensors.get()) return; |
| |
| MachineProvisioningLocation<?> loc = getProvisioningLocation(); |
| if (loc != null) { |
| try { |
| Method method = loc.getClass().getMethod("getProvider"); |
| method.setAccessible(true); |
| String provider = (String) method.invoke(loc); |
| String result = "(nothing special)"; |
| if (provider!=null) { |
| if (provider.contains("rackspace") || provider.contains("cloudservers") || provider.contains("softlayer")) { |
| /* These clouds have 2 NICs and it has to be consistent, so use public IP here to allow external access; |
| * (TODO internal access could be configured to improve performance / lower cost, |
| * if we know all nodes are visible to each other) */ |
| if (getConfig(LISTEN_ADDRESS_SENSOR)==null) |
| config().set(LISTEN_ADDRESS_SENSOR, CassandraNode.ADDRESS.getName()); |
| if (getConfig(BROADCAST_ADDRESS_SENSOR)==null) |
| config().set(BROADCAST_ADDRESS_SENSOR, CassandraNode.ADDRESS.getName()); |
| result = "public IP for both listen and broadcast"; |
| } else if (provider.contains("google-compute")) { |
| /* Google nodes cannot reach themselves/each-other on the public IP, |
| * and there is no hostname, so use private IP here */ |
| if (getConfig(LISTEN_ADDRESS_SENSOR)==null) |
| config().set(LISTEN_ADDRESS_SENSOR, CassandraNode.SUBNET_HOSTNAME.getName()); |
| if (getConfig(BROADCAST_ADDRESS_SENSOR)==null) |
| config().set(BROADCAST_ADDRESS_SENSOR, CassandraNode.SUBNET_HOSTNAME.getName()); |
| result = "private IP for both listen and broadcast"; |
| } |
| } |
| log.debug("Cassandra NICs inferred {} for {}; using location {}, based on provider {}", new Object[] {result, this, loc, provider}); |
| } catch (Exception e) { |
| log.debug("Cassandra NICs auto-detection failed for {} in location {}: {}", new Object[] {this, loc, e}); |
| } |
| } |
| detectedCloudSensors.set(true); |
| } |
| } |
| |
| @Override |
| protected void preStart() { |
| super.preStart(); |
| setCloudPreferredSensorNames(); |
| } |
| |
| // Used for freemarker |
| @Override |
| public String getMajorMinorVersion() { |
| String version = getConfig(CassandraNode.SUGGESTED_VERSION); |
| if (Strings.isBlank(version)) return ""; |
| List<String> versionParts = ImmutableList.copyOf(Splitter.on(".").split(version)); |
| return versionParts.get(0) + (versionParts.size() > 1 ? "."+versionParts.get(1) : ""); |
| } |
| |
| public String getCassandraConfigTemplateUrl() { |
| String templatedUrl = getConfig(CassandraNode.CASSANDRA_CONFIG_TEMPLATE_URL); |
| return TemplateProcessor.processTemplateContents(templatedUrl, this, ImmutableMap.<String, Object>of()); |
| } |
| |
| @Override public Integer getGossipPort() { return getAttribute(CassandraNode.GOSSIP_PORT); } |
| @Override public Integer getSslGossipPort() { return getAttribute(CassandraNode.SSL_GOSSIP_PORT); } |
| @Override public Integer getThriftPort() { return getAttribute(CassandraNode.THRIFT_PORT); } |
| @Override public Integer getNativeTransportPort() { return getAttribute(CassandraNode.NATIVE_TRANSPORT_PORT); } |
| @Override public String getClusterName() { return getAttribute(CassandraNode.CLUSTER_NAME); } |
| |
| @Override public int getNumTokensPerNode() { |
| return getConfig(CassandraNode.NUM_TOKENS_PER_NODE); |
| } |
| |
| @Override public Set<BigInteger> getTokens() { |
| // Prefer an already-set attribute over the config. |
| // Prefer TOKENS over TOKEN. |
| Set<BigInteger> tokens = getAttribute(CassandraNode.TOKENS); |
| if (tokens == null) { |
| tokens = getConfig(CassandraNode.TOKENS); |
| } |
| return tokens; |
| } |
| |
| @Override public String getTokensAsString() { |
| // TODO check what is required when replacing failed node. |
| // with vnodes in Cassandra 2.x, don't bother supplying token |
| Set<BigInteger> tokens = getTokens(); |
| if (tokens == null) return ""; |
| return Joiner.on(",").join(tokens); |
| } |
| |
| @Override public String getListenAddress() { |
| String sensorName = getConfig(LISTEN_ADDRESS_SENSOR); |
| if (Strings.isNonBlank(sensorName)) |
| return Entities.submit(this, DependentConfiguration.attributeWhenReady(this, Sensors.newStringSensor(sensorName))).getUnchecked(); |
| |
| String subnetAddress = getAttribute(CassandraNode.SUBNET_ADDRESS); |
| return Strings.isNonBlank(subnetAddress) ? subnetAddress : getAttribute(CassandraNode.ADDRESS); |
| } |
| |
| @Override public String getBroadcastAddress() { |
| String sensorName = getConfig(BROADCAST_ADDRESS_SENSOR); |
| if (Strings.isNonBlank(sensorName)) |
| return Entities.submit(this, DependentConfiguration.attributeWhenReady(this, Sensors.newStringSensor(sensorName))).getUnchecked(); |
| |
| String snitchName = getConfig(CassandraNode.ENDPOINT_SNITCH_NAME); |
| if (snitchName.equals("Ec2MultiRegionSnitch") || snitchName.contains("MultiCloudSnitch")) { |
| // http://www.datastax.com/documentation/cassandra/2.0/mobile/cassandra/architecture/architectureSnitchEC2MultiRegion_c.html |
| // describes that the listen_address is set to the private IP, and the broadcast_address is set to the public IP. |
| return getAttribute(CassandraNode.ADDRESS); |
| } else if (!getDriver().isClustered()) { |
| return getListenAddress(); |
| } else { |
| // In other situations, prefer the hostname, so other regions can see it |
| // *Unless* hostname resolves at the target to a local-only interface which is different to ADDRESS |
| // (workaround for issue deploying to localhost) |
| String hostname = getAttribute(CassandraNode.HOSTNAME); |
| try { |
| String resolvedAddress = getDriver().getResolvedAddress(hostname); |
| if (resolvedAddress==null) { |
| log.debug("Cassandra using broadcast address "+getListenAddress()+" for "+this+" because hostname "+hostname+" could not be resolved at remote machine"); |
| return getListenAddress(); |
| } |
| if (resolvedAddress.equals("127.0.0.1")) { |
| log.debug("Cassandra using broadcast address "+getListenAddress()+" for "+this+" because hostname "+hostname+" resolves to 127.0.0.1"); |
| return getListenAddress(); |
| } |
| return hostname; |
| } catch (Exception e) { |
| Exceptions.propagateIfFatal(e); |
| log.warn("Error resolving hostname "+hostname+" for "+this+": "+e, e); |
| return hostname; |
| } |
| } |
| } |
| /** not always the private IP, if public IP has been insisted on for broadcast, e.g. setting up a rack topology */ |
| // have not confirmed this does the right thing in all clouds ... only used for rack topology however |
| @Override |
| public String getPrivateIp() { |
| String sensorName = getConfig(BROADCAST_ADDRESS_SENSOR); |
| if (Strings.isNonBlank(sensorName)) { |
| return getAttribute(Sensors.newStringSensor(sensorName)); |
| } else { |
| String subnetAddress = getAttribute(CassandraNode.SUBNET_ADDRESS); |
| return Strings.isNonBlank(subnetAddress) ? subnetAddress : getAttribute(CassandraNode.ADDRESS); |
| } |
| } |
| @Override |
| public String getPublicIp() { |
| // may need to be something else in google |
| return getAttribute(CassandraNode.ADDRESS); |
| } |
| |
| @Override public String getRpcAddress() { |
| String sensorName = getConfig(RPC_ADDRESS_SENSOR); |
| if (Strings.isNonBlank(sensorName)) |
| return Entities.submit(this, DependentConfiguration.attributeWhenReady(this, Sensors.newStringSensor(sensorName))).getUnchecked(); |
| return "0.0.0.0"; |
| } |
| |
| @Override public String getSeeds() { |
| Set<Entity> seeds = getConfig(CassandraNode.INITIAL_SEEDS); |
| if (seeds==null) { |
| log.warn("No seeds available when requested for "+this, new Throwable("source of no Cassandra seeds when requested")); |
| return null; |
| } |
| String snitchName = getConfig(CassandraNode.ENDPOINT_SNITCH_NAME); |
| MutableSet<String> seedsHostnames = MutableSet.of(); |
| for (Entity entity : seeds) { |
| // tried removing ourselves if there are other nodes, but that is a BAD idea! |
| // blows up with a "java.lang.RuntimeException: No other nodes seen!" |
| |
| if (snitchName.equals("Ec2MultiRegionSnitch") || snitchName.contains("MultiCloudSnitch")) { |
| // http://www.datastax.com/documentation/cassandra/2.0/mobile/cassandra/architecture/architectureSnitchEC2MultiRegion_c.html |
| // says the seeds should be public IPs. |
| seedsHostnames.add(entity.getAttribute(CassandraNode.ADDRESS)); |
| } else { |
| String sensorName = getConfig(BROADCAST_ADDRESS_SENSOR); |
| if (Strings.isNonBlank(sensorName)) { |
| seedsHostnames.add(entity.getAttribute(Sensors.newStringSensor(sensorName))); |
| } else { |
| Maybe<String> optionalSeedHostname = Machines.findSubnetOrPublicHostname(entity); |
| if (optionalSeedHostname.isPresent()) { |
| String seedHostname = optionalSeedHostname.get(); |
| seedsHostnames.add(seedHostname); |
| } else { |
| log.warn("In node {}, seed hostname missing for {}; not including in seeds list", this, entity); |
| } |
| } |
| } |
| } |
| |
| String result = Strings.join(seedsHostnames, ","); |
| log.info("Seeds for {}: {}", this, result); |
| return result; |
| } |
| |
| // referenced by cassandra-rackdc.properties, read by some of the cassandra snitches |
| public String getDatacenterName() { |
| String name = getAttribute(CassandraNode.DATACENTER_NAME); |
| if (name == null) { |
| MachineLocation machine = getMachineOrNull(); |
| MachineProvisioningLocation<?> provisioningLocation = getProvisioningLocation(); |
| if (machine != null) { |
| name = machine.getConfig(CloudLocationConfig.CLOUD_REGION_ID); |
| } |
| if (name == null && provisioningLocation != null) { |
| name = provisioningLocation.getConfig(CloudLocationConfig.CLOUD_REGION_ID); |
| } |
| if (name == null) { |
| name = "UNKNOWN_DATACENTER"; |
| } |
| sensors().set(DATACENTER_NAME, name); |
| } |
| return name; |
| } |
| |
| public String getRackName() { |
| String name = getAttribute(CassandraNode.RACK_NAME); |
| if (name == null) { |
| MachineLocation machine = getMachineOrNull(); |
| MachineProvisioningLocation<?> provisioningLocation = getProvisioningLocation(); |
| if (machine != null) { |
| name = machine.getConfig(CloudLocationConfig.CLOUD_AVAILABILITY_ZONE_ID); |
| } |
| if (name == null && provisioningLocation != null) { |
| name = provisioningLocation.getConfig(CloudLocationConfig.CLOUD_AVAILABILITY_ZONE_ID); |
| } |
| if (name == null) { |
| name = "UNKNOWN_RACK"; |
| } |
| sensors().set(RACK_NAME, name); |
| } |
| return name; |
| } |
| |
| @Override |
| public Class<? extends CassandraNodeDriver> getDriverInterface() { |
| return CassandraNodeDriver.class; |
| } |
| |
| @Override |
| public CassandraNodeDriver getDriver() { |
| return (CassandraNodeDriver) super.getDriver(); |
| } |
| |
| private volatile JmxFeed jmxFeed; |
| private volatile FunctionFeed functionFeed; |
| private JmxFeed jmxMxBeanFeed; |
| private JmxHelper jmxHelper; |
| private ObjectName storageServiceMBean = JmxHelper.createObjectName("org.apache.cassandra.db:type=StorageService"); |
| private ObjectName readStageMBean = JmxHelper.createObjectName("org.apache.cassandra.request:type=ReadStage"); |
| private ObjectName mutationStageMBean = JmxHelper.createObjectName("org.apache.cassandra.request:type=MutationStage"); |
| private ObjectName snitchMBean = JmxHelper.createObjectName("org.apache.cassandra.db:type=EndpointSnitchInfo"); |
| |
| |
| @SuppressWarnings({ "unchecked", "rawtypes" }) |
| @Override |
| protected void connectSensors() { |
| // "cassandra" isn't really a protocol, but okay for now |
| sensors().set(DATASTORE_URL, "cassandra://"+getAttribute(HOSTNAME)+":"+getAttribute(THRIFT_PORT)); |
| |
| super.connectSensors(); |
| |
| jmxHelper = new JmxHelper(this); |
| boolean retrieveUsageMetrics = getConfig(RETRIEVE_USAGE_METRICS); |
| |
| if (getDriver().isJmxEnabled()) { |
| jmxFeed = JmxFeed.builder() |
| .entity(this) |
| .period(3000, TimeUnit.MILLISECONDS) |
| .helper(jmxHelper) |
| .pollAttribute(new JmxAttributePollConfig<Boolean>(SERVICE_UP_JMX) |
| .objectName(storageServiceMBean) |
| .attributeName("Initialized") |
| .onSuccess(Functions.forPredicate(Predicates.notNull())) |
| .onException(Functions.constant(false)) |
| .suppressDuplicates(true)) |
| .pollAttribute(new JmxAttributePollConfig<Set<BigInteger>>(TOKENS) |
| .objectName(storageServiceMBean) |
| .attributeName("TokenToEndpointMap") |
| .onSuccess(new Function<Object, Set<BigInteger>>() { |
| @Override |
| public Set<BigInteger> apply(@Nullable Object arg) { |
| Map input = (Map)arg; |
| if (input == null || input.isEmpty()) return null; |
| // FIXME does not work on aws-ec2, uses RFC1918 address |
| Predicate<String> self = Predicates.in(ImmutableList.of(getAttribute(HOSTNAME), getAttribute(ADDRESS), getAttribute(SUBNET_ADDRESS), getAttribute(SUBNET_HOSTNAME))); |
| Set<String> tokens = Maps.filterValues(input, self).keySet(); |
| Set<BigInteger> result = Sets.newLinkedHashSet(); |
| for (String token : tokens) { |
| result.add(new BigInteger(token)); |
| } |
| return result; |
| }}) |
| .onException(Functions.<Set<BigInteger>>constant(null)) |
| .suppressDuplicates(true)) |
| .pollOperation(new JmxOperationPollConfig<String>(DATACENTER_NAME) |
| .period(60, TimeUnit.SECONDS) |
| .objectName(snitchMBean) |
| .operationName("getDatacenter") |
| .operationParams(ImmutableList.of(getBroadcastAddress())) |
| .onException(Functions.<String>constant(null)) |
| .suppressDuplicates(true)) |
| .pollOperation(new JmxOperationPollConfig<String>(RACK_NAME) |
| .period(60, TimeUnit.SECONDS) |
| .objectName(snitchMBean) |
| .operationName("getRack") |
| .operationParams(ImmutableList.of(getBroadcastAddress())) |
| .onException(Functions.<String>constant(null)) |
| .suppressDuplicates(true)) |
| .pollAttribute(new JmxAttributePollConfig<Integer>(PEERS) |
| .objectName(storageServiceMBean) |
| .attributeName("TokenToEndpointMap") |
| .onSuccess(new Function<Object, Integer>() { |
| @Override |
| public Integer apply(@Nullable Object arg) { |
| Map input = (Map)arg; |
| if (input == null || input.isEmpty()) return 0; |
| return input.size(); |
| } |
| }) |
| .onException(Functions.constant(-1))) |
| .pollAttribute(new JmxAttributePollConfig<Integer>(LIVE_NODE_COUNT) |
| .objectName(storageServiceMBean) |
| .attributeName("LiveNodes") |
| .onSuccess(new Function<Object, Integer>() { |
| @Override |
| public Integer apply(@Nullable Object arg) { |
| List input = (List)arg; |
| if (input == null || input.isEmpty()) return 0; |
| return input.size(); |
| } |
| }) |
| .onException(Functions.constant(-1))) |
| .pollAttribute(new JmxAttributePollConfig<Integer>(READ_ACTIVE) |
| .objectName(readStageMBean) |
| .attributeName("ActiveCount") |
| .onException(Functions.constant((Integer)null)) |
| .enabled(retrieveUsageMetrics)) |
| .pollAttribute(new JmxAttributePollConfig<Long>(READ_PENDING) |
| .objectName(readStageMBean) |
| .attributeName("PendingTasks") |
| .onException(Functions.constant((Long)null)) |
| .enabled(retrieveUsageMetrics)) |
| .pollAttribute(new JmxAttributePollConfig<Long>(READ_COMPLETED) |
| .objectName(readStageMBean) |
| .attributeName("CompletedTasks") |
| .onException(Functions.constant((Long)null)) |
| .enabled(retrieveUsageMetrics)) |
| .pollAttribute(new JmxAttributePollConfig<Integer>(WRITE_ACTIVE) |
| .objectName(mutationStageMBean) |
| .attributeName("ActiveCount") |
| .onException(Functions.constant((Integer)null)) |
| .enabled(retrieveUsageMetrics)) |
| .pollAttribute(new JmxAttributePollConfig<Long>(WRITE_PENDING) |
| .objectName(mutationStageMBean) |
| .attributeName("PendingTasks") |
| .onException(Functions.constant((Long)null)) |
| .enabled(retrieveUsageMetrics)) |
| .pollAttribute(new JmxAttributePollConfig<Long>(WRITE_COMPLETED) |
| .objectName(mutationStageMBean) |
| .attributeName("CompletedTasks") |
| .onException(Functions.constant((Long)null)) |
| .enabled(retrieveUsageMetrics)) |
| .build(); |
| |
| jmxMxBeanFeed = JavaAppUtils.connectMXBeanSensors(this); |
| } |
| |
| if (Boolean.TRUE.equals(getConfig(USE_THRIFT_MONITORING))) { |
| functionFeed = FunctionFeed.builder() |
| .entity(this) |
| .period(3000, TimeUnit.MILLISECONDS) |
| .poll(new FunctionPollConfig<Long, Long>(THRIFT_PORT_LATENCY) |
| .onException(Functions.constant(-1L)) |
| .callable(new ThriftLatencyChecker(CassandraNodeImpl.this)) |
| .enabled(retrieveUsageMetrics)) |
| .build(); |
| } |
| |
| connectServiceUpIsRunning(); |
| } |
| |
| protected void connectEnrichers() { |
| connectEnrichers(Duration.TEN_SECONDS); |
| } |
| |
| protected void connectEnrichers(Duration windowPeriod) { |
| JavaAppUtils.connectJavaAppServerPolicies(this); |
| |
| enrichers().add(EnricherSpec.create(TimeWeightedDeltaEnricher.class) |
| .configure("producer", this) |
| .configure("source", READ_COMPLETED) |
| .configure("target", READS_PER_SECOND_LAST) |
| .configure("unitMillis", 1000)); |
| |
| if (windowPeriod!=null) { |
| enrichers().add(EnricherSpec.create(RollingTimeWindowMeanEnricher.class) |
| .configure("producer", this) |
| .configure("source", THRIFT_PORT_LATENCY) |
| .configure("target", THRIFT_PORT_LATENCY_IN_WINDOW) |
| .configure("timePeriod", windowPeriod)); |
| enrichers().add(EnricherSpec.create(RollingTimeWindowMeanEnricher.class) |
| .configure("producer", this) |
| .configure("source", READS_PER_SECOND_LAST) |
| .configure("target", READS_PER_SECOND_IN_WINDOW) |
| .configure("timePeriod", windowPeriod)); |
| enrichers().add(EnricherSpec.create(RollingTimeWindowMeanEnricher.class) |
| .configure("producer", this) |
| .configure("source", WRITES_PER_SECOND_LAST) |
| .configure("target", WRITES_PER_SECOND_IN_WINDOW) |
| .configure("timePeriod", windowPeriod)); |
| } |
| |
| // service-up checks |
| if (Boolean.TRUE.equals(getConfig(USE_THRIFT_MONITORING))) { |
| enrichers().add(Enrichers.builder().updatingMap(Attributes.SERVICE_NOT_UP_INDICATORS) |
| .from(THRIFT_PORT_LATENCY) |
| .computing(Functionals.ifEquals(-1L).value("Thrift latency polling failed") ) |
| .build()); |
| } |
| |
| if (Boolean.TRUE.equals(getConfig(UsesJmx.USE_JMX))) { |
| enrichers().add(Enrichers.builder().updatingMap(Attributes.SERVICE_NOT_UP_INDICATORS) |
| .from(SERVICE_UP_JMX) |
| .computing(Functionals.ifEquals(false).value("JMX reports not up") ) |
| .build()); |
| } |
| } |
| |
| @Override |
| public void disconnectSensors() { |
| super.disconnectSensors(); |
| |
| disconnectServiceUpIsRunning(); |
| if (jmxFeed != null) jmxFeed.stop(); |
| if (jmxMxBeanFeed != null) jmxMxBeanFeed.stop(); |
| if (jmxHelper != null) jmxHelper.terminate(); |
| if (functionFeed != null) functionFeed.stop(); |
| } |
| |
| @Override |
| public void setToken(String token) { |
| try { |
| if (!jmxHelper.isConnected()) jmxHelper.connect();; |
| jmxHelper.operation(storageServiceMBean, "move", token); |
| log.info("Moved server {} to token {}", getId(), token); |
| } catch (IOException ioe) { |
| Throwables.propagate(ioe); |
| } |
| } |
| |
| @Override |
| public String executeScript(String commands) { |
| return getDriver().executeScriptAsync(commands).block().getStdout(); |
| } |
| |
| private static class ThriftLatencyChecker implements Callable<Long> { |
| private final CassandraNode entity; |
| |
| public ThriftLatencyChecker(CassandraNode entity) { |
| this.entity = entity; |
| } |
| @Override |
| public Long call() { |
| Integer privatePort = entity.getThriftPort(); |
| if (privatePort == null) return -1L; |
| |
| HostAndPort hp = BrooklynAccessUtils.getBrooklynAccessibleAddress(entity, privatePort); |
| |
| try { |
| long start = System.currentTimeMillis(); |
| Socket s = new Socket(hp.getHostText(), hp.getPort()); |
| s.close(); |
| long latency = System.currentTimeMillis() - start; |
| return latency; |
| } catch (Exception e) { |
| Exceptions.propagateIfFatal(e); |
| if (log.isDebugEnabled()) |
| log.debug("Cassandra thrift port poll failure: "+e); |
| return -1L; |
| } |
| } |
| } |
| } |
