Add sampler for query time and expose with nodetool patch by Chris Lohfink; reviewed by Dinesh Joshi for CASSANDRA-14436
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f31d1a05 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f31d1a05 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f31d1a05 Branch: refs/heads/trunk Commit: f31d1a05a1f6f85f64c9b965009db814960c4eca Parents: a09dc3a Author: Chris Lohfink <[email protected]> Authored: Fri Aug 17 00:40:54 2018 -0500 Committer: Aleksey Yeshchenko <[email protected]> Committed: Fri Aug 31 18:14:31 2018 +0100 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../apache/cassandra/db/ColumnFamilyStore.java | 67 ++--- .../cassandra/db/ColumnFamilyStoreMBean.java | 4 +- .../cassandra/db/ReadExecutionController.java | 89 ++++--- .../db/SinglePartitionReadCommand.java | 4 +- .../cassandra/metrics/FrequencySampler.java | 103 ++++++++ .../apache/cassandra/metrics/MaxSampler.java | 75 ++++++ .../org/apache/cassandra/metrics/Sampler.java | 97 ++++++++ .../apache/cassandra/metrics/TableMetrics.java | 93 +++++-- .../apache/cassandra/net/MessagingService.java | 2 + .../apache/cassandra/service/StorageProxy.java | 4 +- .../cassandra/service/StorageService.java | 30 ++- .../cassandra/service/StorageServiceMBean.java | 3 +- .../org/apache/cassandra/tools/NodeProbe.java | 17 +- .../org/apache/cassandra/tools/NodeTool.java | 54 ++-- .../cassandra/tools/nodetool/ProfileLoad.java | 192 ++++++++++++++ .../cassandra/tools/nodetool/TopPartitions.java | 121 +-------- .../org/apache/cassandra/utils/TopKSampler.java | 139 ----------- .../cassandra/metrics/MaxSamplerTest.java | 93 +++++++ .../apache/cassandra/metrics/SamplerTest.java | 247 +++++++++++++++++++ .../metrics/TopFrequencySamplerTest.java | 71 ++++++ .../cassandra/tools/TopPartitionsTest.java | 14 +- .../apache/cassandra/utils/TopKSamplerTest.java | 171 ------------- 23 files changed, 1120 insertions(+), 571 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 6489038..475cd48 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Add sampler for query time and expose with nodetool (CASSANDRA-14436) * Clean up Message.Request implementations (CASSANDRA-14677) * Disable old native protocol versions on demand (CASANDRA-14659) * Allow specifying now-in-seconds in native protocol (CASSANDRA-14664) http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/db/ColumnFamilyStore.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 496fc10..5e38584 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -42,7 +42,6 @@ import com.google.common.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.clearspring.analytics.stream.Counter; import org.apache.cassandra.cache.*; import org.apache.cassandra.concurrent.*; import org.apache.cassandra.config.*; @@ -65,11 +64,9 @@ import org.apache.cassandra.exceptions.StartupException; import org.apache.cassandra.index.SecondaryIndexManager; import org.apache.cassandra.index.internal.CassandraIndex; import org.apache.cassandra.index.transactions.UpdateTransaction; -import org.apache.cassandra.io.FSError; import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.sstable.Component; -import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.KeyIterator; import org.apache.cassandra.io.sstable.SSTableMultiWriter; @@ -77,8 +74,10 @@ import org.apache.cassandra.io.sstable.format.*; import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.io.util.RandomAccessReader; +import org.apache.cassandra.metrics.Sampler; +import org.apache.cassandra.metrics.Sampler.Sample; +import org.apache.cassandra.metrics.Sampler.SamplerType; import org.apache.cassandra.metrics.TableMetrics; -import org.apache.cassandra.metrics.TableMetrics.Sampler; import org.apache.cassandra.repair.TableRepairManager; import org.apache.cassandra.schema.*; import org.apache.cassandra.schema.CompactionParams.TombstoneOption; @@ -87,7 +86,6 @@ import org.apache.cassandra.service.CacheService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.TableStreamManager; import org.apache.cassandra.utils.*; -import org.apache.cassandra.utils.TopKSampler.SamplerResult; import org.apache.cassandra.utils.concurrent.OpOrder; import org.apache.cassandra.utils.concurrent.Refs; import org.apache.cassandra.utils.memory.MemtableAllocator; @@ -145,22 +143,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean new NamedThreadFactory("MemtableReclaimMemory"), "internal"); - private static final String[] COUNTER_NAMES = new String[]{"raw", "count", "error", "string"}; + private static final String[] COUNTER_NAMES = new String[]{"table", "count", "error", "value"}; private static final String[] COUNTER_DESCS = new String[] - { "partition key in raw hex bytes", - "value of this partition for given sampler", - "value is within the error bounds plus or minus of this", - "the partition key turned into a human readable format" }; + { "keyspace.tablename", + "number of occurances", + "error bounds", + "value" }; private static final CompositeType COUNTER_COMPOSITE_TYPE; - private static final TabularType COUNTER_TYPE; - - private static final String[] SAMPLER_NAMES = new String[]{"cardinality", "partitions"}; - private static final String[] SAMPLER_DESCS = new String[] - { "cardinality of partitions", - "list of counter results" }; private static final String SAMPLING_RESULTS_NAME = "SAMPLING_RESULTS"; - private static final CompositeType SAMPLING_RESULT; public static final String SNAPSHOT_TRUNCATE_PREFIX = "truncated"; public static final String SNAPSHOT_DROP_PREFIX = "dropped"; @@ -171,13 +162,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { OpenType<?>[] counterTypes = new OpenType[] { SimpleType.STRING, SimpleType.LONG, SimpleType.LONG, SimpleType.STRING }; COUNTER_COMPOSITE_TYPE = new CompositeType(SAMPLING_RESULTS_NAME, SAMPLING_RESULTS_NAME, COUNTER_NAMES, COUNTER_DESCS, counterTypes); - COUNTER_TYPE = new TabularType(SAMPLING_RESULTS_NAME, SAMPLING_RESULTS_NAME, COUNTER_COMPOSITE_TYPE, COUNTER_NAMES); - - OpenType<?>[] samplerTypes = new OpenType[] { SimpleType.LONG, COUNTER_TYPE }; - SAMPLING_RESULT = new CompositeType(SAMPLING_RESULTS_NAME, SAMPLING_RESULTS_NAME, SAMPLER_NAMES, SAMPLER_DESCS, samplerTypes); } catch (OpenDataException e) { - throw Throwables.propagate(e); + throw new RuntimeException(e); } } @@ -1276,7 +1263,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean long timeDelta = mt.put(update, indexer, opGroup); DecoratedKey key = update.partitionKey(); invalidateCachedPartition(key); - metric.samplers.get(Sampler.WRITES).addSample(key.getKey(), key.hashCode(), 1); + metric.topWritePartitionFrequency.addSample(key.getKey(), 1); + if (metric.topWritePartitionSize.isEnabled()) // dont compute datasize if not needed + metric.topWritePartitionSize.addSample(key.getKey(), update.dataSize()); StorageHook.instance.reportWrite(metadata.id, update); metric.writeLatency.addNano(System.nanoTime() - start); // CASSANDRA-11117 - certain resolution paths on memtable put can result in very @@ -1652,30 +1641,28 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } - - public void beginLocalSampling(String sampler, int capacity) + public void beginLocalSampling(String sampler, int capacity, int durationMillis) { - metric.samplers.get(Sampler.valueOf(sampler)).beginSampling(capacity); + metric.samplers.get(SamplerType.valueOf(sampler)).beginSampling(capacity, durationMillis); } - public CompositeData finishLocalSampling(String sampler, int count) throws OpenDataException + @SuppressWarnings({ "rawtypes", "unchecked" }) + public List<CompositeData> finishLocalSampling(String sampler, int count) throws OpenDataException { - SamplerResult<ByteBuffer> samplerResults = metric.samplers.get(Sampler.valueOf(sampler)) - .finishSampling(count); - TabularDataSupport result = new TabularDataSupport(COUNTER_TYPE); - for (Counter<ByteBuffer> counter : samplerResults.topK) + Sampler samplerImpl = metric.samplers.get(SamplerType.valueOf(sampler)); + List<Sample> samplerResults = samplerImpl.finishSampling(count); + List<CompositeData> result = new ArrayList<>(count); + for (Sample counter : samplerResults) { //Not duplicating the buffer for safety because AbstractSerializer and ByteBufferUtil.bytesToHex //don't modify position or limit - ByteBuffer key = counter.getItem(); - result.put(new CompositeDataSupport(COUNTER_COMPOSITE_TYPE, COUNTER_NAMES, new Object[] { - ByteBufferUtil.bytesToHex(key), // raw - counter.getCount(), // count - counter.getError(), // error - metadata().partitionKeyType.getString(key) })); // string - } - return new CompositeDataSupport(SAMPLING_RESULT, SAMPLER_NAMES, new Object[]{ - samplerResults.cardinality, result}); + result.add(new CompositeDataSupport(COUNTER_COMPOSITE_TYPE, COUNTER_NAMES, new Object[] { + keyspace.getName() + "." + name, + counter.count, + counter.error, + samplerImpl.toString(counter.value) })); // string + } + return result; } public boolean isCompactionDiskSpaceCheckEnabled() http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java index 42c791d..fb0d611 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java @@ -198,12 +198,12 @@ public interface ColumnFamilyStoreMBean * begin sampling for a specific sampler with a given capacity. The cardinality may * be larger than the capacity, but depending on the use case it may affect its accuracy */ - public void beginLocalSampling(String sampler, int capacity); + public void beginLocalSampling(String sampler, int capacity, int durationMillis); /** * @return top <i>count</i> items for the sampler since beginLocalSampling was called */ - public CompositeData finishLocalSampling(String sampler, int count) throws OpenDataException; + public List<CompositeData> finishLocalSampling(String sampler, int count) throws OpenDataException; /* Is Compaction space check enabled http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/db/ReadExecutionController.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ReadExecutionController.java b/src/java/org/apache/cassandra/db/ReadExecutionController.java index 537e3f8..29b6fa7 100644 --- a/src/java/org/apache/cassandra/db/ReadExecutionController.java +++ b/src/java/org/apache/cassandra/db/ReadExecutionController.java @@ -17,12 +17,17 @@ */ package org.apache.cassandra.db; -import org.apache.cassandra.schema.TableMetadata; +import java.util.concurrent.TimeUnit; + import org.apache.cassandra.index.Index; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.concurrent.OpOrder; public class ReadExecutionController implements AutoCloseable { + private static final long NO_SAMPLING = Long.MIN_VALUE; + // For every reads private final OpOrder.Group baseOp; private final TableMetadata baseMetadata; // kept to sanity check that we have take the op order on the right table @@ -30,8 +35,17 @@ public class ReadExecutionController implements AutoCloseable // For index reads private final ReadExecutionController indexController; private final WriteContext writeContext; + private final ReadCommand command; + static Clock clock = Clock.instance; + + private final long createdAtNanos; // Only used while sampling - private ReadExecutionController(OpOrder.Group baseOp, TableMetadata baseMetadata, ReadExecutionController indexController, WriteContext writeContext) + private ReadExecutionController(ReadCommand command, + OpOrder.Group baseOp, + TableMetadata baseMetadata, + ReadExecutionController indexController, + WriteContext writeContext, + long createdAtNanos) { // We can have baseOp == null, but only when empty() is called, in which case the controller will never really be used // (which validForReadOn should ensure). But if it's not null, we should have the proper metadata too. @@ -40,6 +54,8 @@ public class ReadExecutionController implements AutoCloseable this.baseMetadata = baseMetadata; this.indexController = indexController; this.writeContext = writeContext; + this.command = command; + this.createdAtNanos = createdAtNanos; } public ReadExecutionController indexReadController() @@ -52,14 +68,14 @@ public class ReadExecutionController implements AutoCloseable return writeContext; } - public boolean validForReadOn(ColumnFamilyStore cfs) + boolean validForReadOn(ColumnFamilyStore cfs) { return baseOp != null && cfs.metadata.id.equals(baseMetadata.id); } public static ReadExecutionController empty() { - return new ReadExecutionController(null, null, null, null); + return new ReadExecutionController(null, null, null, null, null, NO_SAMPLING); } /** @@ -77,41 +93,42 @@ public class ReadExecutionController implements AutoCloseable ColumnFamilyStore baseCfs = Keyspace.openAndGetStore(command.metadata()); ColumnFamilyStore indexCfs = maybeGetIndexCfs(baseCfs, command); + long createdAtNanos = baseCfs.metric.topLocalReadQueryTime.isEnabled() ? clock.nanoTime() : NO_SAMPLING; + if (indexCfs == null) + return new ReadExecutionController(command, baseCfs.readOrdering.start(), baseCfs.metadata(), null, null, createdAtNanos); + + OpOrder.Group baseOp = null; + WriteContext writeContext = null; + ReadExecutionController indexController = null; + // OpOrder.start() shouldn't fail, but better safe than sorry. + try { - return new ReadExecutionController(baseCfs.readOrdering.start(), baseCfs.metadata(), null, null); + baseOp = baseCfs.readOrdering.start(); + indexController = new ReadExecutionController(command, indexCfs.readOrdering.start(), indexCfs.metadata(), null, null, NO_SAMPLING); + /* + * TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try* + * to delete stale entries, without blocking if there's no room + * as it stands, we open a writeOp and keep it open for the duration to ensure that should this CF get flushed to make room we don't block the reclamation of any room being made + */ + writeContext = baseCfs.keyspace.getWriteHandler().createContextForRead(); + return new ReadExecutionController(command, baseOp, baseCfs.metadata(), indexController, writeContext, createdAtNanos); } - else + catch (RuntimeException e) { - OpOrder.Group baseOp = null; - WriteContext writeContext = null; - ReadExecutionController indexController = null; - // OpOrder.start() shouldn't fail, but better safe than sorry. + // Note that must have writeContext == null since ReadOrderGroup ctor can't fail + assert writeContext == null; try { - baseOp = baseCfs.readOrdering.start(); - indexController = new ReadExecutionController(indexCfs.readOrdering.start(), indexCfs.metadata(), null, null); - // TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try* to delete stale entries, without blocking if there's no room - // as it stands, we open a writeOp and keep it open for the duration to ensure that should this CF get flushed to make room we don't block the reclamation of any room being made - writeContext = baseCfs.keyspace.getWriteHandler().createContextForRead(); - return new ReadExecutionController(baseOp, baseCfs.metadata(), indexController, writeContext); + if (baseOp != null) + baseOp.close(); } - catch (RuntimeException e) + finally { - // Note that must have writeContext == null since ReadOrderGroup ctor can't fail - assert writeContext == null; - try - { - if (baseOp != null) - baseOp.close(); - } - finally - { - if (indexController != null) - indexController.close(); - } - throw e; + if (indexController != null) + indexController.close(); } + throw e; } } @@ -147,5 +164,17 @@ public class ReadExecutionController implements AutoCloseable } } } + + if (createdAtNanos != NO_SAMPLING) + addSample(); + } + + private void addSample() + { + String cql = command.toCQLString(); + int timeMicros = (int) Math.min(TimeUnit.NANOSECONDS.toMicros(clock.nanoTime() - createdAtNanos), Integer.MAX_VALUE); + ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(baseMetadata.id); + if (cfs != null) + cfs.metric.topLocalReadQueryTime.addSample(cql, timeMicros); } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java index ba87996..97ab210 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java @@ -723,7 +723,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar if (!merged.isEmpty()) { DecoratedKey key = merged.partitionKey(); - metrics.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1); + metrics.topReadPartitionFrequency.addSample(key.getKey(), 1); } class UpdateSstablesIterated extends Transformation @@ -847,7 +847,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar return EmptyIterators.unfilteredRow(metadata(), partitionKey(), false); DecoratedKey key = result.partitionKey(); - cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1); + cfs.metric.topReadPartitionFrequency.addSample(key.getKey(), 1); StorageHook.instance.reportRead(cfs.metadata.id, partitionKey()); // "hoist up" the requested data into a more recent sstable http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/metrics/FrequencySampler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/FrequencySampler.java b/src/java/org/apache/cassandra/metrics/FrequencySampler.java new file mode 100644 index 0000000..c094347 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/FrequencySampler.java @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.metrics; + +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.clearspring.analytics.stream.StreamSummary; + +/** + * Find the most frequent sample. A sample adds to the sum of its key ie + * <p>add("x", 10); and add("x", 20); will result in "x" = 30</p> This uses StreamSummary to only store the + * approximate cardinality (capacity) of keys. If the number of distinct keys exceed the capacity, the error of the + * sample may increase depending on distribution of keys among the total set. + * + * @param <T> + */ +public abstract class FrequencySampler<T> extends Sampler<T> +{ + private static final Logger logger = LoggerFactory.getLogger(FrequencySampler.class); + private long endTimeMillis = -1; + + private StreamSummary<T> summary; + + /** + * Start to record samples + * + * @param capacity + * Number of sample items to keep in memory, the lower this is + * the less accurate results are. For best results use value + * close to cardinality, but understand the memory trade offs. + */ + public synchronized void beginSampling(int capacity, int durationMillis) + { + if (endTimeMillis == -1 || clock.currentTimeMillis() > endTimeMillis) + { + summary = new StreamSummary<T>(capacity); + endTimeMillis = clock.currentTimeMillis() + durationMillis; + } + else + throw new RuntimeException("Sampling already in progress"); + } + + /** + * Call to stop collecting samples, and gather the results + * @param count Number of most frequent items to return + */ + public synchronized List<Sample<T>> finishSampling(int count) + { + List<Sample<T>> results = Collections.emptyList(); + if (endTimeMillis != -1) + { + endTimeMillis = -1; + results = summary.topK(count) + .stream() + .map(c -> new Sample<T>(c.getItem(), c.getCount(), c.getError())) + .collect(Collectors.toList()); + } + return results; + } + + protected synchronized void insert(final T item, final long value) + { + // samplerExecutor is single threaded but still need + // synchronization against jmx calls to finishSampling + if (value > 0 && clock.currentTimeMillis() <= endTimeMillis) + { + try + { + summary.offer(item, (int) Math.min(value, Integer.MAX_VALUE)); + } catch (Exception e) + { + logger.trace("Failure to offer sample", e); + } + } + } + + public boolean isEnabled() + { + return endTimeMillis != -1 && clock.currentTimeMillis() <= endTimeMillis; + } + +} + http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/metrics/MaxSampler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/MaxSampler.java b/src/java/org/apache/cassandra/metrics/MaxSampler.java new file mode 100644 index 0000000..f4fb873 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/MaxSampler.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.metrics; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; + +import com.google.common.collect.MinMaxPriorityQueue; + +public abstract class MaxSampler<T> extends Sampler<T> +{ + private int capacity; + private MinMaxPriorityQueue<Sample<T>> queue; + private long endTimeMillis = -1; + private final Comparator<Sample<T>> comp = Collections.reverseOrder(Comparator.comparing(p -> p.count)); + + public boolean isEnabled() + { + return endTimeMillis != -1 && clock.currentTimeMillis() <= endTimeMillis; + } + + public synchronized void beginSampling(int capacity, int durationMillis) + { + if (endTimeMillis == -1 || clock.currentTimeMillis() > endTimeMillis) + { + endTimeMillis = clock.currentTimeMillis() + durationMillis; + queue = MinMaxPriorityQueue + .orderedBy(comp) + .maximumSize(Math.max(1, capacity)) + .create(); + this.capacity = capacity; + } + else + throw new RuntimeException("Sampling already in progress"); + } + + public synchronized List<Sample<T>> finishSampling(int count) + { + List<Sample<T>> result = new ArrayList<>(count); + if (endTimeMillis != -1) + { + endTimeMillis = -1; + Sample<T> next; + while ((next = queue.poll()) != null && result.size() <= count) + result.add(next); + } + return result; + } + + @Override + protected synchronized void insert(T item, long value) + { + if (value > 0 && clock.currentTimeMillis() <= endTimeMillis + && (queue.isEmpty() || queue.size() < capacity || queue.peekLast().count < value)) + queue.add(new Sample<T>(item, value, 0)); + } + +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/metrics/Sampler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/Sampler.java b/src/java/org/apache/cassandra/metrics/Sampler.java new file mode 100644 index 0000000..4bff332 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/Sampler.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.metrics; + +import java.io.Serializable; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; +import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.net.MessagingService.Verb; +import org.apache.cassandra.utils.Clock; + +import com.google.common.annotations.VisibleForTesting; + +public abstract class Sampler<T> +{ + public enum SamplerType + { + READS, WRITES, LOCAL_READ_TIME, WRITE_SIZE, CAS_CONTENTIONS + } + + @VisibleForTesting + Clock clock = Clock.instance; + + @VisibleForTesting + static final ThreadPoolExecutor samplerExecutor = new JMXEnabledThreadPoolExecutor(1, 1, + TimeUnit.SECONDS, + new ArrayBlockingQueue<Runnable>(1000), + new NamedThreadFactory("Sampler"), + "internal"); + + static + { + samplerExecutor.setRejectedExecutionHandler((runnable, executor) -> + { + MessagingService.instance().incrementDroppedMessages(Verb._SAMPLE); + }); + } + + public void addSample(final T item, final int value) + { + if (isEnabled()) + samplerExecutor.submit(() -> insert(item, value)); + } + + protected abstract void insert(T item, long value); + + public abstract boolean isEnabled(); + + public abstract void beginSampling(int capacity, int durationMillis); + + public abstract List<Sample<T>> finishSampling(int count); + + public abstract String toString(T value); + + /** + * Represents the ranked items collected during a sample period + */ + public static class Sample<S> implements Serializable + { + public final S value; + public final long count; + public final long error; + + public Sample(S value, long count, long error) + { + this.value = value; + this.count = count; + this.error = error; + } + + @Override + public String toString() + { + return "Sample [value=" + value + ", count=" + count + ", error=" + error + "]"; + } + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/metrics/TableMetrics.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index c0e7f7b..49603ba 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java @@ -17,20 +17,20 @@ */ package org.apache.cassandra.metrics; +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + import java.nio.ByteBuffer; -import java.util.*; +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; import java.util.function.Predicate; -import com.google.common.collect.Iterables; -import com.google.common.collect.Maps; - -import com.codahale.metrics.*; -import com.codahale.metrics.Timer; -import org.apache.cassandra.schema.Schema; -import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.Memtable; @@ -39,11 +39,21 @@ import org.apache.cassandra.index.SecondaryIndexManager; import org.apache.cassandra.io.compress.CompressionMetadata; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.metadata.MetadataCollector; +import org.apache.cassandra.metrics.Sampler.SamplerType; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.utils.EstimatedHistogram; import org.apache.cassandra.utils.Pair; -import org.apache.cassandra.utils.TopKSampler; -import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.Metric; +import com.codahale.metrics.RatioGauge; +import com.codahale.metrics.Timer; +import com.google.common.collect.Iterables; +import com.google.common.collect.Maps; /** * Metrics for {@link ColumnFamilyStore}. @@ -208,6 +218,17 @@ public class TableMetrics public final static LatencyMetrics globalWriteLatency = new LatencyMetrics(globalFactory, globalAliasFactory, "Write"); public final static LatencyMetrics globalRangeLatency = new LatencyMetrics(globalFactory, globalAliasFactory, "Range"); + /** When sampler activated, will track the most frequently read partitions **/ + public final Sampler<ByteBuffer> topReadPartitionFrequency; + /** When sampler activated, will track the most frequently written to partitions **/ + public final Sampler<ByteBuffer> topWritePartitionFrequency; + /** When sampler activated, will track the largest mutations **/ + public final Sampler<ByteBuffer> topWritePartitionSize; + /** When sampler activated, will track the most frequent partitions with cas contention **/ + public final Sampler<ByteBuffer> topCasPartitionContention; + /** When sampler activated, will track the slowest local reads **/ + public final Sampler<String> topLocalReadQueryTime; + private static Pair<Long, Long> totalNonSystemTablesSize(Predicate<SSTableReader> predicate) { long total = 0; @@ -281,7 +302,7 @@ public class TableMetrics public final Meter readRepairRequests; public final Meter shortReadProtectionRequests; - public final Map<Sampler, TopKSampler<ByteBuffer>> samplers; + public final EnumMap<SamplerType, Sampler<?>> samplers; /** * stores metrics that will be rolled into a single global metric */ @@ -342,11 +363,48 @@ public class TableMetrics factory = new TableMetricNameFactory(cfs, "Table"); aliasFactory = new TableMetricNameFactory(cfs, "ColumnFamily"); - samplers = Maps.newHashMap(); - for (Sampler sampler : Sampler.values()) + samplers = new EnumMap<>(SamplerType.class); + topReadPartitionFrequency = new FrequencySampler<ByteBuffer>() { - samplers.put(sampler, new TopKSampler<>()); - } + public String toString(ByteBuffer value) + { + return cfs.metadata().partitionKeyType.getString(value); + } + }; + topWritePartitionFrequency = new FrequencySampler<ByteBuffer>() + { + public String toString(ByteBuffer value) + { + return cfs.metadata().partitionKeyType.getString(value); + } + }; + topWritePartitionSize = new MaxSampler<ByteBuffer>() + { + public String toString(ByteBuffer value) + { + return cfs.metadata().partitionKeyType.getString(value); + } + }; + topCasPartitionContention = new FrequencySampler<ByteBuffer>() + { + public String toString(ByteBuffer value) + { + return cfs.metadata().partitionKeyType.getString(value); + } + }; + topLocalReadQueryTime = new MaxSampler<String>() + { + public String toString(String value) + { + return value; + } + }; + + samplers.put(SamplerType.READS, topReadPartitionFrequency); + samplers.put(SamplerType.WRITES, topWritePartitionFrequency); + samplers.put(SamplerType.WRITE_SIZE, topWritePartitionSize); + samplers.put(SamplerType.CAS_CONTENTIONS, topCasPartitionContention); + samplers.put(SamplerType.LOCAL_READ_TIME, topLocalReadQueryTime); memtableColumnsCount = createTableGauge("MemtableColumnsCount", new Gauge<Long>() { @@ -1155,9 +1213,4 @@ public class TableMetrics return new CassandraMetricsRegistry.MetricName(groupName, type, metricName, "all", mbeanName.toString()); } } - - public enum Sampler - { - READS, WRITES - } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/net/MessagingService.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 7732673..c8fe3b7 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -273,6 +273,7 @@ public final class MessagingService implements MessagingServiceMBean UNUSED_3, UNUSED_4, UNUSED_5, + _SAMPLE // dummy verb so we can use MS.droppedMessagesMap ; // add new verbs after the existing verbs, since we serialize by ordinal. @@ -480,6 +481,7 @@ public final class MessagingService implements MessagingServiceMBean * drop internal messages like bootstrap or repair notifications. */ public static final EnumSet<Verb> DROPPABLE_VERBS = EnumSet.of(Verb._TRACE, + Verb._SAMPLE, Verb.MUTATION, Verb.COUNTER_MUTATION, Verb.HINT, http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/service/StorageProxy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index a011841..ed0cafc 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -240,14 +240,13 @@ public class StorageProxy implements StorageProxyMBean throws UnavailableException, IsBootstrappingException, RequestFailureException, RequestTimeoutException, InvalidRequestException { final long startTimeForMetrics = System.nanoTime(); + TableMetadata metadata = Schema.instance.getTableMetadata(keyspaceName, cfName); int contentions = 0; try { consistencyForPaxos.validateForCas(); consistencyForCommit.validateForCasCommit(keyspaceName); - TableMetadata metadata = Schema.instance.getTableMetadata(keyspaceName, cfName); - long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getCasContentionTimeout()); while (System.nanoTime() - queryStartNanoTime < timeout) { @@ -334,6 +333,7 @@ public class StorageProxy implements StorageProxyMBean finally { recordCasContention(contentions); + Keyspace.open(keyspaceName).getColumnFamilyStore(cfName).metric.topCasPartitionContention.addSample(key.getKey(), contentions); final long latency = System.nanoTime() - startTimeForMetrics; casWriteMetrics.addNano(latency); writeMetricsMap.get(consistencyForPaxos).addNano(latency); http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/service/StorageService.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 09bed8d..9467c9a 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -5291,33 +5291,39 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } /* - * 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}, ...]}, ...]} - * } + * { "sampler_name": [ {table: "", count: i, error: i, value: ""}, ... ] } */ @Override - public Map<String, Map<String, CompositeData>> samplePartitions(long duration, int capacity, int count, List<String> samplers) throws OpenDataException + public Map<String, List<CompositeData>> samplePartitions(int durationMillis, int capacity, int count, + List<String> samplers) throws OpenDataException { + ConcurrentHashMap<String, List<CompositeData>> result = new ConcurrentHashMap<>(); for (String sampler : samplers) { for (ColumnFamilyStore table : ColumnFamilyStore.all()) { - table.beginLocalSampling(sampler, capacity); + table.beginLocalSampling(sampler, capacity, durationMillis); } } + Uninterruptibles.sleepUninterruptibly(durationMillis, TimeUnit.MILLISECONDS); - Uninterruptibles.sleepUninterruptibly(duration, TimeUnit.MILLISECONDS); - ConcurrentHashMap<String, Map<String, CompositeData>> result = new ConcurrentHashMap<>(); for (String sampler : samplers) { + List<CompositeData> topk = new ArrayList<>(); 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)); + topk.addAll(table.finishLocalSampling(sampler, count)); } + Collections.sort(topk, new Ordering<CompositeData>() + { + public int compare(CompositeData left, CompositeData right) + { + return Long.compare((long) right.get("count"), (long) left.get("count")); + } + }); + // sublist is not serializable for jmx + topk = new ArrayList<>(topk.subList(0, Math.min(topk.size(), count))); + result.put(sampler, topk); } return result; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/service/StorageServiceMBean.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index b642c31..0f4c7dd 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -33,7 +33,6 @@ import javax.management.openmbean.TabularData; import org.apache.cassandra.db.ColumnFamilyStoreMBean; import org.apache.cassandra.exceptions.ConfigurationException; -import org.apache.cassandra.metrics.TableMetrics.Sampler; public interface StorageServiceMBean extends NotificationEmitter { @@ -630,7 +629,7 @@ 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; + public Map<String, List<CompositeData>> samplePartitions(int duration, int capacity, int count, List<String> samplers) throws OpenDataException; /** * Returns the configured tracing probability. http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/tools/NodeProbe.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 9e5be28..03b8af0 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -69,7 +69,6 @@ import org.apache.cassandra.db.HintedHandOffManager; import org.apache.cassandra.locator.DynamicEndpointSnitchMBean; import org.apache.cassandra.locator.EndpointSnitchInfoMBean; import org.apache.cassandra.metrics.CassandraMetricsRegistry; -import org.apache.cassandra.metrics.TableMetrics.Sampler; import org.apache.cassandra.metrics.StorageMetrics; import org.apache.cassandra.metrics.TableMetrics; import org.apache.cassandra.metrics.ThreadPoolMetrics; @@ -421,27 +420,25 @@ public class NodeProbe implements AutoCloseable } } } - public Map<String, Map<String, CompositeData>> getPartitionSample(int capacity, int duration, int count, List<String> samplers) throws OpenDataException + public Map<String, List<CompositeData>> getPartitionSample(int capacity, int durationMillis, int count, List<String> samplers) throws OpenDataException { - return ssProxy.samplePartitions(duration, capacity, count, samplers); + return ssProxy.samplePartitions(durationMillis, capacity, count, samplers); } - public Map<String, Map<String, CompositeData>> getPartitionSample(String ks, String cf, int capacity, int duration, int count, List<String> samplers) throws OpenDataException + public Map<String, List<CompositeData>> getPartitionSample(String ks, String cf, int capacity, int durationMillis, int count, List<String> samplers) throws OpenDataException { ColumnFamilyStoreMBean cfsProxy = getCfsProxy(ks, cf); for(String sampler : samplers) { - cfsProxy.beginLocalSampling(sampler, capacity); + cfsProxy.beginLocalSampling(sampler, capacity, durationMillis); } - Uninterruptibles.sleepUninterruptibly(duration, TimeUnit.MILLISECONDS); - Map<String, CompositeData> result = Maps.newHashMap(); + Uninterruptibles.sleepUninterruptibly(durationMillis, TimeUnit.MILLISECONDS); + Map<String, List<CompositeData>> result = Maps.newHashMap(); for(String sampler : samplers) { result.put(sampler, cfsProxy.finishLocalSampling(sampler, count)); } - return new ImmutableMap.Builder<String, Map<String, CompositeData>>() - .put(ks + "." + cf, result) - .build(); + return result; } public void invalidateCounterCache() http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/tools/NodeTool.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java index f2bbdce..c2193d4 100644 --- a/src/java/org/apache/cassandra/tools/NodeTool.java +++ b/src/java/org/apache/cassandra/tools/NodeTool.java @@ -17,30 +17,53 @@ */ package org.apache.cassandra.tools; -import java.io.*; +import static com.google.common.base.Throwables.getStackTraceAsString; +import static com.google.common.collect.Iterables.toArray; +import static com.google.common.collect.Lists.newArrayList; +import static java.lang.Integer.parseInt; +import static java.lang.String.format; +import static org.apache.commons.lang3.ArrayUtils.EMPTY_STRING_ARRAY; +import static org.apache.commons.lang3.StringUtils.EMPTY; +import static org.apache.commons.lang3.StringUtils.isEmpty; +import static org.apache.commons.lang3.StringUtils.isNotEmpty; + +import java.io.Console; +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileWriter; +import java.io.IOError; +import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.text.SimpleDateFormat; -import java.util.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.List; +import java.util.Map; import java.util.Map.Entry; - -import com.google.common.base.Joiner; -import com.google.common.base.Throwables; -import com.google.common.collect.*; - -import io.airlift.airline.*; +import java.util.Scanner; +import java.util.SortedMap; import org.apache.cassandra.locator.EndpointSnitchInfoMBean; import org.apache.cassandra.tools.nodetool.*; import org.apache.cassandra.utils.FBUtilities; -import static com.google.common.base.Throwables.getStackTraceAsString; -import static com.google.common.collect.Iterables.toArray; -import static com.google.common.collect.Lists.newArrayList; -import static java.lang.Integer.parseInt; -import static java.lang.String.format; -import static org.apache.commons.lang3.ArrayUtils.EMPTY_STRING_ARRAY; -import static org.apache.commons.lang3.StringUtils.*; +import com.google.common.base.Joiner; +import com.google.common.base.Throwables; +import com.google.common.collect.Maps; + +import io.airlift.airline.Cli; +import io.airlift.airline.Help; +import io.airlift.airline.Option; +import io.airlift.airline.OptionType; +import io.airlift.airline.ParseArgumentsMissingException; +import io.airlift.airline.ParseArgumentsUnexpectedException; +import io.airlift.airline.ParseCommandMissingException; +import io.airlift.airline.ParseCommandUnrecognizedException; +import io.airlift.airline.ParseOptionConversionException; +import io.airlift.airline.ParseOptionMissingException; +import io.airlift.airline.ParseOptionMissingValueException; public class NodeTool { @@ -105,6 +128,7 @@ public class NodeTool Move.class, PauseHandoff.class, ResumeHandoff.class, + ProfileLoad.class, ProxyHistograms.class, Rebuild.class, Refresh.class, http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java b/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java new file mode 100644 index 0000000..7037969 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.tools.nodetool; + +import static com.google.common.base.Preconditions.checkArgument; +import static org.apache.commons.lang3.StringUtils.join; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; + +import javax.management.openmbean.CompositeData; +import javax.management.openmbean.OpenDataException; + +import org.apache.cassandra.metrics.Sampler.SamplerType; +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 io.airlift.airline.Arguments; +import io.airlift.airline.Command; +import io.airlift.airline.Option; + +@Command(name = "profileload", description = "Low footprint profiling of activity for a period of time") +public class ProfileLoad extends NodeToolCmd +{ + @Arguments(usage = "<keyspace> <cfname> <duration>", description = "The keyspace, column family name, and duration in milliseconds") + private List<String> args = new ArrayList<>(); + + @Option(name = "-s", description = "Capacity of the sampler, higher for more accuracy (Default: 256)") + private int capacity = 256; + + @Option(name = "-k", description = "Number of the top samples to list (Default: 10)") + private int topCount = 10; + + @Option(name = "-a", description = "Comma separated list of samplers to use (Default: all)") + private String samplers = join(SamplerType.values(), ','); + + @Override + public void execute(NodeProbe probe) + { + checkArgument(args.size() == 3 || args.size() == 1 || args.size() == 0, "Invalid arguments, either [keyspace table duration] or [duration] or no args"); + checkArgument(topCount < capacity, "TopK count (-k) option must be smaller then the summary capacity (-s)"); + String keyspace = null; + String table = null; + Integer durationMillis = 10000; + if(args.size() == 3) + { + keyspace = args.get(0); + table = args.get(1); + durationMillis = Integer.valueOf(args.get(2)); + } + else if (args.size() == 1) + { + durationMillis = Integer.valueOf(args.get(0)); + } + // generate the list of samplers + List<String> targets = Lists.newArrayList(); + List<String> available = Arrays.stream(SamplerType.values()).map(Enum::toString).collect(Collectors.toList()); + for (String s : samplers.split(",")) + { + String sampler = s.trim().toUpperCase(); + checkArgument(available.contains(sampler), String.format("'%s' sampler is not available from: %s", s, Arrays.toString(SamplerType.values()))); + targets.add(sampler); + } + + Map<String, List<CompositeData>> results; + try + { + if (keyspace == null) + results = probe.getPartitionSample(capacity, durationMillis, topCount, targets); + else + results = probe.getPartitionSample(keyspace, table, capacity, durationMillis, topCount, targets); + + } catch (OpenDataException e) + { + throw new RuntimeException(e); + } + + AtomicBoolean first = new AtomicBoolean(true); + ResultBuilder rb = new ResultBuilder(first, results, targets); + + for(String sampler : Lists.newArrayList("READS", "WRITES", "CAS_CONTENTIONS")) + { + rb.forType(SamplerType.valueOf(sampler), "Frequency of " + sampler.toLowerCase().replaceAll("_", " ") + " by partition") + .addColumn("Table", "table") + .addColumn("Partition", "value") + .addColumn("Count", "count") + .addColumn("+/-", "error") + .print(); + } + + rb.forType(SamplerType.WRITE_SIZE, "Max mutation size by partition") + .addColumn("Table", "table") + .addColumn("Partition", "value") + .addColumn("Bytes", "count") + .print(); + + rb.forType(SamplerType.LOCAL_READ_TIME, "Longest read query times") + .addColumn("Query", "value") + .addColumn("Microseconds", "count") + .print(); + } + + private class ResultBuilder + { + private SamplerType type; + private String description; + private AtomicBoolean first; + private Map<String, List<CompositeData>> results; + private List<String> targets; + private List<Pair<String, String>> dataKeys; + + public ResultBuilder(AtomicBoolean first, Map<String, List<CompositeData>> results, List<String> targets) + { + super(); + this.first = first; + this.results = results; + this.targets = targets; + this.dataKeys = new ArrayList<>(); + this.dataKeys.add(Pair.create(" ", " ")); + } + + public ResultBuilder forType(SamplerType type, String description) + { + ResultBuilder rb = new ResultBuilder(first, results, targets); + rb.type = type; + rb.description = description; + return rb; + } + + public ResultBuilder addColumn(String title, String key) + { + this.dataKeys.add(Pair.create(title, key)); + return this; + } + + private String get(CompositeData cd, String key) + { + if (cd.containsKey(key)) + return cd.get(key).toString(); + return key; + } + + public void print() + { + if (targets.contains(type.toString())) + { + if (!first.get()) + System.out.println(); + first.set(false); + System.out.println(description + ':'); + TableBuilder out = new TableBuilder(); + out.add(dataKeys.stream().map(p -> p.left).collect(Collectors.toList()).toArray(new String[] {})); + List<CompositeData> topk = results.get(type.toString()); + for (CompositeData cd : topk) + { + out.add(dataKeys.stream().map(p -> get(cd, p.right)).collect(Collectors.toList()).toArray(new String[] {})); + } + if (topk.size() == 0) + { + System.out.println(" Nothing recorded during sampling period..."); + } + else + { + out.printTo(System.out); + } + } + } + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/tools/nodetool/TopPartitions.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tools/nodetool/TopPartitions.java b/src/java/org/apache/cassandra/tools/nodetool/TopPartitions.java index ee03cd2..66ad574 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/TopPartitions.java +++ b/src/java/org/apache/cassandra/tools/nodetool/TopPartitions.java @@ -17,127 +17,10 @@ */ package org.apache.cassandra.tools.nodetool; -import static com.google.common.base.Preconditions.checkArgument; -import static org.apache.commons.lang3.StringUtils.join; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; - -import javax.management.openmbean.CompositeData; -import javax.management.openmbean.OpenDataException; -import javax.management.openmbean.TabularDataSupport; - -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; - -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 +@Deprecated +public class TopPartitions extends ProfileLoad { - @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; - @Option(name = "-k", description = "Number of the top partitions to list (Default: 10)") - private int topCount = 10; - @Option(name = "-a", description = "Comma separated list of samplers to use (Default: all)") - private String samplers = join(TableMetrics.Sampler.values(), ','); - @Override - public void execute(NodeProbe probe) - { - 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 = 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<String> targets = Lists.newArrayList(); - for (String s : samplers.split(",")) - { - try - { - 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<String, Map<String, CompositeData>> results; - try - { - 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(String sampler : targets) - { - if(!first) - System.out.println(); - 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("\t Nothing recorded during sampling period..."); - } - } - } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/src/java/org/apache/cassandra/utils/TopKSampler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/TopKSampler.java b/src/java/org/apache/cassandra/utils/TopKSampler.java deleted file mode 100644 index 37fcb60..0000000 --- a/src/java/org/apache/cassandra/utils/TopKSampler.java +++ /dev/null @@ -1,139 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.cassandra.utils; - -import java.io.Serializable; -import java.util.*; -import java.util.concurrent.*; - -import org.apache.cassandra.concurrent.*; -import org.slf4j.*; - -import com.clearspring.analytics.stream.*; -import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus; -import com.google.common.annotations.VisibleForTesting; - -public class TopKSampler<T> -{ - private static final Logger logger = LoggerFactory.getLogger(TopKSampler.class); - private volatile boolean enabled = false; - - @VisibleForTesting - static final ThreadPoolExecutor samplerExecutor = new JMXEnabledThreadPoolExecutor(1, 1, - TimeUnit.SECONDS, - new LinkedBlockingQueue<Runnable>(), - new NamedThreadFactory("Sampler"), - "internal"); - - private StreamSummary<T> summary; - @VisibleForTesting - HyperLogLogPlus hll; - - /** - * Start to record samples - * - * @param capacity - * Number of sample items to keep in memory, the lower this is - * the less accurate results are. For best results use value - * close to cardinality, but understand the memory trade offs. - */ - public synchronized void beginSampling(int capacity) - { - if (!enabled) - { - summary = new StreamSummary<T>(capacity); - hll = new HyperLogLogPlus(14); - enabled = true; - } - } - - /** - * Call to stop collecting samples, and gather the results - * @param count Number of most frequent items to return - */ - public synchronized SamplerResult<T> finishSampling(int count) - { - List<Counter<T>> results = Collections.EMPTY_LIST; - long cardinality = 0; - if (enabled) - { - enabled = false; - results = summary.topK(count); - cardinality = hll.cardinality(); - } - return new SamplerResult<T>(results, cardinality); - } - - public void addSample(T item) - { - addSample(item, item.hashCode(), 1); - } - - /** - * Adds a sample to statistics collection. This method is non-blocking and will - * use the "Sampler" thread pool to record results if the sampler is enabled. If not - * sampling this is a NOOP - */ - public void addSample(final T item, final long hash, final int value) - { - if (enabled) - { - final Object lock = this; - samplerExecutor.execute(new Runnable() - { - public void run() - { - // samplerExecutor is single threaded but still need - // synchronization against jmx calls to finishSampling - synchronized (lock) - { - if (enabled) - { - try - { - summary.offer(item, value); - hll.offerHashed(hash); - } catch (Exception e) - { - logger.trace("Failure to offer sample", e); - } - } - } - } - }); - } - } - - /** - * Represents the cardinality and the topK ranked items collected during a - * sample period - */ - public static class SamplerResult<S> implements Serializable - { - public final List<Counter<S>> topK; - public final long cardinality; - - public SamplerResult(List<Counter<S>> topK, long cardinality) - { - this.topK = topK; - this.cardinality = cardinality; - } - } - -} - http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/test/unit/org/apache/cassandra/metrics/MaxSamplerTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/metrics/MaxSamplerTest.java b/test/unit/org/apache/cassandra/metrics/MaxSamplerTest.java new file mode 100644 index 0000000..69278b8 --- /dev/null +++ b/test/unit/org/apache/cassandra/metrics/MaxSamplerTest.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.metrics; + +import java.util.List; +import java.util.concurrent.TimeoutException; + +import org.apache.cassandra.metrics.Sampler.Sample; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class MaxSamplerTest extends SamplerTest +{ + @Before + public void setSampler() + { + this.sampler = new MaxSampler<String>() + { + public String toString(String value) + { + return value; + } + }; + } + + @Test + public void testReturnsMax() throws TimeoutException + { + sampler.beginSampling(5, 100000); + add(); + List<Sample<String>> result = sampler.finishSampling(10); + for (int i = 9995 ; i < 10000 ; i ++) + { + final String key = "test" + i; + Assert.assertTrue(result.stream().anyMatch(s -> s.value.equals(key))); + } + } + + @Test + public void testSizeEqualsCapacity() throws TimeoutException + { + sampler.beginSampling(10, 100000); + add(); + List<Sample<String>> result = sampler.finishSampling(10); + for (int i = 9990 ; i < 10000 ; i ++) + { + final String key = "test" + i; + Assert.assertTrue(result.stream().anyMatch(s -> s.value.equals(key))); + } + } + + @Test + public void testCapacityLarger() throws TimeoutException + { + + sampler.beginSampling(100, 100000); + add(); + List<Sample<String>> result = sampler.finishSampling(10); + for (int i = 9990 ; i < 10000 ; i ++) + { + final String key = "test" + i; + Assert.assertTrue(result.stream().anyMatch(s -> s.value.equals(key))); + } + } + + private void add() throws TimeoutException + { + for (int i = 0 ; i < 10000 ; i ++) + { + // dont load shed test data + if (i % 999 == 0) + waitForEmpty(1000); + sampler.addSample("test"+i, i); + } + waitForEmpty(1000); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/test/unit/org/apache/cassandra/metrics/SamplerTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/metrics/SamplerTest.java b/test/unit/org/apache/cassandra/metrics/SamplerTest.java new file mode 100644 index 0000000..3d24c1b --- /dev/null +++ b/test/unit/org/apache/cassandra/metrics/SamplerTest.java @@ -0,0 +1,247 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.metrics; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.metrics.Sampler.Sample; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.utils.FreeRunningClock; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.google.common.collect.Maps; +import com.google.common.util.concurrent.Uninterruptibles; + + +public class SamplerTest +{ + Sampler<String> sampler; + + + @BeforeClass + public static void initMessagingService() throws ConfigurationException + { + // required so the rejection policy doesnt fail on initializing + // static MessagingService resources + DatabaseDescriptor.daemonInitialization(); + } + + @Test + public void sampleLoadshedding() throws Exception + { + // dont need to run this in children tests + if (sampler != null) return; + AtomicInteger called = new AtomicInteger(); + CountDownLatch latch = new CountDownLatch(1); + Sampler<String> waitSampler = new Sampler<String>() + { + protected void insert(String item, long value) + { + called.incrementAndGet(); + try + { + latch.await(1, TimeUnit.MINUTES); + } + catch (InterruptedException e) + { + e.printStackTrace(); + } + } + + public boolean isEnabled() + { + return true; + } + + public void beginSampling(int capacity, int durationMillis) + { + } + + public List<Sample<String>> finishSampling(int count) + { + return null; + } + + public String toString(String value) + { + return ""; + } + }; + // 1000 queued, 1 in progress, 1 to drop + for (int i = 0; i < 1002; i++) + { + waitSampler.addSample("TEST", 1); + } + latch.countDown(); + waitForEmpty(1000); + Assert.assertEquals(1001, called.get()); + Assert.assertEquals(1, MessagingService.instance().getDroppedMessages().get("_SAMPLE").intValue()); + } + + @Test + public void testSamplerOutOfOrder() throws TimeoutException + { + if(sampler == null) return; + sampler.beginSampling(10, 1000000); + insert(sampler); + waitForEmpty(1000); + List<Sample<String>> single = sampler.finishSampling(10); + single = sampler.finishSampling(10); + Assert.assertEquals(0, single.size()); + } + + @Test(expected=RuntimeException.class) + public void testWhileRunning() + { + if(sampler == null) throw new RuntimeException(); + sampler.clock = new FreeRunningClock(); + try + { + sampler.beginSampling(10, 1000000); + } catch (RuntimeException e) + { + Assert.fail(); // shouldnt fail on first call + } + // should throw Exception + sampler.beginSampling(10, 1000000); + } + + @Test + public void testRepeatStartAfterTimeout() + { + if(sampler == null) return; + FreeRunningClock clock = new FreeRunningClock(); + sampler.clock = clock; + try + { + sampler.beginSampling(10, 10); + } catch (RuntimeException e) + { + Assert.fail(); // shouldnt fail on first call + } + clock.advance(11, TimeUnit.MILLISECONDS); + sampler.beginSampling(10, 1000000); + } + + /** + * checking for exceptions if not thread safe (MinMaxPQ and SS/HL are not) + */ + @Test + public void testMultithreadedAccess() throws Exception + { + if(sampler == null) return; + final AtomicBoolean running = new AtomicBoolean(true); + final CountDownLatch latch = new CountDownLatch(1); + + NamedThreadFactory.createThread(new Runnable() + { + public void run() + { + try + { + while (running.get()) + { + insert(sampler); + } + } finally + { + latch.countDown(); + } + } + + } + , "inserter").start(); + try + { + // start/stop in fast iterations + for(int i = 0; i<100; i++) + { + sampler.beginSampling(i, 100000); + sampler.finishSampling(i); + } + // start/stop with pause to let it build up past capacity + for(int i = 0; i<3; i++) + { + sampler.beginSampling(i, 100000); + Thread.sleep(250); + sampler.finishSampling(i); + } + + // with empty results + running.set(false); + latch.await(1, TimeUnit.SECONDS); + waitForEmpty(1000); + for(int i = 0; i<10; i++) + { + sampler.beginSampling(i, 100000); + Thread.sleep(i); + sampler.finishSampling(i); + } + } finally + { + running.set(false); + } + } + + public void insert(Sampler<String> sampler) + { + for(int i = 1; i <= 10; i++) + { + for(int j = 0; j < i; j++) + { + String key = "item" + i; + sampler.addSample(key, 1); + } + } + } + + public void waitForEmpty(int timeoutMs) throws TimeoutException + { + int timeout = 0; + while (!Sampler.samplerExecutor.getQueue().isEmpty()) + { + timeout++; + Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); + if (timeout * 100 > timeoutMs) + { + throw new TimeoutException("sampler executor not cleared within timeout"); + } + } + } + + public <T> Map<T, Long> countMap(List<Sample<T>> target) + { + Map<T, Long> counts = Maps.newHashMap(); + for(Sample<T> counter : target) + { + counts.put(counter.value, counter.count); + } + return counts; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/test/unit/org/apache/cassandra/metrics/TopFrequencySamplerTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/metrics/TopFrequencySamplerTest.java b/test/unit/org/apache/cassandra/metrics/TopFrequencySamplerTest.java new file mode 100644 index 0000000..ae6e3f9 --- /dev/null +++ b/test/unit/org/apache/cassandra/metrics/TopFrequencySamplerTest.java @@ -0,0 +1,71 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.cassandra.metrics; + +import java.util.List; +import java.util.concurrent.TimeoutException; + +import org.apache.cassandra.metrics.Sampler.Sample; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class TopFrequencySamplerTest extends SamplerTest +{ + @Before + public void setSampler() + { + this.sampler = new FrequencySampler<String>() + { + public String toString(String value) + { + return value; + } + }; + } + + @Test + public void testSamplerSingleInsertionsEqualMulti() throws TimeoutException + { + sampler.beginSampling(10, 100000); + insert(sampler); + waitForEmpty(1000); + List<Sample<String>> single = sampler.finishSampling(10); + + FrequencySampler<String> sampler2 = new FrequencySampler<String>() + { + public String toString(String value) + { + return value; + } + }; + sampler2.beginSampling(10, 100000); + for(int i = 1; i <= 10; i++) + { + String key = "item" + i; + sampler2.addSample(key, i); + } + waitForEmpty(1000); + Assert.assertEquals(countMap(single), countMap(sampler2.finishSampling(10))); + } + + +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f31d1a05/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java b/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java index 64cea3a..d02b4c4 100644 --- a/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java +++ b/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java @@ -35,7 +35,7 @@ public class TopPartitionsTest @Test public void testServiceTopPartitionsNoArg() throws Exception { - BlockingQueue<Map<String, Map<String, CompositeData>>> q = new ArrayBlockingQueue<>(1); + BlockingQueue<Map<String, List<CompositeData>>> q = new ArrayBlockingQueue<>(1); ColumnFamilyStore.all(); Executors.newCachedThreadPool().execute(() -> { @@ -48,20 +48,20 @@ public class TopPartitionsTest e.printStackTrace(); } }); + Thread.sleep(100); 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()); + Map<String, List<CompositeData>> result = q.poll(5, TimeUnit.SECONDS); + List<CompositeData> cd = result.get("WRITES"); assertEquals(1, cd.size()); } @Test public void testServiceTopPartitionsSingleTable() throws Exception { - ColumnFamilyStore.getIfExists("system", "local").beginLocalSampling("READS", 5); + ColumnFamilyStore.getIfExists("system", "local").beginLocalSampling("READS", 5, 100000); 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()); + List<CompositeData> result = ColumnFamilyStore.getIfExists("system", "local").finishLocalSampling("READS", 5); + assertEquals(1, result.size()); } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
