natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allow noarg toppartitions 

Closes #214

Patch by Chris Lohfink; Reviewed by Dinesh Joshi for CASSANDRA-14360
This commit is contained in:
Chris Lohfink 2018-05-02 00:52:16 -07:00 committed by Jeff Jirsa
parent dd650c8e86
commit 45f250535a
6 changed files with 189 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0
* Allow nodetool toppartitions without specifying table (CASSANDRA-14360)
* Audit logging for database activity (CASSANDRA-12151)
* Clean up build artifacts in docs container (CASSANDRA-14432)
* Minor network authz improvements (Cassandra-14413)

38
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -33,9 +33,13 @@ import java.util.stream.StreamSupport;
import javax.annotation.Nullable;
import javax.management.*;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.CompositeDataSupport;
import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
import javax.management.openmbean.TabularDataSupport;
import com.clearspring.analytics.stream.Counter;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
@ -76,6 +80,7 @@ import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.*;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.metrics.TableMetrics.Sampler;
import org.apache.cassandra.net.*;
import org.apache.cassandra.repair.*;
import org.apache.cassandra.repair.messages.RepairOption;
@ -99,6 +104,7 @@ import org.apache.cassandra.streaming.*;
import org.apache.cassandra.tracing.TraceKeyspace;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.TopKSampler.SamplerResult;
import org.apache.cassandra.utils.logging.LoggingSupportFactory;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventType;
@ -5269,6 +5275,38 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return sampledKeys;
}
/*
* little hard to parse for JMX MBean requirements, but the output looks something like:
*
* {"keyspace.table":
* {"SAMPLER": [{cardinality:i partitions: [{raw:"", string:"", count:i, error:i}, ...]}, ...]}
* }
*/
@Override
public Map<String, Map<String, CompositeData>> samplePartitions(long duration, int capacity, int count, List<String> samplers) throws OpenDataException
{
for (String sampler : samplers)
{
for (ColumnFamilyStore table : ColumnFamilyStore.all())
{
table.beginLocalSampling(sampler, capacity);
}
}
Uninterruptibles.sleepUninterruptibly(duration, TimeUnit.MILLISECONDS);
ConcurrentHashMap<String, Map<String, CompositeData>> result = new ConcurrentHashMap<>();
for (String sampler : samplers)
{
for (ColumnFamilyStore table : ColumnFamilyStore.all())
{
String name = table.keyspace.getName() + "." + table.name;
Map<String, CompositeData> topk = result.computeIfAbsent(name, x -> new HashMap<>());
topk.put(sampler, table.finishLocalSampling(sampler, count));
}
}
return result;
}
public void rebuildSecondaryIndex(String ksName, String cfName, String... idxNames)
{
String[] indices = asList(idxNames).stream()

5
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -27,9 +27,12 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nullable;
import javax.management.NotificationEmitter;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.metrics.TableMetrics.Sampler;
public interface StorageServiceMBean extends NotificationEmitter
{
@ -628,6 +631,8 @@ public interface StorageServiceMBean extends NotificationEmitter
*/
public void setTraceProbability(double probability);
public Map<String, Map<String, CompositeData>> samplePartitions(long duration, int capacity, int count, List<String> samplers) throws OpenDataException;
/**
* Returns the configured tracing probability.
*/

21
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -89,6 +89,7 @@ import org.apache.cassandra.streaming.management.StreamStateCompositeData;
import com.google.common.base.Function;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
@ -418,21 +419,27 @@ public class NodeProbe implements AutoCloseable
}
}
}
public Map<String, Map<String, CompositeData>> getPartitionSample(int capacity, int duration, int count, List<String> samplers) throws OpenDataException
{
return ssProxy.samplePartitions(duration, capacity, count, samplers);
}
public Map<Sampler, CompositeData> getPartitionSample(String ks, String cf, int capacity, int duration, int count, List<Sampler> samplers) throws OpenDataException
public Map<String, Map<String, CompositeData>> getPartitionSample(String ks, String cf, int capacity, int duration, int count, List<String> samplers) throws OpenDataException
{
ColumnFamilyStoreMBean cfsProxy = getCfsProxy(ks, cf);
for(Sampler sampler : samplers)
for(String sampler : samplers)
{
cfsProxy.beginLocalSampling(sampler.name(), capacity);
cfsProxy.beginLocalSampling(sampler, capacity);
}
Uninterruptibles.sleepUninterruptibly(duration, TimeUnit.MILLISECONDS);
Map<Sampler, CompositeData> result = Maps.newHashMap();
for(Sampler sampler : samplers)
Map<String, CompositeData> result = Maps.newHashMap();
for(String sampler : samplers)
{
result.put(sampler, cfsProxy.finishLocalSampling(sampler.name(), count));
result.put(sampler, cfsProxy.finishLocalSampling(sampler, count));
}
return result;
return new ImmutableMap.Builder<String, Map<String, CompositeData>>()
.put(ks + "." + cf, result)
.build();
}
public void invalidateCounterCache()

102
src/java/org/apache/cassandra/tools/nodetool/TopPartitions.java
View File

@ -19,9 +19,6 @@ package org.apache.cassandra.tools.nodetool;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.commons.lang3.StringUtils.join;
import io.airlift.airline.Arguments;
import io.airlift.airline.Command;
import io.airlift.airline.Option;
import java.util.ArrayList;
import java.util.Collections;
@ -37,14 +34,20 @@ import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.metrics.TableMetrics.Sampler;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
import org.apache.cassandra.tools.nodetool.formatter.TableBuilder;
import org.apache.cassandra.utils.Pair;
import com.google.common.collect.Lists;
import com.google.common.collect.Ordering;
@Command(name = "toppartitions", description = "Sample and print the most active partitions for a given column family")
import io.airlift.airline.Arguments;
import io.airlift.airline.Command;
import io.airlift.airline.Option;
@Command(name = "toppartitions", description = "Sample and print the most active partitions")
public class TopPartitions extends NodeToolCmd
{
@Arguments(usage = "<keyspace> <cfname> <duration>", description = "The keyspace, column family name, and duration in milliseconds")
@Arguments(usage = "[keyspace table] [duration]", description = "The keyspace, table name, and duration in milliseconds")
private List<String> args = new ArrayList<>();
@Option(name = "-s", description = "Capacity of stream summary, closer to the actual cardinality of partitions will yield more accurate results (Default: 256)")
private int size = 256;
@ -55,63 +58,86 @@ public class TopPartitions extends NodeToolCmd
@Override
public void execute(NodeProbe probe)
{
checkArgument(args.size() == 3, "toppartitions requires keyspace, column family name, and duration");
checkArgument(args.size() == 3 || args.size() == 1 || args.size() == 0, "Invalid arguments, either [keyspace table duration] or [duration] or no args");
checkArgument(topCount < size, "TopK count (-k) option must be smaller then the summary capacity (-s)");
String keyspace = args.get(0);
String cfname = args.get(1);
Integer duration = Integer.valueOf(args.get(2));
String keyspace = null;
String table = null;
Integer duration = 10000;
if(args.size() == 3)
{
keyspace = args.get(0);
table = args.get(1);
duration = Integer.valueOf(args.get(2));
}
else if (args.size() == 1)
{
duration = Integer.valueOf(args.get(0));
}
// generate the list of samplers
List<Sampler> targets = Lists.newArrayList();
List<String> targets = Lists.newArrayList();
for (String s : samplers.split(","))
{
try
{
targets.add(Sampler.valueOf(s.toUpperCase()));
targets.add(Sampler.valueOf(s.toUpperCase()).toString());
} catch (Exception e)
{
throw new IllegalArgumentException(s + " is not a valid sampler, choose one of: " + join(Sampler.values(), ", "));
}
}
Map<Sampler, CompositeData> results;
Map<String, Map<String, CompositeData>> results;
try
{
results = probe.getPartitionSample(keyspace, cfname, size, duration, topCount, targets);
if (keyspace == null)
{
results = probe.getPartitionSample(size, duration, topCount, targets);
}
else
{
results = probe.getPartitionSample(keyspace, table, size, duration, topCount, targets);
}
} catch (OpenDataException e)
{
throw new RuntimeException(e);
}
boolean first = true;
for(Entry<Sampler, CompositeData> result : results.entrySet())
for(String sampler : targets)
{
CompositeData sampling = result.getValue();
// weird casting for http://bugs.sun.com/view_bug.do?bug_id=6548436
List<CompositeData> topk = (List<CompositeData>) (Object) Lists.newArrayList(((TabularDataSupport) sampling.get("partitions")).values());
Collections.sort(topk, new Ordering<CompositeData>()
{
public int compare(CompositeData left, CompositeData right)
{
return Long.compare((long) right.get("count"), (long) left.get("count"));
}
});
if(!first)
System.out.println();
System.out.println(result.getKey().toString()+ " Sampler:");
System.out.printf(" Cardinality: ~%d (%d capacity)%n", sampling.get("cardinality"), size);
System.out.printf(" Top %d partitions:%n", topCount);
first = false;
System.out.printf(sampler + " Sampler Top %d partitions:%n", topCount);
TableBuilder out = new TableBuilder();
out.add("\t", "Table", "Partition", "Count", "+/-");
List<Pair<String, CompositeData>> topk = new ArrayList<>(topCount);
for (Entry<String, Map<String, CompositeData>> tableResult : results.entrySet())
{
String tableName = tableResult.getKey();
CompositeData sampling = tableResult.getValue().get(sampler);
// weird casting for http://bugs.sun.com/view_bug.do?bug_id=6548436
for(CompositeData cd : (List<CompositeData>) (Object) Lists.newArrayList(((TabularDataSupport) sampling.get("partitions")).values()))
{
topk.add(Pair.create(tableName, cd));
}
}
Collections.sort(topk, new Ordering<Pair<String, CompositeData>>()
{
public int compare(Pair<String, CompositeData> left, Pair<String, CompositeData> right)
{
return Long.compare((long) right.right.get("count"), (long) left.right.get("count"));
}
});
for (Pair<String, CompositeData> entry : topk.subList(0, Math.min(topk.size(), 10)))
{
CompositeData cd = entry.right;
out.add("\t", entry.left, cd.get("string").toString(), cd.get("count").toString(), cd.get("error").toString());
}
out.printTo(System.out);
if (topk.size() == 0)
{
System.out.println("\tNothing recorded during sampling period...");
} else
{
int offset = 0;
for (CompositeData entry : topk)
offset = Math.max(offset, entry.get("string").toString().length());
System.out.printf("\t%-" + offset + "s%10s%10s%n", "Partition", "Count", "+/-");
for (CompositeData entry : topk)
System.out.printf("\t%-" + offset + "s%10d%10d%n", entry.get("string").toString(), entry.get("count"), entry.get("error"));
System.out.println("\t Nothing recorded during sampling period...");
}
first = false;
}
}
}
}

