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

Add SSTableIntervalTree latency metric 

Patch by Ariel Weisberg; Reviewed by David Capwell for CASSANDRA-20502
This commit is contained in:
Ariel Weisberg 2025-03-31 14:31:58 -04:00
parent 1bcd91f531
commit 67df6a5bff
8 changed files with 47 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Add SSTableIntervalTree latency metric (CASSANDRA-20502)
* Ignore repetitions of semicolon in CQLSH (CASSANDRA-19956)
* Avoid NPE during cms initialization abort (CASSANDRA-20527)
* Avoid failing queries when epoch changes and replica goes up/down (CASSANDRA-20489)

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

@ -512,7 +512,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
{
Directories.SSTableLister sstableFiles = directories.sstableLister(Directories.OnTxnErr.IGNORE).skipTemporary(true);
sstables = SSTableReader.openAll(this, sstableFiles.list().entrySet(), metadata);
data.addInitialSSTablesWithoutUpdatingSize(sstables, this);
data.addInitialSSTablesWithoutUpdatingSize(sstables);
}
// compaction strategy should be created after the CFS has been prepared

4
src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
View File

@ -294,7 +294,7 @@ public class LifecycleTransaction extends Transactional.AbstractTransactional im
// replace all updated readers with a version restored to its original state
List<SSTableReader> restored = restoreUpdatedOriginals();
List<SSTableReader> invalid = Lists.newArrayList(Iterables.concat(logged.update, logged.obsolete));
accumulate = tracker.apply(updateLiveSet(logged.update, restored), accumulate);
accumulate = tracker.apply(updateLiveSet(logged.update, restored, tracker.maybeGetSSTableIntervalTreeLatencyMetrics()), accumulate);
accumulate = tracker.notifySSTablesChanged(invalid, restored, OperationType.COMPACTION, accumulate);
// setReplaced immediately preceding versions that have not been obsoleted
accumulate = setReplaced(logged.update, accumulate);
@ -374,7 +374,7 @@ public class LifecycleTransaction extends Transactional.AbstractTransactional im
// and don't want anyone else messing with them
// apply atomically along with updating the live set of readers
tracker.apply(compose(updateCompacting(emptySet(), fresh),
updateLiveSet(toUpdate, staged.update)));
updateLiveSet(toUpdate, staged.update, tracker.maybeGetSSTableIntervalTreeLatencyMetrics())));
// log the staged changes and our newly marked readers
marked.addAll(fresh);

22
src/java/org/apache/cassandra/db/lifecycle/Tracker.java
View File

