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

When a table attempts to clean up metrics, it was cleaning up all global table metrics 

patch by David Capwell; reviewed by Jon Meredith, Jordan West, Yifan Cai for CASSANDRA-16095
This commit is contained in:
David Capwell 2020-12-02 13:35:56 -08:00
parent 7bdecd2176
commit 876ac8c611
8 changed files with 463 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -15,6 +15,7 @@
* Throw BufferOverflowException from DataOutputBuffer for better visibility (CASSANDRA-16214)
* TLS connections to the storage port on a node without server encryption configured causes java.io.IOException accessing missing keystore (CASSANDRA-16144)
* Internode messaging catches OOMs and does not rethrow (CASSANDRA-15214)
* When a table attempts to clean up metrics, it was cleaning up all global table metrics (CASSANDRA-16095)
Merged from 3.11:
* SASI's `max_compaction_flush_memory_in_mb` settings over 100GB revert to default of 1GB (CASSANDRA-16071)
Merged from 3.0:

71
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -163,7 +163,10 @@ public enum CassandraRelevantProperties
ORG_APACHE_CASSANDRA_DB_VIRTUAL_SYSTEM_PROPERTIES_TABLE_TEST("org.apache.cassandra.db.virtual.SystemPropertiesTableTest"),
/** This property indicates whether disable_mbean_registration is true */
IS_DISABLED_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration");
IS_DISABLED_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration"),
/** what class to use for mbean registeration */
MBEAN_REGISTRATION_CLASS("org.apache.cassandra.mbean_registration_class");
CassandraRelevantProperties(String key, String defaultVal)
{
@ -196,6 +199,28 @@ public enum CassandraRelevantProperties
return value == null ? defaultVal : STRING_CONVERTER.convert(value);
}
/**
* Gets the value of a system property as a String.
* @return system property String value if it exists, overrideDefaultValue otherwise.
*/
public String getString(String overrideDefaultValue)
{
String value = System.getProperty(key);
if (value == null)
return overrideDefaultValue;
return STRING_CONVERTER.convert(value);
}
/**
* Sets the value into system properties.
* @param value to set
*/
public void setString(String value)
{
System.setProperty(key, value);
}
/**
* Gets the value of a system property as a boolean.
* @return system property boolean value if it exists, false otherwise().
@ -207,6 +232,28 @@ public enum CassandraRelevantProperties
return BOOLEAN_CONVERTER.convert(value == null ? defaultVal : value);
}
/**
* Gets the value of a system property as a boolean.
* @return system property boolean value if it exists, overrideDefaultValue otherwise.
*/
public boolean getBoolean(boolean overrideDefaultValue)
{
String value = System.getProperty(key);
if (value == null)
return overrideDefaultValue;
return BOOLEAN_CONVERTER.convert(value);
}
/**
* Sets the value into system properties.
* @param value to set
*/
public void setBoolean(boolean value)
{
System.setProperty(key, Boolean.toString(value));
}
/**
* Gets the value of a system property as a int.
* @return system property int value if it exists, defaultValue otherwise.
@ -218,6 +265,28 @@ public enum CassandraRelevantProperties
return INTEGER_CONVERTER.convert(value == null ? defaultVal : value);
}
/**
* Gets the value of a system property as a int.
* @return system property int value if it exists, overrideDefaultValue otherwise.
*/
public int getInt(int overrideDefaultValue)
{
String value = System.getProperty(key);
if (value == null)
return overrideDefaultValue;
return INTEGER_CONVERTER.convert(value);
}
/**
* Sets the value into system properties.
* @param value to set
*/
public void setInt(int value)
{
System.setProperty(key, Integer.toString(value));
}
private interface PropertyConverter<T>
{
T convert(String value);

22
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -439,12 +439,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (registerBookeeping)
{
// register the mbean
mbeanName = String.format("org.apache.cassandra.db:type=%s,keyspace=%s,table=%s",
isIndex() ? "IndexTables" : "Tables",
keyspace.getName(), name);
oldMBeanName = String.format("org.apache.cassandra.db:type=%s,keyspace=%s,columnfamily=%s",
isIndex() ? "IndexColumnFamilies" : "ColumnFamilies",
keyspace.getName(), name);
mbeanName = getTableMBeanName(keyspace.getName(), name, isIndex());
oldMBeanName = getColumnFamilieMBeanName(keyspace.getName(), name, isIndex());
String[] objectNames = {mbeanName, oldMBeanName};
for (String objectName : objectNames)
@ -461,6 +457,20 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
sstableImporter = new SSTableImporter(this);
}
public static String getTableMBeanName(String ks, String name, boolean isIndex)
{
return String.format("org.apache.cassandra.db:type=%s,keyspace=%s,table=%s",
isIndex ? "IndexTables" : "Tables",
ks, name);
}
public static String getColumnFamilieMBeanName(String ks, String name, boolean isIndex)
{
return String.format("org.apache.cassandra.db:type=%s,keyspace=%s,columnfamily=%s",
isIndex ? "IndexColumnFamilies" : "ColumnFamilies",
ks, name);
}
public void updateSpeculationThreshold()
{
try

78
src/java/org/apache/cassandra/metrics/TableMetrics.java
View File

@ -62,6 +62,18 @@ import com.codahale.metrics.RatioGauge;
*/
public class TableMetrics
{
/**
* stores metrics that will be rolled into a single global metric
*/
private static final ConcurrentMap<String, Set<Metric>> ALL_TABLE_METRICS = Maps.newConcurrentMap();
public static final long[] EMPTY = new long[0];
private static final MetricNameFactory GLOBAL_FACTORY = new AllTableMetricNameFactory("Table");
private static final MetricNameFactory GLOBAL_ALIAS_FACTORY = new AllTableMetricNameFactory("ColumnFamily");
public final static LatencyMetrics GLOBAL_READ_LATENCY = new LatencyMetrics(GLOBAL_FACTORY, GLOBAL_ALIAS_FACTORY, "Read");
public final static LatencyMetrics GLOBAL_WRITE_LATENCY = new LatencyMetrics(GLOBAL_FACTORY, GLOBAL_ALIAS_FACTORY, "Write");
public final static LatencyMetrics GLOBAL_RANGE_LATENCY = new LatencyMetrics(GLOBAL_FACTORY, GLOBAL_ALIAS_FACTORY, "Range");
/** Total amount of data stored in the memtable that resides on-heap, including column related overhead and partitions overwritten. */
public final Gauge<Long> memtableOnHeapDataSize;
/** Total amount of data stored in the memtable that resides off-heap, including column related overhead and partitions overwritten. */
@ -205,8 +217,6 @@ public class TableMetrics
private final MetricNameFactory factory;
private final MetricNameFactory aliasFactory;
private static final MetricNameFactory globalFactory = new AllTableMetricNameFactory("Table");
private static final MetricNameFactory globalAliasFactory = new AllTableMetricNameFactory("ColumnFamily");
public final Counter speculativeRetries;
public final Counter speculativeFailedRetries;
@ -238,10 +248,6 @@ public class TableMetrics
public final TableHistogram repairedDataTrackingOverreadRows;
public final TableTimer repairedDataTrackingOverreadTime;
public final static LatencyMetrics globalReadLatency = new LatencyMetrics(globalFactory, globalAliasFactory, "Read");
public final static LatencyMetrics globalWriteLatency = new LatencyMetrics(globalFactory, globalAliasFactory, "Write");
public final static LatencyMetrics globalRangeLatency = new LatencyMetrics(globalFactory, globalAliasFactory, "Range");
/** When sampler activated, will track the most frequently read partitions **/
public final Sampler<ByteBuffer> topReadPartitionFrequency;
/** When sampler activated, will track the most frequently written to partitions **/
@ -284,7 +290,7 @@ public class TableMetrics
return Pair.create(filtered, total);
}
public static final Gauge<Double> globalPercentRepaired = Metrics.register(globalFactory.createMetricName("PercentRepaired"),
public static final Gauge<Double> globalPercentRepaired = Metrics.register(GLOBAL_FACTORY.createMetricName("PercentRepaired"),
new Gauge<Double>()
{
public Double getValue()
@ -296,15 +302,15 @@ public class TableMetrics
}
});
public static final Gauge<Long> globalBytesRepaired = Metrics.register(globalFactory.createMetricName("BytesRepaired"),
public static final Gauge<Long> globalBytesRepaired = Metrics.register(GLOBAL_FACTORY.createMetricName("BytesRepaired"),
() -> totalNonSystemTablesSize(SSTableReader::isRepaired).left);
public static final Gauge<Long> globalBytesUnrepaired =
Metrics.register(globalFactory.createMetricName("BytesUnrepaired"),
Metrics.register(GLOBAL_FACTORY.createMetricName("BytesUnrepaired"),
() -> totalNonSystemTablesSize(s -> !s.isRepaired() && !s.isPendingRepair()).left);
public static final Gauge<Long> globalBytesPendingRepair =
Metrics.register(globalFactory.createMetricName("BytesPendingRepair"),
Metrics.register(GLOBAL_FACTORY.createMetricName("BytesPendingRepair"),
() -> totalNonSystemTablesSize(SSTableReader::isPendingRepair).left);
public final Meter readRepairRequests;
@ -321,15 +327,11 @@ public class TableMetrics
public final Histogram rfpRowsCachedPerQuery;
public final EnumMap<SamplerType, Sampler<?>> samplers;
/**
* stores metrics that will be rolled into a single global metric
*/
public final static ConcurrentMap<String, Set<Metric>> allTableMetrics = Maps.newConcurrentMap();
/**
* Stores all metrics created that can be used when unregistering
*/
public final static Set<ReleasableMetric> all = Sets.newHashSet();
private final Set<ReleasableMetric> all = Sets.newHashSet();
private interface GetHistogram
{
@ -566,9 +568,9 @@ public class TableMetrics
}
});
readLatency = createLatencyMetrics("Read", cfs.keyspace.metric.readLatency, globalReadLatency);
writeLatency = createLatencyMetrics("Write", cfs.keyspace.metric.writeLatency, globalWriteLatency);
rangeLatency = createLatencyMetrics("Range", cfs.keyspace.metric.rangeLatency, globalRangeLatency);
readLatency = createLatencyMetrics("Read", cfs.keyspace.metric.readLatency, GLOBAL_READ_LATENCY);
writeLatency = createLatencyMetrics("Write", cfs.keyspace.metric.writeLatency, GLOBAL_WRITE_LATENCY);
rangeLatency = createLatencyMetrics("Range", cfs.keyspace.metric.rangeLatency, GLOBAL_RANGE_LATENCY);
pendingFlushes = createTableCounter("PendingFlushes");
bytesFlushed = createTableCounter("BytesFlushed");
@ -605,7 +607,7 @@ public class TableMetrics
public Long getValue()
{
long min = Long.MAX_VALUE;
for (Metric cfGauge : allTableMetrics.get("MinPartitionSize"))
for (Metric cfGauge : ALL_TABLE_METRICS.get("MinPartitionSize"))
{
min = Math.min(min, ((Gauge<? extends Number>) cfGauge).getValue().longValue());
}
@ -629,7 +631,7 @@ public class TableMetrics
public Long getValue()
{
long max = 0;
for (Metric cfGauge : allTableMetrics.get("MaxPartitionSize"))
for (Metric cfGauge : ALL_TABLE_METRICS.get("MaxPartitionSize"))
{
max = Math.max(max, ((Gauge<? extends Number>) cfGauge).getValue().longValue());
}
@ -897,7 +899,7 @@ public class TableMetrics
unleveledSSTables = createTableGauge("UnleveledSSTables", cfs::getUnleveledSSTables, () -> {
// global gauge
int cnt = 0;
for (Metric cfGauge : allTableMetrics.get("UnleveledSSTables"))
for (Metric cfGauge : ALL_TABLE_METRICS.get("UnleveledSSTables"))
{
cnt += ((Gauge<? extends Number>) cfGauge).getValue().intValue();
}
@ -944,7 +946,7 @@ public class TableMetrics
Gauge<T> cfGauge = Metrics.register(factory.createMetricName(name), aliasFactory.createMetricName(alias), gauge);
if (register(name, alias, cfGauge) && globalGauge != null)
{
Metrics.register(globalFactory.createMetricName(name), globalAliasFactory.createMetricName(alias), globalGauge);
Metrics.register(GLOBAL_FACTORY.createMetricName(name), GLOBAL_ALIAS_FACTORY.createMetricName(alias), globalGauge);
}
return cfGauge;
}
@ -969,11 +971,11 @@ public class TableMetrics
if (register(name, name, deprecated, cfGauge))
{
Metrics.register(globalFactory.createMetricName(name),
Metrics.register(GLOBAL_FACTORY.createMetricName(name),
globalGauge,
globalAliasFactory.createMetricName(name),
globalFactory.createMetricName(deprecated),
globalAliasFactory.createMetricName(deprecated));
GLOBAL_ALIAS_FACTORY.createMetricName(name),
GLOBAL_FACTORY.createMetricName(deprecated),
GLOBAL_ALIAS_FACTORY.createMetricName(deprecated));
}
return cfGauge;
}
@ -992,14 +994,14 @@ public class TableMetrics
Counter cfCounter = Metrics.counter(factory.createMetricName(name), aliasFactory.createMetricName(alias));
if (register(name, alias, cfCounter))
{
Metrics.register(globalFactory.createMetricName(name),
globalAliasFactory.createMetricName(alias),
Metrics.register(GLOBAL_FACTORY.createMetricName(name),
GLOBAL_ALIAS_FACTORY.createMetricName(alias),
new Gauge<Long>()
{
public Long getValue()
{
long total = 0;
for (Metric cfGauge : allTableMetrics.get(name))
for (Metric cfGauge : ALL_TABLE_METRICS.get(name))
{
total += ((Counter) cfGauge).getCount();
}
@ -1070,8 +1072,8 @@ public class TableMetrics
register(name, alias, cfHistogram);
return new TableHistogram(cfHistogram,
keyspaceHistogram,
Metrics.histogram(globalFactory.createMetricName(name),
globalAliasFactory.createMetricName(alias),
Metrics.histogram(GLOBAL_FACTORY.createMetricName(name),
GLOBAL_ALIAS_FACTORY.createMetricName(alias),
considerZeroes));
}
@ -1091,7 +1093,7 @@ public class TableMetrics
{
Timer cfTimer = Metrics.timer(factory.createMetricName(name), aliasFactory.createMetricName(name));
register(name, name, keyspaceTimer);
Timer global = Metrics.timer(globalFactory.createMetricName(name), globalAliasFactory.createMetricName(name));
Timer global = Metrics.timer(GLOBAL_FACTORY.createMetricName(name), GLOBAL_ALIAS_FACTORY.createMetricName(name));
return new TableTimer(cfTimer, keyspaceTimer, global);
}
@ -1114,8 +1116,8 @@ public class TableMetrics
register(name, alias, meter);
return new TableMeter(meter,
keyspaceMeter,
Metrics.meter(globalFactory.createMetricName(name),
globalAliasFactory.createMetricName(alias)));
Metrics.meter(GLOBAL_FACTORY.createMetricName(name),
GLOBAL_ALIAS_FACTORY.createMetricName(alias)));
}
private LatencyMetrics createLatencyMetrics(String namePrefix, LatencyMetrics ... parents)
@ -1145,8 +1147,8 @@ public class TableMetrics
*/
private boolean register(String name, String alias, String deprecated, Metric metric)
{
boolean ret = allTableMetrics.putIfAbsent(name, ConcurrentHashMap.newKeySet()) == null;
allTableMetrics.get(name).add(metric);
boolean ret = ALL_TABLE_METRICS.putIfAbsent(name, ConcurrentHashMap.newKeySet()) == null;
ALL_TABLE_METRICS.get(name).add(metric);
all.add(() -> releaseMetric(name, alias, deprecated));
return ret;
}
@ -1159,7 +1161,7 @@ public class TableMetrics
if (metric != null)
{
// Metric will be null if we are releasing a view metric. Views have null for ViewLockAcquireTime and ViewLockReadTime
allTableMetrics.get(tableMetricName).remove(metric);
ALL_TABLE_METRICS.get(tableMetricName).remove(metric);
CassandraMetricsRegistry.MetricName cfAlias = aliasFactory.createMetricName(cfMetricName);
if (tableMetricAlias != null)
@ -1331,7 +1333,7 @@ public class TableMetrics
public Long getValue()
{
long total = 0;
for (Metric cfGauge : allTableMetrics.get(name))
for (Metric cfGauge : ALL_TABLE_METRICS.get(name))
{
total = total + ((Gauge<? extends Number>) cfGauge).getValue().longValue();
}

95
src/java/org/apache/cassandra/utils/MBeanWrapper.java
View File

@ -21,14 +21,14 @@ package org.apache.cassandra.utils;
import java.lang.management.ManagementFactory;
import java.util.function.Consumer;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CassandraRelevantProperties;
import static org.apache.cassandra.config.CassandraRelevantProperties.IS_DISABLED_MBEAN_REGISTRATION;
import static org.apache.cassandra.config.CassandraRelevantProperties.MBEAN_REGISTRATION_CLASS;
/**
* Helper class to avoid catching and rethrowing checked exceptions on MBean and
@ -38,9 +38,18 @@ public interface MBeanWrapper
{
static final Logger logger = LoggerFactory.getLogger(MBeanWrapper.class);
static final MBeanWrapper instance = IS_DISABLED_MBEAN_REGISTRATION.getBoolean() ?
new NoOpMBeanWrapper() :
new PlatformMBeanWrapper();
static final MBeanWrapper instance = create();
static MBeanWrapper create()
{
if (IS_DISABLED_MBEAN_REGISTRATION.getBoolean())
return new NoOpMBeanWrapper();
String klass = MBEAN_REGISTRATION_CLASS.getString();
if (klass == null)
return new PlatformMBeanWrapper();
return FBUtilities.construct(klass, "mbean");
}
// Passing true for graceful will log exceptions instead of rethrowing them
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException);
@ -49,7 +58,13 @@ public interface MBeanWrapper
registerMBean(obj, mbeanName, OnException.THROW);
}
public void registerMBean(Object obj, String mbeanName, OnException onException);
default void registerMBean(Object obj, String mbeanName, OnException onException)
{
ObjectName name = create(mbeanName, onException);
if (name == null)
return;
registerMBean(obj, name, onException);
}
default void registerMBean(Object obj, String mbeanName)
{
registerMBean(obj, mbeanName, OnException.THROW);
@ -61,7 +76,13 @@ public interface MBeanWrapper
return isRegistered(mbeanName, OnException.THROW);
}
public boolean isRegistered(String mbeanName, OnException onException);
default boolean isRegistered(String mbeanName, OnException onException)
{
ObjectName name = create(mbeanName, onException);
if (name == null)
return false;
return isRegistered(name, onException);
}
default boolean isRegistered(String mbeanName)
{
return isRegistered(mbeanName, OnException.THROW);
@ -73,12 +94,31 @@ public interface MBeanWrapper
unregisterMBean(mbeanName, OnException.THROW);
}
public void unregisterMBean(String mbeanName, OnException onException);
default void unregisterMBean(String mbeanName, OnException onException)
{
ObjectName name = create(mbeanName, onException);
if (name == null)
return;
unregisterMBean(name, onException);
}
default void unregisterMBean(String mbeanName)
{
unregisterMBean(mbeanName, OnException.THROW);
}
static ObjectName create(String mbeanName, OnException onException)
{
try
{
return new ObjectName(mbeanName);
}
catch (MalformedObjectNameException e)
{
onException.handler.accept(e);
return null;
}
}
static class NoOpMBeanWrapper implements MBeanWrapper
{
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException) {}
@ -104,18 +144,6 @@ public interface MBeanWrapper
}
}
public void registerMBean(Object obj, String mbeanName, OnException onException)
{
try
{
mbs.registerMBean(obj, new ObjectName(mbeanName));
}
catch (Exception e)
{
onException.handler.accept(e);
}
}
public boolean isRegistered(ObjectName mbeanName, OnException onException)
{
try
@ -129,19 +157,6 @@ public interface MBeanWrapper
return false;
}
public boolean isRegistered(String mbeanName, OnException onException)
{
try
{
return mbs.isRegistered(new ObjectName(mbeanName));
}
catch (Exception e)
{
onException.handler.accept(e);
}
return false;
}
public void unregisterMBean(ObjectName mbeanName, OnException onException)
{
try
@ -153,18 +168,6 @@ public interface MBeanWrapper
onException.handler.accept(e);
}
}
public void unregisterMBean(String mbeanName, OnException onException)
{
try
{
mbs.unregisterMBean(new ObjectName(mbeanName));
}
catch (Exception e)
{
onException.handler.accept(e);
}
}
}
public enum OnException
@ -173,7 +176,7 @@ public interface MBeanWrapper
LOG(e -> { logger.error("Error in MBean wrapper: ", e); }),
IGNORE(e -> {});
private Consumer<Exception> handler;
public final Consumer<Exception> handler;
OnException(Consumer<Exception> handler)
{
this.handler = handler;

8
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
View File

@ -305,6 +305,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
}
}
@Override
public void uncaughtException(Thread thread, Throwable throwable)
{
IInvokableInstance delegate = this.delegate;
@ -313,6 +314,13 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
else
logger.error("uncaught exception in thread {}", thread, throwable);
}
@Override
public String toString()
{
IInvokableInstance delegate = this.delegate;
return delegate == null ? "node" + config.num() : delegate.toString();
}
}
protected AbstractCluster(AbstractBuilder<I, ? extends ICluster<I>, ?> builder)

9
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -489,11 +489,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
GossipHelper.unsafeStatusToNormal(this, (IInstance) peer);
});
StorageService.instance.setUpDistributedSystemKeyspaces();
StorageService.instance.setNormalModeUnsafe();
}
StorageService.instance.ensureTraceKeyspace();
// Populate tokenMetadata for the second time,
// see org.apache.cassandra.service.CassandraDaemon.setup
StorageService.instance.populateTokenMetadata();
@ -737,6 +736,12 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
}).call();
}
@Override
public String toString()
{
return "node" + config.num();
}
private static class CapturingOutput implements Closeable
{
@SuppressWarnings("resource")

272
test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java Normal file
View File

@ -0,0 +1,272 @@
/*
* 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.metric;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import javax.management.InstanceAlreadyExistsException;
import javax.management.InstanceNotFoundException;
import javax.management.ObjectName;
import com.google.common.collect.ImmutableSet;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.auth.AuthKeyspace;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.repair.SystemDistributedKeyspace;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaKeyspace;
import org.apache.cassandra.tracing.TraceKeyspace;
import org.apache.cassandra.utils.MBeanWrapper;
import static org.apache.cassandra.config.CassandraRelevantProperties.IS_DISABLED_MBEAN_REGISTRATION;
import static org.apache.cassandra.config.CassandraRelevantProperties.MBEAN_REGISTRATION_CLASS;
public class TableMetricTest extends TestBaseImpl
{
static
{
MBEAN_REGISTRATION_CLASS.setString(MapMBeanWrapper.class.getName());
IS_DISABLED_MBEAN_REGISTRATION.setBoolean(false);
}
private static volatile Map<String, Collection<String>> SYSTEM_TABLES = null;
private static Set<String> TABLE_METRIC_NAMES = ImmutableSet.of("WriteLatency");
/**
* Makes sure that all system tables have the expected metrics
* @throws IOException
*/
@Test
public void systemTables() throws IOException
{
try (Cluster cluster = Cluster.build(1).start())
{
loadSystemTables(cluster);
assertSystemTableMetrics(cluster);
}
}
/**
* Tests that other table metrics are not modified when a single table is modified/deleted.
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-16095">CASSANDRA-16095</a>
*/
@Test
public void userTables() throws IOException
{
try (Cluster cluster = init(Cluster.build(3).start()))
{
loadSystemTables(cluster);
assertSystemTableMetrics(cluster);
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk bigint PRIMARY KEY)"));
cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
// alter table can change metrics, so monitor for it
cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl WITH comment = 'testing'"));
cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl ADD (value bigint)"));
cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl RENAME pk TO pk2"));
cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
cluster.schemaChange(withKeyspace("ALTER TABLE %s.tbl DROP value"));
cluster.forEach(i -> assertTableMetricsExist(i, KEYSPACE, "tbl"));
// drop and make sure table no longer exists
cluster.schemaChange(withKeyspace("DROP TABLE %s.tbl"));
cluster.forEach(i -> assertTableMetricsDoesNotExist(i, KEYSPACE, "tbl"));
cluster.schemaChange(withKeyspace("DROP KEYSPACE %s"));
cluster.forEach(i -> assertKeyspaceMetricDoesNotExists(i, KEYSPACE));
// no other table impacted?
assertSystemTableMetrics(cluster);
}
}
private static void loadSystemTables(Cluster cluster)
{
SYSTEM_TABLES = cluster.get(1).callOnInstance(() -> {
Map<String, Collection<String>> map = new HashMap<>();
Arrays.asList(SystemKeyspace.metadata(), AuthKeyspace.metadata(), SystemDistributedKeyspace.metadata(),
SchemaKeyspace.metadata(), TraceKeyspace.metadata())
.forEach(meta -> {
Set<String> tables = meta.tables.stream().map(t -> t.name).collect(Collectors.toSet());
map.put(meta.name, tables);
});
return map;
});
}
private static void assertSystemTableMetrics(Cluster cluster)
{
for (String keyspace : SYSTEM_TABLES.keySet())
{
for (String table : SYSTEM_TABLES.get(keyspace))
{
cluster.forEach(i -> assertTableMetricsExist(i, keyspace, table));
}
}
}
private static void assertTableMetricsExist(IInvokableInstance inst, String keyspace, String table)
{
assertTableMBeanExists(inst, keyspace, table);
for (String metric : TABLE_METRIC_NAMES)
assertTableMetricExists(inst, keyspace, table, metric);
}
private static void assertTableMetricsDoesNotExist(IInvokableInstance inst, String keyspace, String table)
{
assertTableMBeanDoesNotExists(inst, keyspace, table);
for (String metric : TABLE_METRIC_NAMES)
assertTableMetricDoesNotExists(inst, keyspace, table, metric);
}
private static void assertKeyspaceMetricDoesNotExists(IInvokableInstance inst, String keyspace)
{
for (String metric : TABLE_METRIC_NAMES)
assertKeyspaceMetricDoesNotExists(inst, keyspace, metric);
}
private static void assertTableMBeanExists(IInvokableInstance inst, String keyspace, String table)
{
inst.runOnInstance(() -> {
// cast only to make sure it linked properly
MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
Assert.assertTrue("Unable to find table mbean for " + keyspace + "." + table,
mbeans.isRegistered(ColumnFamilyStore.getTableMBeanName(keyspace, table, false)));
Assert.assertTrue("Unable to find column family mbean for " + keyspace + "." + table,
mbeans.isRegistered(ColumnFamilyStore.getColumnFamilieMBeanName(keyspace, table, false)));
});
}
private static void assertTableMBeanDoesNotExists(IInvokableInstance inst, String keyspace, String table)
{
inst.runOnInstance(() -> {
// cast only to make sure it linked properly
MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
Assert.assertFalse("Found table mbean for " + keyspace + "." + table,
mbeans.isRegistered(ColumnFamilyStore.getTableMBeanName(keyspace, table, false)));
Assert.assertFalse("Found column family mbean for " + keyspace + "." + table,
mbeans.isRegistered(ColumnFamilyStore.getColumnFamilieMBeanName(keyspace, table, false)));
});
}
private static void assertTableMetricExists(IInvokableInstance inst, String keyspace, String table, String name)
{
inst.runOnInstance(() -> {
// cast only to make sure it linked properly
MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
String mbean = getTableMetricName(keyspace, table, name);
Assert.assertTrue("Unable to find metric " + name + " for " + keyspace + "." + table, mbeans.isRegistered(mbean));
// verify replicated to keyspace
String keyspaceMBean = getKeyspaceMetricName(keyspace, name);
Assert.assertTrue("Unable to find keyspace metric " + keyspaceMBean + " for " + keyspace, mbeans.isRegistered(keyspaceMBean));
});
}
private static void assertTableMetricDoesNotExists(IInvokableInstance inst, String keyspace, String table, String name)
{
inst.runOnInstance(() -> {
// cast only to make sure it linked properly
MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
String mbean = getTableMetricName(keyspace, table, name);
Assert.assertFalse("Found metric " + name + " for " + keyspace + "." + table, mbeans.isRegistered(mbean));
// validate keyspace metric
assertKeyspaceMetricMayExists(mbeans, keyspace, name);
});
}
private static void assertKeyspaceMetricMayExists(MapMBeanWrapper mbeans, String keyspace, String name)
{
String keyspaceMBean = getKeyspaceMetricName(keyspace, name);
boolean keyspaceExists = Schema.instance.getKeyspaceMetadata(keyspace) != null;
String errorMessage = keyspaceExists ?
"Unable to find keyspace metric " + keyspaceMBean + " for " + keyspace :
"Found keyspace metric " + keyspaceMBean + " for " + keyspace;
Assert.assertEquals(errorMessage, keyspaceExists, mbeans.isRegistered(keyspaceMBean));
}
private static void assertKeyspaceMetricDoesNotExists(IInvokableInstance inst, String keyspace, String name)
{
inst.runOnInstance(() -> {
// cast only to make sure it linked properly
MapMBeanWrapper mbeans = (MapMBeanWrapper) MBeanWrapper.instance;
String keyspaceMBean = getKeyspaceMetricName(keyspace, name);
Assert.assertFalse("Found keyspace metric " + keyspaceMBean + " for " + keyspace, mbeans.isRegistered(keyspaceMBean));
});
}
private static String getKeyspaceMetricName(String keyspace, String name)
{
return String.format("org.apache.cassandra.metrics:type=Keyspace,keyspace=%s,name=%s", keyspace, name);
}
private static String getTableMetricName(String keyspace, String table, String name)
{
return String.format("org.apache.cassandra.metrics:type=Table,keyspace=%s,scope=%s,name=%s", keyspace, table, name);
}
public static final class MapMBeanWrapper implements MBeanWrapper
{
private final ConcurrentMap<ObjectName, Object> map = new ConcurrentHashMap<>();
@Override
public void registerMBean(Object obj, ObjectName mbeanName, OnException onException)
{
Object current = map.putIfAbsent(mbeanName, obj);
if (current != null)
onException.handler.accept(new InstanceAlreadyExistsException("MBean " + mbeanName + " already exists"));
}
@Override
public boolean isRegistered(ObjectName mbeanName, OnException onException)
{
return map.containsKey(mbeanName);
}
@Override
public void unregisterMBean(ObjectName mbeanName, OnException onException)
{
Object previous = map.remove(mbeanName);
if (previous == null)
onException.handler.accept(new InstanceNotFoundException("MBean " + mbeanName + " was not found"));
}
}
}