andreacavalli/scylla-jmx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updating the http client and support API exception

Browse Source 
The API returns errors with an HTTP code 400, 404 or 500 depending on
the cause with a json object that contains the failure reason.  The
error message should be populate to the JMX calling client, translated
to the appropriate exception.

This patch adds the ability to detect API failure and throw a runtime
exception with the returned message.

It is up to the calling method what to do with the exception, if it
would do nothing, the calling client would get a RuntimeException,
depends on origin MBean definition, the caller can catch the exception
and throw a specific kind.

Note that any exception that is thrown must be known to the JMX client
or it will not be able to process it.

As a first step, this patch replaces the jersey client to the newer
version under glassfish, which has an easy way of getting a Reply
object, and check its status before returning the results.

The only difference in the method that uses the APIClient is the use of
MultivaluedHashMap.

The following MBean implementation where changed

ColumnFamily
CompactionManager
Gossiper
EndpointSnitchInfo
CacheService
StorageProxy
StorageService

Signed-off-by: Amnon Heiman <amnon@scylladb.com>
This commit is contained in:
Amnon Heiman 2015-10-28 11:25:37 +02:00
parent 1c180c7164
commit 649922fed8
9 changed files with 150 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
pom.xml
View File

@ -16,9 +16,24 @@
<dependencies>
<dependency>
<groupId>com.sun.jersey</groupId>
<groupId>org.glassfish.jersey.core</groupId>
<artifactId>jersey-common</artifactId>
<version>2.22.1</version>
</dependency>
<dependency>
<groupId>javax.ws.rs</groupId>
<artifactId>javax.ws.rs-api</artifactId>
<version>2.0.1</version>
</dependency>
<dependency>
<groupId>javax.ws.rs</groupId>
<artifactId>jsr311-api</artifactId>
<version>1.1.1</version>
</dependency>
<dependency>
<groupId>org.glassfish.jersey.core</groupId>
<artifactId>jersey-client</artifactId>
<version>1.19</version>
<version>2.22.1</version>
</dependency>
<dependency>

81
src/main/java/com/cloudius/urchin/api/APIClient.java
View File

