1758 lines
69 KiB
Java
1758 lines
69 KiB
Java
/*
|
|
* 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.
|
|
*/
|
|
/*
|
|
* Copyright 2015 Cloudius Systems
|
|
*
|
|
* Modified by Cloudius Systems
|
|
*/
|
|
package org.apache.cassandra.service;
|
|
|
|
import static java.util.Arrays.asList;
|
|
|
|
import java.io.IOException;
|
|
import java.net.InetAddress;
|
|
import java.net.UnknownHostException;
|
|
import java.nio.ByteBuffer;
|
|
import java.util.ArrayList;
|
|
import java.util.Collection;
|
|
import java.util.HashMap;
|
|
import java.util.HashSet;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Map.Entry;
|
|
import java.util.Set;
|
|
import java.util.Timer;
|
|
import java.util.TimerTask;
|
|
import java.util.concurrent.ExecutionException;
|
|
import java.util.concurrent.TimeoutException;
|
|
import java.util.concurrent.atomic.AtomicLong;
|
|
import java.util.logging.Logger;
|
|
import java.util.stream.Collectors;
|
|
|
|
import javax.json.JsonArray;
|
|
import javax.json.JsonObject;
|
|
import javax.management.ListenerNotFoundException;
|
|
import javax.management.MBeanNotificationInfo;
|
|
import javax.management.Notification;
|
|
import javax.management.NotificationBroadcaster;
|
|
import javax.management.NotificationBroadcasterSupport;
|
|
import javax.management.NotificationFilter;
|
|
import javax.management.NotificationListener;
|
|
import javax.management.openmbean.CompositeData;
|
|
import javax.management.openmbean.TabularData;
|
|
import javax.ws.rs.core.GenericType;
|
|
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.google.common.base.Joiner;
|
|
import com.scylladb.jmx.api.APIClient;
|
|
import com.scylladb.jmx.metrics.MetricsMBean;
|
|
import com.scylladb.jmx.utils.FileUtils;
|
|
|
|
/**
|
|
* This abstraction contains the token/identifier of this node on the identifier
|
|
* space. This token gets gossiped around. This class will also maintain
|
|
* histograms of the load information of other nodes in the cluster.
|
|
*/
|
|
public class StorageService extends MetricsMBean implements StorageServiceMBean, NotificationBroadcaster {
|
|
private static final Logger logger = Logger.getLogger(StorageService.class.getName());
|
|
private static final Timer timer = new Timer("Storage Service Repair", true);
|
|
|
|
private final NotificationBroadcasterSupport notificationBroadcasterSupport = new NotificationBroadcasterSupport();
|
|
|
|
@Override
|
|
public void addNotificationListener(NotificationListener listener, NotificationFilter filter, Object handback) {
|
|
notificationBroadcasterSupport.addNotificationListener(listener, filter, handback);
|
|
}
|
|
|
|
@Override
|
|
public void removeNotificationListener(NotificationListener listener) throws ListenerNotFoundException {
|
|
notificationBroadcasterSupport.removeNotificationListener(listener);
|
|
}
|
|
|
|
@Override
|
|
public void removeNotificationListener(NotificationListener listener, NotificationFilter filter, Object handback)
|
|
throws ListenerNotFoundException {
|
|
notificationBroadcasterSupport.removeNotificationListener(listener, filter, handback);
|
|
}
|
|
|
|
@Override
|
|
public MBeanNotificationInfo[] getNotificationInfo() {
|
|
return notificationBroadcasterSupport.getNotificationInfo();
|
|
}
|
|
|
|
public void sendNotification(Notification notification) {
|
|
notificationBroadcasterSupport.sendNotification(notification);
|
|
}
|
|
|
|
public static enum RepairStatus {
|
|
STARTED, SESSION_SUCCESS, SESSION_FAILED, FINISHED
|
|
}
|
|
|
|
/* JMX notification serial number counter */
|
|
private final AtomicLong notificationSerialNumber = new AtomicLong();
|
|
|
|
public StorageService(APIClient client) {
|
|
super("org.apache.cassandra.db:type=StorageService", client, new StorageMetrics());
|
|
|
|
}
|
|
|
|
public void log(String str) {
|
|
logger.finest(str);
|
|
}
|
|
|
|
/**
|
|
* Retrieve the list of live nodes in the cluster, where "liveness" is
|
|
* determined by the failure detector of the node being queried.
|
|
*
|
|
* @return set of IP addresses, as Strings
|
|
*/
|
|
@Override
|
|
public List<String> getLiveNodes() {
|
|
log(" getLiveNodes()");
|
|
return client.getListStrValue("/gossiper/endpoint/live");
|
|
}
|
|
|
|
/**
|
|
* Retrieve the list of unreachable nodes in the cluster, as determined by
|
|
* this node's failure detector.
|
|
*
|
|
* @return set of IP addresses, as Strings
|
|
*/
|
|
@Override
|
|
public List<String> getUnreachableNodes() {
|
|
log(" getUnreachableNodes()");
|
|
return client.getListStrValue("/gossiper/endpoint/down");
|
|
}
|
|
|
|
/**
|
|
* Retrieve the list of nodes currently bootstrapping into the ring.
|
|
*
|
|
* @return set of IP addresses, as Strings
|
|
*/
|
|
@Override
|
|
public List<String> getJoiningNodes() {
|
|
log(" getJoiningNodes()");
|
|
return client.getListStrValue("/storage_service/nodes/joining");
|
|
}
|
|
|
|
/**
|
|
* Retrieve the list of nodes currently leaving the ring.
|
|
*
|
|
* @return set of IP addresses, as Strings
|
|
*/
|
|
@Override
|
|
public List<String> getLeavingNodes() {
|
|
log(" getLeavingNodes()");
|
|
return client.getListStrValue("/storage_service/nodes/leaving");
|
|
}
|
|
|
|
/**
|
|
* Retrieve the list of nodes currently moving in the ring.
|
|
*
|
|
* @return set of IP addresses, as Strings
|
|
*/
|
|
@Override
|
|
public List<String> getMovingNodes() {
|
|
log(" getMovingNodes()");
|
|
return client.getListStrValue("/storage_service/nodes/moving");
|
|
}
|
|
|
|
/**
|
|
* Fetch string representations of the tokens for this node.
|
|
*
|
|
* @return a collection of tokens formatted as strings
|
|
*/
|
|
@Override
|
|
public List<String> getTokens() {
|
|
log(" getTokens()");
|
|
try {
|
|
return getTokens(getLocalBroadCastingAddress());
|
|
} catch (UnknownHostException e) {
|
|
// We should never reach here,
|
|
// but it makes the compiler happy
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Fetch string representations of the tokens for a specified node.
|
|
*
|
|
* @param endpoint
|
|
* string representation of an node
|
|
* @return a collection of tokens formatted as strings
|
|
*/
|
|
@Override
|
|
public List<String> getTokens(String endpoint) throws UnknownHostException {
|
|
log(" getTokens(String endpoint) throws UnknownHostException");
|
|
return client.getListStrValue("/storage_service/tokens/" + endpoint);
|
|
}
|
|
|
|
/**
|
|
* Fetch a string representation of the Cassandra version.
|
|
*
|
|
* @return A string representation of the Cassandra version.
|
|
*/
|
|
@Override
|
|
public String getReleaseVersion() {
|
|
log(" getReleaseVersion()");
|
|
return client.getStringValue("/storage_service/release_version");
|
|
}
|
|
|
|
/**
|
|
* Fetch a string representation of the current Schema version.
|
|
*
|
|
* @return A string representation of the Schema version.
|
|
*/
|
|
@Override
|
|
public String getSchemaVersion() {
|
|
log(" getSchemaVersion()");
|
|
return client.getStringValue("/storage_service/schema_version");
|
|
}
|
|
|
|
/**
|
|
* Get the list of all data file locations from conf
|
|
*
|
|
* @return String array of all locations
|
|
*/
|
|
@Override
|
|
public String[] getAllDataFileLocations() {
|
|
log(" getAllDataFileLocations()");
|
|
return client.getStringArrValue("/storage_service/data_file/locations");
|
|
}
|
|
|
|
/**
|
|
* Get location of the commit log
|
|
*
|
|
* @return a string path
|
|
*/
|
|
@Override
|
|
public String getCommitLogLocation() {
|
|
log(" getCommitLogLocation()");
|
|
return client.getStringValue("/storage_service/commitlog");
|
|
}
|
|
|
|
/**
|
|
* Get location of the saved caches dir
|
|
*
|
|
* @return a string path
|
|
*/
|
|
@Override
|
|
public String getSavedCachesLocation() {
|
|
log(" getSavedCachesLocation()");
|
|
return client.getStringValue("/storage_service/saved_caches/location");
|
|
}
|
|
|
|
/**
|
|
* Retrieve a map of range to end points that describe the ring topology of
|
|
* a Cassandra cluster.
|
|
*
|
|
* @return mapping of ranges to end points
|
|
*/
|
|
@Override
|
|
public Map<List<String>, List<String>> getRangeToEndpointMap(String keyspace) {
|
|
log(" getRangeToEndpointMap(String keyspace)");
|
|
return client.getMapListStrValue("/storage_service/range_to_endpoint_map/" + keyspace);
|
|
}
|
|
|
|
/**
|
|
* Retrieve a map of range to rpc addresses that describe the ring topology
|
|
* of a Cassandra cluster.
|
|
*
|
|
* @return mapping of ranges to rpc addresses
|
|
*/
|
|
@Override
|
|
public Map<List<String>, List<String>> getRangeToRpcaddressMap(String keyspace) {
|
|
log(" getRangeToRpcaddressMap(String keyspace)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("rpc", "true");
|
|
return client.getMapListStrValue("/storage_service/range/" + keyspace, queryParams);
|
|
}
|
|
|
|
/**
|
|
* The same as {@code describeRing(String)} but converts TokenRange to the
|
|
* String for JMX compatibility
|
|
*
|
|
* @param keyspace
|
|
* The keyspace to fetch information about
|
|
*
|
|
* @return a List of TokenRange(s) converted to String for the given
|
|
* keyspace
|
|
*/
|
|
@Override
|
|
public List<String> describeRingJMX(String keyspace) throws IOException {
|
|
log(" describeRingJMX(String keyspace) throws IOException");
|
|
JsonArray arr = client.getJsonArray("/storage_service/describe_ring/" + keyspace);
|
|
List<String> res = new ArrayList<String>();
|
|
|
|
for (int i = 0; i < arr.size(); i++) {
|
|
JsonObject obj = arr.getJsonObject(i);
|
|
StringBuilder sb = new StringBuilder();
|
|
sb.append("TokenRange(");
|
|
sb.append("start_token:");
|
|
sb.append(obj.getString("start_token"));
|
|
sb.append(", end_token:");
|
|
sb.append(obj.getString("end_token"));
|
|
sb.append(", endpoints:[");
|
|
JsonArray endpoints = obj.getJsonArray("endpoints");
|
|
for (int j = 0; j < endpoints.size(); j++) {
|
|
if (j > 0) {
|
|
sb.append(", ");
|
|
}
|
|
sb.append(endpoints.getString(j));
|
|
}
|
|
sb.append("], rpc_endpoints:[");
|
|
JsonArray rpc_endpoints = obj.getJsonArray("rpc_endpoints");
|
|
for (int j = 0; j < rpc_endpoints.size(); j++) {
|
|
if (j > 0) {
|
|
sb.append(", ");
|
|
}
|
|
sb.append(rpc_endpoints.getString(j));
|
|
}
|
|
|
|
sb.append("], endpoint_details:[");
|
|
JsonArray endpoint_details = obj.getJsonArray("endpoint_details");
|
|
for (int j = 0; j < endpoint_details.size(); j++) {
|
|
JsonObject detail = endpoint_details.getJsonObject(j);
|
|
if (j > 0) {
|
|
sb.append(", ");
|
|
}
|
|
sb.append("EndpointDetails(");
|
|
sb.append("host:");
|
|
sb.append(detail.getString("host"));
|
|
sb.append(", datacenter:");
|
|
sb.append(detail.getString("datacenter"));
|
|
sb.append(", rack:");
|
|
sb.append(detail.getString("rack"));
|
|
sb.append(')');
|
|
}
|
|
sb.append("])");
|
|
res.add(sb.toString());
|
|
}
|
|
return res;
|
|
}
|
|
|
|
/**
|
|
* Retrieve a map of pending ranges to endpoints that describe the ring
|
|
* topology
|
|
*
|
|
* @param keyspace
|
|
* the keyspace to get the pending range map for.
|
|
* @return a map of pending ranges to endpoints
|
|
*/
|
|
@Override
|
|
public Map<List<String>, List<String>> getPendingRangeToEndpointMap(String keyspace) {
|
|
log(" getPendingRangeToEndpointMap(String keyspace)");
|
|
return client.getMapListStrValue("/storage_service/pending_range/" + keyspace);
|
|
}
|
|
|
|
/**
|
|
* Retrieve a map of tokens to endpoints, including the bootstrapping ones.
|
|
*
|
|
* @return a map of tokens to endpoints in ascending order
|
|
*/
|
|
@Override
|
|
public Map<String, String> getTokenToEndpointMap() {
|
|
log(" getTokenToEndpointMap()");
|
|
return client.getMapStrValue("/storage_service/tokens_endpoint");
|
|
}
|
|
|
|
/** Retrieve this hosts unique ID */
|
|
@Override
|
|
public String getLocalHostId() {
|
|
log(" getLocalHostId()");
|
|
return client.getStringValue("/storage_service/hostid/local");
|
|
}
|
|
|
|
public String getLocalBroadCastingAddress() {
|
|
// FIXME:
|
|
// There is no straight API to get the broadcasting
|
|
// address, instead of trying to figure it out from the configuration
|
|
// we will use the getHostIdToAddressMap with the hostid
|
|
return getHostIdToAddressMap().get(getLocalHostId());
|
|
}
|
|
|
|
/** Retrieve the mapping of endpoint to host ID */
|
|
@Override
|
|
public Map<String, String> getHostIdMap() {
|
|
log(" getHostIdMap()");
|
|
return client.getMapStrValue("/storage_service/host_id");
|
|
}
|
|
|
|
/** Retrieve the mapping of endpoint to host ID */
|
|
public Map<String, String> getHostIdToAddressMap() {
|
|
log(" getHostIdToAddressMap()");
|
|
return client.getReverseMapStrValue("/storage_service/host_id");
|
|
}
|
|
|
|
/**
|
|
* Numeric load value.
|
|
*
|
|
* @see org.apache.cassandra.metrics.StorageMetrics#load
|
|
*/
|
|
@Deprecated
|
|
public double getLoad() {
|
|
log(" getLoad()");
|
|
return client.getDoubleValue("/storage_service/load");
|
|
}
|
|
|
|
/** Human-readable load value */
|
|
@Override
|
|
public String getLoadString() {
|
|
log(" getLoadString()");
|
|
return FileUtils.stringifyFileSize(getLoad());
|
|
}
|
|
|
|
/** Human-readable load value. Keys are IP addresses. */
|
|
@Override
|
|
public Map<String, String> getLoadMap() {
|
|
log(" getLoadMap()");
|
|
Map<String, Double> load = getLoadMapAsDouble();
|
|
Map<String, String> map = new HashMap<>();
|
|
for (Map.Entry<String, Double> entry : load.entrySet()) {
|
|
map.put(entry.getKey(), FileUtils.stringifyFileSize(entry.getValue()));
|
|
}
|
|
return map;
|
|
}
|
|
|
|
public Map<String, Double> getLoadMapAsDouble() {
|
|
log(" getLoadMapAsDouble()");
|
|
return client.getMapStringDouble("/storage_service/load_map");
|
|
}
|
|
|
|
/**
|
|
* Return the generation value for this node.
|
|
*
|
|
* @return generation number
|
|
*/
|
|
@Override
|
|
public int getCurrentGenerationNumber() {
|
|
log(" getCurrentGenerationNumber()");
|
|
return client.getIntValue("/storage_service/generation_number");
|
|
}
|
|
|
|
/**
|
|
* This method returns the N endpoints that are responsible for storing the
|
|
* specified key i.e for replication.
|
|
*
|
|
* @param keyspaceName
|
|
* keyspace name
|
|
* @param cf
|
|
* Column family name
|
|
* @param key
|
|
* - key for which we need to find the endpoint return value -
|
|
* the endpoint responsible for this key
|
|
*/
|
|
@Override
|
|
public List<InetAddress> getNaturalEndpoints(String keyspaceName, String cf, String key) {
|
|
log(" getNaturalEndpoints(String keyspaceName, String cf, String key)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("cf", cf);
|
|
queryParams.add("key", key);
|
|
return client.getListInetAddressValue("/storage_service/natural_endpoints/" + keyspaceName, queryParams);
|
|
}
|
|
|
|
@Override
|
|
public List<InetAddress> getNaturalEndpoints(String keyspaceName, ByteBuffer key) {
|
|
log(" getNaturalEndpoints(String keyspaceName, ByteBuffer key)");
|
|
return client.getListInetAddressValue("");
|
|
}
|
|
|
|
@Override
|
|
public void checkAndRepairCdcStreams() throws IOException {
|
|
log(" checkAndRepairCdcStreams() throws IOException");
|
|
client.post("/storage_service/cdc_streams_check_and_repair");
|
|
}
|
|
|
|
/**
|
|
* Takes the snapshot for the given keyspaces. A snapshot name must be
|
|
* specified.
|
|
*
|
|
* @param tag
|
|
* the tag given to the snapshot; may not be null or empty
|
|
* @param keyspaceNames
|
|
* the name of the keyspaces to snapshot; empty means "all."
|
|
*/
|
|
@Override
|
|
public void takeSnapshot(String tag, String... keyspaceNames) throws IOException {
|
|
takeSnapshot(tag, null, keyspaceNames);
|
|
}
|
|
|
|
@Override
|
|
public void takeSnapshot(String tag, Map<String, String> options, String... keyspaceNames) throws IOException {
|
|
log(" takeSnapshot(String tag, String... keyspaceNames) throws IOException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "tag", tag);
|
|
|
|
if (keyspaceNames.length == 1 && keyspaceNames[0].indexOf('.') != -1) {
|
|
String[] parts = keyspaceNames[0].split("\\.");
|
|
keyspaceNames = new String[] { parts[0] };
|
|
APIClient.set_query_param(queryParams, "cf", parts[1]);
|
|
}
|
|
APIClient.set_query_param(queryParams, "kn", APIClient.join(keyspaceNames));
|
|
// TODO: origin has one recognized option: skip flush. We don't.
|
|
client.post("/storage_service/snapshots", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Takes the snapshot of a specific column family. A snapshot name must be
|
|
* specified.
|
|
*
|
|
* @param keyspaceName
|
|
* the keyspace which holds the specified column family
|
|
* @param columnFamilyName
|
|
* the column family to snapshot
|
|
* @param tag
|
|
* the tag given to the snapshot; may not be null or empty
|
|
*/
|
|
@Override
|
|
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 MultivaluedHashMap<String, String>();
|
|
if (keyspaceName == null) {
|
|
throw new IOException("You must supply a keyspace name");
|
|
}
|
|
if (columnFamilyName == null) {
|
|
throw new IOException("You must supply a table name");
|
|
}
|
|
if (tag == null || tag.equals("")) {
|
|
throw new IOException("You must supply a snapshot name.");
|
|
}
|
|
queryParams.add("tag", tag);
|
|
queryParams.add("kn", keyspaceName);
|
|
queryParams.add("cf", columnFamilyName);
|
|
client.post("/storage_service/snapshots", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Remove the snapshot with the given name from the given keyspaces. If no
|
|
* tag is specified we will remove all snapshots.
|
|
*/
|
|
@Override
|
|
public void clearSnapshot(String tag, String... keyspaceNames) throws IOException {
|
|
log(" clearSnapshot(String tag, String... keyspaceNames) throws IOException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "tag", tag);
|
|
APIClient.set_query_param(queryParams, "kn", APIClient.join(keyspaceNames));
|
|
client.delete("/storage_service/snapshots", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Get the details of all the snapshot
|
|
*
|
|
* @return A map of snapshotName to all its details in Tabular form.
|
|
*/
|
|
@Override
|
|
public Map<String, TabularData> getSnapshotDetails() {
|
|
log(" getSnapshotDetails()");
|
|
return client.getMapStringSnapshotTabularDataValue("/storage_service/snapshots", null);
|
|
}
|
|
|
|
public Map<String, Map<String, Set<String>>> getSnapshotKeyspaceColumnFamily() {
|
|
JsonArray arr = client.getJsonArray("/storage_service/snapshots");
|
|
Map<String, Map<String, Set<String>>> res = new HashMap<String, Map<String, Set<String>>>();
|
|
for (int i = 0; i < arr.size(); i++) {
|
|
JsonObject obj = arr.getJsonObject(i);
|
|
Map<String, Set<String>> kscf = new HashMap<String, Set<String>>();
|
|
JsonArray snapshots = obj.getJsonArray("value");
|
|
for (int j = 0; j < snapshots.size(); j++) {
|
|
JsonObject s = snapshots.getJsonObject(j);
|
|
String ks = s.getString("ks");
|
|
String cf = s.getString("cf");
|
|
if (!kscf.containsKey(ks)) {
|
|
kscf.put(ks, new HashSet<String>());
|
|
}
|
|
kscf.get(ks).add(cf);
|
|
}
|
|
res.put(obj.getString("key"), kscf);
|
|
}
|
|
return res;
|
|
}
|
|
|
|
/**
|
|
* Get the true size taken by all snapshots across all keyspaces.
|
|
*
|
|
* @return True size taken by all the snapshots.
|
|
*/
|
|
@Override
|
|
public long trueSnapshotsSize() {
|
|
log(" trueSnapshotsSize()");
|
|
return client.getLongValue("/storage_service/snapshots/size/true");
|
|
}
|
|
|
|
/**
|
|
* Forces major compaction of a single keyspace
|
|
*/
|
|
@Override
|
|
public void forceKeyspaceCompaction(String keyspaceName, String... columnFamilies)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
log(" forceKeyspaceCompaction(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
client.post("/storage_service/keyspace_compaction/" + keyspaceName, queryParams);
|
|
}
|
|
|
|
@Override
|
|
public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken,
|
|
String... tableNames) throws IOException, ExecutionException, InterruptedException {
|
|
// TODO: actually handle token ranges.
|
|
forceKeyspaceCompaction(keyspaceName, tableNames);
|
|
}
|
|
|
|
/**
|
|
* Trigger a cleanup of keys on a single keyspace
|
|
*/
|
|
@Override
|
|
public int forceKeyspaceCleanup(String keyspaceName, String... columnFamilies)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
log(" forceKeyspaceCleanup(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
return client.postInt("/storage_service/keyspace_cleanup/" + keyspaceName, queryParams);
|
|
}
|
|
|
|
/**
|
|
* Scrub (deserialize + reserialize at the latest version, skipping bad rows
|
|
* if any) the given keyspace. If columnFamilies array is empty, all CFs are
|
|
* scrubbed.
|
|
*
|
|
* Scrubbed CFs will be snapshotted first, if disableSnapshot is false
|
|
*/
|
|
@Override
|
|
public int scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
log(" scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
|
|
return scrub(disableSnapshot, skipCorrupted, true, keyspaceName, columnFamilies);
|
|
}
|
|
|
|
@Override
|
|
public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, String keyspaceName,
|
|
String... columnFamilies) throws IOException, ExecutionException, InterruptedException {
|
|
log(" scrub(boolean disableSnapshot, boolean skipCorrupted, bool checkData, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
|
|
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", skipCorrupted);
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
return client.getIntValue("/storage_service/keyspace_scrub/" + keyspaceName, queryParams);
|
|
}
|
|
|
|
/**
|
|
* Rewrite all sstables to the latest version. Unlike scrub, it doesn't skip
|
|
* bad rows and do not snapshot sstables first.
|
|
*/
|
|
@Override
|
|
public int upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
log(" upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_bool_query_param(queryParams, "exclude_current_version", excludeCurrentVersion);
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
return client.getIntValue("/storage_service/keyspace_upgrade_sstables/" + keyspaceName, queryParams);
|
|
}
|
|
|
|
/**
|
|
* Flush all memtables for the given column families, or all columnfamilies
|
|
* for the given keyspace if none are explicitly listed.
|
|
*
|
|
* @param keyspaceName
|
|
* @param columnFamilies
|
|
* @throws IOException
|
|
*/
|
|
@Override
|
|
public void forceKeyspaceFlush(String keyspaceName, String... columnFamilies)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
log(" forceKeyspaceFlush(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
client.post("/storage_service/keyspace_flush/" + keyspaceName, queryParams);
|
|
}
|
|
|
|
private class CheckRepair extends TimerTask {
|
|
@SuppressWarnings("unused")
|
|
private int id;
|
|
private String keyspace;
|
|
private String message;
|
|
private MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
private int cmd;
|
|
private final boolean legacy;
|
|
|
|
public CheckRepair(int id, String keyspace, boolean legacy) {
|
|
this.id = id;
|
|
this.keyspace = keyspace;
|
|
this.legacy = legacy;
|
|
APIClient.set_query_param(queryParams, "id", Integer.toString(id));
|
|
message = String.format("Repair session %d ", id);
|
|
// The returned id is the command number
|
|
this.cmd = id;
|
|
}
|
|
|
|
@Override
|
|
public void run() {
|
|
String status = client.getStringValue("/storage_service/repair_async/" + keyspace, queryParams);
|
|
if (!status.equals("RUNNING")) {
|
|
cancel();
|
|
if (status.equals("SUCCESSFUL")) {
|
|
sendMessage(cmd, RepairStatus.SESSION_SUCCESS, message, legacy);
|
|
} else {
|
|
sendMessage(cmd, RepairStatus.SESSION_FAILED, message + "failed", legacy);
|
|
}
|
|
sendMessage(cmd, RepairStatus.FINISHED, message + "finished", legacy);
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
/**
|
|
* Sends JMX notification to subscribers.
|
|
*
|
|
* @param type
|
|
* Message type
|
|
* @param message
|
|
* Message itself
|
|
* @param userObject
|
|
* Arbitrary object to attach to notification
|
|
*/
|
|
public void sendNotification(String type, String message, Object userObject) {
|
|
Notification jmxNotification = new Notification(type, getBoundName(),
|
|
notificationSerialNumber.incrementAndGet(), message);
|
|
jmxNotification.setUserData(userObject);
|
|
sendNotification(jmxNotification);
|
|
}
|
|
|
|
public String getRepairMessage(final int cmd, final String keyspace, final int ranges_size,
|
|
final RepairParallelism parallelismDegree, final boolean fullRepair) {
|
|
return String.format(
|
|
"Starting repair command #%d, repairing %d ranges for keyspace %s (parallelism=%s, full=%b)", cmd,
|
|
ranges_size, keyspace, parallelismDegree, fullRepair);
|
|
}
|
|
|
|
/**
|
|
*
|
|
* @param repair
|
|
*/
|
|
private int waitAndNotifyRepair(int cmd, String keyspace, String message, boolean legacy) {
|
|
logger.finest(message);
|
|
|
|
sendMessage(cmd, RepairStatus.STARTED, message, legacy);
|
|
|
|
TimerTask taskToExecute = new CheckRepair(cmd, keyspace, legacy);
|
|
timer.schedule(taskToExecute, 100, 1000);
|
|
return cmd;
|
|
}
|
|
|
|
// See org.apache.cassandra.utils.progress.ProgressEventType
|
|
private static enum ProgressEventType {
|
|
START, PROGRESS, ERROR, ABORT, SUCCESS, COMPLETE, NOTIFICATION
|
|
}
|
|
|
|
private void sendMessage(int cmd, RepairStatus status, String message, boolean legacy) {
|
|
String tag = "repair:" + cmd;
|
|
|
|
ProgressEventType type = ProgressEventType.ERROR;
|
|
int total = 100;
|
|
int count = 0;
|
|
switch (status) {
|
|
case STARTED:
|
|
type = ProgressEventType.START;
|
|
break;
|
|
case FINISHED:
|
|
type = ProgressEventType.COMPLETE;
|
|
count = 100;
|
|
break;
|
|
case SESSION_SUCCESS:
|
|
type = ProgressEventType.SUCCESS;
|
|
count = 100;
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
|
|
Notification jmxNotification = new Notification("progress", tag, notificationSerialNumber.incrementAndGet(),
|
|
message);
|
|
Map<String, Integer> userData = new HashMap<>();
|
|
userData.put("type", type.ordinal());
|
|
userData.put("progressCount", count);
|
|
userData.put("total", total);
|
|
jmxNotification.setUserData(userData);
|
|
sendNotification(jmxNotification);
|
|
|
|
if (legacy) {
|
|
sendNotification("repair", message, new int[] { cmd, status.ordinal() });
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Invoke repair asynchronously. You can track repair progress by
|
|
* subscribing JMX notification sent from this StorageServiceMBean.
|
|
* Notification format is: type: "repair" userObject: int array of length 2,
|
|
* [0]=command number, [1]=ordinal of AntiEntropyService.Status
|
|
*
|
|
* @param keyspace
|
|
* Keyspace name to repair. Should not be null.
|
|
* @param options
|
|
* repair option.
|
|
* @return Repair command number, or 0 if nothing to repair
|
|
*/
|
|
@Override
|
|
public int repairAsync(String keyspace, Map<String, String> options) {
|
|
return repairAsync(keyspace, options, false);
|
|
}
|
|
|
|
@SuppressWarnings("unused")
|
|
private static final String PARALLELISM_KEY = "parallelism";
|
|
private static final String PRIMARY_RANGE_KEY = "primaryRange";
|
|
@SuppressWarnings("unused")
|
|
private static final String INCREMENTAL_KEY = "incremental";
|
|
@SuppressWarnings("unused")
|
|
private static final String JOB_THREADS_KEY = "jobThreads";
|
|
private static final String RANGES_KEY = "ranges";
|
|
private static final String COLUMNFAMILIES_KEY = "columnFamilies";
|
|
private static final String DATACENTERS_KEY = "dataCenters";
|
|
private static final String HOSTS_KEY = "hosts";
|
|
@SuppressWarnings("unused")
|
|
private static final String TRACE_KEY = "trace";
|
|
|
|
private int repairAsync(String keyspace, Map<String, String> options, boolean legacy) {
|
|
log(" repairAsync(String keyspace, Map<String, String> options)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
for (String op : options.keySet()) {
|
|
APIClient.set_query_param(queryParams, op, options.get(op));
|
|
}
|
|
|
|
int cmd = client.postInt("/storage_service/repair_async/" + keyspace, queryParams);
|
|
waitAndNotifyRepair(cmd, keyspace, getRepairMessage(cmd, keyspace, 1, RepairParallelism.SEQUENTIAL, true),
|
|
legacy);
|
|
return cmd;
|
|
}
|
|
|
|
private static String commaSeparated(Collection<?> c) {
|
|
String s = c.toString();
|
|
return s.substring(1, s.length() - 1);
|
|
}
|
|
|
|
private int repairRangeAsync(String beginToken, String endToken, String keyspaceName, Boolean isSequential,
|
|
Collection<String> dataCenters, Collection<String> hosts, Boolean primaryRange, Boolean repairedAt,
|
|
String... columnFamilies) {
|
|
log(" forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean repairedAt, String... columnFamilies) throws IOException");
|
|
|
|
Map<String, String> options = new HashMap<String, String>();
|
|
if (beginToken != null && endToken != null) {
|
|
options.put(RANGES_KEY, beginToken + ":" + endToken);
|
|
}
|
|
if (dataCenters != null) {
|
|
options.put(DATACENTERS_KEY, commaSeparated(dataCenters));
|
|
}
|
|
if (hosts != null) {
|
|
options.put(HOSTS_KEY, commaSeparated(hosts));
|
|
}
|
|
if (columnFamilies != null && columnFamilies.length != 0) {
|
|
options.put(COLUMNFAMILIES_KEY, commaSeparated(asList(columnFamilies)));
|
|
}
|
|
if (primaryRange != null) {
|
|
options.put(PRIMARY_RANGE_KEY, primaryRange.toString());
|
|
}
|
|
|
|
return repairAsync(keyspaceName, options, true);
|
|
}
|
|
|
|
@Override
|
|
@Deprecated
|
|
public int forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters,
|
|
Collection<String> hosts, boolean primaryRange, boolean repairedAt, String... columnFamilies)
|
|
throws IOException {
|
|
log(" forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean repairedAt, String... columnFamilies) throws IOException");
|
|
return repairRangeAsync(null, null, keyspace, isSequential, dataCenters, hosts, primaryRange, repairedAt,
|
|
columnFamilies);
|
|
}
|
|
|
|
@Override
|
|
@Deprecated
|
|
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential,
|
|
Collection<String> dataCenters, Collection<String> hosts, boolean repairedAt, String... columnFamilies) {
|
|
log(" forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean repairedAt, String... columnFamilies) throws IOException");
|
|
return repairRangeAsync(beginToken, endToken, keyspaceName, isSequential, dataCenters, hosts, null, repairedAt,
|
|
columnFamilies);
|
|
}
|
|
|
|
@Override
|
|
@Deprecated
|
|
public int forceRepairAsync(String keyspaceName, boolean isSequential, boolean isLocal, boolean primaryRange,
|
|
boolean fullRepair, String... columnFamilies) {
|
|
log(" forceRepairAsync(String keyspace, boolean isSequential, boolean isLocal, boolean primaryRange, boolean fullRepair, String... columnFamilies)");
|
|
return repairRangeAsync(null, null, keyspaceName, isSequential, null, null, primaryRange, null, columnFamilies);
|
|
}
|
|
|
|
@Override
|
|
@Deprecated
|
|
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential,
|
|
boolean isLocal, boolean repairedAt, String... columnFamilies) {
|
|
log(" forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, boolean isLocal, boolean repairedAt, String... columnFamilies)");
|
|
return forceRepairRangeAsync(beginToken, endToken, keyspaceName, isSequential, null, null, repairedAt,
|
|
columnFamilies);
|
|
}
|
|
|
|
@Override
|
|
public void forceTerminateAllRepairSessions() {
|
|
log(" forceTerminateAllRepairSessions()");
|
|
client.post("/storage_service/force_terminate");
|
|
}
|
|
|
|
/**
|
|
* transfer this node's data to other machines and remove it from service.
|
|
*/
|
|
@Override
|
|
public void decommission() throws InterruptedException {
|
|
log(" decommission() throws InterruptedException");
|
|
client.post("/storage_service/decommission");
|
|
}
|
|
|
|
/**
|
|
* @param newToken
|
|
* token to move this node to. This node will unload its data
|
|
* onto its neighbors, and bootstrap to the new token.
|
|
*/
|
|
@Override
|
|
public void move(String newToken) throws IOException {
|
|
log(" move(String newToken) throws IOException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "new_token", newToken);
|
|
client.post("/storage_service/move", queryParams);
|
|
}
|
|
|
|
/**
|
|
* removeToken removes token (and all data associated with enpoint that had
|
|
* it) from the ring
|
|
*
|
|
* @param hostIdString
|
|
* the host id to remove
|
|
*/
|
|
@Override
|
|
public void removeNode(String hostIdString) {
|
|
log(" removeNode(String token)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "host_id", hostIdString);
|
|
client.post("/storage_service/remove_node", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Get the status of a token removal.
|
|
*/
|
|
@Override
|
|
public String getRemovalStatus() {
|
|
log(" getRemovalStatus()");
|
|
return client.getStringValue("/storage_service/removal_status");
|
|
}
|
|
|
|
/**
|
|
* Force a remove operation to finish.
|
|
*/
|
|
@Override
|
|
public void forceRemoveCompletion() {
|
|
log(" forceRemoveCompletion()");
|
|
client.post("/storage_service/force_remove_completion");
|
|
}
|
|
|
|
/**
|
|
* set the logging level at runtime<br>
|
|
* <br>
|
|
* If both classQualifer and level are empty/null, it will reload the
|
|
* configuration to reset.<br>
|
|
* If classQualifer is not empty but level is empty/null, it will set the
|
|
* level to null for the defined classQualifer<br>
|
|
* If level cannot be parsed, then the level will be defaulted to DEBUG<br>
|
|
* <br>
|
|
* The logback configuration should have < jmxConfigurator /> set
|
|
*
|
|
* @param classQualifier
|
|
* The logger's classQualifer
|
|
* @param level
|
|
* The log level
|
|
* @throws Exception
|
|
*
|
|
* @see ch.qos.logback.classic.Level#toLevel(String)
|
|
*/
|
|
@Override
|
|
public void setLoggingLevel(String classQualifier, String level) throws Exception {
|
|
log(" setLoggingLevel(String classQualifier, String level) throws Exception");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "level", level);
|
|
client.post("/system/logger/" + classQualifier, queryParams);
|
|
}
|
|
|
|
/** get the runtime logging levels */
|
|
@Override
|
|
public Map<String, String> getLoggingLevels() {
|
|
log(" getLoggingLevels()");
|
|
return client.getMapStrValue("/storage_service/logging_level");
|
|
}
|
|
|
|
/**
|
|
* get the operational mode (leaving, joining, normal, decommissioned,
|
|
* client)
|
|
**/
|
|
@Override
|
|
public String getOperationMode() {
|
|
log(" getOperationMode()");
|
|
return client.getStringValue("/storage_service/operation_mode");
|
|
}
|
|
|
|
/** Returns whether the storage service is starting or not */
|
|
@Override
|
|
public boolean isStarting() {
|
|
log(" isStarting()");
|
|
return client.getBooleanValue("/storage_service/is_starting");
|
|
}
|
|
|
|
/** get the progress of a drain operation */
|
|
@Override
|
|
public String getDrainProgress() {
|
|
log(" getDrainProgress()");
|
|
// FIXME
|
|
// This is a workaround so the nodetool would work
|
|
// it should be revert when the drain progress will be implemented
|
|
// return c.getStringValue("/storage_service/drain");
|
|
return String.format("Drained %s/%s ColumnFamilies", 0, 0);
|
|
}
|
|
|
|
/**
|
|
* makes node unavailable for writes, flushes memtables and replays
|
|
* commitlog.
|
|
*/
|
|
@Override
|
|
public void drain() throws IOException, InterruptedException, ExecutionException {
|
|
log(" drain() throws IOException, InterruptedException, ExecutionException");
|
|
client.post("/storage_service/drain");
|
|
}
|
|
|
|
/**
|
|
* Truncates (deletes) the given columnFamily from the provided keyspace.
|
|
* Calling truncate results in actual deletion of all data in the cluster
|
|
* under the given columnFamily and it will fail unless all hosts are up.
|
|
* All data in the given column family will be deleted, but its definition
|
|
* will not be affected.
|
|
*
|
|
* @param keyspace
|
|
* The keyspace to delete from
|
|
* @param columnFamily
|
|
* The column family to delete data from.
|
|
*/
|
|
@Override
|
|
public void truncate(String keyspace, String columnFamily) throws TimeoutException, IOException {
|
|
log(" truncate(String keyspace, String columnFamily)throws TimeoutException, IOException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "cf", columnFamily);
|
|
client.post("/storage_service/truncate/" + keyspace, queryParams);
|
|
}
|
|
|
|
/**
|
|
* given a list of tokens (representing the nodes in the cluster), returns a
|
|
* mapping from "token -> %age of cluster owned by that token"
|
|
*/
|
|
@Override
|
|
public Map<InetAddress, Float> getOwnership() {
|
|
log(" getOwnership()");
|
|
return client.getMapInetAddressFloatValue("/storage_service/ownership/");
|
|
}
|
|
|
|
/**
|
|
* Effective ownership is % of the data each node owns given the keyspace we
|
|
* calculate the percentage using replication factor. If Keyspace == null,
|
|
* this method will try to verify if all the keyspaces in the cluster have
|
|
* the same replication strategies and if yes then we will use the first
|
|
* else a empty Map is returned.
|
|
*/
|
|
@Override
|
|
public Map<InetAddress, Float> effectiveOwnership(String keyspace) throws IllegalStateException {
|
|
log(" effectiveOwnership(String keyspace) throws IllegalStateException");
|
|
try {
|
|
return client.getMapInetAddressFloatValue("/storage_service/ownership/" + keyspace);
|
|
} catch (Exception e) {
|
|
throw new IllegalStateException(
|
|
"Non-system keyspaces don't have the same replication settings, effective ownership information is meaningless");
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public List<String> getKeyspaces() {
|
|
log(" getKeyspaces()");
|
|
return client.getListStrValue("/storage_service/keyspaces");
|
|
}
|
|
|
|
public Map<String, Set<String>> getColumnFamilyPerKeyspace() {
|
|
Map<String, Set<String>> res = new HashMap<String, Set<String>>();
|
|
|
|
JsonArray mbeans = client.getJsonArray("/column_family/");
|
|
|
|
for (int i = 0; i < mbeans.size(); i++) {
|
|
JsonObject mbean = mbeans.getJsonObject(i);
|
|
String ks = mbean.getString("ks");
|
|
String cf = mbean.getString("cf");
|
|
if (!res.containsKey(ks)) {
|
|
res.put(ks, new HashSet<String>());
|
|
}
|
|
res.get(ks).add(cf);
|
|
}
|
|
return res;
|
|
}
|
|
|
|
@Override
|
|
public List<String> getNonSystemKeyspaces() {
|
|
log(" getNonSystemKeyspaces()");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("type", "user");
|
|
return client.getListStrValue("/storage_service/keyspaces", queryParams);
|
|
}
|
|
|
|
@Override
|
|
public Map<String, String> getViewBuildStatuses(String keyspace, String view) {
|
|
log(" getViewBuildStatuses()");
|
|
return client.getMapStrValue("storage_service/view_build_statuses/" + keyspace + "/" + view);
|
|
}
|
|
|
|
/**
|
|
* Change endpointsnitch class and dynamic-ness (and dynamic attributes) at
|
|
* runtime
|
|
*
|
|
* @param epSnitchClassName
|
|
* the canonical path name for a class implementing
|
|
* IEndpointSnitch
|
|
* @param dynamic
|
|
* boolean that decides whether dynamicsnitch is used or not
|
|
* @param dynamicUpdateInterval
|
|
* integer, in ms (default 100)
|
|
* @param dynamicResetInterval
|
|
* integer, in ms (default 600,000)
|
|
* @param dynamicBadnessThreshold
|
|
* double, (default 0.0)
|
|
*/
|
|
@Override
|
|
public void updateSnitch(String epSnitchClassName, Boolean dynamic, 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 MultivaluedHashMap<String, String>();
|
|
APIClient.set_bool_query_param(queryParams, "dynamic", dynamic);
|
|
APIClient.set_query_param(queryParams, "epSnitchClassName", epSnitchClassName);
|
|
if (dynamicUpdateInterval != null) {
|
|
queryParams.add("dynamic_update_interval", dynamicUpdateInterval.toString());
|
|
}
|
|
if (dynamicResetInterval != null) {
|
|
queryParams.add("dynamic_reset_interval", dynamicResetInterval.toString());
|
|
}
|
|
if (dynamicBadnessThreshold != null) {
|
|
queryParams.add("dynamic_badness_threshold", dynamicBadnessThreshold.toString());
|
|
}
|
|
client.post("/storage_service/update_snitch", queryParams);
|
|
}
|
|
|
|
// allows a user to forcibly 'kill' a sick node
|
|
@Override
|
|
public void stopGossiping() {
|
|
log(" stopGossiping()");
|
|
client.delete("/storage_service/gossiping");
|
|
}
|
|
|
|
// allows a user to recover a forcibly 'killed' node
|
|
@Override
|
|
public void startGossiping() {
|
|
log(" startGossiping()");
|
|
client.post("/storage_service/gossiping");
|
|
}
|
|
|
|
// allows a user to see whether gossip is running or not
|
|
@Override
|
|
public boolean isGossipRunning() {
|
|
log(" isGossipRunning()");
|
|
return client.getBooleanValue("/storage_service/gossiping");
|
|
}
|
|
|
|
// allows a user to forcibly completely stop cassandra
|
|
@Override
|
|
public void stopDaemon() {
|
|
log(" stopDaemon()");
|
|
client.post("/storage_service/stop_daemon");
|
|
}
|
|
|
|
// to determine if gossip is disabled
|
|
@Override
|
|
public boolean isInitialized() {
|
|
log(" isInitialized()");
|
|
return client.getBooleanValue("/storage_service/is_initialized");
|
|
}
|
|
|
|
// allows a user to disable thrift
|
|
@Override
|
|
public void stopRPCServer() {
|
|
log(" stopRPCServer()");
|
|
client.delete("/storage_service/rpc_server");
|
|
}
|
|
|
|
// allows a user to reenable thrift
|
|
@Override
|
|
public void startRPCServer() {
|
|
log(" startRPCServer()");
|
|
client.post("/storage_service/rpc_server");
|
|
}
|
|
|
|
// to determine if thrift is running
|
|
@Override
|
|
public boolean isRPCServerRunning() {
|
|
log(" isRPCServerRunning()");
|
|
return client.getBooleanValue("/storage_service/rpc_server");
|
|
}
|
|
|
|
@Override
|
|
public void stopNativeTransport() {
|
|
log(" stopNativeTransport()");
|
|
client.delete("/storage_service/native_transport");
|
|
}
|
|
|
|
@Override
|
|
public void startNativeTransport() {
|
|
log(" startNativeTransport()");
|
|
client.post("/storage_service/native_transport");
|
|
}
|
|
|
|
@Override
|
|
public boolean isNativeTransportRunning() {
|
|
log(" isNativeTransportRunning()");
|
|
return client.getBooleanValue("/storage_service/native_transport");
|
|
}
|
|
|
|
// allows a node that have been started without joining the ring to join it
|
|
@Override
|
|
public void joinRing() throws IOException {
|
|
log(" joinRing() throws IOException");
|
|
client.post("/storage_service/join_ring");
|
|
}
|
|
|
|
@Override
|
|
public boolean isJoined() {
|
|
log(" isJoined()");
|
|
return client.getBooleanValue("/storage_service/join_ring");
|
|
}
|
|
|
|
@Override
|
|
public void setStreamThroughputMbPerSec(int value) {
|
|
log(" setStreamThroughputMbPerSec(int value)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("value", Integer.toString(value));
|
|
client.post("/storage_service/stream_throughput", queryParams);
|
|
}
|
|
|
|
@Override
|
|
public int getStreamThroughputMbPerSec() {
|
|
log(" getStreamThroughputMbPerSec()");
|
|
return client.getIntValue("/storage_service/stream_throughput");
|
|
}
|
|
|
|
public int getCompactionThroughputMbPerSec() {
|
|
log(" getCompactionThroughputMbPerSec()");
|
|
return client.getIntValue("/storage_service/compaction_throughput");
|
|
}
|
|
|
|
@Override
|
|
public void setCompactionThroughputMbPerSec(int value) {
|
|
log(" setCompactionThroughputMbPerSec(int value)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("value", Integer.toString(value));
|
|
client.post("/storage_service/compaction_throughput", queryParams);
|
|
}
|
|
|
|
@Override
|
|
public boolean isIncrementalBackupsEnabled() {
|
|
log(" isIncrementalBackupsEnabled()");
|
|
return client.getBooleanValue("/storage_service/incremental_backups");
|
|
}
|
|
|
|
@Override
|
|
public void setIncrementalBackupsEnabled(boolean value) {
|
|
log(" setIncrementalBackupsEnabled(boolean value)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("value", Boolean.toString(value));
|
|
client.post("/storage_service/incremental_backups", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Initiate a process of streaming data for which we are responsible from
|
|
* other nodes. It is similar to bootstrap except meant to be used on a node
|
|
* which is already in the cluster (typically containing no data) as an
|
|
* alternative to running repair.
|
|
*
|
|
* @param sourceDc
|
|
* Name of DC from which to select sources for streaming or null
|
|
* to pick any node
|
|
*/
|
|
@Override
|
|
public void rebuild(String sourceDc) {
|
|
rebuild(sourceDc, null, null, null);
|
|
}
|
|
|
|
/**
|
|
* Same as {@link #rebuild(String)}, but only for specified keyspace and
|
|
* ranges.
|
|
*
|
|
* @param sourceDc
|
|
* Name of DC from which to select sources for streaming or null
|
|
* to pick any node
|
|
* @param keyspace
|
|
* Name of the keyspace which to rebuild or null to rebuild all
|
|
* keyspaces.
|
|
* @param tokens
|
|
* Range of tokens to rebuild or null to rebuild all token
|
|
* ranges. In the format of:
|
|
* "(start_token_1,end_token_1],(start_token_2,end_token_2],...(start_token_n,end_token_n]"
|
|
*/
|
|
@Override
|
|
public void rebuild(String sourceDc, String keyspace, String tokens, String specificSources) {
|
|
log(" rebuild(String sourceDc, String keyspace, String tokens, String specificSources)");
|
|
if (keyspace != null) {
|
|
throw new UnsupportedOperationException("Rebuild: 'keyspace' not yet supported");
|
|
}
|
|
if (tokens != null) {
|
|
throw new UnsupportedOperationException("Rebuild: 'token range' not yet supported");
|
|
}
|
|
if (specificSources != null) {
|
|
throw new UnsupportedOperationException("Rebuild: 'specific sources' not yet supported");
|
|
}
|
|
if (sourceDc != null) {
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "source_dc", sourceDc);
|
|
client.post("/storage_service/rebuild", queryParams);
|
|
} else {
|
|
client.post("/storage_service/rebuild");
|
|
}
|
|
}
|
|
|
|
/** Starts a bulk load and blocks until it completes. */
|
|
@Override
|
|
public void bulkLoad(String directory) {
|
|
log(" bulkLoad(String directory)");
|
|
client.post("/storage_service/bulk_load/" + directory);
|
|
}
|
|
|
|
/**
|
|
* Starts a bulk load asynchronously and returns the String representation
|
|
* of the planID for the new streaming session.
|
|
*/
|
|
@Override
|
|
public String bulkLoadAsync(String directory) {
|
|
log(" bulkLoadAsync(String directory)");
|
|
return client.getStringValue("/storage_service/bulk_load_async/" + directory);
|
|
}
|
|
|
|
@Override
|
|
public void rescheduleFailedDeletions() {
|
|
log(" rescheduleFailedDeletions()");
|
|
client.post("/storage_service/reschedule_failed_deletions");
|
|
}
|
|
|
|
/**
|
|
* Load new SSTables to the given keyspace/columnFamily
|
|
*
|
|
* @param ksName
|
|
* The parent keyspace name
|
|
* @param cfName
|
|
* The ColumnFamily name where SSTables belong
|
|
*/
|
|
@Override
|
|
public void loadNewSSTables(String ksName, String cfName) {
|
|
log(" loadNewSSTables(String ksName, String cfName)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("cf", cfName);
|
|
client.post("/storage_service/sstables/" + ksName, queryParams);
|
|
}
|
|
|
|
/**
|
|
* Return a List of Tokens representing a sample of keys across all
|
|
* ColumnFamilyStores.
|
|
*
|
|
* Note: this should be left as an operation, not an attribute (methods
|
|
* starting with "get") to avoid sending potentially multiple MB of data
|
|
* when accessing this mbean by default. See CASSANDRA-4452.
|
|
*
|
|
* @return set of Tokens as Strings
|
|
*/
|
|
@Override
|
|
public List<String> sampleKeyRange() {
|
|
log(" sampleKeyRange()");
|
|
return client.getListStrValue("/storage_service/sample_key_range");
|
|
}
|
|
|
|
/**
|
|
* rebuild the specified indexes
|
|
*/
|
|
@Override
|
|
public void rebuildSecondaryIndex(String ksName, String cfName, String... idxNames) {
|
|
log(" rebuildSecondaryIndex(String ksName, String cfName, String... idxNames)");
|
|
}
|
|
|
|
@Override
|
|
public void resetLocalSchema() throws IOException {
|
|
log(" resetLocalSchema() throws IOException");
|
|
client.post("/storage_service/relocal_schema");
|
|
}
|
|
|
|
/**
|
|
* Enables/Disables tracing for the whole system. Only thrift requests can
|
|
* start tracing currently.
|
|
*
|
|
* @param probability
|
|
* ]0,1[ will enable tracing on a partial number of requests with
|
|
* the provided probability. 0 will disable tracing and 1 will
|
|
* enable tracing for all requests (which mich severely cripple
|
|
* the system)
|
|
*/
|
|
@Override
|
|
public void setTraceProbability(double probability) {
|
|
log(" setTraceProbability(double probability)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("probability", Double.toString(probability));
|
|
client.post("/storage_service/trace_probability", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Returns the configured tracing probability.
|
|
*/
|
|
@Override
|
|
public double getTraceProbability() {
|
|
log(" getTraceProbability()");
|
|
return client.getDoubleValue("/storage_service/trace_probability");
|
|
}
|
|
|
|
@Override
|
|
public void disableAutoCompaction(String ks, String... columnFamilies) throws IOException {
|
|
log("disableAutoCompaction(String ks, String... columnFamilies)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
client.delete("/storage_service/auto_compaction/" + ks, queryParams);
|
|
}
|
|
|
|
@Override
|
|
public void enableAutoCompaction(String ks, String... columnFamilies) throws IOException {
|
|
log("enableAutoCompaction(String ks, String... columnFamilies)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
client.post("/storage_service/auto_compaction/" + ks, queryParams);
|
|
}
|
|
|
|
@Override
|
|
public void deliverHints(String host) throws UnknownHostException {
|
|
log(" deliverHints(String host) throws UnknownHostException");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("host", host);
|
|
client.post("/storage_service/deliver_hints", queryParams);
|
|
}
|
|
|
|
/** Returns the name of the cluster */
|
|
@Override
|
|
public String getClusterName() {
|
|
log(" getClusterName()");
|
|
return client.getStringValue("/storage_service/cluster_name");
|
|
}
|
|
|
|
/** Returns the cluster partitioner */
|
|
@Override
|
|
public String getPartitionerName() {
|
|
log(" getPartitionerName()");
|
|
return client.getStringValue("/storage_service/partitioner_name");
|
|
}
|
|
|
|
/** Returns the threshold for warning of queries with many tombstones */
|
|
@Override
|
|
public int getTombstoneWarnThreshold() {
|
|
log(" getTombstoneWarnThreshold()");
|
|
return client.getIntValue("/storage_service/tombstone_warn_threshold");
|
|
}
|
|
|
|
/** Sets the threshold for warning queries with many tombstones */
|
|
@Override
|
|
public void setTombstoneWarnThreshold(int tombstoneDebugThreshold) {
|
|
log(" setTombstoneWarnThreshold(int tombstoneDebugThreshold)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("debug_threshold", Integer.toString(tombstoneDebugThreshold));
|
|
client.post("/storage_service/tombstone_warn_threshold", queryParams);
|
|
}
|
|
|
|
/** Returns the threshold for abandoning queries with many tombstones */
|
|
@Override
|
|
public int getTombstoneFailureThreshold() {
|
|
log(" getTombstoneFailureThreshold()");
|
|
return client.getIntValue("/storage_service/tombstone_failure_threshold");
|
|
}
|
|
|
|
/** Sets the threshold for abandoning queries with many tombstones */
|
|
@Override
|
|
public void setTombstoneFailureThreshold(int tombstoneDebugThreshold) {
|
|
log(" setTombstoneFailureThreshold(int tombstoneDebugThreshold)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("debug_threshold", Integer.toString(tombstoneDebugThreshold));
|
|
client.post("/storage_service/tombstone_failure_threshold", queryParams);
|
|
}
|
|
|
|
/** Returns the threshold for rejecting queries due to a large batch size */
|
|
@Override
|
|
public int getBatchSizeFailureThreshold() {
|
|
log(" getBatchSizeFailureThreshold()");
|
|
return client.getIntValue("/storage_service/batch_size_failure_threshold");
|
|
}
|
|
|
|
/** Sets the threshold for rejecting queries due to a large batch size */
|
|
@Override
|
|
public void setBatchSizeFailureThreshold(int batchSizeDebugThreshold) {
|
|
log(" setBatchSizeFailureThreshold(int batchSizeDebugThreshold)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("threshold", Integer.toString(batchSizeDebugThreshold));
|
|
client.post("/storage_service/batch_size_failure_threshold", queryParams);
|
|
}
|
|
|
|
/**
|
|
* Sets the hinted handoff throttle in kb per second, per delivery thread.
|
|
*/
|
|
@Override
|
|
public void setHintedHandoffThrottleInKB(int throttleInKB) {
|
|
log(" setHintedHandoffThrottleInKB(int throttleInKB)");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("throttle", Integer.toString(throttleInKB));
|
|
client.post("/storage_service/hinted_handoff", queryParams);
|
|
|
|
}
|
|
|
|
@Override
|
|
public void takeMultipleColumnFamilySnapshot(String tag, String... columnFamilyList) throws IOException {
|
|
log(" takeMultipleColumnFamilySnapshot");
|
|
Map<String, List<String>> keyspaceColumnfamily = new HashMap<String, List<String>>();
|
|
Map<String, Set<String>> kss = getColumnFamilyPerKeyspace();
|
|
Map<String, Map<String, Set<String>>> snapshots = getSnapshotKeyspaceColumnFamily();
|
|
for (String columnFamily : columnFamilyList) {
|
|
String splittedString[] = columnFamily.split("\\.");
|
|
if (splittedString.length == 2) {
|
|
String keyspaceName = splittedString[0];
|
|
String columnFamilyName = splittedString[1];
|
|
|
|
if (keyspaceName == null) {
|
|
throw new IOException("You must supply a keyspace name");
|
|
}
|
|
if (columnFamilyName == null) {
|
|
throw new IOException("You must supply a column family name");
|
|
}
|
|
if (tag == null || tag.equals("")) {
|
|
throw new IOException("You must supply a snapshot name.");
|
|
}
|
|
if (!kss.containsKey(keyspaceName)) {
|
|
throw new IOException("Keyspace " + keyspaceName + " does not exist");
|
|
}
|
|
if (!kss.get(keyspaceName).contains(columnFamilyName)) {
|
|
throw new IllegalArgumentException(
|
|
String.format("Unknown keyspace/cf pair (%s.%s)", keyspaceName, columnFamilyName));
|
|
}
|
|
// As there can be multiple column family from same keyspace
|
|
// check if snapshot exist for that specific
|
|
// columnfamily and not for whole keyspace
|
|
|
|
if (snapshots.containsKey(tag) && snapshots.get(tag).containsKey(keyspaceName)
|
|
&& snapshots.get(tag).get(keyspaceName).contains(columnFamilyName)) {
|
|
throw new IOException("Snapshot " + tag + " already exists.");
|
|
}
|
|
|
|
if (!keyspaceColumnfamily.containsKey(keyspaceName)) {
|
|
keyspaceColumnfamily.put(keyspaceName, new ArrayList<String>());
|
|
}
|
|
|
|
// Add Keyspace columnfamily to map in order to support
|
|
// atomicity for snapshot process.
|
|
// So no snapshot should happen if any one of the above
|
|
// conditions fail for any keyspace or columnfamily
|
|
keyspaceColumnfamily.get(keyspaceName).add(columnFamilyName);
|
|
|
|
} else {
|
|
throw new IllegalArgumentException(
|
|
"Cannot take a snapshot on secondary index or invalid column family name. You must supply a column family name in the form of keyspace.columnfamily");
|
|
}
|
|
}
|
|
|
|
for (Entry<String, List<String>> entry : keyspaceColumnfamily.entrySet()) {
|
|
for (String columnFamily : entry.getValue()) {
|
|
takeColumnFamilySnapshot(entry.getKey(), columnFamily, tag);
|
|
}
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public int forceRepairAsync(String keyspace, int parallelismDegree, Collection<String> dataCenters,
|
|
Collection<String> hosts, boolean primaryRange, boolean fullRepair, String... columnFamilies) {
|
|
log(" forceRepairAsync(keyspace, parallelismDegree, dataCenters, hosts, primaryRange, fullRepair, columnFamilies)");
|
|
Map<String, String> options = new HashMap<String, String>();
|
|
Joiner commas = Joiner.on(",");
|
|
options.put("parallelism", Integer.toString(parallelismDegree));
|
|
if (dataCenters != null) {
|
|
options.put("dataCenters", commas.join(dataCenters));
|
|
}
|
|
if (hosts != null) {
|
|
options.put("hosts", commas.join(hosts));
|
|
}
|
|
options.put("primaryRange", Boolean.toString(primaryRange));
|
|
options.put("incremental", Boolean.toString(!fullRepair));
|
|
if (columnFamilies != null && columnFamilies.length > 0) {
|
|
options.put("columnFamilies", commas.join(columnFamilies));
|
|
}
|
|
return repairAsync(keyspace, options);
|
|
}
|
|
|
|
@Override
|
|
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, int parallelismDegree,
|
|
Collection<String> dataCenters, Collection<String> hosts, boolean fullRepair, String... columnFamilies) {
|
|
log(" forceRepairRangeAsync(beginToken, endToken, keyspaceName, parallelismDegree, dataCenters, hosts, fullRepair, columnFamilies)");
|
|
Map<String, String> options = new HashMap<String, String>();
|
|
Joiner commas = Joiner.on(",");
|
|
options.put("parallelism", Integer.toString(parallelismDegree));
|
|
if (dataCenters != null) {
|
|
options.put("dataCenters", commas.join(dataCenters));
|
|
}
|
|
if (hosts != null) {
|
|
options.put("hosts", commas.join(hosts));
|
|
}
|
|
options.put("incremental", Boolean.toString(!fullRepair));
|
|
options.put("startToken", beginToken);
|
|
options.put("endToken", endToken);
|
|
return repairAsync(keyspaceName, options);
|
|
}
|
|
|
|
@Override
|
|
public Map<String, String> getEndpointToHostId() {
|
|
return getHostIdMap();
|
|
}
|
|
|
|
@Override
|
|
public Map<String, String> getHostIdToEndpoint() {
|
|
return getHostIdToAddressMap();
|
|
}
|
|
|
|
@Override
|
|
public void refreshSizeEstimates() throws ExecutionException {
|
|
// TODO Auto-generated method stub
|
|
log(" refreshSizeEstimates");
|
|
}
|
|
|
|
@Override
|
|
public void forceKeyspaceCompaction(boolean splitOutput, String keyspaceName, String... tableNames)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
// "splitOutput" afaik not relevant for scylla (yet?...)
|
|
forceKeyspaceCompaction(keyspaceName, tableNames);
|
|
}
|
|
|
|
@Override
|
|
public int forceKeyspaceCleanup(int jobs, String keyspaceName, String... tables)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
// "jobs" not (yet) relevant for scylla. (though possibly useful...)
|
|
return forceKeyspaceCleanup(keyspaceName, tables);
|
|
}
|
|
|
|
@Override
|
|
public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, int jobs, String keyspaceName,
|
|
String... columnFamilies) throws IOException, ExecutionException, InterruptedException {
|
|
// "jobs" not (yet) relevant for scylla. (though possibly useful...)
|
|
return scrub(disableSnapshot, skipCorrupted, checkData, keyspaceName, columnFamilies);
|
|
}
|
|
|
|
@Override
|
|
public int verify(boolean extendedVerify, String keyspaceName, String... tableNames)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
// TODO Auto-generated method stub
|
|
log(" verify");
|
|
return 0;
|
|
}
|
|
|
|
@Override
|
|
public int upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, int jobs, String... tableNames)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
// "jobs" not (yet) relevant for scylla. (though possibly useful...)
|
|
return upgradeSSTables(keyspaceName, excludeCurrentVersion, tableNames);
|
|
}
|
|
|
|
@Override
|
|
public List<String> getNonLocalStrategyKeyspaces() {
|
|
log(" getNonLocalStrategyKeyspaces");
|
|
MultivaluedMap<String, String> queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("type", "non_local_strategy");
|
|
return client.getListStrValue("/storage_service/keyspaces", queryParams);
|
|
}
|
|
|
|
@Override
|
|
public void setInterDCStreamThroughputMbPerSec(int value) {
|
|
// TODO Auto-generated method stub
|
|
log(" setInterDCStreamThroughputMbPerSec");
|
|
}
|
|
|
|
@Override
|
|
public int getInterDCStreamThroughputMbPerSec() {
|
|
// TODO Auto-generated method stub
|
|
log(" getInterDCStreamThroughputMbPerSec");
|
|
return 0;
|
|
}
|
|
|
|
@Override
|
|
public boolean resumeBootstrap() {
|
|
log(" resumeBootstrap");
|
|
return false;
|
|
}
|
|
|
|
@Override
|
|
public List<CompositeData> getSSTableInfo(String keyspace, String table) {
|
|
if (keyspace == null && table != null) {
|
|
throw new IllegalArgumentException("Missing keyspace name");
|
|
}
|
|
MultivaluedMap<String, String> queryParams = null;
|
|
|
|
if (keyspace != null) {
|
|
queryParams = new MultivaluedHashMap<String, String>();
|
|
queryParams.add("keyspace", keyspace);
|
|
}
|
|
if (table != null) {
|
|
queryParams.add("cf", table);
|
|
}
|
|
|
|
return client.get("/storage_service/sstable_info", queryParams)
|
|
.get(new GenericType<List<PerTableSSTableInfo>>() {
|
|
}).stream().map((i) -> i.toCompositeData()).collect(Collectors.toList());
|
|
}
|
|
|
|
@Override
|
|
public List<CompositeData> getSSTableInfo() {
|
|
return getSSTableInfo(null, null);
|
|
}
|
|
|
|
@Override
|
|
public int scrub(boolean disableSnapshot, boolean skipCorrupted, boolean checkData, boolean reinsertOverflowedTTL,
|
|
int jobs, String keyspaceName, String... columnFamilies)
|
|
throws IOException, ExecutionException, InterruptedException {
|
|
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", skipCorrupted);
|
|
APIClient.set_query_param(queryParams, "cf", APIClient.join(columnFamilies));
|
|
return client.getIntValue("/storage_service/keyspace_scrub/" + keyspaceName, queryParams);
|
|
}
|
|
@Override
|
|
public long getUptime() {
|
|
log("getUptime()");
|
|
return client.getLongValue("/system/uptime_ms");
|
|
}
|
|
}
|