diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 05161d7cf2..2cfb3b87e6 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 predicate, boolean ephemeral, Duration ttl, RateLimiter rateLimiter) + public TableSnapshot snapshotWithoutFlush(String snapshotName, Predicate 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 sstables) { + protected TableSnapshot createSnapshot(String tag, boolean ephemeral, Duration ttl, Set sstables, Instant creationTime) { Set 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 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 predicate, boolean ephemeral, boolean skipFlush, Duration ttl, RateLimiter rateLimiter) + public TableSnapshot snapshot(String snapshotName, Predicate 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) diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java index 66eb1e02f8..e5cf7413be 100644 --- a/src/java/org/apache/cassandra/db/Keyspace.java +++ b/src/java/org/apache/cassandra/db/Keyspace.java @@ -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()); } /** diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 30671425e4..e07101c2a2 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -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); } } diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index 13c97253bc..19a0698967 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -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())) { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index a976a281d8..dcd2598a35 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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> 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) diff --git a/src/java/org/apache/cassandra/service/snapshot/SnapshotManifest.java b/src/java/org/apache/cassandra/service/snapshot/SnapshotManifest.java index 8fbc619b9d..5d44acbf4f 100644 --- a/src/java/org/apache/cassandra/service/snapshot/SnapshotManifest.java +++ b/src/java/org/apache/cassandra/service/snapshot/SnapshotManifest.java @@ -60,10 +60,10 @@ public class SnapshotManifest this.expiresAt = null; } - public SnapshotManifest(List files, Duration ttl) + public SnapshotManifest(List files, Duration ttl, Instant creationTime) { this.files = files; - this.createdAt = Instant.now(); + this.createdAt = creationTime; this.expiresAt = ttl == null ? null : createdAt.plusMillis(ttl.toMilliseconds()); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/SnapshotsTTLTest.java b/test/distributed/org/apache/cassandra/distributed/test/SnapshotsTest.java similarity index 88% rename from test/distributed/org/apache/cassandra/distributed/test/SnapshotsTTLTest.java rename to test/distributed/org/apache/cassandra/distributed/test/SnapshotsTest.java index ace6c11fc8..c56adfe7b1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/SnapshotsTTLTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/SnapshotsTest.java @@ -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); } } diff --git a/test/unit/org/apache/cassandra/db/DirectoriesTest.java b/test/unit/org/apache/cassandra/db/DirectoriesTest.java index fbecc69d2d..4c282a6bb3 100644 --- a/test/unit/org/apache/cassandra/db/DirectoriesTest.java +++ b/test/unit/org/apache/cassandra/db/DirectoriesTest.java @@ -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 dirs = new HashSet<>(); diff --git a/test/unit/org/apache/cassandra/service/snapshot/SnapshotManifestTest.java b/test/unit/org/apache/cassandra/service/snapshot/SnapshotManifestTest.java index 8760270886..2dc73bff66 100644 --- a/test/unit/org/apache/cassandra/service/snapshot/SnapshotManifestTest.java +++ b/test/unit/org/apache/cassandra/service/snapshot/SnapshotManifestTest.java @@ -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);