mirror of https://github.com/apache/cassandra
Add ability to log load profiles at fixed intervals
Patch by Yifan Cai; reviewed by Josh McKenzie, Dinesh Joshi, and Chris Lohfink for CASSANDRA-17821 Co-authored-by: Yifan Cai <ycai@apache.org> Co-authored-by: Josh McKenzie <jmckenzie@apache.org>
This commit is contained in:
Josh McKenzie
parent
fd7b1bf81e
commit
4526b3fcbd
|
|
@ -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)
|
||||
|
|
|
|||
4
NEWS.txt
4
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.
|
||||
|
|
|
|||
|
|
@ -33,33 +33,31 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
|
|||
* <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.
|
||||
*
|
||||
* 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 <T>
|
||||
*/
|
||||
public abstract class FrequencySampler<T> extends Sampler<T>
|
||||
{
|
||||
private static final Logger logger = LoggerFactory.getLogger(FrequencySampler.class);
|
||||
private long endTimeNanos = -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.
|
||||
* @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<T> extends Sampler<T>
|
|||
public synchronized List<Sample<T>> finishSampling(int count)
|
||||
{
|
||||
List<Sample<T>> results = Collections.emptyList();
|
||||
if (endTimeNanos != -1)
|
||||
if (isEnabled())
|
||||
{
|
||||
endTimeNanos = -1;
|
||||
disable();
|
||||
results = summary.topK(count)
|
||||
.stream()
|
||||
.map(c -> new Sample<T>(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<T> extends Sampler<T>
|
|||
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<T> extends Sampler<T>
|
||||
{
|
||||
private int capacity;
|
||||
private MinMaxPriorityQueue<Sample<T>> queue;
|
||||
private long endTimeNanos = -1;
|
||||
private final Comparator<Sample<T>> 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<Sample<T>> finishSampling(int count)
|
||||
{
|
||||
List<Sample<T>> result = new ArrayList<>(count);
|
||||
if (endTimeNanos != -1)
|
||||
if (isEnabled())
|
||||
{
|
||||
endTimeNanos = -1;
|
||||
disable();
|
||||
Sample<T> next;
|
||||
while ((next = queue.poll()) != null && result.size() <= count)
|
||||
result.add(next);
|
||||
|
|
@ -69,9 +65,12 @@ public abstract class MaxSampler<T> extends Sampler<T>
|
|||
@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<T>(item, value, 0));
|
||||
}
|
||||
|
||||
private boolean permitsValue(long value)
|
||||
{
|
||||
return value > 0 && (queue.isEmpty() || queue.size() < capacity || queue.peekLast().count < value);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<T>
|
||||
{
|
||||
private static long DISABLED = -1L;
|
||||
|
||||
private static final BiFunction<SamplerType, SamplingManager.ResultBuilder, SamplingManager.ResultBuilder>
|
||||
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<SamplerType, SamplingManager.ResultBuilder, SamplingManager.ResultBuilder> formatter;
|
||||
|
||||
SamplerType(String description, BiFunction<SamplerType, SamplingManager.ResultBuilder, SamplingManager.ResultBuilder> 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<T>
|
|||
.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<T>
|
|||
|
||||
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<Sample<T>> finishSampling(int count);
|
||||
|
||||
public abstract String toString(T value);
|
||||
|
|
|
|||
|
|
@ -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<JobId, Future<?>> activeSamplingTasks = new ConcurrentHashMap<>();
|
||||
|
||||
/** Tasks that are actively being cancelled */
|
||||
private final Set<JobId> 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<ColumnFamilyStore> 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<String> 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<String> allJobs()
|
||||
{
|
||||
return jobIds().stream()
|
||||
.map(JobId::toString)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
private Set<JobId> jobIds()
|
||||
{
|
||||
Set<JobId> 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<JobId> 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<ColumnFamilyStore> tables, int duration, int interval, int capacity, int count, List<String> 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<String> 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<ColumnFamilyStore> tables, int duration, int interval, int capacity, int count, List<String> samplers)
|
||||
{
|
||||
return () ->
|
||||
{
|
||||
Map<String, List<CompositeData>> results = new HashMap<>();
|
||||
for (String sampler : samplers)
|
||||
{
|
||||
List<CompositeData> 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<String, List<CompositeData>> results;
|
||||
protected List<String> targets;
|
||||
protected List<Pair<String, String>> dataKeys;
|
||||
|
||||
public ResultBuilder(AtomicBoolean first, Map<String, List<CompositeData>> results, List<String> 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<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)
|
||||
{
|
||||
ps.println(" Nothing recorded during sampling period...");
|
||||
}
|
||||
else
|
||||
{
|
||||
out.printTo(ps);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String, List<CompositeData>> samplePartitions(int duration, int capacity, int count, List<String> samplers) throws OpenDataException {
|
||||
return samplePartitions(null, duration, capacity, count, samplers);
|
||||
}
|
||||
|
||||
/*
|
||||
* { "sampler_name": [ {table: "", count: i, error: i, value: ""}, ... ] }
|
||||
*/
|
||||
@Override
|
||||
public Map<String, List<CompositeData>> samplePartitions(int durationMillis, int capacity, int count,
|
||||
List<String> samplers) throws OpenDataException
|
||||
public Map<String, List<CompositeData>> samplePartitions(String keyspace, int durationMillis, int capacity, int count,
|
||||
List<String> samplers) throws OpenDataException
|
||||
{
|
||||
ConcurrentHashMap<String, List<CompositeData>> result = new ConcurrentHashMap<>();
|
||||
Iterable<ColumnFamilyStore> 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<CompositeData> 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<String> 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<Sampler.SamplerType> 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<String> getSampleTasks()
|
||||
{
|
||||
return samplingManager.allJobs();
|
||||
}
|
||||
|
||||
public void rebuildSecondaryIndex(String ksName, String cfName, String... idxNames)
|
||||
{
|
||||
String[] indices = asList(idxNames).stream()
|
||||
|
|
|
|||
|
|
@ -795,6 +795,36 @@ public interface StorageServiceMBean extends NotificationEmitter
|
|||
|
||||
public Map<String, List<CompositeData>> samplePartitions(int duration, int capacity, int count, List<String> samplers) throws OpenDataException;
|
||||
|
||||
public Map<String, List<CompositeData>> samplePartitions(String keyspace, int duration, int capacity, int count, List<String> 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<String> 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<String> getSampleTasks();
|
||||
|
||||
/**
|
||||
* Returns the configured tracing probability.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -498,9 +498,29 @@ public class NodeProbe implements AutoCloseable
|
|||
}
|
||||
}
|
||||
}
|
||||
public Map<String, List<CompositeData>> getPartitionSample(int capacity, int durationMillis, int count, List<String> samplers) throws OpenDataException
|
||||
|
||||
public boolean handleScheduledSampling(String ks,
|
||||
String table,
|
||||
int capacity,
|
||||
int count,
|
||||
int durationMillis,
|
||||
int intervalMillis,
|
||||
List<String> 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<String> getSampleTasks()
|
||||
{
|
||||
return ssProxy.getSampleTasks();
|
||||
}
|
||||
|
||||
public Map<String, List<CompositeData>> getPartitionSample(String ks, int capacity, int durationMillis, int count, List<String> samplers) throws OpenDataException
|
||||
{
|
||||
return ssProxy.samplePartitions(ks, durationMillis, capacity, count, samplers);
|
||||
}
|
||||
|
||||
public Map<String, List<CompositeData>> getPartitionSample(String ks, String cf, int capacity, int durationMillis, int count, List<String> samplers) throws OpenDataException
|
||||
|
|
|
|||
|
|
@ -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 = "<keyspace> <cfname> <duration>", description = "The keyspace, column family name, and duration in milliseconds")
|
||||
@Arguments(usage = "<keyspace> <cfname> <duration>", description = "The keyspace, column family name, and duration in milliseconds (Default: 10000)")
|
||||
private List<String> 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 <keyspace> and <cfname>. 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<String> targets = Lists.newArrayList();
|
||||
List<String> available = Arrays.stream(SamplerType.values()).map(Enum::toString).collect(Collectors.toList());
|
||||
Set<String> 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<String, List<CompositeData>> 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<Pair<String, String>> 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<String, List<CompositeData>> results;
|
||||
private List<String> targets;
|
||||
private List<Pair<String, String>> dataKeys;
|
||||
return intervalMillis != -1;
|
||||
}
|
||||
|
||||
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(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<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)
|
||||
{
|
||||
outStream.println(" Nothing recorded during sampling period...");
|
||||
}
|
||||
else
|
||||
{
|
||||
out.printTo(outStream);
|
||||
}
|
||||
}
|
||||
}
|
||||
private String nullifyWildcard(String input)
|
||||
{
|
||||
return input != null && input.equals("*") ? null : input;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String> 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<String> 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());
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
{
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<CompositeData> 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<String> 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));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue