diff --git a/CHANGES.txt b/CHANGES.txt index e2e4d7ddde..255b46d7aa 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.2 + * Add ability to log load profiles at fixed intervals (CASSANDRA-17821) * Protect against Gossip backing up due to a quarantined endpoint without version information (CASSANDRA-17830) * NPE in org.apache.cassandra.cql3.Attributes.getTimeToLive (CASSANDRA-17822) * Add guardrail for column size (CASSANDRA-17151) diff --git a/NEWS.txt b/NEWS.txt index 780cd7b88f..d8e9318a41 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -75,10 +75,12 @@ New features - Whether DROP KEYSPACE commands are allowed. - Column value size - It is possible to list ephemeral snapshots by nodetool listsnaphots command when flag "-e" is specified. + - Added a new flag to `nodetool profileload` and JMX endpoint to set up recurring profile load generation on specified + intervals (see CASSANDRA-17821) Upgrading --------- - - Emphemeral marker files for snapshots done by repairs are not created anymore, + - Ephemeral marker files for snapshots done by repairs are not created anymore, there is a dedicated flag in snapshot manifest instead. On upgrade of a node to version 4.2, on node's start, in case there are such ephemeral snapshots on disk, they will be deleted (same behaviour as before) and any new ephemeral snapshots will stop to create ephemeral marker files as flag in a snapshot manifest was introduced instead. diff --git a/src/java/org/apache/cassandra/metrics/FrequencySampler.java b/src/java/org/apache/cassandra/metrics/FrequencySampler.java index 8a8918b9fa..d4dfe860a4 100644 --- a/src/java/org/apache/cassandra/metrics/FrequencySampler.java +++ b/src/java/org/apache/cassandra/metrics/FrequencySampler.java @@ -33,33 +33,31 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; *

