WIP: Index size records

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

@@ -146,7 +146,7 @@ public final EngineConfig config() {
         return engineConfig;
     }
 
-    protected abstract SegmentInfos getLastCommittedSegmentInfos();
+    public abstract SegmentInfos getLastCommittedSegmentInfos();
 
     public MergeStats getMergeStats() {
         return new MergeStats();

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

@@ -2478,7 +2478,7 @@ protected boolean maybeFailEngine(String source, Exception e) {
     }
 
     @Override
-    protected SegmentInfos getLastCommittedSegmentInfos() {
+    public SegmentInfos getLastCommittedSegmentInfos() {
         return lastCommittedSegmentInfos;
     }
 

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

@@ -287,7 +287,7 @@ protected ReferenceManager<ElasticsearchDirectoryReader> getReferenceManager(Sea
     }
 
     @Override
-    protected SegmentInfos getLastCommittedSegmentInfos() {
+    public SegmentInfos getLastCommittedSegmentInfos() {
         return lastCommittedSegmentInfos;
     }
 

x-pack/plugin/metering-ix/build.gradle

@@ -0,0 +1,24 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+apply plugin: 'elasticsearch.internal-es-plugin'
+
+esplugin {
+  name 'metering-ix'
+  description 'Metering IX'
+  classname 'org.elasticsearch.xpack.meter.ix.MeterIXPlugin'
+  extendedPlugins = ['x-pack-core']
+}
+
+base {
+  archivesName = 'x-pack-monitoring'
+}
+
+dependencies {
+  compileOnly project(":server")
+  compileOnly project(path: xpackModule('core'))
+}

x-pack/plugin/metering-ix/src/main/java/org/elasticsearch/xpack/meter/ix/MeterIXPlugin.java

@@ -0,0 +1,50 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.meter.ix;
+
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.routing.allocation.AllocationService;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.script.ScriptService;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.tracing.Tracer;
+import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Supplier;
+
+public class MeterIXPlugin extends Plugin {
+    @Override
+    public Collection<Object> createComponents(
+        Client client,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ResourceWatcherService resourceWatcherService,
+        ScriptService scriptService,
+        NamedXContentRegistry xContentRegistry,
+        Environment environment,
+        NodeEnvironment nodeEnvironment,
+        NamedWriteableRegistry namedWriteableRegistry,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<RepositoriesService> repositoriesServiceSupplier,
+        Tracer tracer,
+        AllocationService allocationService,
+        IndicesService indicesService
+    ) {
+        return List.of(new MeterIXPoller(threadPool, indicesService));
+    }
+}

x-pack/plugin/metering-ix/src/main/java/org/elasticsearch/xpack/meter/ix/MeterIXPoller.java

@@ -0,0 +1,153 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.meter.ix;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.lucene.index.SegmentCommitInfo;
+import org.apache.lucene.util.StringHelper;
+import org.elasticsearch.common.component.Lifecycle;
+import org.elasticsearch.common.component.LifecycleComponent;
+import org.elasticsearch.common.component.LifecycleListener;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.IndexService;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.threadpool.Scheduler;
+import org.elasticsearch.threadpool.ThreadPool;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class MeterIXPoller implements LifecycleComponent {
+    private static final Logger logger = LogManager.getLogger(MeterIXPoller.class);
+    final ThreadPool threadPool;
+    final IndicesService indicesService;
+    Lifecycle.State state = Lifecycle.State.INITIALIZED;
+    Scheduler.ScheduledCancellable next = null;
+    private final ReentrantLock mutex = new ReentrantLock();
+
+    public MeterIXPoller(ThreadPool threadPool, IndicesService indicesService) {
+        this.threadPool = threadPool;
+        this.indicesService = indicesService;
+    }
+
+    @Override
+    public void close() {
+        mutex.lock();
+        try {
+            if (state != Lifecycle.State.STOPPED && state != Lifecycle.State.CLOSED) {
+                state = Lifecycle.State.CLOSED;
+                if (next != null) {
+                    next.cancel();
+                }
+                next = null;
+            }
+        } finally {
+            mutex.unlock();
+        }
+    }
+
+    @Override
+    public Lifecycle.State lifecycleState() {
+        return null;
+    }
+
+    @Override
+    public void addLifecycleListener(LifecycleListener listener) {
+
+    }
+
+    @Override
+    public void start() {
+        mutex.lock();
+        try {
+            if (state == Lifecycle.State.INITIALIZED) {
+                state = Lifecycle.State.STARTED;
+                setScheduleNext();
+            }
+        } finally {
+            mutex.unlock();
+        }
+    }
+
+    @Override
+    public void stop() {
+        mutex.lock();
+        try {
+            if (state != Lifecycle.State.STOPPED && state != Lifecycle.State.CLOSED) {
+                state = Lifecycle.State.STOPPED;
+                if (next != null) {
+                    next.cancel();
+                }
+                next = null;
+            }
+        } finally {
+            mutex.unlock();
+        }
+    }
+
+    private void setScheduleNext() {
+        next = threadPool.schedule(this::logAndSchedule, TimeValue.timeValueSeconds(20), ThreadPool.Names.GENERIC);
+    }
+
+    public void logAndSchedule() {
+        logIndexStats();
+        mutex.lock();
+        try {
+            if (state == Lifecycle.State.STARTED) {
+                setScheduleNext();
+            }
+        } finally {
+            mutex.unlock();
+        }
+    }
+
+    public void logIndexStats() {
+        logger.warn("[IX] logIndexStats");
+        collectRecords();
+        logger.warn("[IX] done logIndexStats");
+    }
+
+    public List<IXRecord> collectRecords() {
+        List<IXRecord> records = new ArrayList<>();
+        for (final IndexService indexService : indicesService) {
+            String index = indexService.index().getName();
+            for (final IndexShard shard : indexService) {
+                Engine engine = shard.getEngineOrNull();
+                if (engine == null) {
+                    continue;
+                }
+                long term = shard.getOperationPrimaryTerm();
+                int shardId = shard.shardId().id();
+                for (final SegmentCommitInfo commitInfo : engine.getLastCommittedSegmentInfos()) {
+                    try {
+                        var record = new IXRecord(
+                            index,
+                            shardId,
+                            term,
+                            commitInfo.getDocValuesGen(),
+                            StringHelper.idToString(commitInfo.getId()),
+                            commitInfo.sizeInBytes()
+                        );
+                        records.add(record);
+                        logger.info("{}", record);
+                    } catch (IOException err) {
+                        logger.warn("error: [{}], index: [{}], shard: [{}], id: [{}]", err, index, shardId, commitInfo.getId());
+                    }
+                }
+            }
+        }
+        return records;
+    }
+
+    private record IXRecord(String index, int shardId, long term, long generation, String commitId, long size) {}
+}