Fix recovery percentage > 100%

The recovery API was sometimes misreporting the recovered byte
percentages of index files. This was caused by summing up total file
lengths on each file chunk transfer. It should have been summing the
lengths of each transfer request.

Closes #6113

src/main/java/org/elasticsearch/indices/recovery/RecoveryState.java

@@ -658,6 +658,10 @@ public class RecoveryState implements ToXContent, Streamable {
             }
         }
 
+        public float percentFilesRecovered() {
+            return percentFilesRecovered(recoveredFileCount.get());
+        }
+
         public int numberOfRecoveredFiles() {
             return totalFileCount - reusedFileCount;
         }
@@ -699,6 +703,10 @@ public class RecoveryState implements ToXContent, Streamable {
             }
         }
 
+        public float percentBytesRecovered() {
+            return percentBytesRecovered(recoveredByteCount.get());
+        }
+
         public int reusedFileCount() {
             return reusedFileCount;
         }

src/main/java/org/elasticsearch/indices/recovery/RecoveryTarget.java

@@ -635,7 +635,7 @@ public class RecoveryTarget extends AbstractComponent {
                             content = content.toBytesArray();
                         }
                         indexOutput.writeBytes(content.array(), content.arrayOffset(), content.length());
-                        onGoingRecovery.recoveryState.getIndex().addRecoveredByteCount(request.length());
+                        onGoingRecovery.recoveryState.getIndex().addRecoveredByteCount(content.length());
                         RecoveryState.File file = onGoingRecovery.recoveryState.getIndex().file(request.name());
                         if (file != null) {
                             file.updateRecovered(request.length());

src/test/java/org/elasticsearch/indices/recovery/IndexRecoveryTests.java

@@ -25,6 +25,7 @@ import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotR
 import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
 import org.elasticsearch.action.admin.indices.recovery.ShardRecoveryResponse;
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.index.shard.ShardId;
@@ -36,12 +37,12 @@ import org.junit.Test;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 
 import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
 import static org.elasticsearch.test.ElasticsearchIntegrationTest.*;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.*;
 
 /**
  *
@@ -50,10 +51,12 @@ import static org.hamcrest.Matchers.greaterThan;
 public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
 
     private static final String INDEX_NAME = "test-idx-1";
+    private static final String INDEX_TYPE = "test-type-1";
     private static final String REPO_NAME = "test-repo-1";
     private static final String SNAP_NAME = "test-snap-1";
 
-    private static final int DOC_COUNT = 100;
+    private static final int MIN_DOC_COUNT = 500;
+    private static final int MAX_DOC_COUNT = 1000;
     private static final int SHARD_COUNT = 1;
     private static final int REPLICA_COUNT = 0;
 
@@ -84,6 +87,8 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
         assertThat(node, equalTo(state.getSourceNode().getName()));
         assertThat(node, equalTo(state.getTargetNode().getName()));
         assertNull(state.getRestoreSource());
+
+        validateIndexRecoveryState(state.getIndex());
     }
 
     @Test
@@ -141,6 +146,7 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
         assertThat(nodeAShardResponse.recoveryState().getTargetNode().getName(), equalTo(nodeA));
         assertThat(nodeAShardResponse.recoveryState().getType(), equalTo(RecoveryState.Type.GATEWAY));
         assertThat(nodeAShardResponse.recoveryState().getStage(), equalTo(RecoveryState.Stage.DONE));
+        validateIndexRecoveryState(nodeAShardResponse.recoveryState().getIndex());
 
         // validate node B recovery
         ShardRecoveryResponse nodeBShardResponse = nodeBResponses.get(0);
@@ -149,6 +155,7 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
         assertThat(nodeBShardResponse.recoveryState().getTargetNode().getName(), equalTo(nodeB));
         assertThat(nodeBShardResponse.recoveryState().getType(), equalTo(RecoveryState.Type.REPLICA));
         assertThat(nodeBShardResponse.recoveryState().getStage(), equalTo(RecoveryState.Stage.DONE));
+        validateIndexRecoveryState(nodeBShardResponse.recoveryState().getIndex());
     }
 
     @Test
@@ -184,6 +191,7 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
         assertThat(nodeA, equalTo(state.getSourceNode().getName()));
         assertThat(nodeB, equalTo(state.getTargetNode().getName()));
         assertNull(state.getRestoreSource());
+        validateIndexRecoveryState(state.getIndex());
     }
 
     @Test
@@ -194,8 +202,8 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
         logger.info("--> create repository");
         assertAcked(client().admin().cluster().preparePutRepository(REPO_NAME)
                 .setType("fs").setSettings(ImmutableSettings.settingsBuilder()
-                        .put("location", newTempDir(LifecycleScope.SUITE))
-                        .put("compress", false)
+                                .put("location", newTempDir(LifecycleScope.SUITE))
+                                .put("compress", false)
                 ).get());
 
         ensureGreen();
@@ -237,6 +245,7 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
                 assertThat(shardResponse.recoveryState().getStage(), equalTo(RecoveryState.Stage.DONE));
                 assertNotNull(shardResponse.recoveryState().getRestoreSource());
                 assertThat(shardResponse.recoveryState().getTargetNode().getName(), equalTo(nodeA));
+                validateIndexRecoveryState(shardResponse.recoveryState().getIndex());
             }
         }
     }
@@ -251,18 +260,36 @@ public class IndexRecoveryTests extends ElasticsearchIntegrationTest {
         return nodeResponses;
     }
 
-    private IndicesStatsResponse createAndPopulateIndex(String name, int nodeCount, int shardCount, int replicaCount) {
+    private IndicesStatsResponse createAndPopulateIndex(String name, int nodeCount, int shardCount, int replicaCount)
+            throws ExecutionException, InterruptedException {
+
         logger.info("--> creating test index: {}", name);
         assertAcked(prepareCreate(name, nodeCount, settingsBuilder().put("number_of_shards", shardCount)
                 .put("number_of_replicas", replicaCount)));
         ensureGreen();
 
         logger.info("--> indexing sample data");
-        for (int i = 0; i < DOC_COUNT; i++) {
-            index(INDEX_NAME, "x", Integer.toString(i), "foo-" + i, "bar-" + i);
+        final int numDocs = between(MIN_DOC_COUNT, MAX_DOC_COUNT);
+        final IndexRequestBuilder[] docs = new IndexRequestBuilder[numDocs];
+
+        for (int i = 0; i < numDocs; i++) {
+            docs[i] = client().prepareIndex(INDEX_NAME, INDEX_TYPE).
+                    setSource("foo-int-" + i, randomInt(),
+                              "foo-string-" + i, randomAsciiOfLength(32),
+                              "foo-float-" + i, randomFloat());
         }
-        refresh();
-        assertThat(client().prepareCount(INDEX_NAME).get().getCount(), equalTo((long) DOC_COUNT));
+
+        indexRandom(true, docs);
+        flush();
+        assertThat(client().prepareCount(INDEX_NAME).get().getCount(), equalTo((long) numDocs));
         return client().admin().indices().prepareStats(INDEX_NAME).execute().actionGet();
     }
+
+    private void validateIndexRecoveryState(RecoveryState.Index indexState) {
+        assertThat(indexState.time(), greaterThanOrEqualTo(0L));
+        assertThat(indexState.percentFilesRecovered(), greaterThanOrEqualTo(0.0f));
+        assertThat(indexState.percentFilesRecovered(), lessThanOrEqualTo(100.0f));
+        assertThat(indexState.percentBytesRecovered(), greaterThanOrEqualTo(0.0f));
+        assertThat(indexState.percentBytesRecovered(), lessThanOrEqualTo(100.0f));
+    }
 }