@ -45,6 +45,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.metrics.LatencyMetrics;
import org.apache.cassandra.metrics.StorageMetrics;
import org.apache.cassandra.notifications.INotification;
import org.apache.cassandra.notifications.INotificationConsumer;
@ -251,14 +252,14 @@ public class Tracker
addSSTablesInternal(sstables, true, false, true);
}
public void addInitialSSTablesWithoutUpdatingSize(Collection<SSTableReader> sstables, ColumnFamilyStore cfs)
public void addInitialSSTablesWithoutUpdatingSize(Collection<SSTableReader> sstables)
{
if (!isDummy())
{
for (SSTableReader reader : sstables)
reader.setupOnline();
}
apply(updateLiveSet(emptySet(), sstables));
apply(updateLiveSet(emptySet(), sstables, maybeGetSSTableIntervalTreeLatencyMetrics()));
notifyAdded(sstables, true);
}
@ -279,7 +280,7 @@ public class Tracker
{
if (!isDummy())
setupOnline(sstables);
apply(updateLiveSet(emptySet(), sstables));
apply(updateLiveSet(emptySet(), sstables, maybeGetSSTableIntervalTreeLatencyMetrics()));
if(updateSize)
maybeFail(updateSizeTracking(emptySet(), sstables, null));
if (maybeIncrementallyBackup)
@ -332,7 +333,7 @@ public class Tracker
{
Pair<View, View> result = apply(view -> {
Set<SSTableReader> toremove = copyOf(filter(view.sstables, and(remove, notIn(view.compacting))));
return updateLiveSet(toremove, emptySet()).apply(view);
return updateLiveSet(toremove, emptySet(), maybeGetSSTableIntervalTreeLatencyMetrics()).apply(view);
});
Set<SSTableReader> removed = Sets.difference(result.left.sstables, result.right.sstables);
@ -434,7 +435,7 @@ public class Tracker
{
// sstable may be null if we flushed batchlog and nothing needed to be retained
// if it's null, we don't care what state the cfstore is in, we just replace it and continue
apply(View.replaceFlushed(memtable, null));
apply(View.replaceFlushed(memtable, null, maybeGetSSTableIntervalTreeLatencyMetrics()));
return;
}
@ -442,7 +443,7 @@ public class Tracker
// back up before creating a new Snapshot (which makes the new one eligible for compaction)
maybeIncrementallyBackup(sstables);
apply(View.replaceFlushed(memtable, sstables));
apply(View.replaceFlushed(memtable, sstables, maybeGetSSTableIntervalTreeLatencyMetrics()));
Throwable fail;
fail = updateSizeTracking(emptySet(), sstables, null);
@ -625,6 +626,13 @@ public class Tracker
@VisibleForTesting
public void removeUnsafe(Set<SSTableReader> toRemove)
{
Pair<View, View> result = apply(view -> updateLiveSet(toRemove, emptySet()).apply(view));
Pair<View, View> result = apply(view -> updateLiveSet(toRemove, emptySet(), maybeGetSSTableIntervalTreeLatencyMetrics()).apply(view));
}
public LatencyMetrics maybeGetSSTableIntervalTreeLatencyMetrics()
{
if (cfstore == null)
return null;
return cfstore.metric != null ? cfstore.metric.viewSSTableIntervalTree : null;
}
}

21
src/java/org/apache/cassandra/db/lifecycle/View.java
View File

@ -23,6 +23,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
@ -36,6 +37,8 @@ import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.metrics.LatencyMetrics;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.Interval;
import static com.google.common.base.Predicates.equalTo;
@ -297,7 +300,7 @@ public class View
}
// construct a function to change the liveset in a Snapshot
static Function<View, View> updateLiveSet(final Set<SSTableReader> remove, final Collection<SSTableReader> add)
static Function<View, View> updateLiveSet(final Set<SSTableReader> remove, final Collection<SSTableReader> add, @Nullable LatencyMetrics sstableIntervalTreeLatency)
{
if (remove.isEmpty() && Iterables.isEmpty(add))
return Functions.identity();
@ -306,8 +309,11 @@ public class View
public View apply(View view)
{
Map<SSTableReader, SSTableReader> sstableMap = replace(view.sstablesMap, remove, add);
return new View(view.liveMemtables, view.flushingMemtables, sstableMap, view.compactingMap,
SSTableIntervalTree.update(view.intervalTree, remove, add));
long treeBuildStart = Clock.Global.nanoTime();
SSTableIntervalTree sstableIntervalTree = SSTableIntervalTree.update(view.intervalTree, remove, add);
if (sstableIntervalTreeLatency != null)
sstableIntervalTreeLatency.addNano(Clock.Global.nanoTime() - treeBuildStart);
return new View(view.liveMemtables, view.flushingMemtables, sstableMap, view.compactingMap, sstableIntervalTree);
}
};
}
@ -346,7 +352,7 @@ public class View
}
// called after flush: removes memtable from flushingMemtables, and inserts flushed into the live sstable set
static Function<View, View> replaceFlushed(final Memtable memtable, final Collection<SSTableReader> flushed)
static Function<View, View> replaceFlushed(final Memtable memtable, final Collection<SSTableReader> flushed, @Nullable LatencyMetrics sstableIntervalTreeLatency)
{
return new Function<View, View>()
{
@ -360,8 +366,11 @@ public class View
view.compactingMap, view.intervalTree);
Map<SSTableReader, SSTableReader> sstableMap = replace(view.sstablesMap, emptySet(), flushed);
return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap,
SSTableIntervalTree.update(view.intervalTree, null, flushed));
long treeBuildStart = Clock.Global.nanoTime();
SSTableIntervalTree sstableIntervalTree = SSTableIntervalTree.update(view.intervalTree, null, flushed);
if (sstableIntervalTreeLatency != null)
sstableIntervalTreeLatency.addNano(Clock.Global.nanoTime() - treeBuildStart);
return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap, sstableIntervalTree);
}
};
}

4
src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java
View File

