honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Shared Storage Gateway (fs/s3): Wrong snapshotting of transaction log, closes #375.

This commit is contained in:
kimchy 2010-09-18 21:33:38 +02:00
parent a2011e0151
commit 2372f481aa
4 changed files with 9 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGateway.java
View File

@ -70,16 +70,14 @@ public interface IndexShardGateway extends IndexShardComponent, CloseableIndexCo
private final long lastIndexVersion; private final long lastIndexVersion;
private final long lastTranslogId; private final long lastTranslogId;
private final long lastTranslogPosition;
private final long lastTranslogLength; private final long lastTranslogLength;
private final int lastTotalTranslogOperations; private final int lastTotalTranslogOperations;
public Snapshot(SnapshotIndexCommit indexCommit, Translog.Snapshot translogSnapshot, long lastIndexVersion, long lastTranslogId, long lastTranslogPosition, long lastTranslogLength, int lastTotalTranslogOperations) { public Snapshot(SnapshotIndexCommit indexCommit, Translog.Snapshot translogSnapshot, long lastIndexVersion, long lastTranslogId, long lastTranslogLength, int lastTotalTranslogOperations) {
this.indexCommit = indexCommit; this.indexCommit = indexCommit;
this.translogSnapshot = translogSnapshot; this.translogSnapshot = translogSnapshot;
this.lastIndexVersion = lastIndexVersion; this.lastIndexVersion = lastIndexVersion;
this.lastTranslogId = lastTranslogId; this.lastTranslogId = lastTranslogId;
this.lastTranslogPosition = lastTranslogPosition;
this.lastTranslogLength = lastTranslogLength; this.lastTranslogLength = lastTranslogLength;
this.lastTotalTranslogOperations = lastTotalTranslogOperations; this.lastTotalTranslogOperations = lastTotalTranslogOperations;
} }
@ -127,10 +125,6 @@ public interface IndexShardGateway extends IndexShardComponent, CloseableIndexCo
return lastTranslogId; return lastTranslogId;
} }
public long lastTranslogPosition() {
return lastTranslogPosition;
}
public long lastTranslogLength() { public long lastTranslogLength() {
return lastTranslogLength; return lastTranslogLength;
} }

8
modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java
View File

