Recovery: add total operations to the _recovery API

This commit adds the current total number of translog operations to the recovery reporting API.  We also expose the recovered / total percentage:

```
"translog": {
     "recovered": 536,
     "total": 986,
     "percent": "54.3%",
     "total_time": "2ms",
     "total_time_in_millis": 2
},
```

Closes elastic#9368
Closes elastic#10042

Author: bleskes

docs/reference/indices/recovery.asciidoc

@@ -65,6 +65,9 @@ coming[1.5.0, this syntax was change to fix inconsistencies with other API]
       },
       "translog" : {
         "recovered" : 0,
+        "total" : 0,
+        "percent" : "100.0%",
+        "total_on_start" : 0,
         "total_time" : "0s",
         "total_time_in_millis" : 0
       },

rest-api-spec/test/cat.recovery/10_basic.yaml

@@ -40,6 +40,9 @@
                 \d+\.\d+%   \s+                                 # bytes_percent
                 \d+         \s+                                 # total_files
                 \d+         \s+                                 # total_bytes
+                \d+         \s+                                 # translog
+                -?\d+\.\d+% \s+                                 # translog_percent
+                -?\d+       \s+                                 # total_translog
                 \n
               )+
               $/

rest-api-spec/test/indices.recovery/10_basic.yaml

@@ -33,6 +33,8 @@
   - gte:   { test_1.shards.0.index.size.recovered_in_bytes:     0                       }
   - match: { test_1.shards.0.index.size.percent:                /^\d+\.\d\%$/           }
   - gte:   { test_1.shards.0.translog.recovered:                0                       }
+  - gte:   { test_1.shards.0.translog.total:                    -1                      }
+  - gte:   { test_1.shards.0.translog.total_on_start:           0                       }
   - gte:   { test_1.shards.0.translog.total_time_in_millis:     0                       }
   - gte:   { test_1.shards.0.start.check_index_time_in_millis:  0                       }
   - gte:   { test_1.shards.0.start.total_time_in_millis:        0                       }

src/main/java/org/elasticsearch/action/admin/indices/status/TransportIndicesStatusAction.java

@@ -202,7 +202,7 @@ protected ShardStatus shardOperation(IndexShardStatusRequest request) throws Ela
                             recoveryState.getTimer().time(),
                             index.totalBytes(),
                             index.reusedBytes(),
-                            index.recoveredBytes(), recoveryState.getTranslog().currentTranslogOperations());
+                            index.recoveredBytes(), recoveryState.getTranslog().recoveredOperations());
                 } else if (recoveryState.getType() == RecoveryState.Type.GATEWAY) {
                     GatewayRecoveryStatus.Stage stage;
                     switch (recoveryState.getStage()) {
@@ -222,7 +222,7 @@ protected ShardStatus shardOperation(IndexShardStatusRequest request) throws Ela
                             stage = GatewayRecoveryStatus.Stage.INIT;
                     }
                     shardStatus.gatewayRecoveryStatus = new GatewayRecoveryStatus(stage, recoveryState.getTimer().startTime(), recoveryState.getTimer().time(),
-                            index.totalBytes(), index.reusedBytes(), index.recoveredBytes(), recoveryState.getTranslog().currentTranslogOperations());
+                            index.totalBytes(), index.reusedBytes(), index.recoveredBytes(), recoveryState.getTranslog().recoveredOperations());
                 }
             }
         }

src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java

@@ -130,7 +130,7 @@ public void run() {
                                 .append(new ByteSizeValue(index.reusedBytes())).append("]\n");
                         sb.append("    start    : took [").append(TimeValue.timeValueMillis(recoveryState.getStart().time())).append("], check_index [")
                                 .append(timeValueMillis(recoveryState.getStart().checkIndexTime())).append("]\n");
-                        sb.append("    translog : number_of_operations [").append(recoveryState.getTranslog().currentTranslogOperations())
+                        sb.append("    translog : number_of_operations [").append(recoveryState.getTranslog().recoveredOperations())
                                 .append("], took [").append(TimeValue.timeValueMillis(recoveryState.getTranslog().time())).append("]");
                         logger.trace(sb.toString());
                     } else if (logger.isDebugEnabled()) {

src/main/java/org/elasticsearch/index/gateway/local/LocalIndexShardGateway.java

@@ -56,7 +56,6 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
-import java.nio.file.StandardCopyOption;
 import java.util.Arrays;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -229,6 +228,8 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
             if (recoveringTranslogFile == null || !recoveringTranslogFile.exists()) {
                 // no translog to recovery from, start and bail
                 // no translog files, bail
+                recoveryState.getTranslog().totalOperations(0);
+                recoveryState.getTranslog().totalOperationsOnStart(0);
                 indexShard.finalizeRecovery();
                 indexShard.postRecovery("post recovery from gateway, no translog");
                 // no index, just start the shard and bail
@@ -273,7 +274,7 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
                                 typesToUpdate.add(potentialIndexOperation.docMapper().type());
                             }
                         }
