Add num docs and size to logsdb telemetry

docs/changelog/116128.yaml

@@ -0,0 +1,5 @@
+pr: 116128
+summary: Add num docs and size to logsdb telemetry
+area: Logs
+type: enhancement
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/monitor/metrics/IndicesMetricsIT.java

@@ -11,9 +11,11 @@
 
 import org.elasticsearch.action.admin.indices.stats.CommonStatsFlags;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.mapper.OnScriptError;
 import org.elasticsearch.index.query.RangeQueryBuilder;
 import org.elasticsearch.indices.IndicesService;
@@ -329,6 +331,10 @@ public void testIndicesMetrics() {
                 equalTo(0L)
             )
         );
+
+        verifyStatsPerIndexMode(
+            Map.of(IndexMode.STANDARD, numStandardDocs, IndexMode.LOGSDB, numLogsdbDocs, IndexMode.TIME_SERIES, numTimeSeriesDocs)
+        );
     }
 
     void collectThenAssertMetrics(TestTelemetryPlugin telemetry, int times, Map<String, Matcher<Long>> matchers) {
@@ -434,6 +440,16 @@ int populateLogsdbIndices(long numIndices) {
         return totalDocs;
     }
 
+    private void verifyStatsPerIndexMode(Map<IndexMode, Long> expectedDocs) {
+        var nodes = clusterService().state().nodes().stream().toArray(DiscoveryNode[]::new);
+        var request = new IndexModeStatsActionType.StatsRequest(nodes);
+        var resp = client().execute(IndexModeStatsActionType.TYPE, request).actionGet();
+        var stats = resp.stats();
+        for (Map.Entry<IndexMode, Long> e : expectedDocs.entrySet()) {
+            assertThat(stats.get(e.getKey()).numDocs(), equalTo(e.getValue()));
+        }
+    }
+
     private Map<String, Object> parseMapping(String mapping) throws IOException {
         try (XContentParser parser = createParser(JsonXContent.jsonXContent, mapping)) {
             return parser.map();

server/src/main/java/module-info.java

@@ -470,5 +470,6 @@
             org.elasticsearch.serverless.apifiltering;
     exports org.elasticsearch.lucene.spatial;
     exports org.elasticsearch.inference.configuration;
+    exports org.elasticsearch.monitor.metrics;
 
 }

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -187,6 +187,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion QUERY_RULES_RETRIEVER = def(8_782_00_0);
     public static final TransportVersion ESQL_CCS_EXEC_INFO_WITH_FAILURES = def(8_783_00_0);
     public static final TransportVersion LOGSDB_TELEMETRY = def(8_784_00_0);
+    public static final TransportVersion LOGSDB_TELEMETRY_STATS = def(8_785_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -243,6 +243,7 @@
 import org.elasticsearch.injection.guice.AbstractModule;
 import org.elasticsearch.injection.guice.TypeLiteral;
 import org.elasticsearch.injection.guice.multibindings.MapBinder;
+import org.elasticsearch.monitor.metrics.IndexModeStatsActionType;
 import org.elasticsearch.persistent.CompletionPersistentTaskAction;
 import org.elasticsearch.persistent.RemovePersistentTaskAction;
 import org.elasticsearch.persistent.StartPersistentTaskAction;
@@ -628,6 +629,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(TransportNodesFeaturesAction.TYPE, TransportNodesFeaturesAction.class);
         actions.register(RemoteClusterNodesAction.TYPE, RemoteClusterNodesAction.TransportAction.class);
         actions.register(TransportNodesStatsAction.TYPE, TransportNodesStatsAction.class);
+        actions.register(IndexModeStatsActionType.TYPE, IndexModeStatsActionType.TransportAction.class);
         actions.register(TransportNodesUsageAction.TYPE, TransportNodesUsageAction.class);
         actions.register(TransportNodesHotThreadsAction.TYPE, TransportNodesHotThreadsAction.class);
         actions.register(TransportListTasksAction.TYPE, TransportListTasksAction.class);

server/src/main/java/org/elasticsearch/monitor/metrics/IndexModeStatsActionType.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.monitor.metrics;
+
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.nodes.BaseNodeResponse;
+import org.elasticsearch.action.support.nodes.BaseNodesRequest;
+import org.elasticsearch.action.support.nodes.BaseNodesResponse;
+import org.elasticsearch.action.support.nodes.TransportNodesAction;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+
+public final class IndexModeStatsActionType extends ActionType<IndexModeStatsActionType.StatsResponse> {
+    public static final IndexModeStatsActionType TYPE = new IndexModeStatsActionType();
+
+    private IndexModeStatsActionType() {
+        super("cluster:monitor/nodes/index_mode_stats");
+    }
+
+    public static final class StatsRequest extends BaseNodesRequest {
+        public StatsRequest(String[] nodesIds) {
+            super(nodesIds);
+        }
+
+        public StatsRequest(DiscoveryNode... concreteNodes) {
+            super(concreteNodes);
+        }
+    }
+
+    public static final class StatsResponse extends BaseNodesResponse<NodeResponse> {
+        StatsResponse(ClusterName clusterName, List<NodeResponse> nodes, List<FailedNodeException> failures) {
+            super(clusterName, nodes, failures);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            assert false : "must be local";
+            throw new UnsupportedOperationException("must be local");
+        }
+
+        @Override
+        protected List<NodeResponse> readNodesFrom(StreamInput in) throws IOException {
+            assert false : "must be local";
+            throw new UnsupportedOperationException("must be local");
+        }
+
+        @Override
+        protected void writeNodesTo(StreamOutput out, List<NodeResponse> nodes) throws IOException {
+            assert false : "must be local";
+            throw new UnsupportedOperationException("must be local");
+        }
+
+        public Map<IndexMode, IndexStats> stats() {
+            final Map<IndexMode, IndexStats> stats = new EnumMap<>(IndexMode.class);
+            for (IndexMode mode : IndexMode.values()) {
+                stats.put(mode, new IndexStats());
+            }
+            for (NodeResponse node : getNodes()) {
+                for (Map.Entry<IndexMode, IndexStats> e : node.stats.entrySet()) {
+                    stats.get(e.getKey()).add(e.getValue());
+                }
+            }
+            return stats;
+        }
+    }
+
+    public static final class NodeRequest extends TransportRequest {
+        NodeRequest() {
+
+        }
+
+        NodeRequest(StreamInput in) throws IOException {
+            super(in);
+        }
+    }
+
+    public static class NodeResponse extends BaseNodeResponse {
+        private final Map<IndexMode, IndexStats> stats;
+
+        NodeResponse(DiscoveryNode node, Map<IndexMode, IndexStats> stats) {
+            super(node);
+            this.stats = stats;
+        }
+
+        NodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
+            super(in, node);
+            stats = in.readMap(IndexMode::readFrom, IndexStats::new);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeMap(stats, (o, m) -> IndexMode.writeTo(m, o), (o, s) -> s.writeTo(o));
+        }
+    }
+
+    public static class TransportAction extends TransportNodesAction<StatsRequest, StatsResponse, NodeRequest, NodeResponse, Void> {
+        private final IndicesService indicesService;
+
+        @Inject
+        public TransportAction(
+            ClusterService clusterService,
+            TransportService transportService,
+            ActionFilters actionFilters,
+            IndicesService indicesService
+        ) {
+            super(
+                TYPE.name(),
+                clusterService,
+                transportService,
+                actionFilters,
+                NodeRequest::new,
+                transportService.getThreadPool().executor(ThreadPool.Names.MANAGEMENT)
+            );
+            this.indicesService = indicesService;
+        }
+
+        @Override
+        protected StatsResponse newResponse(StatsRequest request, List<NodeResponse> nodeResponses, List<FailedNodeException> failures) {
+            return new StatsResponse(ClusterName.DEFAULT, nodeResponses, failures);
+        }
+
+        @Override
+        protected NodeRequest newNodeRequest(StatsRequest request) {
+            return new NodeRequest();
+        }
+
+        @Override
+        protected NodeResponse newNodeResponse(StreamInput in, DiscoveryNode node) throws IOException {
+            return new NodeResponse(in, node);
+        }
+
+        @Override
+        protected NodeResponse nodeOperation(NodeRequest request, Task task) {
+            return new NodeResponse(clusterService.localNode(), IndicesMetrics.getStatsWithoutCache(indicesService));
+        }
+    }
+}

server/src/main/java/org/elasticsearch/monitor/metrics/IndexStats.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.monitor.metrics;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.index.search.stats.SearchStats;
+import org.elasticsearch.index.shard.IndexingStats;
+
+import java.io.IOException;
+
+public final class IndexStats implements Writeable {
+    int numIndices = 0;
+    long numDocs = 0;
+    long numBytes = 0;
+    SearchStats.Stats search = new SearchStats().getTotal();
+    IndexingStats.Stats indexing = new IndexingStats().getTotal();
+
+    IndexStats() {
+
+    }
+
+    IndexStats(StreamInput in) throws IOException {
+        this.numIndices = in.readVInt();
+        this.numDocs = in.readVLong();
+        this.numBytes = in.readVLong();
+        this.search = SearchStats.Stats.readStats(in);
+        this.indexing = new IndexingStats.Stats(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeVInt(numIndices);
+        out.writeVLong(numDocs);
+        out.writeVLong(numBytes);
+        search.writeTo(out);
+        indexing.writeTo(out);
+    }
+
+    void add(IndexStats other) {
+        this.numIndices += other.numIndices;
+        this.numDocs += other.numDocs;
+        this.numBytes += other.numBytes;
+        this.search.add(other.search);
+        this.indexing.add(other.indexing);
+    }
+
+    public int numIndices() {
+        return numIndices;
+    }
+
+    public long numDocs() {
+        return numDocs;
+    }
+
+    public long numBytes() {
+        return numBytes;
+    }
+}