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

make creation timestamp consistent for all tables of a snapshot 

patch by Stefan Miklosovic; reviewed by Paulo Motta and Aleksei Zotov for CASSANDRA-16920
This commit is contained in:
Stefan Miklosovic 2021-09-08 15:40:55 +02:00
parent 99246fb24f
commit e86ae7fbe5
9 changed files with 74 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -24,6 +24,7 @@ import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.time.Instant;
import java.util.*;
import java.util.Objects;
import java.util.concurrent.*;
@ -1524,7 +1525,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
// skip snapshot creation during scrub, SEE JIRA 5891
if(!disableSnapshot)
snapshotWithoutFlush("pre-scrub-" + System.currentTimeMillis());
{
Instant creationTime = Instant.now();
String snapshotName = "pre-scrub-" + creationTime.toEpochMilli();
snapshotWithoutFlush(snapshotName, creationTime);
}
try
{
@ -1839,13 +1844,18 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public TableSnapshot snapshotWithoutFlush(String snapshotName)
{
return snapshotWithoutFlush(snapshotName, null, false, null, null);
return snapshotWithoutFlush(snapshotName, Instant.now());
}
public TableSnapshot snapshotWithoutFlush(String snapshotName, Instant creationTime)
{
return snapshotWithoutFlush(snapshotName, null, false, null, null, creationTime);
}
/**
* @param ephemeral If this flag is set to true, the snapshot will be cleaned during next startup
*/
public TableSnapshot snapshotWithoutFlush(String snapshotName, Predicate<SSTableReader> predicate, boolean ephemeral, Duration ttl, RateLimiter rateLimiter)
public TableSnapshot snapshotWithoutFlush(String snapshotName, Predicate<SSTableReader> predicate, boolean ephemeral, Duration ttl, RateLimiter rateLimiter, Instant creationTime)
{
if (ephemeral && ttl != null)
{
@ -1873,17 +1883,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
return createSnapshot(snapshotName, ephemeral, ttl, snapshottedSSTables);
return createSnapshot(snapshotName, ephemeral, ttl, snapshottedSSTables, creationTime);
}
protected TableSnapshot createSnapshot(String tag, boolean ephemeral, Duration ttl, Set<SSTableReader> sstables) {
protected TableSnapshot createSnapshot(String tag, boolean ephemeral, Duration ttl, Set<SSTableReader> sstables, Instant creationTime) {
Set<File> snapshotDirs = sstables.stream()
.map(s -> Directories.getSnapshotDirectory(s.descriptor, tag).getAbsoluteFile())
.filter(dir -> !Directories.isSecondaryIndexFolder(dir)) // Remove secondary index subdirectory
.collect(Collectors.toCollection(HashSet::new));
// Create and write snapshot manifest
SnapshotManifest manifest = new SnapshotManifest(mapToDataFilenames(sstables), ttl);
SnapshotManifest manifest = new SnapshotManifest(mapToDataFilenames(sstables), ttl, creationTime);
File manifestFile = getDirectories().getSnapshotManifestFile(tag);
writeSnapshotManifest(manifest, manifestFile);
snapshotDirs.add(manifestFile.getParentFile().getAbsoluteFile()); // manifest may create empty snapshot dir
@ -2030,7 +2040,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public TableSnapshot snapshot(String snapshotName)
{
return snapshot(snapshotName, false, null, null);
return snapshot(snapshotName, false, null, null, Instant.now());
}
/**
@ -2038,11 +2048,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*
* @param snapshotName the name of the associated with the snapshot
* @param skipFlush Skip blocking flush of memtable
* @param ttl duration after which the taken snapshot is removed automatically, if supplied with null, it will never be automatically removed
* @param rateLimiter Rate limiter for hardlinks-per-second
* @param creationTime time when this snapshot was taken
*/
public TableSnapshot snapshot(String snapshotName, boolean skipFlush, Duration ttl, RateLimiter rateLimiter)
public TableSnapshot snapshot(String snapshotName, boolean skipFlush, Duration ttl, RateLimiter rateLimiter, Instant creationTime)
{
return snapshot(snapshotName, null, false, skipFlush, ttl, rateLimiter);
return snapshot(snapshotName, null, false, skipFlush, ttl, rateLimiter, creationTime);
}
@ -2052,21 +2064,23 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public TableSnapshot snapshot(String snapshotName, Predicate<SSTableReader> predicate, boolean ephemeral, boolean skipFlush)
{
return snapshot(snapshotName, predicate, ephemeral, skipFlush, null, null);
return snapshot(snapshotName, predicate, ephemeral, skipFlush, null, null, Instant.now());
}
/**
* @param ephemeral If this flag is set to true, the snapshot will be cleaned up during next startup
* @param skipFlush Skip blocking flush of memtable
* @param ttl duration after which the taken snapshot is removed automatically, if supplied with null, it will never be automatically removed
* @param rateLimiter Rate limiter for hardlinks-per-second
* @param creationTime time when this snapshot was taken
*/
public TableSnapshot snapshot(String snapshotName, Predicate<SSTableReader> predicate, boolean ephemeral, boolean skipFlush, Duration ttl, RateLimiter rateLimiter)
public TableSnapshot snapshot(String snapshotName, Predicate<SSTableReader> predicate, boolean ephemeral, boolean skipFlush, Duration ttl, RateLimiter rateLimiter, Instant creationTime)
{
if (!skipFlush)
{
forceBlockingFlush();
}
return snapshotWithoutFlush(snapshotName, predicate, ephemeral, ttl, rateLimiter);
return snapshotWithoutFlush(snapshotName, predicate, ephemeral, ttl, rateLimiter, creationTime);
}
public boolean snapshotExists(String snapshotName)

7
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.db;
import java.io.File;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@ -248,7 +249,7 @@ public class Keyspace
* @param rateLimiter Rate limiter for hardlinks-per-second
* @throws IOException if the column family doesn't exist
*/
public void snapshot(String snapshotName, String columnFamilyName, boolean skipFlush, Duration ttl, RateLimiter rateLimiter) throws IOException
public void snapshot(String snapshotName, String columnFamilyName, boolean skipFlush, Duration ttl, RateLimiter rateLimiter, Instant creationTime) throws IOException
{
assert snapshotName != null;
boolean tookSnapShot = false;
@ -257,7 +258,7 @@ public class Keyspace
if (columnFamilyName == null || cfStore.name.equals(columnFamilyName))
{
tookSnapShot = true;
cfStore.snapshot(snapshotName, skipFlush, ttl, rateLimiter);
cfStore.snapshot(snapshotName, skipFlush, ttl, rateLimiter, creationTime);
}
}
@ -275,7 +276,7 @@ public class Keyspace
*/
public void snapshot(String snapshotName, String columnFamilyName) throws IOException
{
snapshot(snapshotName, columnFamilyName, false, null, null);
snapshot(snapshotName, columnFamilyName, false, null, null, Instant.now());
}
/**

5
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -22,6 +22,7 @@ import java.io.IOError;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@ -1473,8 +1474,10 @@ public final class SystemKeyspace
String snapshotName = Keyspace.getTimestampedSnapshotName(format("upgrade-%s-%s",
previous,
next));
Instant creationTime = Instant.now();
for (String keyspace : SchemaConstants.LOCAL_SYSTEM_KEYSPACE_NAMES)
Keyspace.open(keyspace).snapshot(snapshotName, null, false, null, null);
Keyspace.open(keyspace).snapshot(snapshotName, null, false, null, null, creationTime);
}
}

7
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.db.compaction;
import java.time.Instant;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
@ -115,7 +116,11 @@ public class CompactionTask extends AbstractCompactionTask
CompactionStrategyManager strategy = cfs.getCompactionStrategyManager();
if (DatabaseDescriptor.isSnapshotBeforeCompaction())
cfs.snapshotWithoutFlush(System.currentTimeMillis() + "-compact-" + cfs.name);
{
Instant creationTime = Instant.now();
cfs.snapshotWithoutFlush(creationTime.toEpochMilli() + "-compact-" + cfs.name, creationTime);
}
try (CompactionController controller = getCompactionController(transaction.originals()))
{

8
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -22,6 +22,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.file.Paths;
import java.time.Instant;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.*;
@ -3873,10 +3874,11 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
RateLimiter snapshotRateLimiter = DatabaseDescriptor.getSnapshotRateLimiter();
Instant creationTime = Instant.now();
for (Keyspace keyspace : keyspaces)
{
keyspace.snapshot(tag, null, skipFlush, ttl, snapshotRateLimiter);
keyspace.snapshot(tag, null, skipFlush, ttl, snapshotRateLimiter, creationTime);
}
}
@ -3938,13 +3940,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
RateLimiter snapshotRateLimiter = DatabaseDescriptor.getSnapshotRateLimiter();
Instant creationTime = Instant.now();
for (Entry<Keyspace, List<String>> entry : keyspaceColumnfamily.entrySet())
{
for (String table : entry.getValue())
entry.getKey().snapshot(tag, table, skipFlush, ttl, snapshotRateLimiter);
entry.getKey().snapshot(tag, table, skipFlush, ttl, snapshotRateLimiter, creationTime);
}
}
private void verifyKeyspaceIsValid(String keyspaceName)

4
src/java/org/apache/cassandra/service/snapshot/SnapshotManifest.java
View File

@ -60,10 +60,10 @@ public class SnapshotManifest
this.expiresAt = null;
}
public SnapshotManifest(List<String> files, Duration ttl)
public SnapshotManifest(List<String> files, Duration ttl, Instant creationTime)
{
this.files = files;
this.createdAt = Instant.now();
this.createdAt = creationTime;
this.expiresAt = ttl == null ? null : createdAt.plusMillis(ttl.toMilliseconds());
}

31
test/distributed/org/apache/cassandra/distributed/test/SnapshotsTTLTest.java → test/distributed/org/apache/cassandra/distributed/test/SnapshotsTest.java
View File

@ -19,8 +19,10 @@
package org.apache.cassandra.distributed.test;
import java.io.IOException;
import java.util.Arrays;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
@ -32,10 +34,9 @@ import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.NodeToolResult;
import org.apache.cassandra.distributed.shared.WithProperties;
import static java.lang.String.format;
import static org.apache.cassandra.distributed.shared.ClusterUtils.stopUnchecked;
public class SnapshotsTTLTest extends TestBaseImpl
public class SnapshotsTest extends TestBaseImpl
{
public static final Integer SNAPSHOT_CLEANUP_PERIOD_SECONDS = 1;
public static final Integer FIVE_SECONDS = 5;
@ -157,7 +158,8 @@ public class SnapshotsTTLTest extends TestBaseImpl
}
@Test
public void testSecondaryIndexCleanup() throws Exception {
public void testSecondaryIndexCleanup() throws Exception
{
cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS default WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};");
cluster.schemaChange("CREATE TABLE default.tbl (key int, value text, PRIMARY KEY (key))");
cluster.schemaChange("CREATE INDEX value_idx ON default.tbl (value)");
@ -180,10 +182,25 @@ public class SnapshotsTTLTest extends TestBaseImpl
listSnapshotsResult.stdoutNotContains("first");
}
private void populate(Cluster cluster) {
for (int i = 0; i < 100; i++) {
cluster.coordinator(1).execute("INSERT INTO default.tbl (key, value) VALUES (?, 'txt')", ConsistencyLevel.ONE, i);
}
@Test
public void testSameTimestampOnEachTableOfSnaphot()
{
cluster.get(1).nodetoolResult("snapshot", "-t", "sametimestamp").asserts().success();
NodeToolResult result = cluster.get(1).nodetoolResult("listsnapshots");
long distinctTimestamps = Arrays.stream(result.getStdout().split("\n"))
.filter(line -> line.startsWith("sametimestamp"))
.map(line -> line.replaceAll(" +", " ").split(" ")[7])
.distinct()
.count();
// assert all dates are same so there is just one value accross all individual tables
Assert.assertEquals(1, distinctTimestamps);
}
private void populate(Cluster cluster)
{
for (int i = 0; i < 100; i++)
cluster.coordinator(1).execute("INSERT INTO default.tbl (key, value) VALUES (?, 'txt')", ConsistencyLevel.ONE, i);
}
}

4
test/unit/org/apache/cassandra/db/DirectoriesTest.java
View File

@ -185,7 +185,7 @@ public class DirectoriesTest
if (createManifest)
{
File manifestFile = Directories.getSnapshotManifestFile(snapshotDir);
manifest = new SnapshotManifest(Collections.singletonList(sstableDesc.filenameFor(Component.DATA)), new Duration("1m"));
manifest = new SnapshotManifest(Collections.singletonList(sstableDesc.filenameFor(Component.DATA)), new Duration("1m"), Instant.now());
manifest.serializeToJsonFile(manifestFile);
}
@ -310,7 +310,7 @@ public class DirectoriesTest
File manifestFile = directories.getSnapshotManifestFile(tag);
SnapshotManifest manifest = new SnapshotManifest(files, new Duration("1m"));
SnapshotManifest manifest = new SnapshotManifest(files, new Duration("1m"), Instant.now());
manifest.serializeToJsonFile(manifestFile);
Set<File> dirs = new HashSet<>();

2
test/unit/org/apache/cassandra/service/snapshot/SnapshotManifestTest.java
View File

@ -106,7 +106,7 @@ public class SnapshotManifestTest
@Test
public void testSerializeAndDeserialize() throws Exception {
SnapshotManifest manifest = new SnapshotManifest(Arrays.asList("db1", "db2", "db3"), new Duration("2m"));
SnapshotManifest manifest = new SnapshotManifest(Arrays.asList("db1", "db2", "db3"), new Duration("2m"), Instant.now());
File manifestFile = tempFolder.newFile("manifest.json");
manifest.serializeToJsonFile(manifestFile);
manifest = SnapshotManifest.deserializeFromJsonFile(manifestFile);