-                        recoveryState.getTranslog().addTranslogOperations(1);
+                        recoveryState.getTranslog().incrementRecoveredOperations();
                     } catch (ElasticsearchException e) {
                         if (e.status() == RestStatus.BAD_REQUEST) {
                             // mainly for MapperParsingException and Failure to detect xcontent

src/main/java/org/elasticsearch/index/gateway/none/NoneIndexShardGateway.java

@@ -65,6 +65,8 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
         } finally {
             indexShard.store().decRef();
         }
+        recoveryState.getTranslog().totalOperations(0);
+        recoveryState.getTranslog().totalOperationsOnStart(0);
         indexShard.prepareForTranslogRecovery();
         indexShard.finalizeRecovery();
         indexShard.postRecovery("post recovery from gateway");

src/main/java/org/elasticsearch/index/snapshots/IndexShardSnapshotAndRestoreService.java

@@ -115,6 +115,8 @@ public void restore(final RecoveryState recoveryState) {
             logger.trace("[{}] restoring shard  [{}]", restoreSource.snapshotId(), shardId);
         }
         try {
+            recoveryState.getTranslog().totalOperations(0);
+            recoveryState.getTranslog().totalOperationsOnStart(0);
             indexShard.prepareForIndexRecovery();
             IndexShardRepository indexShardRepository = repositoriesService.indexShardRepository(restoreSource.snapshotId().getRepository());
             ShardId snapshotShardId = shardId;

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

@@ -40,14 +40,16 @@ class RecoveryCleanFilesRequest extends TransportRequest {
     private ShardId shardId;
     private Set<String> legacySnapshotFiles; // legacy - we moved to a real snapshot in 1.5
     private Store.MetadataSnapshot snapshotFiles;
+    private int totalTranslogOps = RecoveryState.Translog.UNKNOWN;
 
     RecoveryCleanFilesRequest() {
     }
 
-    RecoveryCleanFilesRequest(long recoveryId, ShardId shardId, Store.MetadataSnapshot snapshotFiles) {
+    RecoveryCleanFilesRequest(long recoveryId, ShardId shardId, Store.MetadataSnapshot snapshotFiles, int totalTranslogOps) {
         this.recoveryId = recoveryId;
         this.shardId = shardId;
         this.snapshotFiles = snapshotFiles;
+        this.totalTranslogOps = totalTranslogOps;
     }
 
     public long recoveryId() {
@@ -65,6 +67,7 @@ public void readFrom(StreamInput in) throws IOException {
         shardId = ShardId.readShardId(in);
         if (in.getVersion().onOrAfter(Version.V_1_5_0)) {
             snapshotFiles = Store.MetadataSnapshot.read(in);
+            totalTranslogOps = in.readVInt();
         } else {
             int size = in.readVInt();
             legacySnapshotFiles = Sets.newHashSetWithExpectedSize(size);
@@ -82,6 +85,7 @@ public void writeTo(StreamOutput out) throws IOException {
         shardId.writeTo(out);
         if (out.getVersion().onOrAfter(Version.V_1_5_0)) {
             snapshotFiles.writeTo(out);
+            out.writeVInt(totalTranslogOps);
         } else {
             out.writeVInt(snapshotFiles.size());
             for (StoreFileMetaData snapshotFile : snapshotFiles) {
@@ -98,4 +102,8 @@ public Store.MetadataSnapshot sourceMetaSnapshot() {
     public Set<String> legacySnapshotFiles() {
         return legacySnapshotFiles;
     }
+
+    public int totalTranslogOps() {
+        return totalTranslogOps;
+    }
 }

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

@@ -42,16 +42,20 @@ public final class RecoveryFileChunkRequest extends TransportRequest {  // publi
     private BytesReference content;
     private StoreFileMetaData metaData;
 
+    private int totalTranslogOps;
+
     RecoveryFileChunkRequest() {
     }
 
-    public RecoveryFileChunkRequest(long recoveryId, ShardId shardId, StoreFileMetaData metaData, long position, BytesReference content, boolean lastChunk) {
+    public RecoveryFileChunkRequest(long recoveryId, ShardId shardId, StoreFileMetaData metaData, long position, BytesReference content,
+                                    boolean lastChunk, int totalTranslogOps) {
         this.recoveryId = recoveryId;
         this.shardId = shardId;
         this.metaData = metaData;
         this.position = position;
         this.content = content;
         this.lastChunk = lastChunk;
+        this.totalTranslogOps = totalTranslogOps;
     }
 
     public long recoveryId() {
@@ -83,6 +87,10 @@ public BytesReference content() {
         return content;
     }
 
+    public int totalTranslogOps() {
+        return totalTranslogOps;
+    }
+
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
@@ -104,6 +112,12 @@ public void readFrom(StreamInput in) throws IOException {
         } else {
             lastChunk = false;
         }
+
+        if (in.getVersion().onOrAfter(org.elasticsearch.Version.V_1_5_0)) {
+            totalTranslogOps = in.readVInt();
+        } else {
+            totalTranslogOps = RecoveryState.Translog.UNKNOWN;
+        }
     }
 
     @Override
@@ -122,6 +136,10 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(org.elasticsearch.Version.V_1_4_0_Beta1)) {
             out.writeBoolean(lastChunk);
         }
+        if (out.getVersion().onOrAfter(org.elasticsearch.Version.V_1_5_0)) {
+            out.writeVInt(totalTranslogOps);
+        }
+
     }
 
     @Override