Move native_allocator under native_memory; replace PluginNodeStats SPI with typed NodeStats field (opensearch-project#21820)

* Move native_allocator under native_memory; replace PluginNodeStats SPI

Consolidates Arrow native allocator stats under
nodes.<id>.native_memory.native_allocator (was a top-level sibling at
nodes.<id>.native_allocator) and replaces the experimental
PluginNodeStats SPI with a typed NativeAllocatorPoolStats field on
NodeStats.

Strips peak_bytes, child_count, and humanReadableField byte-string
duplicates from the allocator JSON. Pool and root blocks now expose
only allocated_bytes and limit_bytes.

Adds new metric: GET /_nodes/stats/native_allocator
Removes metric:  GET /_nodes/stats/plugin_stats (no production consumer)

PluginNodeStats and Plugin.nodeStats() were @experimentalapi with one
production consumer (arrow-base) and an unreleased wire format
(Version.V_3_7_0). Replaced with a small ArrowAllocatorPlugin SPI in
libs/arrow-spi so server can inject the allocator into NodeService
without depending on the arrow-base plugin.

Signed-off-by: Gaurav Singh <snghsvn@amazon.com>

Author: gaurav-amz

libs/arrow-spi/src/main/java/org/opensearch/arrow/spi/NativeAllocator.java

@@ -52,11 +52,6 @@ public interface NativeAllocator extends Closeable {
      */
     void setRootLimit(long limit);
 
-    /**
-     * Collects a point-in-time stats snapshot across all pools.
-     */
-    NativeAllocatorPoolStats stats();
-
     /**
      * Opaque handle to a memory pool. Plugins downcast to the concrete type
      * (e.g., Arrow's {@code BufferAllocator}) in the implementation layer.

plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowBasePlugin.java

@@ -20,9 +20,10 @@
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
 import org.opensearch.node.resource.tracker.ResourceTrackerSettings;
+import org.opensearch.plugin.stats.NativeAllocatorPoolStats;
+import org.opensearch.plugin.stats.NativeAllocatorStatsRegistry;
 import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.Plugin;
-import org.opensearch.plugins.PluginNodeStats;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ThreadPool;
@@ -252,7 +253,15 @@ public Collection<Object> createComponents(
         ClusterSettings cs = clusterService.getClusterSettings();
         ArrowNativeAllocator built = buildAllocator(settings, cs);
         this.allocator = built;
-        return List.of(built);
+        // Publish a NativeAllocatorStatsRegistry alongside the allocator so the server-side
+        // NodeService can discover the supplier via pluginComponents (instanceof filter) without
+        // taking a compile-time dependency on this plugin. The lambda re-reads `this.allocator`
+        // each invocation, so after close() nulls the field, the supplier returns null cleanly.
+        Supplier<NativeAllocatorPoolStats> statsSupplier = () -> {
+            ArrowNativeAllocator a = this.allocator;
+            return a != null ? a.stats() : null;
+        };
+        return List.of(built, new NativeAllocatorStatsRegistry(statsSupplier));
     }
 
     /**
@@ -347,21 +356,6 @@ public List<Setting<?>> getSettings() {
         );
     }
 
-    @Override
-    public List<PluginNodeStats> nodeStats() {
-        if (allocator == null) {
-            return List.of();
-        }
-        return List.of(new NativeAllocatorPluginStats(allocator.stats()));
-    }
-
-    @Override
-    public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
-        return List.of(
-            new NamedWriteableRegistry.Entry(PluginNodeStats.class, NativeAllocatorPluginStats.NAME, NativeAllocatorPluginStats::new)
-        );
-    }
-
     private static void validateMinMax(String poolName, long min, long max) {
         if (min > max) {
             throw new IllegalArgumentException("Pool '" + poolName + "' min (" + min + ") exceeds max (" + max + ")");

plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/ArrowNativeAllocator.java

@@ -11,7 +11,7 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.opensearch.arrow.spi.NativeAllocator;
-import org.opensearch.arrow.spi.NativeAllocatorPoolStats;
+import org.opensearch.plugin.stats.NativeAllocatorPoolStats;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -176,7 +176,12 @@ public void setRootLimit(long limit) {
         root.setLimit(limit);
     }
 
-    @Override
+    /**
+     * Returns a point-in-time stats snapshot across all pools. Used by the
+     * {@code NativeAllocatorStatsRegistry} component published from
+     * {@code ArrowBasePlugin.createComponents()} and wired into {@code NodeService} to
+     * render allocator state under {@code _nodes/stats[/native_allocator]}.
+     */
     public NativeAllocatorPoolStats stats() {
         List<NativeAllocatorPoolStats.PoolStats> poolStats = new ArrayList<>();
         for (var entry : pools.entrySet()) {
@@ -186,8 +191,7 @@ public NativeAllocatorPoolStats stats() {
                     entry.getKey(),
                     alloc.getAllocatedMemory(),
                     alloc.getPeakMemoryAllocation(),
-                    alloc.getLimit(),
-                    alloc.getChildAllocators().size()
+                    alloc.getLimit()
                 )
             );
         }

plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/NativeAllocatorPluginStats.java

@@ -10,7 +10,7 @@
 
 import org.apache.arrow.memory.BufferAllocator;
 import org.opensearch.arrow.spi.NativeAllocator;
-import org.opensearch.arrow.spi.NativeAllocatorPoolStats;
+import org.opensearch.plugin.stats.NativeAllocatorPoolStats;
 import org.opensearch.test.OpenSearchTestCase;
 
 public class ArrowNativeAllocatorTests extends OpenSearchTestCase {
@@ -86,7 +86,8 @@ public void testStats() {
         assertEquals("stats-pool", poolStats.getName());
         assertEquals(64 * 1024 * 1024, poolStats.getLimitBytes());
         assertEquals(0, poolStats.getAllocatedBytes());
-        assertEquals(0, poolStats.getChildCount());
+        // child_count is no longer rendered in stats; getPoolAllocator(...).getChildAllocators()
+        // is the runtime accessor for that detail if needed.
     }
 
     public void testStatsMultiplePools() {