@ -12,6 +12,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import javax.json.Json;
@ -22,17 +23,17 @@ import javax.json.JsonReaderFactory;
import javax.json.JsonString;
import javax.management.openmbean.TabularData;
import javax.management.openmbean.TabularDataSupport;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.Invocation;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.UriBuilder;
import javax.ws.rs.core.Response;
import org.glassfish.jersey.client.ClientConfig;
import com.cloudius.urchin.utils.EstimatedHistogram;
import com.cloudius.urchin.utils.SnapshotDetailsTabularData;
import com.sun.jersey.api.client.Client;
import com.sun.jersey.api.client.WebResource;
import com.sun.jersey.api.client.WebResource.Builder;
import com.sun.jersey.api.client.config.ClientConfig;
import com.sun.jersey.api.client.config.DefaultClientConfig;
import com.yammer.metrics.core.HistogramValues;
public class APIClient {
@ -77,43 +78,44 @@ public class APIClient {
+ System.getProperty("apiport", "10000");
}
public Builder get(String path) {
ClientConfig config = new DefaultClientConfig();
Client client = Client.create(config);
WebResource service = client.resource(UriBuilder.fromUri(getBaseUrl())
.build());
return service.path(path).accept(MediaType.APPLICATION_JSON);
}
public Builder get(String path, MultivaluedMap<String, String> queryParams) {
if (queryParams == null) {
return get(path);
}
ClientConfig config = new DefaultClientConfig();
Client client = Client.create(config);
WebResource service = client.resource(UriBuilder.fromUri(getBaseUrl())
.build());
return service.queryParams(queryParams).path(path)
.accept(MediaType.APPLICATION_JSON);
}
public void post(String path, MultivaluedMap<String, String> queryParams) {
public Invocation.Builder get(String path, MultivaluedMap<String, String> queryParams) {
Client client = ClientBuilder.newClient( new ClientConfig());
WebTarget webTarget = client.target(getBaseUrl()).path(path);
if (queryParams != null) {
get(path, queryParams).post();
return;
for (Entry<String, List<String>> qp : queryParams.entrySet()) {
for (String e : qp.getValue()) {
webTarget = webTarget.queryParam(qp.getKey(), e);
}
}
}
get(path).post();
return webTarget.request(MediaType.APPLICATION_JSON);
}
public Invocation.Builder get(String path) {
return get(path, null);
}
public Response post(String path, MultivaluedMap<String, String> queryParams) {
Response response = get(path, queryParams).post(Entity.entity(null, MediaType.TEXT_PLAIN));
if (response.getStatus() != Response.Status.OK.getStatusCode() ) {
throw getException(response.readEntity(String.class));
}
return response;
}
public void post(String path) {
post(path, null);
}
public RuntimeException getException(String txt) {
JsonReader reader = factory.createReader(new StringReader(txt));
JsonObject res = reader.readObject();
return new RuntimeException(res.getString("message"));
}
public String postGetVal(String path, MultivaluedMap<String, String> queryParams) {
if (queryParams != null) {
return get(path, queryParams).post(String.class);
}
return get(path).post(String.class);
return post(path, queryParams).readEntity(String.class);
}
public int postInt(String path, MultivaluedMap<String, String> queryParams) {
@ -146,8 +148,15 @@ public class APIClient {
if (res != null) {
return res;
}
Response response = get(string, queryParams).get(Response.class);
res = get(string, queryParams).get(String.class);
if (response.getStatus() != Response.Status.OK.getStatusCode() ) {
// TBD
// We are currently not caching errors,
// it should be reconsider.
throw getException(response.readEntity(String.class));
}
res = response.readEntity(String.class);
if (duration > 0) {
cache.put(key, new CacheEntry(res));
}

32
src/main/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -24,6 +24,7 @@
package org.apache.cassandra.db;
import java.lang.management.ManagementFactory;
import java.net.ConnectException;
import java.util.*;
import java.util.concurrent.*;
@ -32,13 +33,14 @@ import javax.json.JsonObject;
import javax.management.*;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.OpenDataException;
import javax.ws.rs.ProcessingException;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import org.apache.cassandra.metrics.ColumnFamilyMetrics;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.api.client.ClientHandlerException;
import com.sun.jersey.core.util.MultivaluedMapImpl;
import com.google.common.base.Throwables;
public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
private static final java.util.logging.Logger logger = java.util.logging.Logger
@ -130,10 +132,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
}
}
missed_response = 0;
} catch (ClientHandlerException e) {
if (missed_response++ > MAX_RETRY) {
System.err.println("API is not available, JMX is shuting down");
System.exit(-1);
} catch (ProcessingException e) {
if (Throwables.getRootCause(e) instanceof ConnectException) {
if (missed_response++ > MAX_RETRY) {
System.err.println("API is not available, JMX is shuting down");
System.exit(-1);
}
} else {
// ignoring exceptions, will retry on the next interval
}
} catch (Exception e) {
// ignoring exceptions, will retry on the next interval
@ -474,7 +480,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public void setMinimumCompactionThreshold(int threshold) {
log(" setMinimumCompactionThreshold(int threshold)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("value", Integer.toString(threshold));
c.post("column_family/minimum_compaction/" + getCFName(), queryParams);
}
@ -493,7 +499,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public void setCompactionThresholds(int minThreshold, int maxThreshold) {
log(" setCompactionThresholds(int minThreshold, int maxThreshold)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("minimum", Integer.toString(minThreshold));
queryParams.add("maximum", Integer.toString(maxThreshold));
c.post("column_family/compaction" + getCFName(), queryParams);
@ -504,7 +510,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public void setMaximumCompactionThreshold(int threshold) {
log(" setMaximumCompactionThreshold(int threshold)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("value", Integer.toString(threshold));
c.post("column_family/maximum_compaction/" + getCFName(), queryParams);
}
@ -517,7 +523,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public void setCompactionStrategyClass(String className) {
log(" setCompactionStrategyClass(String className)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("class_name", className);
c.post("column_family/compaction_strategy/" + getCFName(), queryParams);
}
@ -548,7 +554,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public void setCompressionParameters(Map<String, String> opts) {
log(" setCompressionParameters(Map<String,String> opts)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("opts", APIClient.mapToString(opts));
c.post("column_family/compression_parameters/" + getCFName(),
queryParams);
@ -559,7 +565,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public void setCrcCheckChance(double crcCheckChance) {
log(" setCrcCheckChance(double crcCheckChance)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("check_chance", Double.toString(crcCheckChance));
c.post("column_family/crc_check_chance/" + getCFName(), queryParams);
}
@ -633,7 +639,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean {
*/
public List<String> getSSTablesForKey(String key) {
log(" getSSTablesForKey(String key)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("key", key);
return c.getListStrValue("column_family/sstables/by_key/" + getCFName(),
queryParams);

8
src/main/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -23,12 +23,12 @@ import java.util.*;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.management.openmbean.TabularData;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import org.apache.cassandra.metrics.CompactionMetrics;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.core.util.MultivaluedMapImpl;
/**
* A singleton which manages a private executor of ongoing compactions.
@ -140,7 +140,7 @@ public class CompactionManager implements CompactionManagerMBean {
*/
public void forceUserDefinedCompaction(String dataFiles) {
log(" forceUserDefinedCompaction(String dataFiles)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("dataFiles", dataFiles);
c.post("compaction_manager/compaction_manager/force_user_defined_compaction",
queryParams);
@ -148,14 +148,14 @@ public class CompactionManager implements CompactionManagerMBean {
/**
* Stop all running compaction-like tasks having the provided {@code type}.
*
*
* @param type
* the type of compaction to stop. Can be one of: - COMPACTION -
* VALIDATION - CLEANUP - SCRUB - INDEX_BUILD
*/
public void stopCompaction(String type) {
log(" stopCompaction(String type)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("type", type);
c.post("compaction_manager/compaction_manager/stop_compaction",
queryParams);

4
src/main/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -28,10 +28,10 @@ import java.net.UnknownHostException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.core.util.MultivaluedMapImpl;
/**
* This module is responsible for Gossiping information for the local endpoint.
@ -91,7 +91,7 @@ public class Gossiper implements GossiperMBean {
public void unsafeAssassinateEndpoint(String address)
throws UnknownHostException {
log(" unsafeAssassinateEndpoint(String address) throws UnknownHostException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("unsafe", "True");
c.post("gossiper/assassinate/" + address, queryParams);
}

6
src/main/java/org/apache/cassandra/locator/EndpointSnitchInfo.java
View File

@ -23,10 +23,10 @@ import java.net.UnknownHostException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.core.util.MultivaluedMapImpl;
public class EndpointSnitchInfo implements EndpointSnitchInfoMBean {
private static final java.util.logging.Logger logger = java.util.logging.Logger
@ -64,7 +64,7 @@ public class EndpointSnitchInfo implements EndpointSnitchInfoMBean {
@Override
public String getRack(String host) throws UnknownHostException {
log("getRack(String host) throws UnknownHostException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
if (host == null) {
host = InetAddress.getLoopbackAddress().getHostAddress();
}
@ -82,7 +82,7 @@ public class EndpointSnitchInfo implements EndpointSnitchInfoMBean {
@Override
public String getDatacenter(String host) throws UnknownHostException {
log(" getDatacenter(String host) throws UnknownHostException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
if (host == null) {
host = InetAddress.getLoopbackAddress().getHostAddress();
}

20
src/main/java/org/apache/cassandra/service/CacheService.java
View File

@ -29,12 +29,12 @@ import java.util.concurrent.ExecutionException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import org.apache.cassandra.metrics.CacheMetrics;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.core.util.MultivaluedMapImpl;
public class CacheService implements CacheServiceMBean {
private static final java.util.logging.Logger logger = java.util.logging.Logger
@ -77,7 +77,7 @@ public class CacheService implements CacheServiceMBean {
public void setRowCacheSavePeriodInSeconds(int rcspis) {
log(" setRowCacheSavePeriodInSeconds(int rcspis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("period", Integer.toString(rcspis));
c.post("cache_service/row_cache_save_period", queryParams);
}
@ -89,7 +89,7 @@ public class CacheService implements CacheServiceMBean {
public void setKeyCacheSavePeriodInSeconds(int kcspis) {
log(" setKeyCacheSavePeriodInSeconds(int kcspis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("period", Integer.toString(kcspis));
c.post("cache_service/key_cache_save_period", queryParams);
}
@ -101,7 +101,7 @@ public class CacheService implements CacheServiceMBean {
public void setCounterCacheSavePeriodInSeconds(int ccspis) {
log(" setCounterCacheSavePeriodInSeconds(int ccspis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("ccspis", Integer.toString(ccspis));
c.post("cache_service/counter_cache_save_period", queryParams);
}
@ -113,7 +113,7 @@ public class CacheService implements CacheServiceMBean {
public void setRowCacheKeysToSave(int rckts) {
log(" setRowCacheKeysToSave(int rckts)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("rckts", Integer.toString(rckts));
c.post("cache_service/row_cache_keys_to_save", queryParams);
}
@ -125,7 +125,7 @@ public class CacheService implements CacheServiceMBean {
public void setKeyCacheKeysToSave(int kckts) {
log(" setKeyCacheKeysToSave(int kckts)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("kckts", Integer.toString(kckts));
c.post("cache_service/key_cache_keys_to_save", queryParams);
}
@ -137,7 +137,7 @@ public class CacheService implements CacheServiceMBean {
public void setCounterCacheKeysToSave(int cckts) {
log(" setCounterCacheKeysToSave(int cckts)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("cckts", Integer.toString(cckts));
c.post("cache_service/counter_cache_keys_to_save", queryParams);
}
@ -165,21 +165,21 @@ public class CacheService implements CacheServiceMBean {
public void setRowCacheCapacityInMB(long capacity) {
log(" setRowCacheCapacityInMB(long capacity)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("capacity", Long.toString(capacity));
c.post("cache_service/row_cache_capacity", queryParams);
}
public void setKeyCacheCapacityInMB(long capacity) {
log(" setKeyCacheCapacityInMB(long capacity)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("capacity", Long.toString(capacity));
c.post("cache_service/key_cache_capacity", queryParams);
}
public void setCounterCacheCapacityInMB(long capacity) {
log(" setCounterCacheCapacityInMB(long capacity)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("capacity", Long.toString(capacity));
c.post("cache_service/counter_cache_capacity_in_mb", queryParams);
}

24
src/main/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -28,10 +28,10 @@ import java.util.*;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.core.util.MultivaluedMapImpl;
import org.apache.cassandra.metrics.*;
@ -203,14 +203,14 @@ public class StorageProxy implements StorageProxyMBean {
public void setHintedHandoffEnabled(boolean b) {
log(" setHintedHandoffEnabled(boolean b)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("enable", Boolean.toString(b));
c.post("storage_proxy/hinted_handoff_enabled", queryParams);
}
public void setHintedHandoffEnabledByDCList(String dcs) {
log(" setHintedHandoffEnabledByDCList(String dcs)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("dcs", dcs);
c.post("storage_proxy/hinted_handoff_enabled_by_dc_list");
}
@ -222,7 +222,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setMaxHintWindow(int ms) {
log(" setMaxHintWindow(int ms)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("ms", Integer.toString(ms));
c.post("storage_proxy/max_hint_window", queryParams);
}
@ -234,7 +234,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setMaxHintsInProgress(int qs) {
log(" setMaxHintsInProgress(int qs)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("qs", Integer.toString(qs));
c.post("storage_proxy/max_hints_in_progress", queryParams);
}
@ -251,7 +251,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setRpcTimeout(Long timeoutInMillis) {
log(" setRpcTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/rpc_timeout", queryParams);
}
@ -263,7 +263,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setReadRpcTimeout(Long timeoutInMillis) {
log(" setReadRpcTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/read_rpc_timeout", queryParams);
}
@ -275,7 +275,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setWriteRpcTimeout(Long timeoutInMillis) {
log(" setWriteRpcTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/write_rpc_timeout", queryParams);
}
@ -287,7 +287,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setCounterWriteRpcTimeout(Long timeoutInMillis) {
log(" setCounterWriteRpcTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/counter_write_rpc_timeout", queryParams);
}
@ -299,7 +299,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setCasContentionTimeout(Long timeoutInMillis) {
log(" setCasContentionTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/cas_contention_timeout", queryParams);
}
@ -311,7 +311,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setRangeRpcTimeout(Long timeoutInMillis) {
log(" setRangeRpcTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/range_rpc_timeout", queryParams);
}
@ -323,7 +323,7 @@ public class StorageProxy implements StorageProxyMBean {
public void setTruncateRpcTimeout(Long timeoutInMillis) {
log(" setTruncateRpcTimeout(Long timeoutInMillis)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("timeout", Long.toString(timeoutInMillis));
c.post("storage_proxy/truncate_rpc_timeout", queryParams);
}

72
src/main/java/org/apache/cassandra/service/StorageService.java
View File

@ -33,12 +33,12 @@ import java.util.concurrent.atomic.AtomicLong;
import javax.management.*;
import javax.management.openmbean.TabularData;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.repair.RepairParallelism;
import com.cloudius.urchin.api.APIClient;
import com.sun.jersey.core.util.MultivaluedMapImpl;
/**
* This abstraction contains the token/identifier of this node on the identifier
@ -234,7 +234,7 @@ public class StorageService extends NotificationBroadcasterSupport
public Map<List<String>, List<String>> getRangeToRpcaddressMap(
String keyspace) {
log(" getRangeToRpcaddressMap(String keyspace)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("rpc", "true");
return c.getMapListStrValue("/storage_service/range/" + keyspace,
queryParams);
@ -340,7 +340,7 @@ public class StorageService extends NotificationBroadcasterSupport
public List<InetAddress> getNaturalEndpoints(String keyspaceName, String cf,
String key) {
log(" getNaturalEndpoints(String keyspaceName, String cf, String key)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("cf", cf);
queryParams.add("key", key);
return c.getListInetAddressValue(
@ -366,7 +366,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void takeSnapshot(String tag, String... keyspaceNames)
throws IOException {
log(" takeSnapshot(String tag, String... keyspaceNames) throws IOException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "tag", tag);
APIClient.set_query_param(queryParams, "kn",
APIClient.join(keyspaceNames));
@ -387,7 +387,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void takeColumnFamilySnapshot(String keyspaceName,
String columnFamilyName, String tag) throws IOException {
log(" takeColumnFamilySnapshot(String keyspaceName, String columnFamilyName, String tag) throws IOException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
if (keyspaceName == null)
throw new IOException("You must supply a keyspace name");
if (columnFamilyName == null)
@ -407,7 +407,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void clearSnapshot(String tag, String... keyspaceNames)
throws IOException {
log(" clearSnapshot(String tag, String... keyspaceNames) throws IOException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "tag", tag);
APIClient.set_query_param(queryParams, "kn",
APIClient.join(keyspaceNames));
@ -442,7 +442,7 @@ public class StorageService extends NotificationBroadcasterSupport
String... columnFamilies) throws IOException, ExecutionException,
InterruptedException {
log(" forceKeyspaceCompaction(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "cf",
APIClient.join(columnFamilies));
c.post("/storage_service/keyspace_compaction/" + keyspaceName,
@ -456,7 +456,7 @@ public class StorageService extends NotificationBroadcasterSupport
String... columnFamilies) throws IOException, ExecutionException,
InterruptedException {
log(" forceKeyspaceCleanup(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "cf",
APIClient.join(columnFamilies));
return c.postInt("/storage_service/keyspace_cleanup/" + keyspaceName,
@ -474,7 +474,7 @@ public class StorageService extends NotificationBroadcasterSupport
String keyspaceName, String... columnFamilies) throws IOException,
ExecutionException, InterruptedException {
log(" scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_bool_query_param(queryParams, "disable_snapshot",
disableSnapshot);
APIClient.set_bool_query_param(queryParams, "skip_corrupted",
@ -493,7 +493,7 @@ public class StorageService extends NotificationBroadcasterSupport
throws IOException, ExecutionException,
InterruptedException {
log(" upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_bool_query_param(queryParams, "exclude_current_version",
excludeCurrentVersion);
APIClient.set_query_param(queryParams, "cf",
@ -514,7 +514,7 @@ public class StorageService extends NotificationBroadcasterSupport
String... columnFamilies) throws IOException, ExecutionException,
InterruptedException {
log(" forceKeyspaceFlush(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "cf",
APIClient.join(columnFamilies));
c.post("/storage_service/keyspace_flush/" + keyspaceName, queryParams);
@ -525,7 +525,7 @@ public class StorageService extends NotificationBroadcasterSupport
int id;
String keyspace;
String message;
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
int cmd;
public CheckRepair(int id, String keyspace) {
this.id = id;
@ -598,7 +598,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public int repairAsync(String keyspace, Map<String, String> options) {
log(" repairAsync(String keyspace, Map<String, String> options)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
String opts = "";
for (String op : options.keySet()) {
if (!opts.equals("")) {
@ -681,7 +681,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public void move(String newToken) throws IOException {
log(" move(String newToken) throws IOException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "new_token", newToken);
c.post("/storage_service/move");
}
@ -693,7 +693,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public void removeNode(String hostIdString) {
log(" removeNode(String token)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "host_id", hostIdString);
c.post("/storage_service/remove_node", queryParams);
}
@ -736,7 +736,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void setLoggingLevel(String classQualifier, String level)
throws Exception {
log(" setLoggingLevel(String classQualifier, String level) throws Exception");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "level", level);
APIClient.set_query_param(queryParams, "class_qualifier",
classQualifier);
@ -795,7 +795,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void truncate(String keyspace, String columnFamily)
throws TimeoutException, IOException {
log(" truncate(String keyspace, String columnFamily)throws TimeoutException, IOException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "cf", columnFamily);
c.post("/storage_service/truncate/" + keyspace, queryParams);
}
@ -828,7 +828,7 @@ public class StorageService extends NotificationBroadcasterSupport
public List<String> getKeyspaces() {
log(" getKeyspaces()");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("non_system", "true");
return c.getListStrValue("/storage_service/keyspaces", queryParams);
}
@ -858,7 +858,7 @@ public class StorageService extends NotificationBroadcasterSupport
Integer dynamicUpdateInterval, Integer dynamicResetInterval,
Double dynamicBadnessThreshold) throws ClassNotFoundException {
log(" updateSnitch(String epSnitchClassName, Boolean dynamic, Integer dynamicUpdateInterval, Integer dynamicResetInterval, Double dynamicBadnessThreshold) throws ClassNotFoundException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_bool_query_param(queryParams, "dynamic", dynamic);
APIClient.set_query_param(queryParams, "epSnitchClassName",
epSnitchClassName);
@ -959,10 +959,9 @@ public class StorageService extends NotificationBroadcasterSupport
public void setStreamThroughputMbPerSec(int value) {
log(" setStreamThroughputMbPerSec(int value)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("value", Integer.toString(value));
c.post("/storage_service/stream_throughput", queryParams);
}
public int getStreamThroughputMbPerSec() {
@ -977,7 +976,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void setCompactionThroughputMbPerSec(int value) {
log(" setCompactionThroughputMbPerSec(int value)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("value", Integer.toString(value));
c.post("/storage_service/compaction_throughput", queryParams);
}
@ -989,7 +988,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void setIncrementalBackupsEnabled(boolean value) {
log(" setIncrementalBackupsEnabled(boolean value)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("value", Boolean.toString(value));
c.post("/storage_service/incremental_backups", queryParams);
}
@ -1006,7 +1005,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public void rebuild(String sourceDc) {
log(" rebuild(String sourceDc)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "value", sourceDc);
c.post("/storage_service/rebuild", queryParams);
}
@ -1042,7 +1041,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public void loadNewSSTables(String ksName, String cfName) {
log(" loadNewSSTables(String ksName, String cfName)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("cf", cfName);
c.post("/storage_service/sstables/" + ksName, queryParams);
}
@ -1087,7 +1086,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public void setTraceProbability(double probability) {
log(" setTraceProbability(double probability)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("probability", Double.toString(probability));
c.post("/storage_service/trace_probability", queryParams);
}
@ -1103,7 +1102,7 @@ public class StorageService extends NotificationBroadcasterSupport
public void disableAutoCompaction(String ks, String... columnFamilies)
throws IOException {
log("disableAutoCompaction(String ks, String... columnFamilies)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "cf",
APIClient.join(columnFamilies));
c.delete("/storage_service/auto_compaction/", queryParams);
@ -1112,16 +1111,21 @@ public class StorageService extends NotificationBroadcasterSupport
public void enableAutoCompaction(String ks, String... columnFamilies)
throws IOException {
log("enableAutoCompaction(String ks, String... columnFamilies)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
APIClient.set_query_param(queryParams, "cf",
APIClient.join(columnFamilies));
c.post("/storage_service/auto_compaction/", queryParams);
try {
c.post("/storage_service/auto_compaction/", queryParams);
} catch (RuntimeException e) {
// FIXME should throw the right exception
throw new IOException(e.getMessage());
}
}
public void deliverHints(String host) throws UnknownHostException {
log(" deliverHints(String host) throws UnknownHostException");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("host", host);
c.post("/storage_service/deliver_hints", queryParams);
}
@ -1147,7 +1151,7 @@ public class StorageService extends NotificationBroadcasterSupport
/** Sets the threshold for warning queries with many tombstones */
public void setTombstoneWarnThreshold(int tombstoneDebugThreshold) {
log(" setTombstoneWarnThreshold(int tombstoneDebugThreshold)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("debug_threshold",
Integer.toString(tombstoneDebugThreshold));
c.post("/storage_service/tombstone_warn_threshold", queryParams);
@ -1162,7 +1166,7 @@ public class StorageService extends NotificationBroadcasterSupport
/** Sets the threshold for abandoning queries with many tombstones */
public void setTombstoneFailureThreshold(int tombstoneDebugThreshold) {
log(" setTombstoneFailureThreshold(int tombstoneDebugThreshold)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("debug_threshold",
Integer.toString(tombstoneDebugThreshold));
c.post("/storage_service/tombstone_failure_threshold", queryParams);
@ -1177,7 +1181,7 @@ public class StorageService extends NotificationBroadcasterSupport
/** Sets the threshold for rejecting queries due to a large batch size */
public void setBatchSizeFailureThreshold(int batchSizeDebugThreshold) {
log(" setBatchSizeFailureThreshold(int batchSizeDebugThreshold)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("threshold", Integer.toString(batchSizeDebugThreshold));
c.post("/storage_service/batch_size_failure_threshold", queryParams);
}
@ -1187,7 +1191,7 @@ public class StorageService extends NotificationBroadcasterSupport
*/
public void setHintedHandoffThrottleInKB(int throttleInKB) {
log(" setHintedHandoffThrottleInKB(int throttleInKB)");
MultivaluedMap<String, String> queryParams = new MultivaluedMapImpl();
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
queryParams.add("throttle", Integer.toString(throttleInKB));
c.post("/storage_service/hinted_handoff", queryParams);