Expose retention leases in shard stats

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.CommitStats;
+import org.elasticsearch.index.seqno.RetentionLeaseStats;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.indices.IndicesService;
@@ -101,21 +102,25 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
                     // only report on fully started shards
                     CommitStats commitStats;
                     SeqNoStats seqNoStats;
+                    RetentionLeaseStats retentionLeaseStats;
                     try {
                         commitStats = indexShard.commitStats();
                         seqNoStats = indexShard.seqNoStats();
-                    } catch (AlreadyClosedException e) {
+                        retentionLeaseStats = indexShard.getRetentionLeaseStats();
+                    } catch (final AlreadyClosedException e) {
                         // shard is closed - no stats is fine
                         commitStats = null;
                         seqNoStats = null;
+                        retentionLeaseStats = null;
                     }
                     shardsStats.add(
-                        new ShardStats(
-                            indexShard.routingEntry(),
-                            indexShard.shardPath(),
-                            new CommonStats(indicesService.getIndicesQueryCache(), indexShard, SHARD_STATS_FLAGS),
-                            commitStats,
-                            seqNoStats));
+                            new ShardStats(
+                                    indexShard.routingEntry(),
+                                    indexShard.shardPath(),
+                                    new CommonStats(indicesService.getIndicesQueryCache(), indexShard, SHARD_STATS_FLAGS),
+                                    commitStats,
+                                    seqNoStats,
+                                    retentionLeaseStats));
                 }
             }
         }

server/src/main/java/org/elasticsearch/action/admin/indices/stats/ShardStats.java