add("x", 10); and add("x", 20); will result in "x" = 30

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. + * + * Note: {@link Sampler#samplerExecutor} is single threaded but we still need to synchronize as we have access + * from both internal and the external JMX context that can cause races. * * @param */ public abstract class FrequencySampler extends Sampler { private static final Logger logger = LoggerFactory.getLogger(FrequencySampler.class); - private long endTimeNanos = -1; private StreamSummary 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. + * @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) + public synchronized void beginSampling(int capacity, long durationMillis) { - if (endTimeNanos == -1 || clock.now() > endTimeNanos) - { - summary = new StreamSummary<>(capacity); - endTimeNanos = clock.now() + MILLISECONDS.toNanos(durationMillis); - } - else + if (isActive()) throw new RuntimeException("Sampling already in progress"); + updateEndTime(clock.now() + MILLISECONDS.toNanos(durationMillis)); + summary = new StreamSummary<>(capacity); } /** @@ -69,12 +67,12 @@ public abstract class FrequencySampler extends Sampler public synchronized List> finishSampling(int count) { List> results = Collections.emptyList(); - if (endTimeNanos != -1) + if (isEnabled()) { - endTimeNanos = -1; + disable(); results = summary.topK(count) .stream() - .map(c -> new Sample(c.getItem(), c.getCount(), c.getError())) + .map(c -> new Sample<>(c.getItem(), c.getCount(), c.getError())) .collect(Collectors.toList()); } return results; @@ -82,24 +80,16 @@ public abstract class FrequencySampler extends Sampler 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.now() <= endTimeNanos) + if (value > 0 && isActive()) { try { summary.offer(item, (int) Math.min(value, Integer.MAX_VALUE)); - } catch (Exception e) + } + catch (Exception e) { logger.trace("Failure to offer sample", e); } } } - - public boolean isEnabled() - { - return endTimeNanos != -1 && clock.now() <= endTimeNanos; - } - } - diff --git a/src/java/org/apache/cassandra/metrics/MaxSampler.java b/src/java/org/apache/cassandra/metrics/MaxSampler.java index df24bb9629..0593e341d6 100644 --- a/src/java/org/apache/cassandra/metrics/MaxSampler.java +++ b/src/java/org/apache/cassandra/metrics/MaxSampler.java @@ -26,39 +26,35 @@ import com.google.common.collect.MinMaxPriorityQueue; import static java.util.concurrent.TimeUnit.MILLISECONDS; +/** + * Note: {@link Sampler#samplerExecutor} is single threaded but we still need to synchronize as we have access + * from both internal and the external JMX context that can cause races. + */ public abstract class MaxSampler extends Sampler { private int capacity; private MinMaxPriorityQueue> queue; - private long endTimeNanos = -1; private final Comparator> comp = Collections.reverseOrder(Comparator.comparing(p -> p.count)); - public boolean isEnabled() + @Override + public synchronized void beginSampling(int capacity, long durationMillis) { - return endTimeNanos != -1 && clock.now() <= endTimeNanos; - } - - public synchronized void beginSampling(int capacity, int durationMillis) - { - if (endTimeNanos == -1 || clock.now() > endTimeNanos) - { - endTimeNanos = clock.now() + MILLISECONDS.toNanos(durationMillis); - queue = MinMaxPriorityQueue - .orderedBy(comp) - .maximumSize(Math.max(1, capacity)) - .create(); - this.capacity = capacity; - } - else + if (isActive()) throw new RuntimeException("Sampling already in progress"); + updateEndTime(clock.now() + MILLISECONDS.toNanos(durationMillis)); + queue = MinMaxPriorityQueue.orderedBy(comp) + .maximumSize(Math.max(1, capacity)) + .create(); + this.capacity = capacity; } + @Override public synchronized List> finishSampling(int count) { List> result = new ArrayList<>(count); - if (endTimeNanos != -1) + if (isEnabled()) { - endTimeNanos = -1; + disable(); Sample next; while ((next = queue.poll()) != null && result.size() <= count) result.add(next); @@ -69,9 +65,12 @@ public abstract class MaxSampler extends Sampler @Override protected synchronized void insert(T item, long value) { - if (value > 0 && clock.now() <= endTimeNanos - && (queue.isEmpty() || queue.size() < capacity || queue.peekLast().count < value)) + if (isActive() && permitsValue(value)) queue.add(new Sample(item, value, 0)); } + private boolean permitsValue(long value) + { + return value > 0 && (queue.isEmpty() || queue.size() < capacity || queue.peekLast().count < value); + } } diff --git a/src/java/org/apache/cassandra/metrics/Sampler.java b/src/java/org/apache/cassandra/metrics/Sampler.java index b3d0f21aa5..de7f0b2e1a 100644 --- a/src/java/org/apache/cassandra/metrics/Sampler.java +++ b/src/java/org/apache/cassandra/metrics/Sampler.java @@ -17,10 +17,12 @@ */ package org.apache.cassandra.metrics; +import java.io.PrintStream; import java.io.Serializable; import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.BiFunction; import org.apache.cassandra.concurrent.ExecutorPlus; import org.apache.cassandra.net.MessagingService; @@ -34,9 +36,44 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac public abstract class Sampler { + private static long DISABLED = -1L; + + private static final BiFunction + FrequencySamplerFomatter = (type, resultBuilder) -> + resultBuilder.forType(type, type.description) + .addColumn("Table", "table") + .addColumn("Partition", "value") + .addColumn("Count", "count") + .addColumn("+/-", "error"); + public enum SamplerType { - READS, WRITES, LOCAL_READ_TIME, WRITE_SIZE, CAS_CONTENTIONS + READS("Frequency of reads by partition", FrequencySamplerFomatter), + WRITES("Frequency of writes by partition", FrequencySamplerFomatter), + LOCAL_READ_TIME("Longest read query times", ((samplerType, resultBuilder) -> + resultBuilder.forType(samplerType, samplerType.description) + .addColumn("Query", "value") + .addColumn("Microseconds", "count"))), + WRITE_SIZE("Max mutation size by partition", ((samplerType, resultBuilder) -> + resultBuilder.forType(samplerType, samplerType.description) + .addColumn("Table", "table") + .addColumn("Partition", "value") + .addColumn("Bytes", "count"))), + CAS_CONTENTIONS("Frequency of CAS contention by partition", FrequencySamplerFomatter); + + private final String description; + private final BiFunction formatter; + + SamplerType(String description, BiFunction formatter) + { + this.description = description; + this.formatter = formatter; + } + + void format(SamplingManager.ResultBuilder resultBuilder, PrintStream ps) + { + formatter.apply(this, resultBuilder).print(ps); + } } @VisibleForTesting @@ -50,6 +87,8 @@ public abstract class Sampler .withRejectedExecutionHandler((runnable, executor) -> MessagingService.instance().metrics.recordSelfDroppedMessage(Verb._SAMPLE)) .build(); + private long endTimeNanos = -1; + public void addSample(final T item, final int value) { if (isEnabled()) @@ -58,10 +97,53 @@ public abstract class Sampler protected abstract void insert(T item, long value); - public abstract boolean isEnabled(); + /** + * A sampler is enabled between {@link this#beginSampling} and {@link this#finishSampling} + * @return true if the sampler is enabled. + */ + public boolean isEnabled() + { + return endTimeNanos != DISABLED; + } - public abstract void beginSampling(int capacity, int durationMillis); + public void disable() + { + endTimeNanos = DISABLED; + } + /** + * @return true if the sampler is active. + * A sampler is active only if it is enabled and the current time is within the `durationMillis` when beginning sampling. + */ + public boolean isActive() + { + return isEnabled() && clock.now() <= endTimeNanos; + } + + /** + * Update the end time for the sampler. Implicitly, calling this method enables the sampler. + */ + public void updateEndTime(long endTimeMillis) + { + this.endTimeNanos = endTimeMillis; + } + + /** + * Begin sampling with the configured capacity and duration + * @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. + * @param durationMillis Upperbound duration in milliseconds for sampling. The sampler + * stops accepting new samples after exceeding the duration + * even if {@link #finishSampling(int)}} is not called. + */ + public abstract void beginSampling(int capacity, long durationMillis); + + /** + * Stop sampling and return the results + * @param count The number of the samples requested to retrieve from the sampler + * @return a list of samples, the size is the minimum of the total samples and {@param count}. + */ public abstract List> finishSampling(int count); public abstract String toString(T value); diff --git a/src/java/org/apache/cassandra/metrics/SamplingManager.java b/src/java/org/apache/cassandra/metrics/SamplingManager.java new file mode 100644 index 0000000000..37d8d355c2 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/SamplingManager.java @@ -0,0 +1,391 @@ +/* + * 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.ByteArrayOutputStream; +import java.io.PrintStream; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; +import javax.management.openmbean.CompositeData; +import javax.management.openmbean.OpenDataException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.concurrent.ScheduledExecutors; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.tools.nodetool.ProfileLoad; +import org.apache.cassandra.tools.nodetool.formatter.TableBuilder; +import org.apache.cassandra.utils.Pair; + +public class SamplingManager +{ + private static final Logger logger = LoggerFactory.getLogger(SamplingManager.class); + + /** + * Tracks the active scheduled sampling tasks. + * The key of the map is a {@link JobId}, which is effectively a keyspace + table abstracted behind some syntactic + * sugar so we can use them without peppering Pairs throughout this class. Both keyspace and table are nullable, + * a paradigm we inherit from {@link ProfileLoad} so need to accommodate here. + * + * The value of the map is the current scheduled task. + */ + private final ConcurrentHashMap> activeSamplingTasks = new ConcurrentHashMap<>(); + + /** Tasks that are actively being cancelled */ + private final Set cancelingTasks = ConcurrentHashMap.newKeySet(); + + public static String formatResult(ResultBuilder resultBuilder) + { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (PrintStream ps = new PrintStream(baos)) + { + for (Sampler.SamplerType samplerType : Sampler.SamplerType.values()) + { + samplerType.format(resultBuilder, ps); + } + return baos.toString(); + } + } + + public static Iterable getTables(String ks, String table) + { + // null KEYSPACE == all the tables + if (ks == null) + return ColumnFamilyStore.all(); + + Keyspace keyspace = Keyspace.open(ks); + + // KEYSPACE defined w/null table == all the tables on that KEYSPACE + if (table == null) + return keyspace.getColumnFamilyStores(); + // Or we just have a specific ks+table combo we're looking to profile + else + return Collections.singletonList(keyspace.getColumnFamilyStore(table)); + } + + /** + * Register the samplers for the keyspace and table. + * @param ks Keyspace. Nullable. If null, the scheduled sampling is on all keyspaces and tables + * @param table Nullable. If null, the scheduled sampling is on all tables of the specified keyspace + * @param duration Duration of each scheduled sampling job in milliseconds + * @param interval Interval of each scheduled sampling job in milliseconds + * @param capacity Capacity of the sampler, higher for more accuracy + * @param count Number of the top samples to list + * @param samplers a list of samplers to enable + * @return true if the scheduled sampling is started successfully. Otherwise return fasle + */ + public boolean register(String ks, String table, int duration, int interval, int capacity, int count, List samplers) + { + JobId jobId = new JobId(ks, table); + logger.info("Registering samplers {} for {}", samplers, jobId); + + if (!canSchedule(jobId)) + { + logger.info("Unable to register {} due to existing ongoing sampling.", jobId); + return false; + } + + // 'begin' tasks are chained to finish before their paired 'finish' + activeSamplingTasks.put(jobId, ScheduledExecutors.optionalTasks.submit( + createSamplingBeginRunnable(jobId, getTables(ks, table), duration, interval, capacity, count, samplers) + )); + return true; + } + + public boolean unregister(String ks, String table) + { + // unregister all + // return true when all tasks are cancelled successfully + if (ks == null && table == null) + { + boolean res = true; + for (JobId id : activeSamplingTasks.keySet()) + { + res = cancelTask(id) & res; + } + return res; + } + else + { + return cancelTask(new JobId(ks, table)); + } + } + + public List allJobs() + { + return jobIds().stream() + .map(JobId::toString) + .collect(Collectors.toList()); + } + + private Set jobIds() + { + Set all = new HashSet<>(); + all.addAll(activeSamplingTasks.keySet()); + all.addAll(cancelingTasks); + return all; + } + + /** + * Validate if a schedule on the keyspace and table is permitted + * @param jobId + * @return true if possible, false if there are overlapping tables already being sampled + */ + private boolean canSchedule(JobId jobId) + { + Set allJobIds = jobIds(); + // There is a schedule that works on all tables. Overlapping guaranteed. + if (allJobIds.contains(JobId.ALL_KS_AND_TABLES) || (!allJobIds.isEmpty() && jobId.equals(JobId.ALL_KS_AND_TABLES))) + return false; + // there is an exactly duplicated schedule + else if (allJobIds.contains(jobId)) + return false; + else + // make sure has no overlapping tables under the keyspace + return !allJobIds.contains(JobId.createForAllTables(jobId.keyspace)); + } + + /** + * Cancel a task by its id. The corresponding task will be stopped once its final sampling completes. + * @param jobId + * @return true if the task exists, false if not found + */ + private boolean cancelTask(JobId jobId) + { + Future task = activeSamplingTasks.remove(jobId); + if (task != null) + cancelingTasks.add(jobId); + return task != null; + } + + /** + * Begin sampling and schedule a future task to end the sampling task + */ + private Runnable createSamplingBeginRunnable(JobId jobId, Iterable tables, int duration, int interval, int capacity, int count, List samplers) + { + return () -> + { + if (cancelingTasks.contains(jobId)) + { + logger.debug("The sampling job of {} is currently canceling. Not issuing a new run.", jobId); + activeSamplingTasks.remove(jobId); + return; + } + List tableNames = StreamSupport.stream(tables.spliterator(), false) + .map(cfs -> String.format("%s.%s", cfs.keyspace, cfs.name)) + .collect(Collectors.toList()); + logger.info("Starting to sample tables {} with the samplers {} for {} ms", tableNames, samplers, duration); + for (String sampler : samplers) + { + for (ColumnFamilyStore cfs : tables) + { + cfs.beginLocalSampling(sampler, capacity, duration); + } + } + Future fut = ScheduledExecutors.optionalTasks.schedule( + createSamplingEndRunnable(jobId, tables, duration, interval, capacity, count, samplers), + interval, + TimeUnit.MILLISECONDS); + // reached to the end of the current runnable + // update the referenced future to SamplingFinish + activeSamplingTasks.put(jobId, fut); + }; + } + + /** + * Finish the sampling and begin a new one immediately after. + * + * NOTE: Do not call this outside the context of {@link this#createSamplingBeginRunnable}, as we need to preserve + * ordering between a "start" and "end" runnable + */ + private Runnable createSamplingEndRunnable(JobId jobId, Iterable tables, int duration, int interval, int capacity, int count, List samplers) + { + return () -> + { + Map> results = new HashMap<>(); + for (String sampler : samplers) + { + List topk = new ArrayList<>(); + for (ColumnFamilyStore cfs : tables) + { + try + { + topk.addAll(cfs.finishLocalSampling(sampler, count)); + } + catch (OpenDataException e) + { + logger.warn("Failed to retrieve the sampled data. Abort the background sampling job: {}.", jobId, e); + activeSamplingTasks.remove(jobId); + cancelingTasks.remove(jobId); + return; + } + } + + topk.sort((left, right) -> 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))); + results.put(sampler, topk); + } + AtomicBoolean first = new AtomicBoolean(false); + ResultBuilder rb = new ResultBuilder(first, results, samplers); + logger.info(formatResult(rb)); + + // If nobody has canceled us, we ping-pong back to a "begin" runnable to run another profile load + if (!cancelingTasks.contains(jobId)) + { + Future fut = ScheduledExecutors.optionalTasks.submit( + createSamplingBeginRunnable(jobId, tables, duration, interval, capacity, count, samplers)); + activeSamplingTasks.put(jobId, fut); + } + // If someone *has* canceled us, we need to remove the runnable from activeSampling and also remove the + // cancellation sentinel so subsequent re-submits of profiling don't get blocked immediately + else + { + logger.info("The sampling job {} has been cancelled.", jobId); + activeSamplingTasks.remove(jobId); + cancelingTasks.remove(jobId); + } + }; + } + + private static class JobId + { + public static final JobId ALL_KS_AND_TABLES = new JobId(null, null); + + public final String keyspace; + public final String table; + + public JobId(String ks, String tb) + { + keyspace = ks; + table = tb; + } + + public static JobId createForAllTables(String keyspace) + { + return new JobId(keyspace, null); + } + + @Override + public String toString() + { + return maybeWildCard(keyspace) + '.' + maybeWildCard(table); + } + + private String maybeWildCard(String input) + { + return input == null ? "*" : input; + } + + @Override + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + JobId jobId = (JobId) o; + return Objects.equals(keyspace, jobId.keyspace) && Objects.equals(table, jobId.table); + } + + @Override + public int hashCode() + { + return Objects.hash(keyspace, table); + } + } + + public static class ResultBuilder + { + protected Sampler.SamplerType type; + protected String description; + protected AtomicBoolean first; + protected Map> results; + protected List targets; + protected List> dataKeys; + + public ResultBuilder(AtomicBoolean first, Map> results, List targets) + { + this.first = first; + this.results = results; + this.targets = targets; + this.dataKeys = new ArrayList<>(); + this.dataKeys.add(Pair.create(" ", " ")); + } + + public SamplingManager.ResultBuilder forType(Sampler.SamplerType type, String description) + { + SamplingManager.ResultBuilder rb = new SamplingManager.ResultBuilder(first, results, targets); + rb.type = type; + rb.description = description; + return rb; + } + + public SamplingManager.ResultBuilder addColumn(String title, String key) + { + this.dataKeys.add(Pair.create(title, key)); + return this; + } + + protected String get(CompositeData cd, String key) + { + if (cd.containsKey(key)) + return cd.get(key).toString(); + return key; + } + + public void print(PrintStream ps) + { + if (targets.contains(type.toString())) + { + if (!first.get()) + ps.println(); + first.set(false); + ps.println(description + ':'); + TableBuilder out = new TableBuilder(); + out.add(dataKeys.stream().map(p -> p.left).collect(Collectors.toList()).toArray(new String[] {})); + List 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) + { + ps.println(" Nothing recorded during sampling period..."); + } + else + { + out.printTo(ps); + } + } + } + } +} diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index b7756c0c21..80ffa80a8c 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -69,6 +69,8 @@ import org.apache.cassandra.fql.FullQueryLoggerOptions; import org.apache.cassandra.fql.FullQueryLoggerOptionsCompositeData; import org.apache.cassandra.io.util.File; import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict; +import org.apache.cassandra.metrics.Sampler; +import org.apache.cassandra.metrics.SamplingManager; import org.apache.cassandra.schema.Keyspaces; import org.apache.cassandra.service.disk.usage.DiskUsageBroadcaster; import org.apache.cassandra.service.snapshot.SnapshotLoader; @@ -142,6 +144,7 @@ import org.apache.cassandra.utils.progress.ProgressEventType; import org.apache.cassandra.utils.progress.jmx.JMXBroadcastExecutor; import org.apache.cassandra.utils.progress.jmx.JMXProgressSupport; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.collect.Iterables.transform; import static com.google.common.collect.Iterables.tryFind; import static java.util.Arrays.asList; @@ -225,6 +228,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public static final StorageService instance = new StorageService(); + private final SamplingManager samplingManager = new SamplingManager(); + @Deprecated public boolean isInShutdownHook() { @@ -5978,17 +5983,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return sampledKeys; } + @Override + public Map> samplePartitions(int duration, int capacity, int count, List samplers) throws OpenDataException { + return samplePartitions(null, duration, capacity, count, samplers); + } + /* * { "sampler_name": [ {table: "", count: i, error: i, value: ""}, ... ] } */ @Override - public Map> samplePartitions(int durationMillis, int capacity, int count, - List samplers) throws OpenDataException + public Map> samplePartitions(String keyspace, int durationMillis, int capacity, int count, + List samplers) throws OpenDataException { ConcurrentHashMap> result = new ConcurrentHashMap<>(); + Iterable tables = SamplingManager.getTables(keyspace, null); for (String sampler : samplers) { - for (ColumnFamilyStore table : ColumnFamilyStore.all()) + for (ColumnFamilyStore table : tables) { table.beginLocalSampling(sampler, capacity, durationMillis); } @@ -5998,7 +6009,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE for (String sampler : samplers) { List topk = new ArrayList<>(); - for (ColumnFamilyStore table : ColumnFamilyStore.all()) + for (ColumnFamilyStore table : tables) { topk.addAll(table.finishLocalSampling(sampler, count)); } @@ -6016,6 +6027,44 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return result; } + @Override // Note from parent javadoc: ks and table are nullable + public boolean startSamplingPartitions(String ks, String table, int duration, int interval, int capacity, int count, List samplers) + { + Preconditions.checkArgument(duration > 0, "Sampling duration %s must be positive.", duration); + + Preconditions.checkArgument(interval <= 0 || interval >= duration, + "Sampling interval %s should be greater then or equals to duration %s if defined.", + interval, duration); + + Preconditions.checkArgument(capacity > 0 && capacity <= 1024, + "Sampling capacity %s must be positive and the max value is 1024 (inclusive).", + capacity); + + Preconditions.checkArgument(count > 0 && count < capacity, + "Sampling count %s must be positive and smaller than capacity %s.", + count, capacity); + + Preconditions.checkArgument(!samplers.isEmpty(), "Samplers cannot be empty."); + + Set available = EnumSet.allOf(Sampler.SamplerType.class); + samplers.forEach((x) -> checkArgument(available.contains(Sampler.SamplerType.valueOf(x)), + "'%s' sampler is not available from: %s", + x, Arrays.toString(Sampler.SamplerType.values()))); + return samplingManager.register(ks, table, duration, interval, capacity, count, samplers); + } + + @Override + public boolean stopSamplingPartitions(String ks, String table) + { + return samplingManager.unregister(ks, table); + } + + @Override + public List getSampleTasks() + { + return samplingManager.allJobs(); + } + public void rebuildSecondaryIndex(String ksName, String cfName, String... idxNames) { String[] indices = asList(idxNames).stream() diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index ac2ff68411..02485274a3 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -795,6 +795,36 @@ public interface StorageServiceMBean extends NotificationEmitter public Map> samplePartitions(int duration, int capacity, int count, List samplers) throws OpenDataException; + public Map> samplePartitions(String keyspace, int duration, int capacity, int count, List samplers) throws OpenDataException; + + /** + * Start a scheduled sampling + * @param ks Keyspace. Nullable. If null, the scheduled sampling is on all keyspaces and tables + * @param table Nullable. If null, the scheduled sampling is on all tables of the specified keyspace + * @param duration Duration of each scheduled sampling job in milliseconds + * @param interval Interval of each scheduled sampling job in milliseconds + * @param capacity Capacity of the sampler, higher for more accuracy + * @param count Number of the top samples to list + * @param samplers a list of samplers to enable + * @return true if the scheduled sampling is started successfully. Otherwise return false + */ + public boolean startSamplingPartitions(String ks, String table, int duration, int interval, int capacity, int count, List samplers) throws OpenDataException; + + /** + * Stop a scheduled sampling + * @param ks Keyspace. Nullable. If null, the scheduled sampling is on all keysapces and tables + * @param table Nullable. If null, the scheduled sampling is on all tables of the specified keyspace + * @return true if the scheduled sampling is stopped. False is returned if the sampling task is not found + */ + public boolean stopSamplingPartitions(String ks, String table) throws OpenDataException; + + /** + * @return a list of qualified table names that have active scheduled sampling tasks. The format of the name is `KEYSPACE.TABLE` + * The wild card symbol (*) indicates all keyspace/table. For example, "*.*" indicates all tables in all keyspaces. "foo.*" indicates + * all tables under keyspace 'foo'. "foo.bar" indicates the scheduled sampling is enabled for the table 'bar' + */ + public List getSampleTasks(); + /** * Returns the configured tracing probability. */ diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index d8dc2ab8e2..daf4eb2518 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -498,9 +498,29 @@ public class NodeProbe implements AutoCloseable } } } - public Map> getPartitionSample(int capacity, int durationMillis, int count, List samplers) throws OpenDataException + + public boolean handleScheduledSampling(String ks, + String table, + int capacity, + int count, + int durationMillis, + int intervalMillis, + List samplers, + boolean shouldStop) throws OpenDataException { - return ssProxy.samplePartitions(durationMillis, capacity, count, samplers); + return shouldStop ? + ssProxy.stopSamplingPartitions(ks, table) : + ssProxy.startSamplingPartitions(ks, table, durationMillis, intervalMillis, capacity, count, samplers); + } + + public List getSampleTasks() + { + return ssProxy.getSampleTasks(); + } + + public Map> getPartitionSample(String ks, int capacity, int durationMillis, int count, List samplers) throws OpenDataException + { + return ssProxy.samplePartitions(ks, durationMillis, capacity, count, samplers); } public Map> getPartitionSample(String ks, String cf, int capacity, int durationMillis, int count, List samplers) throws OpenDataException diff --git a/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java b/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java index 487f14a0a0..45cade7560 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java +++ b/src/java/org/apache/cassandra/tools/nodetool/ProfileLoad.java @@ -17,36 +17,36 @@ */ package org.apache.cassandra.tools.nodetool; -import static com.google.common.base.Preconditions.checkArgument; -import static org.apache.commons.lang3.StringUtils.join; - import java.io.PrintStream; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.Set; 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; +import org.apache.cassandra.metrics.Sampler.SamplerType; +import org.apache.cassandra.metrics.SamplingManager; +import org.apache.cassandra.tools.NodeProbe; +import org.apache.cassandra.tools.NodeTool.NodeToolCmd; +import org.apache.cassandra.utils.Pair; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkState; +import static org.apache.commons.lang3.StringUtils.join; @Command(name = "profileload", description = "Low footprint profiling of activity for a period of time") public class ProfileLoad extends NodeToolCmd { - @Arguments(usage = " ", description = "The keyspace, column family name, and duration in milliseconds") + @Arguments(usage = " ", description = "The keyspace, column family name, and duration in milliseconds (Default: 10000)") private List args = new ArrayList<>(); @Option(name = "-s", description = "Capacity of the sampler, higher for more accuracy (Default: 256)") @@ -58,27 +58,59 @@ public class ProfileLoad extends NodeToolCmd @Option(name = "-a", description = "Comma separated list of samplers to use (Default: all)") private String samplers = join(SamplerType.values(), ','); + @Option(name = {"-i", "--interval"}, description = "Schedule a new job that samples every interval milliseconds (Default: disabled) in the background") + private int intervalMillis = -1; // -1 for disabled. + + @Option(name = {"-t", "--stop"}, description = "Stop the scheduled sampling job identified by and . Jobs are stopped until the last schedules complete.") + private boolean shouldStop = false; + + @Option(name = {"-l", "--list"}, description = "List the scheduled sampling jobs") + private boolean shouldList = false; + @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)"); + checkArgument(args.size() == 3 || args.size() == 2 || args.size() == 1 || args.size() == 0, + "Invalid arguments, either [keyspace table/* duration] or [keyspace table/*] or [duration] or no args.\n" + + "Optionally, use * to represent all tables under the keyspace."); + checkArgument(topCount > 0, "TopK count (-k) option must have positive value"); + checkArgument(topCount < capacity, + "TopK count (-k) option must be smaller then the summary capacity (-s)"); + checkArgument(capacity <= 1024, "Capacity (-s) cannot exceed 1024."); String keyspace = null; String table = null; - Integer durationMillis = 10000; - if(args.size() == 3) + int durationMillis = 10000; + /* There are 3 possible outcomes after processing the args. + * - keyspace == null && table == null. We need to sample all tables + * - keyspace == KEYSPACE && table == *. We need to sample all tables under the specified KEYSPACE + * - keyspace = KEYSPACE && table == TABLE. Sample the specific KEYSPACE.table combination + */ + if (args.size() == 3) + { + keyspace = args.get(0); + table = args.get(1); + durationMillis = Integer.parseInt(args.get(2)); + } + else if (args.size() == 2) { 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)); + durationMillis = Integer.parseInt(args.get(0)); } + keyspace = nullifyWildcard(keyspace); + table = nullifyWildcard(table); + + checkArgument(durationMillis > 0, "Duration: %s must be positive", durationMillis); + + checkArgument(!hasInterval() || intervalMillis >= durationMillis, + "Invalid scheduled sampling interval. Expecting interval >= duration, but interval: %s ms; duration: %s ms", + intervalMillis, durationMillis); // generate the list of samplers List targets = Lists.newArrayList(); - List available = Arrays.stream(SamplerType.values()).map(Enum::toString).collect(Collectors.toList()); + Set available = Arrays.stream(SamplerType.values()).map(Enum::toString).collect(Collectors.toSet()); for (String s : samplers.split(",")) { String sampler = s.trim().toUpperCase(); @@ -86,108 +118,70 @@ public class ProfileLoad extends NodeToolCmd targets.add(sampler); } + PrintStream out = probe.output().out; + Map> results; try { - if (keyspace == null) - results = probe.getPartitionSample(capacity, durationMillis, topCount, targets); + // handle scheduled samplings, i.e. start or stop + if (hasInterval() || shouldStop) + { + // keyspace and table are nullable + boolean opSuccess = probe.handleScheduledSampling(keyspace, table, capacity, topCount, durationMillis, intervalMillis, targets, shouldStop); + if (!opSuccess) + { + if (shouldStop) + out.printf("Unable to stop the non-existent scheduled sampling for keyspace: %s, table: %s%n", keyspace, table); + else + out.printf("Unable to schedule sampling for keyspace: %s, table: %s due to existing samplings. " + + "Stop the existing sampling jobs first.%n", keyspace, table); + } + return; + } + else if (shouldList) + { + List> sampleTasks = new ArrayList<>(); + int maxKsLength = "KEYSPACE".length(); + int maxTblLength = "TABLE".length(); + for (String fullTableName : probe.getSampleTasks()) + { + String[] parts = fullTableName.split("\\."); + checkState(parts.length == 2, "Unable to parse the full table name: %s", fullTableName); + sampleTasks.add(Pair.create(parts[0], parts[1])); + maxKsLength = Math.max(maxKsLength, parts[0].length()); + } + // print the header line and put enough space between KEYSPACE AND TABLE. + String lineFormat = "%" + maxKsLength + "s %" + maxTblLength + "s%n"; + out.printf(lineFormat, "KEYSPACE", "TABLE"); + sampleTasks.forEach(pair -> out.printf(lineFormat, pair.left, pair.right)); + return; + } else - results = probe.getPartitionSample(keyspace, table, capacity, durationMillis, topCount, targets); - - } catch (OpenDataException e) + { + // blocking sample all the tables or all the tables under a keyspace + if (keyspace == null || table == null) + results = probe.getPartitionSample(keyspace, capacity, durationMillis, topCount, targets); + else // blocking sample the specific table + 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(probe.output().out); - } - - rb.forType(SamplerType.WRITE_SIZE, "Max mutation size by partition") - .addColumn("Table", "table") - .addColumn("Partition", "value") - .addColumn("Bytes", "count") - .print(probe.output().out); - - rb.forType(SamplerType.LOCAL_READ_TIME, "Longest read query times") - .addColumn("Query", "value") - .addColumn("Microseconds", "count") - .print(probe.output().out); + SamplingManager.ResultBuilder rb = new SamplingManager.ResultBuilder(first, results, targets); + out.println(SamplingManager.formatResult(rb)); } - private class ResultBuilder + private boolean hasInterval() { - private SamplerType type; - private String description; - private AtomicBoolean first; - private Map> results; - private List targets; - private List> dataKeys; + return intervalMillis != -1; + } - public ResultBuilder(AtomicBoolean first, Map> results, List 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(PrintStream outStream) - { - if (targets.contains(type.toString())) - { - if (!first.get()) - outStream.println(); - first.set(false); - outStream.println(description + ':'); - TableBuilder out = new TableBuilder(); - out.add(dataKeys.stream().map(p -> p.left).collect(Collectors.toList()).toArray(new String[] {})); - List 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) - { - outStream.println(" Nothing recorded during sampling period..."); - } - else - { - out.printTo(outStream); - } - } - } + private String nullifyWildcard(String input) + { + return input != null && input.equals("*") ? null : input; } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/ProfileLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/ProfileLoadTest.java new file mode 100644 index 0000000000..a83ea7cedc --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ProfileLoadTest.java @@ -0,0 +1,151 @@ +/* + * 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.distributed.test; + +import com.google.common.util.concurrent.Uninterruptibles; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +public class ProfileLoadTest extends TestBaseImpl +{ + @Test + public void testScheduledSamplingTaskLogs() throws IOException + { + try (Cluster cluster = init(Cluster.build(1).start())) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + // start the scheduled profileload task that samples for 1 second and every second. + cluster.get(1).nodetoolResult("profileload", "1000", "-i", "1000").asserts().success(); + + Random rnd = new Random(); + // 800 * 2ms = 1.6 seconds. It logs every second. So it logs at least once. + for (int i = 0; i < 800; i++) + { + cluster.coordinator(1) + .execute(withKeyspace("INSERT INTO %s.tbl (pk, ck, v) VALUES (?,?,?)"), + ConsistencyLevel.QUORUM, rnd.nextInt(), rnd.nextInt(), i); + Uninterruptibles.sleepUninterruptibly(2, TimeUnit.MILLISECONDS); + } + // --list should display all active tasks. + String expectedOutput = String.format("KEYSPACE TABLE%n" + "%8s %5s", "*", "*"); + cluster.get(1).nodetoolResult("profileload", "--list") + .asserts() + .success() + .stdoutContains(expectedOutput); + + // loop assert the log contains the frequency readout; give this 15 seconds which should be plenty of time + // even on very badly underprovisioned environments + int timeout = 15; + boolean testPassed = false; + while (timeout-- > 0) + { + List freqHeadings = cluster.get(1) + .logs() + .grep("Frequency of (reads|writes|cas contentions) by partition") + .getResult(); + if (freqHeadings.size() > 3) + { + testPassed = true; + break; + } + Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); + } + Assert.assertTrue("The scheduled task should at least run and log once", testPassed); + + List startSamplingLogs = cluster.get(1) + .logs() + .grep("Starting to sample tables") + .getResult(); + Assert.assertTrue("It should start sampling at least once", startSamplingLogs.size() > 0); + + // stop the scheduled sampling + cluster.get(1).nodetoolResult("profileload", "--stop").asserts().success(); + + // wait for the last schedule to be stopped. --list should list nothing after stopping + assertListEmpty(cluster.get(1)); + + // schedule on the specific table + cluster.get(1).nodetoolResult("profileload", KEYSPACE, "tbl", "1000", "-i", "1000").asserts().success(); + expectedOutput = String.format("%" + KEYSPACE.length() + "s %5s%n" + + "%s %5s", + "KEYSPACE", "TABLE", + KEYSPACE, "tbl"); + cluster.get(1).nodetoolResult("profileload", "--list") + .asserts() + .success() + .stdoutContains(expectedOutput); + // stop all should stop the task scheduled with the specific table + cluster.get(1).nodetoolResult("profileload", "--stop") + .asserts().success(); + assertListEmpty(cluster.get(1)); + } + } + + @Test + public void testPreventDuplicatedSchedule() throws IOException + { + try (Cluster cluster = init(Cluster.build(1).start())) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + // New sampling; we are good + cluster.get(1).nodetoolResult("profileload", KEYSPACE, "tbl", "1000", "-i", "1000") + .asserts() + .success() + .stdoutNotContains("Unable to schedule sampling for keyspace"); + + // Duplicated sampling (against the same table) but different interval. Nodetool should reject + cluster.get(1).nodetoolResult("profileload", KEYSPACE, "tbl", "1000", "-i", "1000") + .asserts() + .success() + .stdoutContains("Unable to schedule sampling for keyspace"); + + // The "sampling all" request creates overlaps, so it should be rejected too + cluster.get(1).nodetoolResult("profileload", "1000", "-i", "1000") + .asserts() + .success() + .stdoutContains("Unable to schedule sampling for keyspace"); + + cluster.get(1).nodetoolResult("profileload", KEYSPACE, "tbl", "--stop").asserts().success(); + assertListEmpty(cluster.get(1)); + + cluster.get(1).nodetoolResult("profileload", "nonexistks", "nonexisttbl", "--stop") + .asserts() + .success() + .stdoutContains("Unable to stop the non-existent scheduled sampling"); + } + } + + private void assertListEmpty(IInvokableInstance instance) + { + Uninterruptibles.sleepUninterruptibly(1500, TimeUnit.MILLISECONDS); + Assert.assertEquals("--list should list nothing", + "KEYSPACE TABLE\n", + instance.nodetoolResult("profileload", "--list").getStdout()); + } +} diff --git a/test/unit/org/apache/cassandra/metrics/SamplerTest.java b/test/unit/org/apache/cassandra/metrics/SamplerTest.java index dba19a3064..862abbb05b 100644 --- a/test/unit/org/apache/cassandra/metrics/SamplerTest.java +++ b/test/unit/org/apache/cassandra/metrics/SamplerTest.java @@ -79,7 +79,12 @@ public class SamplerTest return true; } - public void beginSampling(int capacity, int durationMillis) + public boolean isActive() + { + return true; + } + + public void beginSampling(int capacity, long durationMillis) { } diff --git a/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java b/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java index 975ad15776..934d4d814c 100644 --- a/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java +++ b/test/unit/org/apache/cassandra/tools/TopPartitionsTest.java @@ -18,16 +18,20 @@ package org.apache.cassandra.tools; +import java.util.Arrays; +import java.util.Collections; 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 java.util.stream.Collectors; import javax.management.openmbean.CompositeData; import com.google.common.collect.Lists; +import com.google.common.util.concurrent.Uninterruptibles; import org.junit.BeforeClass; import org.junit.Test; @@ -36,12 +40,18 @@ 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.metrics.Sampler; import org.apache.cassandra.service.StorageService; import static java.lang.String.format; import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +/** + * Includes test cases for both the 'toppartitions' command and its successor 'profileload' + */ public class TopPartitionsTest { @BeforeClass @@ -59,7 +69,7 @@ public class TopPartitionsTest { try { - q.put(StorageService.instance.samplePartitions(1000, 100, 10, Lists.newArrayList("READS", "WRITES"))); + q.put(StorageService.instance.samplePartitions(null, 1000, 100, 10, Lists.newArrayList("READS", "WRITES"))); } catch (Exception e) { @@ -82,4 +92,30 @@ public class TopPartitionsTest List result = ColumnFamilyStore.getIfExists("system", "local").finishLocalSampling("READS", 5); assertEquals("If this failed you probably have to raise the beginLocalSampling duration", 1, result.size()); } + + @Test + public void testStartAndStopScheduledSampling() + { + List allSamplers = Arrays.stream(Sampler.SamplerType.values()).map(Enum::toString).collect(Collectors.toList()); + StorageService ss = StorageService.instance; + + assertTrue("Scheduling new sampled tasks should be allowed", + ss.startSamplingPartitions(null, null, 10, 10, 100, 10, allSamplers)); + + assertEquals(Collections.singletonList("*.*"), ss.getSampleTasks()); + + assertFalse("Sampling with duplicate keys should be disallowed", + ss.startSamplingPartitions(null, null, 20, 20, 100, 10, allSamplers)); + + assertTrue("Existing scheduled sampling tasks should be cancellable", ss.stopSamplingPartitions(null, null)); + + int timeout = 10; + while (timeout-- > 0 && ss.getSampleTasks().size() > 0) + Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS); + + assertEquals("Scheduled sampled tasks should be removed", Collections.emptyList(), ss.getSampleTasks()); + + assertTrue("When nothing is scheduled, you should be able to stop all scheduled sampling tasks", + ss.stopSamplingPartitions(null, null)); + } }