@ -183,6 +183,8 @@ public class KeyspaceMetrics
public final Meter tooManySSTableIndexesReadWarnings;
public final Meter tooManySSTableIndexesReadAborts;
public final LatencyMetrics viewSSTableIntervalTree;
public final ImmutableMap<SSTableFormat<?, ?>, ImmutableMap<String, Gauge<? extends Number>>> formatSpecificGauges;
private final KeyspaceMetricNameFactory factory;
@ -291,6 +293,8 @@ public class KeyspaceMetrics
outOfRangeTokenReads = createKeyspaceCounter("ReadOutOfRangeToken");
outOfRangeTokenWrites = createKeyspaceCounter("WriteOutOfRangeToken");
outOfRangeTokenPaxosRequests = createKeyspaceCounter("PaxosOutOfRangeToken");
viewSSTableIntervalTree = createLatencyMetrics("ViewSSTableIntervalTree");
}
/**

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

@ -289,6 +289,9 @@ public class TableMetrics
public final ImmutableMap<SSTableFormat<?, ?>, ImmutableMap<String, Gauge<? extends Number>>> formatSpecificGauges;
// Time spent building SSTableIntervalTree when constructing a new View under the Tracker lock
public final LatencyMetrics viewSSTableIntervalTree;
private static Pair<Long, Long> totalNonSystemTablesSize(Predicate<SSTableReader> predicate)
{
long total = 0;
@ -861,6 +864,8 @@ public class TableMetrics
tooManySSTableIndexesReadWarnings = createTableMeter("TooManySSTableIndexesReadWarnings", cfs.keyspace.metric.tooManySSTableIndexesReadWarnings);
tooManySSTableIndexesReadAborts = createTableMeter("TooManySSTableIndexesReadAborts", cfs.keyspace.metric.tooManySSTableIndexesReadAborts);
viewSSTableIntervalTree = createLatencyMetrics("ViewSSTableIntervalTree", cfs.keyspace.metric.viewSSTableIntervalTree);
formatSpecificGauges = createFormatSpecificGauges(cfs);
}

8
test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
View File

@ -119,7 +119,7 @@ public class ViewTest
testFailure(View.updateCompacting(emptySet(), of(r2)), cur);
// update one compacting, one non-compacting, of the liveset to another instance of the same readers;
// confirm liveset changes but compacting does not
cur = View.updateLiveSet(copyOf(readers.subList(1, 3)), of(r1, r2)).apply(cur);
cur = View.updateLiveSet(copyOf(readers.subList(1, 3)), of(r1, r2), cfs.metric.viewSSTableIntervalTree).apply(cur);
Assert.assertSame(readers.get(0), cur.sstablesMap.get(r0));
Assert.assertSame(r1, cur.sstablesMap.get(r1));
Assert.assertSame(r2, cur.sstablesMap.get(r2));
@ -179,7 +179,7 @@ public class ViewTest
Assert.assertEquals(memtable2, cur.liveMemtables.get(1));
Assert.assertEquals(memtable3, cur.getCurrentMemtable());
testFailure(View.replaceFlushed(memtable2, null), cur);
testFailure(View.replaceFlushed(memtable2, null, cfs.metric.viewSSTableIntervalTree), cur);
cur = View.markFlushing(memtable2).apply(cur);
Assert.assertTrue(cur.flushingMemtables.contains(memtable2));
@ -196,14 +196,14 @@ public class ViewTest
Assert.assertEquals(memtable2, cur.flushingMemtables.get(1));
Assert.assertEquals(memtable3, cur.getCurrentMemtable());
cur = View.replaceFlushed(memtable2, null).apply(cur);
cur = View.replaceFlushed(memtable2, null, cfs.metric.viewSSTableIntervalTree).apply(cur);
Assert.assertEquals(1, cur.liveMemtables.size());
Assert.assertEquals(1, cur.flushingMemtables.size());
Assert.assertEquals(memtable1, cur.flushingMemtables.get(0));
Assert.assertEquals(memtable3, cur.getCurrentMemtable());
SSTableReader sstable = MockSchema.sstable(1, cfs);
cur = View.replaceFlushed(memtable1, singleton(sstable)).apply(cur);
cur = View.replaceFlushed(memtable1, singleton(sstable), cfs.metric.viewSSTableIntervalTree).apply(cur);
Assert.assertEquals(0, cur.flushingMemtables.size());
Assert.assertEquals(1, cur.liveMemtables.size());
Assert.assertEquals(memtable3, cur.getCurrentMemtable());