more work on reusing work node level data, now when recovering from p…

…eers
elastic · Jun 24, 2010 · edf0075 · edf0075
1 parent 57169d4
commit edf0075
Show file tree

Hide file tree

Showing 7 changed files with 254 additions and 94 deletions.
diff --git a/config/logging.yml b/config/logging.yml
@@ -1,7 +1,14 @@
 rootLogger: INFO, console, file
 logger:
   # log action execution errors for easier debugging
-  action : DEBUG
+  action: DEBUG
+
+  # gateway
+  #gateway: DEBUG
+  #index.gateway: DEBUG
+
+  # peer shard recovery
+  #index.shard.recovery: DEBUG
 
 appender:
   console:

diff --git a/...elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java b/...elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java
@@ -137,8 +137,8 @@ public synchronized void recover() throws IndexShardGatewayRecoveryException, Ig
                 if (logger.isDebugEnabled()) {
                     StringBuilder sb = new StringBuilder();
                     sb.append("recovery completed from ").append(shardGateway).append(", took [").append(stopWatch.totalTime()).append("], throttling_wait [").append(throttlingWaitTime.totalTime()).append("]\n");
-                    sb.append("    index    : number_of_files [").append(recoveryStatus.index().numberOfFiles()).append("] with total_size [").append(recoveryStatus.index().totalSize()).append("], throttling_wait [").append(recoveryStatus.index().throttlingWaitTime()).append("]\n");
-                    sb.append("             : reusing_files [").append(recoveryStatus.index().numberOfExistingFiles()).append("] with total_size [").append(recoveryStatus.index().existingTotalSize()).append("]\n");
+                    sb.append("    index    : recovered_files [").append(recoveryStatus.index().numberOfFiles()).append("] with total_size [").append(recoveryStatus.index().totalSize()).append("], throttling_wait [").append(recoveryStatus.index().throttlingWaitTime()).append("]\n");
+                    sb.append("             : reusing_files   [").append(recoveryStatus.index().numberOfExistingFiles()).append("] with total_size [").append(recoveryStatus.index().existingTotalSize()).append("]\n");
                     sb.append("    translog : translog_id [").append(recoveryStatus.translog().translogId()).append("], number_of_operations [").append(recoveryStatus.translog().numberOfOperations()).append("] with total_size[").append(recoveryStatus.translog().totalSize()).append("]");
                     logger.debug(sb.toString());
                 }

diff --git a/...h/src/main/java/org/elasticsearch/index/gateway/blobstore/BlobStoreIndexShardGateway.java b/...h/src/main/java/org/elasticsearch/index/gateway/blobstore/BlobStoreIndexShardGateway.java
@@ -381,9 +381,11 @@ private RecoveryStatus.Translog recoverTranslog() throws IndexShardGatewayRecove
                         break;
                     }
                 }
-                indexShard.performRecovery(operations);
+                indexShard.performRecoveryPrepareForTranslog();
+                indexShard.performRecoveryOperations(operations);
+                indexShard.performRecoveryFinalization();
 
-                // clean all the other translogs
+                // clean all the other translog
                 for (Long translogIdToDelete : translogIds) {
                     if (!translogId.equals(translogIdToDelete)) {
                         try {

diff --git a/...es/elasticsearch/src/main/java/org/elasticsearch/index/shard/recovery/RecoveryAction.java b/...es/elasticsearch/src/main/java/org/elasticsearch/index/shard/recovery/RecoveryAction.java
diff --git a/...elasticsearch/src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java b/...elasticsearch/src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java
@@ -400,35 +400,30 @@ private void innerDeleteByQuery(byte[] querySource, String queryParserName, Stri
         }
     }
 
-    public void performRecovery(Iterable<Translog.Operation> operations) throws ElasticSearchException {
+    /**
+     * After the store has been recovered, we need to start the engine in order to apply operations
+     */
+    public void performRecoveryPrepareForTranslog() throws ElasticSearchException {
         if (state != IndexShardState.RECOVERING) {
             throw new IndexShardNotRecoveringException(shardId, state);
         }
         engine.start();
-        applyTranslogOperations(operations);
+    }
+
+    public void performRecoveryFinalization() throws ElasticSearchException {
         synchronized (mutex) {
-            logger.debug("Moved to state [STARTED] post recovery (from gateway)");
+            logger.debug("Moved to state [STARTED] post recovery (from another shard)");
             state = IndexShardState.STARTED;
         }
         scheduleRefresherIfNeeded();
+        engine.refresh(new Engine.Refresh(true));
     }
 
-    public void performRecovery(Translog.Snapshot snapshot, boolean phase3) throws ElasticSearchException {
+    public void performRecoveryOperations(Iterable<Translog.Operation> operations) throws ElasticSearchException {
         if (state != IndexShardState.RECOVERING) {
             throw new IndexShardNotRecoveringException(shardId, state);
         }
-        if (!phase3) {
-            // start the engine, but the shard is not started yet...
-            engine.start();
-        }
-        applyTranslogOperations(snapshot);
-        if (phase3) {
-            synchronized (mutex) {
-                logger.debug("Moved to state [STARTED] post recovery (from another shard)");
-                state = IndexShardState.STARTED;
-            }
-            scheduleRefresherIfNeeded();
-        }
+        applyTranslogOperations(operations);
     }
 
     private void applyTranslogOperations(Iterable<Translog.Operation> snapshot) {

diff --git a/modules/elasticsearch/src/main/java/org/elasticsearch/index/translog/TranslogStreams.java b/modules/elasticsearch/src/main/java/org/elasticsearch/index/translog/TranslogStreams.java
@@ -25,7 +25,7 @@
 import java.io.IOException;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public class TranslogStreams {
 

diff --git a/...ins/cloud/src/main/java/org/elasticsearch/index/gateway/cloud/CloudIndexShardGateway.java b/...ins/cloud/src/main/java/org/elasticsearch/index/gateway/cloud/CloudIndexShardGateway.java
@@ -461,7 +461,10 @@ private RecoveryStatus.Translog recoverTranslog() throws IndexShardGatewayRecove
             }
             currentTranslogPartToWrite = index;
 
-            indexShard.performRecovery(operations);
+            indexShard.performRecoveryPrepareForTranslog();
+            indexShard.performRecoveryOperations(operations);
+            indexShard.performRecoveryFinalization();
+
             return new RecoveryStatus.Translog(latestTranslogId, operations.size(), new ByteSizeValue(size, ByteSizeUnit.BYTES));
         } catch (Exception e) {
             throw new IndexShardGatewayRecoveryException(shardId(), "Failed to perform recovery of translog", e);