Expose external refreshes through the stats API (elastic#38643)

Right now, the stats API only provides refresh metrics regarding
internal refreshes. This isn't very useful and somewhat misleading for
cluster administrators since the internal refreshes are not indicative
of documents being available for search.

In this PR I added a new metric for collecting external refreshes as
they occur and exposing them through the stats API. Now, calling an
endpoint for stats will yield external refresh metrics as well.

Relates elastic#36712

Author: Conor Landry

rest-api-spec/src/main/resources/rest-api-spec/test/cat.shards/10_basic.yml

@@ -1,6 +1,8 @@
 ---
 "Help":
-
+  - skip:
+      version: " - 7.0.99"
+      reason:  external refresh stats were added in 7.1.0
   - do:
       cat.shards:
         help: true
@@ -52,6 +54,8 @@
                     merges.total_time                .+   \n
                     refresh.total                    .+   \n
                     refresh.time                     .+   \n
+                    refresh.external_total           .+   \n
+                    refresh.external_time            .+   \n
                     refresh.listeners                .+   \n
                     search.fetch_current             .+   \n
                     search.fetch_time                .+   \n

server/src/main/java/org/elasticsearch/index/refresh/RefreshStats.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.index.refresh;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
@@ -36,6 +37,10 @@ public class RefreshStats implements Streamable, Writeable, ToXContentFragment {
 
     private long totalTimeInMillis;
 
+    private long externalTotal;
+
+    private long externalTotalTimeInMillis;
+
     /**
      * Number of waiting refresh listeners.
      */
@@ -47,12 +52,29 @@ public RefreshStats() {
     public RefreshStats(StreamInput in) throws IOException {
         total = in.readVLong();
         totalTimeInMillis = in.readVLong();
+        if (in.getVersion().onOrAfter(Version.V_7_1_0)) {
+            externalTotal = in.readVLong();
+            externalTotalTimeInMillis = in.readVLong();
+        }
         listeners = in.readVInt();
     }
 
-    public RefreshStats(long total, long totalTimeInMillis, int listeners) {
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeVLong(total);
+        out.writeVLong(totalTimeInMillis);
+        if (out.getVersion().onOrAfter(Version.V_7_1_0)) {
+            out.writeVLong(externalTotal);
+            out.writeVLong(externalTotalTimeInMillis);
+        }
+        out.writeVInt(listeners);
+    }
+
+    public RefreshStats(long total, long totalTimeInMillis, long externalTotal, long externalTotalTimeInMillis, int listeners) {
         this.total = total;
         this.totalTimeInMillis = totalTimeInMillis;
+        this.externalTotal = externalTotal;
+        this.externalTotalTimeInMillis = externalTotalTimeInMillis;
         this.listeners = listeners;
     }
 
@@ -66,6 +88,8 @@ public void addTotals(RefreshStats refreshStats) {
         }
         this.total += refreshStats.total;
         this.totalTimeInMillis += refreshStats.totalTimeInMillis;
+        this.externalTotal += refreshStats.externalTotal;
+        this.externalTotalTimeInMillis += refreshStats.externalTotalTimeInMillis;
         this.listeners += refreshStats.listeners;
     }
 
@@ -76,20 +100,38 @@ public long getTotal() {
         return this.total;
     }
 
+    /*
+     * The total number of external refresh executed.
+     */
+    public long getExternalTotal() { return this.externalTotal; }
+
     /**
-     * The total time merges have been executed (in milliseconds).
+     * The total time spent executing refreshes (in milliseconds).
      */
     public long getTotalTimeInMillis() {
         return this.totalTimeInMillis;
     }
 
     /**
-     * The total time merges have been executed.
+     * The total time spent executing external refreshes (in milliseconds).
+     */
+    public long getExternalTotalTimeInMillis() {
+        return this.externalTotalTimeInMillis;
+    }
+
+    /**
+     * The total time refreshes have been executed.
      */
     public TimeValue getTotalTime() {
         return new TimeValue(totalTimeInMillis);
     }
 
+    /**
+     * The total time external refreshes have been executed.
+     */
+    public TimeValue getExternalTotalTime() {
+        return new TimeValue(externalTotalTimeInMillis);
+    }
     /**
      * The number of waiting refresh listeners.
      */
@@ -102,6 +144,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.startObject("refresh");
         builder.field("total", total);
         builder.humanReadableField("total_time_in_millis", "total_time", getTotalTime());
+        builder.field("external_total", externalTotal);
+        builder.humanReadableField("external_total_time_in_millis", "external_total_time", getExternalTotalTime());
         builder.field("listeners", listeners);
         builder.endObject();
         return builder;
@@ -112,13 +156,6 @@ public void readFrom(StreamInput in) throws IOException {
         throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
     }
 
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeVLong(total);
-        out.writeVLong(totalTimeInMillis);
-        out.writeVInt(listeners);
-    }
-
     @Override
     public boolean equals(Object obj) {
         if (obj == null || obj.getClass() != RefreshStats.class) {
@@ -127,11 +164,13 @@ public boolean equals(Object obj) {
         RefreshStats rhs = (RefreshStats) obj;
         return total == rhs.total
                 && totalTimeInMillis == rhs.totalTimeInMillis
+                && externalTotal == rhs.externalTotal
+                && externalTotalTimeInMillis == rhs.externalTotalTimeInMillis
                 && listeners == rhs.listeners;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(total, totalTimeInMillis, listeners);
+        return Objects.hash(total, totalTimeInMillis, externalTotal, externalTotalTimeInMillis, listeners);
     }
 }

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -222,6 +222,7 @@ Runnable getGlobalCheckpointSyncer() {
 
     private final RecoveryStats recoveryStats = new RecoveryStats();
     private final MeanMetric refreshMetric = new MeanMetric();
+    private final MeanMetric externalRefreshMetric = new MeanMetric();
     private final MeanMetric flushMetric = new MeanMetric();
     private final CounterMetric periodicFlushMetric = new CounterMetric();
 
@@ -932,7 +933,12 @@ public long getWritingBytes() {
 
     public RefreshStats refreshStats() {
         int listeners = refreshListeners.pendingCount();
-        return new RefreshStats(refreshMetric.count(), TimeUnit.NANOSECONDS.toMillis(refreshMetric.sum()), listeners);
+        return new RefreshStats(
+            refreshMetric.count(),
+            TimeUnit.NANOSECONDS.toMillis(refreshMetric.sum()),
+            externalRefreshMetric.count(),
+            TimeUnit.NANOSECONDS.toMillis(externalRefreshMetric.sum()),
+            listeners);
     }
 
     public FlushStats flushStats() {
@@ -2900,7 +2906,8 @@ private RefreshListeners buildRefreshListeners() {
             indexSettings::getMaxRefreshListeners,
             () -> refresh("too_many_listeners"),
             threadPool.executor(ThreadPool.Names.LISTENER)::execute,
-            logger, threadPool.getThreadContext());
+            logger, threadPool.getThreadContext(),
+            externalRefreshMetric);
     }
 
     /**

server/src/main/java/org/elasticsearch/index/shard/RefreshListeners.java

@@ -23,6 +23,7 @@
 import org.apache.lucene.search.ReferenceManager;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.metrics.MeanMetric;
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.index.translog.Translog;
@@ -50,6 +51,12 @@ public final class RefreshListeners implements ReferenceManager.RefreshListener,
     private final Executor listenerExecutor;
     private final Logger logger;
     private final ThreadContext threadContext;
+    private final MeanMetric refreshMetric;
+
+    /**
+     * Time in nanosecond when beforeRefresh() is called. Used for calculating refresh metrics.
+     */
+    private long currentRefreshStartTime;
 
     /**
      * Is this closed? If true then we won't add more listeners and have flushed all pending listeners.
@@ -76,12 +83,13 @@ public final class RefreshListeners implements ReferenceManager.RefreshListener,
     private volatile Translog.Location lastRefreshedLocation;
 
     public RefreshListeners(IntSupplier getMaxRefreshListeners, Runnable forceRefresh, Executor listenerExecutor, Logger logger,
-                            ThreadContext threadContext) {
+                            ThreadContext threadContext, MeanMetric refreshMetric) {
         this.getMaxRefreshListeners = getMaxRefreshListeners;
         this.forceRefresh = forceRefresh;
         this.listenerExecutor = listenerExecutor;
         this.logger = logger;
         this.threadContext = threadContext;
+        this.refreshMetric = refreshMetric;
     }
 
     /**
@@ -204,10 +212,14 @@ public void setCurrentRefreshLocationSupplier(Supplier<Translog.Location> curren
     @Override
     public void beforeRefresh() throws IOException {
         currentRefreshLocation = currentRefreshLocationSupplier.get();
+        currentRefreshStartTime = System.nanoTime();
     }
 
     @Override
     public void afterRefresh(boolean didRefresh) throws IOException {
+        // Increment refresh metric before communicating to listeners.
+        refreshMetric.inc(System.nanoTime() - currentRefreshStartTime);
+
         /* We intentionally ignore didRefresh here because our timing is a little off. It'd be a useful flag if we knew everything that made
          * it into the refresh, but the way we snapshot the translog position before the refresh, things can sneak into the refresh that we
          * don't know about. */

server/src/main/java/org/elasticsearch/rest/action/cat/RestIndicesAction.java

@@ -289,6 +289,14 @@ protected Table getTableWithHeader(final RestRequest request) {
         table.addCell("refresh.time", "sibling:pri;alias:rti,refreshTime;default:false;text-align:right;desc:time spent in refreshes");
         table.addCell("pri.refresh.time", "default:false;text-align:right;desc:time spent in refreshes");
 
+        table.addCell("refresh.external_total",
+            "sibling:pri;alias:rto,refreshTotal;default:false;text-align:right;desc:total external refreshes");
+        table.addCell("pri.refresh.external_total", "default:false;text-align:right;desc:total external refreshes");
+
+        table.addCell("refresh.external_time",
+            "sibling:pri;alias:rti,refreshTime;default:false;text-align:right;desc:time spent in external refreshes");
+        table.addCell("pri.refresh.external_time", "default:false;text-align:right;desc:time spent in external refreshes");
+
         table.addCell("refresh.listeners",
             "sibling:pri;alias:rli,refreshListeners;default:false;text-align:right;desc:number of pending refresh listeners");
         table.addCell("pri.refresh.listeners", "default:false;text-align:right;desc:number of pending refresh listeners");
@@ -562,6 +570,12 @@ Table buildTable(final RestRequest request,
             table.addCell(totalStats.getRefresh() == null ? null : totalStats.getRefresh().getTotalTime());
             table.addCell(primaryStats.getRefresh() == null ? null : primaryStats.getRefresh().getTotalTime());
 
+            table.addCell(totalStats.getRefresh() == null ? null : totalStats.getRefresh().getExternalTotal());
+            table.addCell(primaryStats.getRefresh() == null ? null : primaryStats.getRefresh().getExternalTotal());
+
+            table.addCell(totalStats.getRefresh() == null ? null : totalStats.getRefresh().getExternalTotalTime());
+            table.addCell(primaryStats.getRefresh() == null ? null : primaryStats.getRefresh().getExternalTotalTime());
+
             table.addCell(totalStats.getRefresh() == null ? null : totalStats.getRefresh().getListeners());
             table.addCell(primaryStats.getRefresh() == null ? null : primaryStats.getRefresh().getListeners());
 

server/src/main/java/org/elasticsearch/rest/action/cat/RestNodesAction.java

@@ -201,6 +201,9 @@ protected Table getTableWithHeader(final RestRequest request) {
 
         table.addCell("refresh.total", "alias:rto,refreshTotal;default:false;text-align:right;desc:total refreshes");
         table.addCell("refresh.time", "alias:rti,refreshTime;default:false;text-align:right;desc:time spent in refreshes");
+        table.addCell("refresh.external_total", "alias:rto,refreshTotal;default:false;text-align:right;desc:total external refreshes");
+        table.addCell("refresh.external_time",
+            "alias:rti,refreshTime;default:false;text-align:right;desc:time spent in external refreshes");
         table.addCell("refresh.listeners", "alias:rli,refreshListeners;default:false;text-align:right;"
                 + "desc:number of pending refresh listeners");
 
@@ -378,6 +381,8 @@ Table buildTable(boolean fullId, RestRequest req, ClusterStateResponse state, No
             RefreshStats refreshStats = indicesStats == null ? null : indicesStats.getRefresh();
             table.addCell(refreshStats == null ? null : refreshStats.getTotal());
             table.addCell(refreshStats == null ? null : refreshStats.getTotalTime());
+            table.addCell(refreshStats == null ? null : refreshStats.getExternalTotal());
+            table.addCell(refreshStats == null ? null : refreshStats.getExternalTotalTime());
             table.addCell(refreshStats == null ? null : refreshStats.getListeners());
 
             ScriptStats scriptStats = stats == null ? null : stats.getScriptStats();

server/src/main/java/org/elasticsearch/rest/action/cat/RestShardsAction.java

@@ -164,6 +164,9 @@ protected Table getTableWithHeader(final RestRequest request) {
 
         table.addCell("refresh.total", "alias:rto,refreshTotal;default:false;text-align:right;desc:total refreshes");
         table.addCell("refresh.time", "alias:rti,refreshTime;default:false;text-align:right;desc:time spent in refreshes");
+        table.addCell("refresh.external_total", "alias:rto,refreshTotal;default:false;text-align:right;desc:total external refreshes");
+        table.addCell("refresh.external_time",
+            "alias:rti,refreshTime;default:false;text-align:right;desc:time spent in external refreshes");
         table.addCell("refresh.listeners",
             "alias:rli,refreshListeners;default:false;text-align:right;desc:number of pending refresh listeners");
 
@@ -319,6 +322,8 @@ private Table buildTable(RestRequest request, ClusterStateResponse state, Indice
 
             table.addCell(getOrNull(commonStats, CommonStats::getRefresh, RefreshStats::getTotal));
             table.addCell(getOrNull(commonStats, CommonStats::getRefresh, RefreshStats::getTotalTime));
+            table.addCell(getOrNull(commonStats, CommonStats::getRefresh, RefreshStats::getExternalTotal));
+            table.addCell(getOrNull(commonStats, CommonStats::getRefresh, RefreshStats::getExternalTotalTime));
             table.addCell(getOrNull(commonStats, CommonStats::getRefresh, RefreshStats::getListeners));
 
             table.addCell(getOrNull(commonStats, CommonStats::getSearch, i -> i.getTotal().getFetchCurrent()));

server/src/test/java/org/elasticsearch/index/refresh/RefreshStatsTests.java

@@ -28,14 +28,17 @@
 public class RefreshStatsTests extends ESTestCase {
 
     public void testSerialize() throws IOException {
-        RefreshStats stats = new RefreshStats(randomNonNegativeLong(), randomNonNegativeLong(), between(0, Integer.MAX_VALUE));
+        RefreshStats stats = new RefreshStats(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(),
+            randomNonNegativeLong(), between(0, Integer.MAX_VALUE));
         BytesStreamOutput out = new BytesStreamOutput();
         stats.writeTo(out);
         StreamInput input = out.bytes().streamInput();
         RefreshStats read = new RefreshStats(input);
         assertEquals(-1, input.read());
         assertEquals(stats.getTotal(), read.getTotal());
+        assertEquals(stats.getExternalTotal(), read.getExternalTotal());
         assertEquals(stats.getListeners(), read.getListeners());
         assertEquals(stats.getTotalTimeInMillis(), read.getTotalTimeInMillis());
+        assertEquals(stats.getExternalTotalTimeInMillis(), read.getExternalTotalTimeInMillis());
     }
 }

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -1513,6 +1513,33 @@ public void testRefreshMetric() throws IOException {
         closeShards(shard);
     }
 
+    public void testExternalRefreshMetric() throws IOException {
+        IndexShard shard = newStartedShard();
+        assertThat(shard.refreshStats().getExternalTotal(), equalTo(2L)); // refresh on: finalize and end of recovery
+        long initialTotalTime = shard.refreshStats().getExternalTotalTimeInMillis();
+        // check time advances
+        for (int i = 1; shard.refreshStats().getExternalTotalTimeInMillis() == initialTotalTime; i++) {
+            indexDoc(shard, "_doc", "test");
+            assertThat(shard.refreshStats().getExternalTotal(), equalTo(2L + i - 1));
+            shard.refresh("test");
+            assertThat(shard.refreshStats().getExternalTotal(), equalTo(2L + i));
+            assertThat(shard.refreshStats().getExternalTotalTimeInMillis(), greaterThanOrEqualTo(initialTotalTime));
+        }
+        long externalRefreshCount = shard.refreshStats().getExternalTotal();
+
+        indexDoc(shard, "_doc", "test");
+        try (Engine.GetResult ignored = shard.get(new Engine.Get(true, false, "_doc", "test",
+            new Term(IdFieldMapper.NAME, Uid.encodeId("test"))))) {
+            assertThat(shard.refreshStats().getExternalTotal(), equalTo(externalRefreshCount));
+            assertThat(shard.refreshStats().getExternalTotal(), equalTo(shard.refreshStats().getTotal() - 1));
+        }
+        indexDoc(shard, "_doc", "test");
+        shard.writeIndexingBuffer();
+        assertThat(shard.refreshStats().getExternalTotal(), equalTo(externalRefreshCount));
+        assertThat(shard.refreshStats().getExternalTotal(), equalTo(shard.refreshStats().getTotal() - 2));
+        closeShards(shard);
+    }
+
     public void testIndexingOperationsListeners() throws IOException {
         IndexShard shard = newStartedShard(true);
         indexDoc(shard, "_doc", "0", "{\"foo\" : \"bar\"}");

server/src/test/java/org/elasticsearch/index/shard/RefreshListenersTests.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.lucene.uid.Versions;
+import org.elasticsearch.common.metrics.MeanMetric;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.BigArrays;
@@ -89,20 +90,23 @@ public class RefreshListenersTests extends ESTestCase {
     private volatile int maxListeners;
     private ThreadPool threadPool;
     private Store store;
+    private MeanMetric refreshMetric;
 
     @Before
     public void setupListeners() throws Exception {
         // Setup dependencies of the listeners
         maxListeners = randomIntBetween(1, 1000);
         // Now setup the InternalEngine which is much more complicated because we aren't mocking anything
         threadPool = new TestThreadPool(getTestName());
+        refreshMetric = new MeanMetric();
         listeners = new RefreshListeners(
                 () -> maxListeners,
                 () -> engine.refresh("too-many-listeners"),
                 // Immediately run listeners rather than adding them to the listener thread pool like IndexShard does to simplify the test.
                 Runnable::run,
                 logger,
-                threadPool.getThreadContext());
+                threadPool.getThreadContext(),
+                refreshMetric);
 
         IndexSettings indexSettings = IndexSettingsModule.newIndexSettings("index", Settings.EMPTY);
         ShardId shardId = new ShardId(new Index("index", "_na_"), 1);