Add fdb metrics to the indexMerger log messages (#3686)

To get better visibility of the index merge process, add fdb metrics to
the merger's log messages.

    Resolves #3685

Author: jjezra

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java

@@ -510,6 +510,7 @@ protected CompletableFuture<Boolean> doneOrThrottleDelayAndMaybeLogProgress(bool
             StoreTimer metricsDiff = null;
             if (timer != null) {
                 metricsDiff = lastProgressSnapshot == null ? timer : StoreTimer.getDifference(timer, lastProgressSnapshot);
+                // Note: this fdb metrics represents all the fdb activity (including merges, heartbeats, etc.) since the previous log message.
                 lastProgressSnapshot = StoreTimerSnapshot.from(timer);
             }
             LOGGER.info(KeyValueLogMessage.build("Indexer: Built Range",

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingMerger.java

@@ -25,12 +25,17 @@
 import com.apple.foundationdb.record.logging.KeyValueLogMessage;
 import com.apple.foundationdb.record.logging.LogMessageKeys;
 import com.apple.foundationdb.record.metadata.Index;
+import com.apple.foundationdb.record.provider.common.StoreTimer;
+import com.apple.foundationdb.record.provider.common.StoreTimerSnapshot;
 import com.apple.foundationdb.record.util.Result;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
 import javax.annotation.ParametersAreNonnullByDefault;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -52,6 +57,8 @@ public class IndexingMerger {
     private final IndexingCommon common;
     private int repartitionDocumentCount = 0;
     private int repartitionSecondChances = 0;
+    private StoreTimerSnapshot lastProgressSnapshot = null;
+
 
     public IndexingMerger(final Index index,  IndexingCommon common, long initialMergesCountLimit) {
         this.index = index;
@@ -75,6 +82,7 @@ CompletableFuture<Void> mergeIndex() {
                 // Note: this runAsync will retry according to the runner's "maxAttempts" setting
                 common.getRunner().runAsync(context -> openRecordStore(context)
                                 .thenCompose(store -> {
+                                    startFdbMetricsMeasurement(timer);
                                     mergeStartTime.set(System.nanoTime());
                                     final IndexDeferredMaintenanceControl mergeControl = store.getIndexDeferredMaintenanceControl();
                                     mergeControlRef.set(mergeControl);
@@ -229,7 +237,7 @@ private void giveUpMerging(final IndexDeferredMaintenanceControl mergeControl, T
         throw common.getRunner().getDatabase().mapAsyncToSyncException(e);
     }
 
-    List<Object> mergerKeysAndValues(final IndexDeferredMaintenanceControl mergeControl) {
+    private List<Object> mergerKeysAndValues(final IndexDeferredMaintenanceControl mergeControl) {
         return List.of(
                 LogMessageKeys.INDEX_NAME, index.getName(),
                 LogMessageKeys.INDEX_MERGES_LAST_LIMIT, mergeControl.getMergesLimit(),
@@ -242,14 +250,27 @@ List<Object> mergerKeysAndValues(final IndexDeferredMaintenanceControl mergeCont
         );
     }
 
-    String mergerLogMessage(String ttl, final IndexDeferredMaintenanceControl mergeControl) {
+    private String mergerLogMessage(String ttl, final IndexDeferredMaintenanceControl mergeControl) {
         final KeyValueLogMessage msg = KeyValueLogMessage.build(ttl);
         msg.addKeysAndValues(mergerKeysAndValues(mergeControl));
+        // Note: this fdb metrics represents fdb activity within a single merge operation
+        msg.addKeysAndValues(getFdbMetricsMeasurement(common.getRunner().getTimer()));
         SubspaceProvider subspaceProvider = common.getRecordStoreBuilder().getSubspaceProvider();
         if (subspaceProvider != null) {
             msg.addKeyAndValue(subspaceProvider.logKey(), subspaceProvider);
         }
         return msg.toString();
     }
 
+    private void startFdbMetricsMeasurement(@Nullable FDBStoreTimer timer) {
+        if (timer != null) {
+            lastProgressSnapshot = StoreTimerSnapshot.from(timer);
+        }
+    }
+
+    private Map<String, Number> getFdbMetricsMeasurement(@Nullable FDBStoreTimer timer) {
+        return (timer == null || lastProgressSnapshot == null) ?
+               Collections.emptyMap() :
+               StoreTimer.getDifference(timer, lastProgressSnapshot).getKeysAndValues();
+    }
 }