[7.x] Enhanced segment files sizes information in Nodes Stats/Indices Stats APIs (elastic#71725)

Since elastic#16661 it is possible to know the total sizes for some Lucene segment files 
by using the Node Stats or Indices Stats API with the include_segment_file_sizes 
parameter, and the list of file extensions has been extended in elastic#71416.

This commit adds a bit more information about file sizes like the number of files 
(count), the min, max and average file sizes in bytes that share the same extension.

Here is a sample:
"cfs" : {
  "description" : "Compound Files",
  "size_in_bytes" : 2260,
  "min_size_in_bytes" : 2260,
  "max_size_in_bytes" : 2260,
  "average_size_in_bytes" : 2260,
  "count" : 1
}

This commit also simplifies how compound file sizes were computed: before 
compound segment files were extracted and sizes aggregated with regular 
non-compound files sizes (which can be confusing and out of the scope of 
the original issue elastic#6728), now CFS/CFE files appears as distinct files.

These new information are provided to give a better view of the segment 
files and are useful in many cases, specially with frozen searchable snapshots 
whose segment stats can now be introspected thanks to the 
include_unloaded_segments parameter.

Backport of elastic#71643

Author: tlrx

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.stats/30_segments.yml

@@ -61,3 +61,42 @@ setup:
         forbid_closed_indices: false
 
   - match: { indices.test.primaries.segments.count: $num_segments_after_flush }
+
+
+---
+"Indices Stats API with extended files stats":
+
+  - skip:
+      version: " - 7.12.99"
+      reason: segment files stats extended in 7.13.0
+
+  - do:
+      index:
+        index: test
+        id:    1
+        body:  { "foo": "bar" }
+
+  - do:
+      indices.flush:
+        index: test
+
+  - do:
+      indices.stats:
+        metric: [ segments ]
+        include_segment_file_sizes: true
+
+  - is_true: _all.total.segments.file_sizes
+  - is_true: _all.total.segments.file_sizes.si
+  - gt: { _all.total.segments.file_sizes.si.count: 0 }
+  - gt: { _all.total.segments.file_sizes.si.size_in_bytes: 0 }
+  - gt: { _all.total.segments.file_sizes.si.min_size_in_bytes: 0 }
+  - gt: { _all.total.segments.file_sizes.si.max_size_in_bytes: 0 }
+  - gt: { _all.total.segments.file_sizes.si.average_size_in_bytes: 0 }
+
+  - is_true: indices.test.primaries.segments.file_sizes
+  - is_true: indices.test.primaries.segments.file_sizes.si
+  - gt: { indices.test.primaries.segments.file_sizes.si.count: 0 }
+  - gt: { indices.test.primaries.segments.file_sizes.si.size_in_bytes: 0 }
+  - gt: { indices.test.primaries.segments.file_sizes.si.min_size_in_bytes: 0 }
+  - gt: { indices.test.primaries.segments.file_sizes.si.max_size_in_bytes: 0 }
+  - gt: { indices.test.primaries.segments.file_sizes.si.average_size_in_bytes: 0 }

server/src/internalClusterTest/java/org/elasticsearch/indices/stats/IndexStatsIT.java

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.indices.stats;
 
+import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
 import org.apache.lucene.util.LuceneTestCase.SuppressCodecs;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
@@ -39,6 +40,7 @@
 import org.elasticsearch.index.MergeSchedulerConfig;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.cache.query.QueryCacheStats;
+import org.elasticsearch.index.engine.SegmentsStats;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.shard.IndexShard;
@@ -601,11 +603,23 @@ public void testSegmentsStats() {
         client().admin().indices().prepareFlush().get();
         client().admin().indices().prepareForceMerge().setMaxNumSegments(1).execute().actionGet();
         client().admin().indices().prepareRefresh().get();
-        stats = client().admin().indices().prepareStats().setSegments(true).get();
+
+        final boolean includeSegmentFileSizes = randomBoolean();
+        stats = client().admin().indices().prepareStats().setSegments(true).setIncludeSegmentFileSizes(includeSegmentFileSizes).get();
 
         assertThat(stats.getTotal().getSegments(), notNullValue());
         assertThat(stats.getTotal().getSegments().getCount(), equalTo((long) test1.totalNumShards));
         assertThat(stats.getTotal().getSegments().getMemoryInBytes(), greaterThan(0L));
+        if (includeSegmentFileSizes) {
+            assertThat(stats.getTotal().getSegments().getFiles().size(), greaterThan(0));
+            for (ObjectObjectCursor<String, SegmentsStats.FileStats> cursor : stats.getTotal().getSegments().getFiles()) {
+                assertThat(cursor.value.getExt(), notNullValue());
+                assertThat(cursor.value.getTotal(), greaterThan(0L));
+                assertThat(cursor.value.getCount(), greaterThan(0L));
+                assertThat(cursor.value.getMin(), greaterThan(0L));
+                assertThat(cursor.value.getMax(), greaterThan(0L));
+            }
+        }
     }
 
     public void testAllFlags() throws Exception {

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -25,8 +25,6 @@
 import org.apache.lucene.search.ReferenceManager;
 import org.apache.lucene.search.similarities.Similarity;
 import org.apache.lucene.store.AlreadyClosedException;
-import org.apache.lucene.store.Directory;
-import org.apache.lucene.store.IOContext;
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.Accountables;
 import org.apache.lucene.util.SetOnce;
@@ -69,10 +67,8 @@
 import org.elasticsearch.search.suggest.completion.CompletionStats;
 
 import java.io.Closeable;
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.nio.file.NoSuchFileException;
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.Comparator;
@@ -139,7 +135,7 @@ protected Engine(EngineConfig engineConfig) {
         this.store = engineConfig.getStore();
         // we use the engine class directly here to make sure all subclasses have the same logger name
         this.logger = Loggers.getLogger(Engine.class,
-                engineConfig.getShardId());
+            engineConfig.getShardId());
         this.eventListener = engineConfig.getEventListener();
     }
 
@@ -182,7 +178,7 @@ public DocsStats docStats() {
         // when indexing but not refreshing in general. Yet, if a refresh happens the internal searcher is refresh as well so we are
         // safe here.
         try (Searcher searcher = acquireSearcher("docStats", SearcherScope.INTERNAL)) {
-           return docsStats(searcher.getIndexReader());
+            return docsStats(searcher.getIndexReader());
         }
     }
 
@@ -289,12 +285,14 @@ boolean throttleLockIsHeldByCurrentThread() { // to be used in assertions and te
 
     /**
      * Returns the <code>true</code> iff this engine is currently under index throttling.
+     *
      * @see #getIndexThrottleTimeInMillis()
      */
     public abstract boolean isThrottled();
 
     /**
      * Trims translog for terms below <code>belowTerm</code> and seq# above <code>aboveSeqNo</code>
+     *
      * @see Translog#trimOperations(long, long)
      */
     public abstract void trimOperationsFromTranslog(long belowTerm, long aboveSeqNo) throws EngineException;
@@ -819,86 +817,38 @@ protected void fillSegmentStats(SegmentReader segmentReader, boolean includeSegm
         stats.addNormsMemoryInBytes(guardedRamBytesUsed(segmentReader.getNormsReader()));
         stats.addPointsMemoryInBytes(guardedRamBytesUsed(segmentReader.getPointsReader()));
         stats.addDocValuesMemoryInBytes(guardedRamBytesUsed(segmentReader.getDocValuesReader()));
-
         if (includeSegmentFileSizes) {
-            // TODO: consider moving this to StoreStats
-            stats.addFileSizes(getSegmentFileSizes(segmentReader));
+            stats.addFiles(getSegmentFileSizes(segmentReader));
         }
     }
 
-    private ImmutableOpenMap<String, Long> getSegmentFileSizes(SegmentReader segmentReader) {
-        Directory directory = null;
-        SegmentCommitInfo segmentCommitInfo = segmentReader.getSegmentInfo();
-        boolean useCompoundFile = segmentCommitInfo.info.getUseCompoundFile();
-        if (useCompoundFile) {
-            try {
-                directory = engineConfig.getCodec().compoundFormat().getCompoundReader(segmentReader.directory(),
-                    segmentCommitInfo.info, IOContext.READ);
-            } catch (IOException e) {
-                logger.warn(() -> new ParameterizedMessage("Error when opening compound reader for Directory [{}] and " +
-                    "SegmentCommitInfo [{}]", segmentReader.directory(), segmentCommitInfo), e);
-
-                return ImmutableOpenMap.of();
-            }
-        } else {
-            directory = segmentReader.directory();
-        }
-
-        assert directory != null;
-
-        String[] files;
-        if (useCompoundFile) {
-            try {
-                files = directory.listAll();
-            } catch (IOException e) {
-                final Directory finalDirectory = directory;
-                logger.warn(() ->
-                    new ParameterizedMessage("Couldn't list Compound Reader Directory [{}]", finalDirectory), e);
-                return ImmutableOpenMap.of();
-            }
-        } else {
-            try {
-                files = segmentReader.getSegmentInfo().files().toArray(new String[]{});
-            } catch (IOException e) {
-                logger.warn(() ->
-                    new ParameterizedMessage("Couldn't list Directory from SegmentReader [{}] and SegmentInfo [{}]",
-                        segmentReader, segmentReader.getSegmentInfo()), e);
-                return ImmutableOpenMap.of();
-            }
-        }
-
-        ImmutableOpenMap.Builder<String, Long> map = ImmutableOpenMap.builder();
-        for (String file : files) {
-            String extension = IndexFileNames.getExtension(file);
-            long length = 0L;
-            try {
-                length = directory.fileLength(file);
-            } catch (NoSuchFileException | FileNotFoundException e) {
-                final Directory finalDirectory = directory;
-                logger.warn(() -> new ParameterizedMessage("Tried to query fileLength but file is gone [{}] [{}]",
-                    finalDirectory, file), e);
-            } catch (IOException e) {
-                final Directory finalDirectory = directory;
-                logger.warn(() -> new ParameterizedMessage("Error when trying to query fileLength [{}] [{}]",
-                    finalDirectory, file), e);
-            }
-            if (length == 0L) {
-                continue;
-            }
-            map.put(extension, length);
-        }
-
-        if (useCompoundFile) {
-            try {
-                directory.close();
-            } catch (IOException e) {
-                final Directory finalDirectory = directory;
-                logger.warn(() -> new ParameterizedMessage("Error when closing compound reader on Directory [{}]",
-                    finalDirectory), e);
+    private ImmutableOpenMap<String, SegmentsStats.FileStats> getSegmentFileSizes(SegmentReader segmentReader) {
+        try {
+            final ImmutableOpenMap.Builder<String, SegmentsStats.FileStats> files = ImmutableOpenMap.builder();
+            final SegmentCommitInfo segmentCommitInfo = segmentReader.getSegmentInfo();
+            for (String fileName : segmentCommitInfo.files()) {
+                String fileExtension = IndexFileNames.getExtension(fileName);
+                if (fileExtension != null) {
+                    try {
+                        long fileLength = segmentReader.directory().fileLength(fileName);
+                        files.put(fileExtension, new SegmentsStats.FileStats(fileExtension, fileLength, 1L, fileLength, fileLength));
+                    } catch (IOException ioe) {
+                        logger.warn(() ->
+                            new ParameterizedMessage("Error when retrieving file length for [{}]", fileName), ioe);
+                    } catch (AlreadyClosedException ace) {
+                        logger.warn(() ->
+                            new ParameterizedMessage("Error when retrieving file length for [{}], directory is closed", fileName), ace);
+                        return ImmutableOpenMap.of();
+                    }
+                }
             }
+            return files.build();
+        } catch (IOException e) {
+            logger.warn(() ->
+                new ParameterizedMessage("Error when listing files for segment reader [{}] and segment info [{}]",
+                    segmentReader, segmentReader.getSegmentInfo()), e);
+            return ImmutableOpenMap.of();
         }
-
-        return map.build();
     }
 
     protected void writerSegmentStats(SegmentsStats stats) {

server/src/main/java/org/elasticsearch/index/engine/NoOpEngine.java

@@ -110,7 +110,7 @@ public SegmentsStats segmentsStats(boolean includeSegmentFileSizes, boolean incl
             final SegmentsStats stats = new SegmentsStats();
             stats.add(this.segmentsStats);
             if (includeSegmentFileSizes == false) {
-                stats.clearFileSizes();
+                stats.clearFiles();
             }
             return stats;
         } else {