cassandra/test/unit/org/apache/cassandra/tcm/GetLogStateTest.java

449 lines
24 KiB
Java

/*
* 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.tcm;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.NavigableMap;
import java.util.TreeMap;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.listeners.MetadataSnapshotListener;
import org.apache.cassandra.tcm.listeners.SchemaListener;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogState;
import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.log.SystemKeyspaceStorage;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
import org.apache.cassandra.tcm.transformations.CustomTransformation;
import org.apache.cassandra.tcm.transformations.cms.Initialize;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.psjava.util.AssertStatus.assertTrue;
public class GetLogStateTest
{
Epoch maxEpoch = Epoch.EMPTY;
NavigableMap<Epoch, ClusterMetadata> epochToSnapshot = new TreeMap<>();
@BeforeClass
public static void setup() throws IOException
{
DatabaseDescriptor.daemonInitialization();
StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance);
ServerTestUtils.cleanupAndLeaveDirs();
CommitLog.instance.start();
LogStorage logStorage = LogStorage.SystemKeyspace;
MetadataSnapshots snapshots = new MetadataSnapshots.SystemKeyspaceMetadataSnapshots();
ClusterMetadata initial = new ClusterMetadata(DatabaseDescriptor.getPartitioner());
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.sync()
.withInitialState(initial)
.withStorage(logStorage)
.withLogListener(new MetadataSnapshotListener())
.withListener(new SchemaListener(true));
LocalLog log = logSpec.createLog();
ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(),
snapshots,
log,
new AtomicLongBackedProcessor(log),
Commit.Replicator.NO_OP,
false);
ClusterMetadataService.unsetInstance();
ClusterMetadataService.setInstance(cms);
log.readyUnchecked();
NodeAddresses addresses = NodeAddresses.current();
Location location = DatabaseDescriptor.getInitialLocationProvider().initialLocation();
log.unsafeBootstrapForTesting(addresses.broadcastAddress);
// register first node & make it CMS as this affects how the meta strategy placements are calculated
ClusterMetadata metadata = ClusterMetadata.current().forceInitializedState(addresses.broadcastAddress.hashCode(),
addresses,
NodeVersion.CURRENT,
location);
Initialize initialize = new Initialize(metadata);
cms.commit(initialize);
}
@Test
public void testGetLogState()
{
clearAndPopulate();
// Starting at the current epoch, iterate backwards. For each preceding epoch, fetch and verify a LogState
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(maxEpoch);
assertEquals(logState, LogState.EMPTY);
testGetLogStateHelper();
}
@Test
public void testIncompleteLog1()
{
clearAndPopulate();
// delete an entry from the log which precedes the latest snapshot but is after toQuery. In this case we would
// prefer a simple list of entries, but will get the latest snapshot + subsequent entries instead.
Epoch lastSnapshot = epochToSnapshot.lastKey();
Epoch toDelete = Epoch.create(lastSnapshot.getEpoch() - 2);
Epoch toQuery = Epoch.create(lastSnapshot.getEpoch() - 3);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, epochToSnapshot.lastEntry().getValue(), null, maxEpoch);
}
@Test
public void testIncompleteLog2()
{
clearAndPopulate();
// delete an entry from the log with multiple snapshots between toQuery and current. The deleted entry is one
// which comes after all available snapshots. Without losing a log entry, we should expect a LogState containing
// the most recent snapshot plus subsequent entries.
ClusterMetadata lastSnapshot = epochToSnapshot.lastEntry().getValue();
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
for (int i=0; i<3; i++)
snapshots.next();
Epoch toQuery = Epoch.create(snapshots.next().getEpoch() - 1);
Epoch toDelete = Epoch.create(maxEpoch.getEpoch() - 2);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, lastSnapshot, lastSnapshot.epoch, maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertEquals(lastSnapshot, logState.baseState);
boolean containsExpectedEntries = logState.entries.stream()
.map(entry -> entry.epoch.getEpoch())
.allMatch(e -> e > lastSnapshot.epoch.getEpoch()
&& e <= maxEpoch.getEpoch()
&& e != toDelete.getEpoch());
assertTrue(containsExpectedEntries);
}
@Test
public void testIncompleteLog3()
{
clearAndPopulate();
// delete an entry from the log where we have multiple snapshots after the deleted entry. In this case we should
// get the latest snapshot + subsequent entries and the deleted entry should not affect this.
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
ClusterMetadata lastSnapshot = epochToSnapshot.get(snapshots.next());
// Jump back over a couple more snapshots to give us points to query for and delete
snapshots.next();
Epoch e = snapshots.next();
Epoch toQuery = Epoch.create(e.getEpoch() + 1);
Epoch toDelete = Epoch.create(toQuery.getEpoch() + 1);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, lastSnapshot, lastSnapshot.epoch, maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, lastSnapshot, lastSnapshot.epoch, maxEpoch);
}
@Test
public void testIncompleteLog4()
{
clearAndPopulate();
// delete an entry from the log which comes after the most recent snapshot, which itself comes after toQuery.
// We would normally expect to skip over the intervening snapshot and return just a list of entries from toQuery
// to current. Because we cannot make that list continuous, instead we get the most recent snapshot and an
// incomplete list.
ClusterMetadata lastSnapshot = epochToSnapshot.lastEntry().getValue();
Epoch toQuery = Epoch.create(lastSnapshot.epoch.getEpoch() - 2);
Epoch toDelete = Epoch.create(maxEpoch.getEpoch() - 1);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertEquals(lastSnapshot, logState.baseState);
boolean containsExpectedEntries = logState.entries.stream()
.map(entry -> entry.epoch.getEpoch())
.allMatch(e -> e > lastSnapshot.epoch.getEpoch()
&& e <= maxEpoch.getEpoch()
&& e != toDelete.getEpoch());
assertTrue(containsExpectedEntries);
}
@Test
public void testCorruptSnapshot1()
{
clearAndPopulate();
// with a single snapshot following toQuery, corrupt it so that it cannot be read. In this scenario, we already
// prefer to return a list of consecutive entries and no base snapshot, so the corruption should not matter.
Epoch lastSnapshot = epochToSnapshot.lastKey();
Epoch toQuery = Epoch.create(lastSnapshot.getEpoch() - 3);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", lastSnapshot.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
}
@Test
public void testCorruptSnapshot2()
{
clearAndPopulate();
// with a multiple snapshots following toQuery, corrupt the most recent. Ordinarily, we would expect the
// returned LogState to include the most recent snapshot plus subsequent entries. With the corruption, the base
// state should be the previous, valid snapshot plus all subsequent entries.
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
ClusterMetadata lastSnapshot = epochToSnapshot.get(snapshots.next());
ClusterMetadata previousSnapshot = epochToSnapshot.get(snapshots.next());
Epoch toQuery = Epoch.create(previousSnapshot.epoch.getEpoch() - 3);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, lastSnapshot, lastSnapshot.epoch, maxEpoch);
QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", lastSnapshot.epoch.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, previousSnapshot, previousSnapshot.epoch, maxEpoch);
}
@Test
public void testCorruptSnapshot3()
{
clearAndPopulate();
// with a multiple snapshots following toQuery, corrupt the n-most recent. Ordinarily, we would expect the
// returned LogState to include the most recent snapshot plus subsequent entries. With the corruption, the base
// state should be the first valid snapshot plus all subsequent entries.
List<ClusterMetadata> toCorrupt = new ArrayList<>();
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
ClusterMetadata lastSnapshot = epochToSnapshot.get(snapshots.next());
toCorrupt.add(lastSnapshot);
for ( int i=0; i < 3; i++)
toCorrupt.add(epochToSnapshot.get(snapshots.next()));
ClusterMetadata validSnapshot = epochToSnapshot.get(snapshots.next());
Epoch toQuery = Epoch.create(validSnapshot.epoch.getEpoch() - 3);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, lastSnapshot, lastSnapshot.epoch, maxEpoch);
toCorrupt.forEach(cm -> QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", cm.epoch.getEpoch()));
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, validSnapshot, validSnapshot.epoch, maxEpoch);
}
@Test
public void testCorruptSnapshot4()
{
clearAndPopulate();
// Corrupt every snapshot following toQuery, corrupt the n-most recent. Ordinarily, we would expect the
// returned LogState to include the most recent snapshot plus subsequent entries. With the corruption, the log
// state should contain just a list of all entries from toQuery to current.
List<ClusterMetadata> toCorrupt = new ArrayList<>();
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
ClusterMetadata lastSnapshot = epochToSnapshot.get(snapshots.next());
toCorrupt.add(lastSnapshot);
for ( int i=0; i < 3; i++)
toCorrupt.add(epochToSnapshot.get(snapshots.next()));
ClusterMetadata validSnapshot = epochToSnapshot.get(snapshots.next());
Epoch toQuery = Epoch.create(validSnapshot.epoch.getEpoch() + 1);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, lastSnapshot, lastSnapshot.epoch, maxEpoch);
toCorrupt.forEach(cm -> QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", cm.epoch.getEpoch()));
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
}
@Test
public void testIncompleteLogAndCorruptSnapshot1()
{
clearAndPopulate();
// delete an entry from the log which precedes the last snapshot and also corrupt that snapshot. Both the deleted
// entry and snapshot are after toQuery so it's not possible to construct a contiguous sequence from since to
// current, so instead we get just the entries which follow the deleted one
Epoch lastSnapshot = epochToSnapshot.lastKey();
Epoch toDelete = Epoch.create(lastSnapshot.getEpoch() - 2);
Epoch toQuery = Epoch.create(lastSnapshot.getEpoch() - 3);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", lastSnapshot.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toDelete, maxEpoch);
}
@Test
public void testIncompleteLogAndCorruptSnapshot2()
{
clearAndPopulate();
// delete an entry from the log which comes after the last snapshot and also corrupt that snapshot.
// We would normally expect to skip over the intervening snapshot and return just a list of entries from toQuery
// to current. Because we cannot make that list continuous, but we also cannot read the snapshot, instead we get
// just the entries which we are present and have an epoch > toQuery.
Epoch lastSnapshot = epochToSnapshot.lastKey();
Epoch toDelete = Epoch.create(lastSnapshot.getEpoch() + 2);
Epoch toQuery = Epoch.create(lastSnapshot.getEpoch() - 3);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, null, toQuery, maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", lastSnapshot.getEpoch());
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertNull(logState.baseState);
boolean containsExpectedEntries = logState.entries.stream()
.map(entry -> entry.epoch.getEpoch())
.allMatch(e -> e > toQuery.getEpoch()
&& e <= maxEpoch.getEpoch()
&& e != toDelete.getEpoch());
assertTrue(containsExpectedEntries);
}
@Test
public void testIncompleteLogAndCorruptSnapshot3()
{
clearAndPopulate();
// delete an entry from the log which is followed by multiple corrupt snapshots. Without corruption, querying
// from a point before the deleted entry would return a LogState with the most recent snapshot plus subsequent
// entries. Where all the relevant snapshots are corrupt, expect a LogState with only the non-continuous list
// of available entries between toQuery and current.
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
List<Epoch> toCorrupt = new ArrayList<>();
for (int i=0; i<3; i++)
toCorrupt.add(snapshots.next());
Epoch toQuery = Epoch.create(snapshots.next().getEpoch() + 1);
Epoch toDelete = Epoch.create(toQuery.getEpoch() + 2);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, epochToSnapshot.lastEntry().getValue(), epochToSnapshot.lastKey(), maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
toCorrupt.forEach(e -> QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", e.getEpoch()));
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertNull(logState.baseState);
boolean containsExpectedEntries = logState.entries.stream()
.map(entry -> entry.epoch.getEpoch())
.allMatch(e -> e > toQuery.getEpoch()
&& e <= maxEpoch.getEpoch()
&& e != toDelete.getEpoch());
assertTrue(containsExpectedEntries);
}
@Test
public void testIncompleteLogAndCorruptSnapshot4()
{
clearAndPopulate();
// Corrupt multiple snapshots and delete an entry which follows them in the log. Without corruption, querying
// from an epoch before those snapshots deleted entry would return a LogState with the most recent snapshot plus
// subsequent, non-consecutive entries. Where all the relevant snapshots are corrupt, expect a LogState with
// only the list of available, and still non-consecutive entries between toQuery and current.
Iterator<Epoch> snapshots = epochToSnapshot.descendingKeySet().iterator();
List<Epoch> toCorrupt = new ArrayList<>();
for (int i=0; i<3; i++)
toCorrupt.add(snapshots.next());
Epoch toQuery = Epoch.create(snapshots.next().getEpoch() + 1);
Epoch toDelete = Epoch.create(maxEpoch.getEpoch() - 2);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertCorrectLogState(logState, epochToSnapshot.lastEntry().getValue(), epochToSnapshot.lastKey(), maxEpoch);
QueryProcessor.executeInternal("DELETE FROM system.local_metadata_log WHERE epoch = ?", toDelete.getEpoch());
toCorrupt.forEach(e -> QueryProcessor.executeInternal("DELETE snapshot FROM system.metadata_snapshots WHERE epoch = ?", e.getEpoch()));
logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(toQuery);
assertNull(logState.baseState);
boolean containsExpectedEntries = logState.entries.stream()
.map(entry -> entry.epoch.getEpoch())
.allMatch(e -> e > toQuery.getEpoch()
&& e <= maxEpoch.getEpoch()
&& e != toDelete.getEpoch());
assertTrue(containsExpectedEntries);
}
private void testGetLogStateHelper()
{
// If there is only a single snapshot between the starting point and the current epoch, we prefer a LogState
// with no base snapshot and just the consecutive entries. In all other cases, the LogState should include the
// most recent snapshot that can be successfully read and was taken after the starting epoch along with any
// subsequent entries.
Epoch lastSnapshotAt = epochToSnapshot.lastKey();
Epoch lastEligibleSnapshotAt = epochToSnapshot.lowerKey(lastSnapshotAt);
for (int i = 1; i < 20; i++)
{
Epoch start = Epoch.create(maxEpoch.getEpoch() - i);
LogState logState = SystemKeyspaceStorage.SystemKeyspace.getLogState(start);
ClusterMetadata expectedBase;
Epoch expectedStart;
if (start.isEqualOrAfter(lastEligibleSnapshotAt))
{
expectedBase = null;
expectedStart = start;
}
else
{
expectedBase = epochToSnapshot.lastEntry().getValue();
expectedStart = null;
}
assertCorrectLogState(logState, expectedBase, expectedStart, maxEpoch);
}
}
private static void assertCorrectLogState(LogState logState, ClusterMetadata expectedBase, Epoch expectedStart, Epoch endEpoch)
{
Epoch prev;
if (expectedBase != null)
{
assertEquals(expectedBase, logState.baseState);
prev = expectedBase.epoch;
}
else
{
// exclusive start of entries;
assertEquals(expectedStart.nextEpoch(), logState.entries.get(0).epoch);
prev = expectedStart;
}
for (Entry e : logState.entries)
{
assertEquals(prev.nextEpoch(), e.epoch);
prev = e.epoch;
}
assertEquals(prev, endEpoch);
}
private void clearAndPopulate()
{
epochToSnapshot.clear();
for (int i = 0; i < 10; i++)
{
ClusterMetadata metadata = ClusterMetadataService.instance().triggerSnapshot();
epochToSnapshot.put(metadata.epoch, metadata);
for (int j = 0; j < 4; j++)
{
Transformation transform = new CustomTransformation(CustomTransformation.PokeInt.NAME,
new CustomTransformation.PokeInt((int) ClusterMetadata.current().epoch.getEpoch()));
maxEpoch = ClusterMetadataService.instance().commit(transform).epoch;
}
}
}
}