@ -60,8 +60,6 @@ public class IndexShardGatewayService extends AbstractIndexShardComponent implem
private volatile long lastTranslogId = -1; private volatile long lastTranslogId = -1;
private volatile long lastTranslogPosition;
private volatile int lastTotalTranslogOperations; private volatile int lastTotalTranslogOperations;
private volatile long lastTranslogLength; private volatile long lastTranslogLength;
@ -150,7 +148,6 @@ public class IndexShardGatewayService extends AbstractIndexShardComponent implem
lastIndexVersion = recoveryStatus.index().version(); lastIndexVersion = recoveryStatus.index().version();
lastTranslogId = -1; lastTranslogId = -1;
lastTranslogPosition = 0;
lastTranslogLength = 0; lastTranslogLength = 0;
lastTotalTranslogOperations = recoveryStatus.translog().currentTranslogOperations(); lastTotalTranslogOperations = recoveryStatus.translog().currentTranslogOperations();
@ -230,11 +227,10 @@ public class IndexShardGatewayService extends AbstractIndexShardComponent implem
logger.debug("snapshot ({}) to {} ...", reason, shardGateway); logger.debug("snapshot ({}) to {} ...", reason, shardGateway);
SnapshotStatus snapshotStatus = SnapshotStatus snapshotStatus =
shardGateway.snapshot(new IndexShardGateway.Snapshot(snapshotIndexCommit, translogSnapshot, lastIndexVersion, lastTranslogId, lastTranslogPosition, lastTranslogLength, lastTotalTranslogOperations)); shardGateway.snapshot(new IndexShardGateway.Snapshot(snapshotIndexCommit, translogSnapshot, lastIndexVersion, lastTranslogId, lastTranslogLength, lastTotalTranslogOperations));
lastIndexVersion = snapshotIndexCommit.getVersion(); lastIndexVersion = snapshotIndexCommit.getVersion();
lastTranslogId = translogSnapshot.translogId(); lastTranslogId = translogSnapshot.translogId();
lastTranslogPosition = translogSnapshot.position();
lastTranslogLength = translogSnapshot.length(); lastTranslogLength = translogSnapshot.length();
lastTotalTranslogOperations = translogSnapshot.totalOperations(); lastTotalTranslogOperations = translogSnapshot.totalOperations();
return snapshotStatus; return snapshotStatus;
@ -247,7 +243,7 @@ public class IndexShardGatewayService extends AbstractIndexShardComponent implem
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
sb.append("snapshot (").append(reason).append(") completed to ").append(shardGateway).append(", took [").append(TimeValue.timeValueMillis(snapshotStatus.time())).append("]\n"); sb.append("snapshot (").append(reason).append(") completed to ").append(shardGateway).append(", took [").append(TimeValue.timeValueMillis(snapshotStatus.time())).append("]\n");
sb.append(" index : version [").append(lastIndexVersion).append("], number_of_files [").append(snapshotStatus.index().numberOfFiles()).append("] with total_size [").append(new ByteSizeValue(snapshotStatus.index().totalSize())).append("], took [").append(TimeValue.timeValueMillis(snapshotStatus.index().time())).append("]\n"); sb.append(" index : version [").append(lastIndexVersion).append("], number_of_files [").append(snapshotStatus.index().numberOfFiles()).append("] with total_size [").append(new ByteSizeValue(snapshotStatus.index().totalSize())).append("], took [").append(TimeValue.timeValueMillis(snapshotStatus.index().time())).append("]\n");
sb.append(" translog : id [").append(lastTranslogId).append("], number_of_operations [" + snapshotStatus.translog().expectedNumberOfOperations() + "], took [").append(TimeValue.timeValueMillis(snapshotStatus.translog().time())).append("]"); sb.append(" translog : id [").append(lastTranslogId).append("], number_of_operations [").append(snapshotStatus.translog().expectedNumberOfOperations()).append("], took [").append(TimeValue.timeValueMillis(snapshotStatus.translog().time())).append("]");
logger.debug(sb.toString()); logger.debug(sb.toString());
} }
} }

2
modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/blobstore/BlobStoreIndexShardGateway.java
View File

@ -257,7 +257,7 @@ public abstract class BlobStoreIndexShardGateway extends AbstractIndexShardCompo
if (allTranslogFilesExists) { if (allTranslogFilesExists) {
translogCommitPointFiles.addAll(commitPoint.translogFiles()); translogCommitPointFiles.addAll(commitPoint.translogFiles());
if (snapshot.sameTranslogNewOperations()) { if (snapshot.sameTranslogNewOperations()) {
translogSnapshot.seekForward(snapshot.lastTranslogPosition()); translogSnapshot.seekForward(snapshot.lastTranslogLength());
if (translogSnapshot.lengthInBytes() > 0) { if (translogSnapshot.lengthInBytes() > 0) {
snapshotRequired = true; snapshotRequired = true;
expectedNumberOfOperations = translogSnapshot.totalOperations() - snapshot.lastTotalTranslogOperations(); expectedNumberOfOperations = translogSnapshot.totalOperations() - snapshot.lastTotalTranslogOperations();

5
modules/test/integration/src/test/java/org/elasticsearch/test/integration/gateway/fs/AbstractSimpleIndexGatewayTests.java
View File

@ -249,6 +249,11 @@ public abstract class AbstractSimpleIndexGatewayTests extends AbstractNodesTests
client("server1").prepareIndex("test", "type1", Long.toString(i)) client("server1").prepareIndex("test", "type1", Long.toString(i))
.setCreate(true) // make sure we use create, so if we recover wrongly, we will get increments... .setCreate(true) // make sure we use create, so if we recover wrongly, we will get increments...
.setSource(MapBuilder.<String, Object>newMapBuilder().put("test", "value" + i).map()).execute().actionGet(); .setSource(MapBuilder.<String, Object>newMapBuilder().put("test", "value" + i).map()).execute().actionGet();
// snapshot every 100 so we get some actions going on in the gateway
if ((i % 100) == 0) {
client("server1").admin().indices().prepareGatewaySnapshot().execute().actionGet();
}
} }
logger.info("--> refreshing and checking count"); logger.info("--> refreshing and checking count");