You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by am...@apache.org on 2017/10/13 10:40:06 UTC
svn commit: r1812108 - in /jackrabbit/oak/trunk/oak-blob-plugins/src:
main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTracker.java
test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTrackerTest.java
Author: amitj
Date: Fri Oct 13 10:40:06 2017
New Revision: 1812108
URL: http://svn.apache.org/viewvc?rev=1812108&view=rev
Log:
OAK-6550: Make BlobTracker snapshot process smarter
- Enabled snapshot after removal
- And skipping regular snapshot if happened within the last interval
Modified:
jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTracker.java
jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTrackerTest.java
Modified: jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTracker.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTracker.java?rev=1812108&r1=1812107&r2=1812108&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTracker.java (original)
+++ jackrabbit/oak/trunk/oak-blob-plugins/src/main/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTracker.java Fri Oct 13 10:40:06 2017
@@ -35,10 +35,12 @@ import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.io.Files;
+import org.apache.commons.io.FileUtils;
import org.apache.jackrabbit.core.data.DataRecord;
import org.apache.jackrabbit.oak.commons.FileIOUtils.FileLineDifferenceIterator;
import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
import org.apache.jackrabbit.oak.plugins.blob.SharedDataStore;
+import org.apache.jackrabbit.oak.stats.Clock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,11 +93,13 @@ public class BlobIdTracker implements Cl
private static final String datastoreMeta = "blobids";
private static final String fileNamePrefix = "blob";
private static final String mergedFileSuffix = ".refs";
+ private static final String snapshotMarkerSuffix = ".snapshot";
/* Local instance identifier */
private final String instanceId = randomUUID().toString();
private final SharedDataStore datastore;
+ private final long snapshotInterval;
protected BlobIdStore store;
@@ -103,6 +107,8 @@ public class BlobIdTracker implements Cl
private String prefix;
+ private File rootDir;
+
public BlobIdTracker(String path, String repositoryId,
long snapshotIntervalSecs, SharedDataStore datastore) throws IOException {
this(path, repositoryId, newSingleThreadScheduledExecutor(),
@@ -113,10 +119,10 @@ public class BlobIdTracker implements Cl
long snapshotDelaySecs, long snapshotIntervalSecs, SharedDataStore datastore)
throws IOException {
String root = concat(path, datastoreMeta);
- File rootDir = new File(root);
+ this.rootDir = new File(root);
this.datastore = datastore;
this.scheduler = scheduler;
-
+ this.snapshotInterval = SECONDS.toMillis(snapshotIntervalSecs);
try {
forceMkdir(rootDir);
prefix = fileNamePrefix + "-" + repositoryId;
@@ -134,11 +140,13 @@ public class BlobIdTracker implements Cl
@Override
public void remove(File recs) throws IOException {
store.removeRecords(recs);
+ snapshot(true);
}
@Override
public void remove(Iterator<String> recs) throws IOException {
store.removeRecords(recs);
+ snapshot(true);
}
@Override
@@ -250,21 +258,27 @@ public class BlobIdTracker implements Cl
}
}
+ private void snapshot() throws IOException {
+ snapshot(false);
+ }
+
/**
* Takes a snapshot on the tracker.
* to other cluster nodes/repositories connected to the DataStore.
*
* @throws IOException
*/
- private void snapshot() throws IOException {
+ private void snapshot(boolean skipStoreSnapshot) throws IOException {
try {
if (!SKIP_TRACKER) {
Stopwatch watch = Stopwatch.createStarted();
- store.snapshot();
- LOG.debug("Completed snapshot in [{}]", watch.elapsed(TimeUnit.MILLISECONDS));
- watch = Stopwatch.createStarted();
+ if (!skipStoreSnapshot) {
+ store.snapshot();
+ LOG.debug("Completed snapshot in [{}]", watch.elapsed(TimeUnit.MILLISECONDS));
+ }
+ watch = Stopwatch.createStarted();
File recs = store.getBlobRecordsFile();
datastore.addMetadataRecord(recs,
(prefix + instanceId + System.currentTimeMillis() + mergedFileSuffix));
@@ -274,8 +288,12 @@ public class BlobIdTracker implements Cl
try {
forceDelete(recs);
LOG.info("Deleted blob record file after snapshot and upload {}", recs);
+
+ // Update the timestamp for the snapshot marker
+ FileUtils.touch(getSnapshotMarkerFile());
+ LOG.info("Updated snapshot marker");
} catch (IOException e) {
- LOG.debug("Failed to delete file {}", recs, e);
+ LOG.debug("Failed to in cleaning up {}", recs, e);
}
}
} catch (Exception e) {
@@ -284,6 +302,11 @@ public class BlobIdTracker implements Cl
}
}
+ private File getSnapshotMarkerFile() {
+ File snapshotMarker = new File(rootDir, prefix + snapshotMarkerSuffix);
+ return snapshotMarker;
+ }
+
/**
* Closes the tracker and the underlying store.
*
@@ -646,15 +669,38 @@ public class BlobIdTracker implements Cl
* Job which calls the snapshot on the tracker.
*/
class SnapshotJob implements Runnable {
+ private long interval;
+ private Clock clock;
+
+ public SnapshotJob() {
+ this.interval = snapshotInterval;
+ this.clock = Clock.SIMPLE;
+ }
+
+ public SnapshotJob(long millis, Clock clock) {
+ this.interval = millis;
+ this.clock = clock;
+ }
+
@Override
public void run() {
- try {
- snapshot();
- LOG.info("Finished taking snapshot");
- } catch (Exception e) {
- LOG.warn("Failure in taking snapshot", e);
+ if (!skip()) {
+ try {
+ snapshot();
+ LOG.info("Finished taking snapshot");
+ } catch(Exception e){
+ LOG.warn("Failure in taking snapshot", e);
+ }
}
}
- }
+ private boolean skip() {
+ File snapshotMarker = getSnapshotMarkerFile();
+ if (snapshotMarker.exists() &&
+ (snapshotMarker.lastModified() > (clock.getTime() - interval))) {
+ return true;
+ }
+ return false;
+ }
+ }
}
Modified: jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTrackerTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTrackerTest.java?rev=1812108&r1=1812107&r2=1812108&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTrackerTest.java (original)
+++ jackrabbit/oak/trunk/oak-blob-plugins/src/test/java/org/apache/jackrabbit/oak/plugins/blob/datastore/BlobIdTrackerTest.java Fri Oct 13 10:40:06 2017
@@ -31,20 +31,21 @@ import java.util.concurrent.TimeUnit;
import com.google.common.collect.Iterators;
import com.google.common.io.Closer;
-import org.apache.commons.io.FileUtils;
import org.apache.jackrabbit.core.data.DataRecord;
import org.apache.jackrabbit.core.data.DataStoreException;
import org.apache.jackrabbit.oak.commons.FileIOUtils;
import org.apache.jackrabbit.oak.commons.concurrent.ExecutorCloser;
import org.apache.jackrabbit.oak.plugins.blob.SharedDataStore;
+import org.apache.jackrabbit.oak.stats.Clock;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static com.google.common.collect.Lists.newArrayList;
import static com.google.common.collect.Sets.newHashSet;
@@ -67,6 +68,8 @@ import static org.junit.Assume.assumeTha
* Test for BlobIdTracker to test addition, retrieval and removal of blob ids.
*/
public class BlobIdTrackerTest {
+ private static final Logger LOG = LoggerFactory.getLogger(BlobIdTrackerTest.class);
+
File root;
SharedDataStore dataStore;
BlobIdTracker tracker;
@@ -106,6 +109,7 @@ public class BlobIdTrackerTest {
@Test
public void addSnapshot() throws Exception {
+ LOG.info("In addSnapshot");
Set<String> initAdd = add(tracker, range(0, 4));
ScheduledFuture<?> scheduledFuture =
scheduler.schedule(tracker.new SnapshotJob(), 0, TimeUnit.MILLISECONDS);
@@ -119,19 +123,64 @@ public class BlobIdTrackerTest {
@Test
public void addSnapshotRemove() throws Exception {
+ LOG.info("In addSnapshotRemove");
+ snapshotRemove(tracker.new SnapshotJob());
+ }
+
+ @Test
+ public void snapshotIgnoreAfterRemove() throws Exception {
+ LOG.info("In snapshotIgnoreAfterRemove");
+ BlobIdTracker.SnapshotJob job = tracker.new SnapshotJob();
+
+ snapshotRemove(job);
+
+ // Since already retrieved the datastore should be empty unless the snapshot has actually run
+ ScheduledFuture<?> scheduledFuture =
+ scheduler.schedule(job, 0, TimeUnit.MILLISECONDS);
+ scheduledFuture.get();
+ assertTrue("Snapshot not skipped",
+ read(dataStore.getAllMetadataRecords(BLOBREFERENCES.getType())).isEmpty());
+ }
+
+ @Test
+ public void snapshotExecuteAfterRemove() throws Exception {
+ LOG.info("In snapshotExecuteAfterRemove");
+
+ Clock clock = Clock.ACCURATE;
+ BlobIdTracker.SnapshotJob job = tracker.new SnapshotJob(100, clock);
+
+ Set<String> present = snapshotRemove(job);
+
+ clock.waitUntil(System.currentTimeMillis() + 100);
+
+ // Since already retrieved the datastore should not be empty unless the snapshot is ignored
+ ScheduledFuture<?> scheduledFuture =
+ scheduler.schedule(job, 0, TimeUnit.MILLISECONDS);
+ scheduledFuture.get();
+
+ assertEquals("Elements not equal after snapshot after remove", present,
+ read(dataStore.getAllMetadataRecords(BLOBREFERENCES.getType())));
+ }
+
+ private Set<String> snapshotRemove(BlobIdTracker.SnapshotJob job) throws Exception {
Set<String> initAdd = add(tracker, range(0, 4));
ScheduledFuture<?> scheduledFuture =
- scheduler.schedule(tracker.new SnapshotJob(), 0, TimeUnit.MILLISECONDS);
+ scheduler.schedule(job, 0, TimeUnit.MILLISECONDS);
scheduledFuture.get();
assertEquals("Extra elements after add", initAdd, retrieve(tracker));
remove(tracker, folder.newFile(), initAdd, range(1, 2));
+ assertEquals("Extra elements after removes synced with datastore",
+ initAdd, read(dataStore.getAllMetadataRecords(BLOBREFERENCES.getType())));
assertEquals("Extra elements after remove", initAdd, retrieve(tracker));
+ return initAdd;
}
@Test
public void snapshotRetrieveIgnored() throws Exception {
+ LOG.info("In snapshotRetrieveIgnored");
+
System.setProperty("oak.datastore.skipTracker", "true");
// Close and open a new object to use the system property
@@ -161,6 +210,8 @@ public class BlobIdTrackerTest {
@Test
public void externalAddOffline() throws Exception {
+ LOG.info("In externalAddOffline");
+
// Close and open a new object to use the system property
closer.close();