@@ -26,37 +26,58 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.xcontent.ToXContent.Params;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.engine.CommitStats;
+import org.elasticsearch.index.seqno.RetentionLeaseStats;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.shard.ShardPath;
 
 import java.io.IOException;
 
 public class ShardStats implements Streamable, Writeable, ToXContentFragment {
+
     private ShardRouting shardRouting;
     private CommonStats commonStats;
     @Nullable
     private CommitStats commitStats;
     @Nullable
     private SeqNoStats seqNoStats;
+
+    @Nullable
+    private RetentionLeaseStats retentionLeaseStats;
+
+    /**
+     * Gets the current retention lease stats.
+     *
+     * @return the current retention lease stats
+     */
+    public RetentionLeaseStats getRetentionLeaseStats() {
+        return retentionLeaseStats;
+    }
+
     private String dataPath;
     private String statePath;
     private boolean isCustomDataPath;
 
     ShardStats() {
     }
 
-    public ShardStats(ShardRouting routing, ShardPath shardPath, CommonStats commonStats, CommitStats commitStats, SeqNoStats seqNoStats) {
+    public ShardStats(
+            final ShardRouting routing,
+            final ShardPath shardPath,
+            final CommonStats commonStats,
+            final CommitStats commitStats,
+            final SeqNoStats seqNoStats,
+            final RetentionLeaseStats retentionLeaseStats) {
         this.shardRouting = routing;
         this.dataPath = shardPath.getRootDataPath().toString();
         this.statePath = shardPath.getRootStatePath().toString();
         this.isCustomDataPath = shardPath.isCustomDataPath();
         this.commitStats = commitStats;
         this.commonStats = commonStats;
         this.seqNoStats = seqNoStats;
+        this.retentionLeaseStats = retentionLeaseStats;
     }
 
     /**
@@ -109,6 +130,9 @@ public void readFrom(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
             seqNoStats = in.readOptionalWriteable(SeqNoStats::new);
         }
+        if (in.getVersion().onOrAfter(Version.V_7_0_0)) {
+            retentionLeaseStats = in.readOptionalWriteable(RetentionLeaseStats::new);
+        }
     }
 
     @Override
@@ -122,6 +146,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
             out.writeOptionalWriteable(seqNoStats);
         }
+        if (out.getVersion().onOrAfter(Version.V_7_0_0)) {
+            out.writeOptionalWriteable(retentionLeaseStats);
+        }
     }
 
     @Override
@@ -140,6 +167,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (seqNoStats != null) {
             seqNoStats.toXContent(builder, params);
         }
+        if (retentionLeaseStats != null) {
+            retentionLeaseStats.toXContent(builder, params);
+        }
         builder.startObject(Fields.SHARD_PATH);
         builder.field(Fields.STATE_PATH, statePath);
         builder.field(Fields.DATA_PATH, dataPath);
@@ -159,4 +189,5 @@ static final class Fields {
         static final String NODE = "node";
         static final String RELOCATING_NODE = "relocating_node";
     }
+
 }

server/src/main/java/org/elasticsearch/action/admin/indices/stats/TransportIndicesStatsAction.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.CommitStats;
+import org.elasticsearch.index.seqno.RetentionLeaseStats;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardNotFoundException;
@@ -106,15 +107,23 @@ protected ShardStats shardOperation(IndicesStatsRequest request, ShardRouting sh
         CommonStats commonStats = new CommonStats(indicesService.getIndicesQueryCache(), indexShard, request.flags());
         CommitStats commitStats;
         SeqNoStats seqNoStats;
+        RetentionLeaseStats retentionLeaseStats;
         try {
             commitStats = indexShard.commitStats();
             seqNoStats = indexShard.seqNoStats();
-        } catch (AlreadyClosedException e) {
+            retentionLeaseStats = indexShard.getRetentionLeaseStats();
+        } catch (final AlreadyClosedException e) {
             // shard is closed - no stats is fine
             commitStats = null;
             seqNoStats = null;
+            retentionLeaseStats = null;
         }
-        return new ShardStats(indexShard.routingEntry(), indexShard.shardPath(), commonStats,
-            commitStats, seqNoStats);
+        return new ShardStats(
+                indexShard.routingEntry(),
+                indexShard.shardPath(),
+                commonStats,
+                commitStats,
+                seqNoStats,
+                retentionLeaseStats);
     }
 }

server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseStats.java

@@ -0,0 +1,102 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.seqno;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ToXContentFragment;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Objects;
+
+/**
+ * Represents retention lease stats.
+ */
+public class RetentionLeaseStats implements ToXContentFragment, Writeable {
+
+    private final Collection<RetentionLease> leases;
+
+    /**
+     * Constructs a new retention lease stats object from the specified leases.
+     *
+     * @param leases the leases
+     */
+    public RetentionLeaseStats(final Collection<RetentionLease> leases) {
+        this.leases = Objects.requireNonNull(leases);
+    }
+
+    /**
+     * Constructs a new retention lease stats object from a stream. The retention lease stats should have been written via
+     * {@link #writeTo(StreamOutput)}.
+     *
+     * @param in the stream to construct the retention lease stats from
+     * @throws IOException if an I/O exception occurs reading from the stream
+     */
+    public RetentionLeaseStats(final StreamInput in) throws IOException {
+        leases = in.readList(RetentionLease::new);
+    }
+
+    /**
+     * Writes a retention lease stats object to a stream in a manner suitable for later reconstruction via
+     * {@link #RetentionLeaseStats(StreamInput)} (StreamInput)}.
+     *
+     * @param out the stream to write the retention lease stats to
+     * @throws IOException if an I/O exception occurs writing to the stream
+     */
+    @Override
+    public void writeTo(final StreamOutput out) throws IOException {
+        out.writeCollection(leases);
+    }
+
+    /**
+     * Converts the retention lease stats to {@link org.elasticsearch.common.xcontent.XContent} using the specified builder and pararms.
+     *
+     * @param builder the builder
+     * @param params  the params
+     * @return the builder that these retention leases were converted to {@link org.elasticsearch.common.xcontent.XContent} into
+     * @throws IOException if an I/O exception occurs writing to the builder
+     */
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        builder.startObject("retention_leases");
+        {
+            builder.startArray("leases");
+            {
+                for (final RetentionLease retentionLease : leases) {
+                    builder.startObject();
+                    {
+                        builder.field("id", retentionLease.id());
+                        builder.field("retaining_sequence_number", retentionLease.retainingSequenceNumber());
+                        builder.field("timestamp", retentionLease.timestamp());
+                        builder.field("source", retentionLease.source());
+                    }
+                    builder.endObject();
+                }
+            }
+            builder.endArray();
+        }
+        builder.endObject();
+        return builder;
+    }
+
+}

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

@@ -108,6 +108,7 @@
 import org.elasticsearch.index.search.stats.ShardSearchStats;
 import org.elasticsearch.index.seqno.ReplicationTracker;
 import org.elasticsearch.index.seqno.RetentionLease;
+import org.elasticsearch.index.seqno.RetentionLeaseStats;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.PrimaryReplicaSyncer.ResyncTask;
@@ -1895,6 +1896,11 @@ public Collection<RetentionLease> getRetentionLeases() {
         return replicationTracker.getRetentionLeases();
     }
 
+    public RetentionLeaseStats getRetentionLeaseStats() {
+        verifyNotClosed();
+        return new RetentionLeaseStats(getRetentionLeases());
+    }
+
     /**
      * Adds a new retention lease.
      *

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -94,6 +94,7 @@
 import org.elasticsearch.index.recovery.RecoveryStats;
 import org.elasticsearch.index.refresh.RefreshStats;
 import org.elasticsearch.index.search.stats.SearchStats;
+import org.elasticsearch.index.seqno.RetentionLeaseStats;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.shard.IllegalIndexShardStateException;
@@ -366,23 +367,29 @@ IndexShardStats indexShardStats(final IndicesService indicesService, final Index
 
         CommitStats commitStats;
         SeqNoStats seqNoStats;
+        RetentionLeaseStats retentionLeaseStats;
         try {
             commitStats = indexShard.commitStats();
             seqNoStats = indexShard.seqNoStats();
+            retentionLeaseStats = indexShard.getRetentionLeaseStats();
         } catch (AlreadyClosedException e) {
             // shard is closed - no stats is fine
             commitStats = null;
             seqNoStats = null;
+            retentionLeaseStats = null;
         }
 
-        return new IndexShardStats(indexShard.shardId(),
-                                   new ShardStats[] {
-                                       new ShardStats(indexShard.routingEntry(),
-                                                      indexShard.shardPath(),
-                                                      new CommonStats(indicesService.getIndicesQueryCache(), indexShard, flags),
-                                                      commitStats,
-                                                      seqNoStats)
-                                   });
+        return new IndexShardStats(
+                indexShard.shardId(),
+                new ShardStats[]{
+                        new ShardStats(
+                                indexShard.routingEntry(),
+                                indexShard.shardPath(),
+                                new CommonStats(indicesService.getIndicesQueryCache(), indexShard, flags),
+                                commitStats,
+                                seqNoStats,
+                                retentionLeaseStats)
+                });
     }
 
     /**

server/src/test/java/org/elasticsearch/cluster/DiskUsageTests.java

@@ -113,8 +113,8 @@ public void testFillShardLevelInfo() {
         CommonStats commonStats1 = new CommonStats();
         commonStats1.store = new StoreStats(1000);
         ShardStats[] stats  = new ShardStats[] {
-                new ShardStats(test_0, new ShardPath(false, test0Path, test0Path, test_0.shardId()), commonStats0 , null, null),
-                new ShardStats(test_1, new ShardPath(false, test1Path, test1Path, test_1.shardId()), commonStats1 , null, null)
+                new ShardStats(test_0, new ShardPath(false, test0Path, test0Path, test_0.shardId()), commonStats0 , null, null, null),
+                new ShardStats(test_1, new ShardPath(false, test1Path, test1Path, test_1.shardId()), commonStats1 , null, null, null)
         };
         ImmutableOpenMap.Builder<String, Long> shardSizes = ImmutableOpenMap.builder();
         ImmutableOpenMap.Builder<ShardRouting, String> routingToPath = ImmutableOpenMap.builder();

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

@@ -1385,8 +1385,13 @@ public void testMinimumCompatVersion() throws IOException {
     public void testShardStats() throws IOException {
 
         IndexShard shard = newStartedShard();
-        ShardStats stats = new ShardStats(shard.routingEntry(), shard.shardPath(),
-            new CommonStats(new IndicesQueryCache(Settings.EMPTY), shard, new CommonStatsFlags()), shard.commitStats(), shard.seqNoStats());
+        ShardStats stats = new ShardStats(
+                shard.routingEntry(),
+                shard.shardPath(),
+                new CommonStats(new IndicesQueryCache(Settings.EMPTY), shard, new CommonStatsFlags()),
+                shard.commitStats(),
+                shard.seqNoStats(),
+                shard.getRetentionLeaseStats());
         assertEquals(shard.shardPath().getRootDataPath().toString(), stats.getDataPath());
         assertEquals(shard.shardPath().getRootStatePath().toString(), stats.getStatePath());
         assertEquals(shard.shardPath().isCustomDataPath(), stats.isCustomDataPath());

server/src/test/java/org/elasticsearch/rest/action/cat/RestIndicesActionTests.java

@@ -163,7 +163,7 @@ private IndicesStatsResponse randomIndicesStatsResponse(final Index[] indices) {
                 stats.get = new GetStats();
                 stats.flush = new FlushStats();
                 stats.warmer = new WarmerStats();
-                shardStats.add(new ShardStats(shardRouting, new ShardPath(false, path, path, shardId), stats, null, null));
+                shardStats.add(new ShardStats(shardRouting, new ShardPath(false, path, path, shardId), stats, null, null, null));
             }
         }
         return IndicesStatsTests.newIndicesStatsResponse(

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/indices/IndicesStatsMonitoringDocTests.java

@@ -47,10 +47,10 @@ public void setUp() throws Exception {
         super.setUp();
         indicesStats = Collections.singletonList(new IndexStats("index-0", "dcvO5uZATE-EhIKc3tk9Bg", new ShardStats[] {
                 // Primaries
-                new ShardStats(mockShardRouting(true), mockShardPath(), mockCommonStats(), null, null),
-                new ShardStats(mockShardRouting(true), mockShardPath(), mockCommonStats(), null, null),
+                new ShardStats(mockShardRouting(true), mockShardPath(), mockCommonStats(), null, null, null),
+                new ShardStats(mockShardRouting(true), mockShardPath(), mockCommonStats(), null, null, null),
                 // Replica
-                new ShardStats(mockShardRouting(false), mockShardPath(), mockCommonStats(), null, null)
+                new ShardStats(mockShardRouting(false), mockShardPath(), mockCommonStats(), null, null, null)
         }));
     }