67
test/unit/org/apache/cassandra/tools/TopPartitionsTest.java Normal file
View File

@ -0,0 +1,67 @@
package org.apache.cassandra.tools;
import static java.lang.String.format;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.junit.Assert.assertEquals;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.TabularDataSupport;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.service.StorageService;
import org.junit.BeforeClass;
import org.junit.Test;
import com.google.common.collect.Lists;
public class TopPartitionsTest
{
@BeforeClass
public static void loadSchema() throws ConfigurationException
{
SchemaLoader.prepareServer();
}
@Test
public void testServiceTopPartitionsNoArg() throws Exception
{
BlockingQueue<Map<String, Map<String, CompositeData>>> q = new ArrayBlockingQueue<>(1);
ColumnFamilyStore.all();
Executors.newCachedThreadPool().execute(() ->
{
try
{
q.put(StorageService.instance.samplePartitions(1000, 100, 10, Lists.newArrayList("READS", "WRITES")));
}
catch (Exception e)
{
e.printStackTrace();
}
});
SystemKeyspace.persistLocalMetadata();
Map<String, Map<String, CompositeData>> result = q.poll(11, TimeUnit.SECONDS);
List<CompositeData> cd = (List<CompositeData>) (Object) Lists.newArrayList(((TabularDataSupport) result.get("system.local").get("WRITES").get("partitions")).values());
assertEquals(1, cd.size());
}
@Test
public void testServiceTopPartitionsSingleTable() throws Exception
{
ColumnFamilyStore.getIfExists("system", "local").beginLocalSampling("READS", 5);
String req = "SELECT * FROM system.%s WHERE key='%s'";
executeInternal(format(req, SystemKeyspace.LOCAL, SystemKeyspace.LOCAL));
CompositeData result = ColumnFamilyStore.getIfExists("system", "local").finishLocalSampling("READS", 5);
List<CompositeData> cd = (List<CompositeData>) (Object) Lists.newArrayList(((TabularDataSupport) result.get("partitions")).values());
assertEquals(1, cd.size());
}
}