Add composite-engine stats and parquet ingest rejection/pool metrics

sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/BaseStatsIT.java

@@ -35,6 +35,14 @@ protected Map<String, Object> luceneIndexStats(String index, String... queryPara
         return StatsITHelpers.luceneIndexStats(getRestClient(), index, queryParams);
     }
 
+    protected Map<String, Object> compositeIndexStats(String index, String... queryParams) throws IOException {
+        return StatsITHelpers.compositeIndexStats(getRestClient(), index, queryParams);
+    }
+
+    protected Map<String, Object> compositeNodeStats(String nodeIdOrEmpty, String... queryParams) throws IOException {
+        return StatsITHelpers.compositeNodeStats(getRestClient(), nodeIdOrEmpty, queryParams);
+    }
+
     protected Map<String, Object> parquetNodeStats(String nodeIdOrEmpty, String... queryParams) throws IOException {
         return StatsITHelpers.parquetNodeStats(getRestClient(), nodeIdOrEmpty, queryParams);
     }

sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeStatsEndpointIT.java

@@ -0,0 +1,116 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.composite;
+
+import org.opensearch.test.OpenSearchIntegTestCase.ClusterScope;
+import org.opensearch.test.OpenSearchIntegTestCase.Scope;
+
+import java.util.Map;
+
+/**
+ * Integration tests for the composite-engine per-format stats endpoint
+ * ({@code /_plugins/composite/...}) and the parquet {@code native_write_rejections} counter.
+ *
+ * @opensearch.experimental
+ */
+@ClusterScope(scope = Scope.SUITE, numDataNodes = 1)
+public class CompositeStatsEndpointIT extends BaseStatsIT {
+
+    /** Fresh composite index: endpoint responds and all counters are zero. */
+    public void testCompositeStatsZeroOnFreshIndex() throws Exception {
+        String idx = "composite-zero-idx";
+        createCompositeIndex(idx, true);
+
+        Map<String, Object> c = compositeIndexStats(idx);
+        assertCounter("fresh refresh_total", c, "indices." + idx + ".refresh.refresh_total", 0L);
+        assertCounter("fresh refresh_merge_total", c, "indices." + idx + ".refresh.refresh_merge_total", 0L);
+        assertCounter("fresh refresh_merge_failures", c, "indices." + idx + ".refresh.refresh_merge_failures", 0L);
+        assertCounter("fresh merge_total", c, "indices." + idx + ".merge.merge_total", 0L);
+        assertCounter("fresh merge_failures", c, "indices." + idx + ".merge.merge_failures", 0L);
+        assertCounter("fresh write_total", c, "indices." + idx + ".write.write_total", 0L);
+        assertCounter("fresh write_primary_failures", c, "indices." + idx + ".write.write_primary_failures", 0L);
+        assertCounter("fresh write_secondary_failures", c, "indices." + idx + ".write.write_secondary_failures", 0L);
+        assertCounter("fresh mapping_update_executed_total", c, "indices." + idx + ".mapping.mapping_update_executed_total", 0L);
+    }
+
+    /** After indexing + refresh, composite refresh counters increment and failures stay zero. */
+    public void testCompositeRefreshAndMergeCountersIncrement() throws Exception {
+        String idx = "composite-refresh-idx";
+        createCompositeIndex(idx, true);
+        indexDocs(idx, 100, 0);
+        refreshIndex(idx);
+
+        Map<String, Object> c = compositeIndexStats(idx);
+        assertCounterAtLeast("refresh_total", c, "indices." + idx + ".refresh.refresh_total", 1L);
+        assertCounterAtLeast("refresh_time_millis", c, "indices." + idx + ".refresh.refresh_time_millis", 0L);
+        // write_total counts every indexed doc attempt; 100 docs were indexed.
+        assertCounterAtLeast("write_total", c, "indices." + idx + ".write.write_total", 100L);
+        // Happy path: no write or merge failures.
+        assertCounter("write_primary_failures", c, "indices." + idx + ".write.write_primary_failures", 0L);
+        assertCounter("write_secondary_failures", c, "indices." + idx + ".write.write_secondary_failures", 0L);
+        assertCounter("merge_failures", c, "indices." + idx + ".merge.merge_failures", 0L);
+        assertCounter("refresh_merge_failures", c, "indices." + idx + ".refresh.refresh_merge_failures", 0L);
+
+        // refresh_merge_total is the merge-on-refresh subset of merge_total — it must never exceed it.
+        long refreshMerge = StatsITHelpers.getCounter(c, "indices." + idx + ".refresh.refresh_merge_total");
+        long mergeTotal = StatsITHelpers.getCounter(c, "indices." + idx + ".merge.merge_total");
+        assertTrue(
+            "refresh_merge_total (" + refreshMerge + ") must not exceed merge_total (" + mergeTotal + ")",
+            refreshMerge <= mergeTotal
+        );
+    }
+
+    /** The composite per-node endpoint aggregates and responds after indexing. */
+    public void testCompositeNodeStatsEndpoint() throws Exception {
+        String idx = "composite-node-idx";
+        createCompositeIndex(idx, true);
+        indexDocs(idx, 50, 0);
+        refreshIndex(idx);
+
+        Map<String, Object> n = compositeNodeStats("");
+        // A successful per-node response carries a "nodes" object (one entry per responding node).
+        assertTrue("per-node response must contain a 'nodes' object", n.get("nodes") instanceof Map);
+        assertFalse("per-node response 'nodes' object must not be empty", ((Map<?, ?>) n.get("nodes")).isEmpty());
+    }
+
+    /** Parquet native_write_rejections is wired and reads zero on the happy path. */
+    public void testParquetNativeWriteRejectionsZeroOnHappyPath() throws Exception {
+        String idx = "parquet-rejection-idx";
+        createCompositeIndex(idx, true);
+        indexDocs(idx, 100, 0);
+        refreshIndex(idx);
+
+        Map<String, Object> p = parquetIndexStats(idx);
+        assertCounter("native_write_rejections zero", p, "indices." + idx + ".native_write.native_write_rejections", 0L);
+    }
+
+    /** The parquet per-node endpoint exposes the live native_ingest_pool block (queue/active/rejected). */
+    public void testParquetNodeStatsExposesIngestPool() throws Exception {
+        String idx = "parquet-ingest-pool-idx";
+        createCompositeIndex(idx, true);
+        indexDocs(idx, 50, 0);
+        refreshIndex(idx);
+
+        Map<String, Object> n = parquetNodeStats("");
+        // Each responding node must carry a native_ingest_pool block with the pool fields.
+        Map<?, ?> nodes = (Map<?, ?>) n.get("nodes");
+        assertNotNull("per-node response must contain 'nodes'", nodes);
+        assertFalse("'nodes' must not be empty", nodes.isEmpty());
+        boolean sawPool = false;
+        for (Object node : nodes.values()) {
+            Object pool = ((Map<?, ?>) node).get("native_ingest_pool");
+            if (pool instanceof Map) {
+                assertTrue("native_ingest_pool must report queue_depth", ((Map<?, ?>) pool).containsKey("queue_depth"));
+                assertTrue("native_ingest_pool must report rejected", ((Map<?, ?>) pool).containsKey("rejected"));
+                sawPool = true;
+            }
+        }
+        assertTrue("at least one node must expose native_ingest_pool", sawPool);
+    }
+}

sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/StatsITHelpers.java

@@ -41,6 +41,17 @@ static Map<String, Object> luceneIndexStats(RestClient rest, String index, Strin
         return fetchStats(rest, "/_plugins/lucene/" + index + "/_stats", queryParams);
     }
 
+    static Map<String, Object> compositeIndexStats(RestClient rest, String index, String... queryParams) throws IOException {
+        return fetchStats(rest, "/_plugins/composite/" + index + "/_stats", queryParams);
+    }
+
+    static Map<String, Object> compositeNodeStats(RestClient rest, String nodeIdOrEmpty, String... queryParams) throws IOException {
+        String path = nodeIdOrEmpty.isEmpty()
+            ? "/_plugins/composite/_nodes/_stats"
+            : "/_plugins/composite/_nodes/" + nodeIdOrEmpty + "/_stats";
+        return fetchStats(rest, path, queryParams);
+    }
+
     static Map<String, Object> parquetNodeStats(RestClient rest, String nodeIdOrEmpty, String... queryParams) throws IOException {
         String path = nodeIdOrEmpty.isEmpty() ? "/_plugins/parquet/_nodes/_stats" : "/_plugins/parquet/_nodes/" + nodeIdOrEmpty + "/_stats";
         return fetchStats(rest, path, queryParams);

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormat.java

@@ -26,6 +26,9 @@
 @ExperimentalApi
 public class CompositeDataFormat extends DataFormat {
 
+    /** Canonical format name for the composite engine. */
+    public static final String COMPOSITE_FORMAT_NAME = "composite";
+
     private final DataFormat primaryDataFormat;
     private final List<DataFormat> dataFormats;
 
@@ -68,7 +71,7 @@ public DataFormat getPrimaryDataFormat() {
 
     @Override
     public String name() {
-        return "composite";
+        return COMPOSITE_FORMAT_NAME;
     }
 
     @Override

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java

@@ -10,13 +10,25 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.action.ActionRequest;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
+import org.opensearch.cluster.node.DiscoveryNodes;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.ValidationException;
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.IndexScopedSettings;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.settings.SettingsFilter;
+import org.opensearch.composite.stats.CompositeStatsProvider;
+import org.opensearch.composite.stats.transport.CompositeNodeStatsActionType;
+import org.opensearch.composite.stats.transport.CompositeNodeStatsRestAction;
+import org.opensearch.composite.stats.transport.CompositeNodeStatsTransportAction;
+import org.opensearch.composite.stats.transport.CompositeStatsActionType;
+import org.opensearch.composite.stats.transport.CompositeStatsRestAction;
+import org.opensearch.composite.stats.transport.CompositeStatsTransportAction;
+import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.common.unit.ByteSizeUnit;
 import org.opensearch.core.common.unit.ByteSizeValue;
@@ -39,10 +51,13 @@
 import org.opensearch.indices.IndexCreationException;
 import org.opensearch.indices.IndicesService;
 import org.opensearch.plugin.stats.DataFormatStatsProviderRegistry;
+import org.opensearch.plugins.ActionPlugin;
 import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.MapperPlugin;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.rest.RestController;
+import org.opensearch.rest.RestHandler;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.client.Client;
@@ -96,7 +111,7 @@
  * @opensearch.experimental
  */
 @ExperimentalApi
-public class CompositeDataFormatPlugin extends Plugin implements DataFormatPlugin, ExtensiblePlugin, MapperPlugin {
+public class CompositeDataFormatPlugin extends Plugin implements DataFormatPlugin, ExtensiblePlugin, MapperPlugin, ActionPlugin {
 
     private static final Logger logger = LogManager.getLogger(CompositeDataFormatPlugin.class);
 
@@ -214,9 +229,34 @@ public Collection<Object> createComponents(
         Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
         this.clusterService = clusterService;
+        // Eagerly construct the provider so the registry is populated before the engine and
+        // transport-action layers attempt lookups. The engine self-registers its per-shard
+        // tracker via CompositeStatsProvider.getInstance() on construction.
+        new CompositeStatsProvider();
         return Collections.emptyList();
     }
 
+    @Override
+    public List<ActionPlugin.ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
+        return List.of(
+            new ActionPlugin.ActionHandler<>(CompositeStatsActionType.INSTANCE, CompositeStatsTransportAction.class),
+            new ActionPlugin.ActionHandler<>(CompositeNodeStatsActionType.INSTANCE, CompositeNodeStatsTransportAction.class)
+        );
+    }
+
+    @Override
+    public List<RestHandler> getRestHandlers(
+        Settings settings,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster
+    ) {
+        return List.of(new CompositeStatsRestAction(), new CompositeNodeStatsRestAction());
+    }
+
     /**
      * Stamps the cluster-scope defaults for {@link #PRIMARY_DATA_FORMAT} and
      * {@link #SECONDARY_DATA_FORMATS} into newly created indices when those index-level settings

sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java

@@ -14,6 +14,9 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.composite.merge.CompositeMerger;
+import org.opensearch.composite.stats.CompositeShardStatsTracker;
+import org.opensearch.composite.stats.CompositeStatsProvider;
+import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.engine.dataformat.DataFormat;
 import org.opensearch.index.engine.dataformat.DataFormatPlugin;
@@ -37,6 +40,7 @@
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.store.FormatChecksumStrategy;
 import org.opensearch.index.store.Store;
+import org.opensearch.plugin.stats.StatsRecorder;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -79,6 +83,8 @@ public class CompositeIndexingExecutionEngine implements IndexingExecutionEngine
     private final Committer committer;
     private final IndexSettings indexSettings;
     private final CompositeMerger merger;
+    private final CompositeShardStatsTracker statsTracker = new CompositeShardStatsTracker();
+    private final ShardId shardId;
     private volatile Map<String, Collection<String>> pendingDeletes = new ConcurrentHashMap<>();
 
     /**
@@ -151,6 +157,27 @@ public CompositeIndexingExecutionEngine(
         this.committer = committer;
         this.indexSettings = indexSettings;
         this.merger = new CompositeMerger(this, compositeDataFormat);
+        this.shardId = store != null ? store.shardId() : null;
+
+        // Register the per-shard tracker so REST endpoints can read live counters; unregistered
+        // in close(). Rolls back the registration if anything below throws, to avoid leaking it.
+        CompositeStatsProvider provider = CompositeStatsProvider.getInstance();
+        boolean registered = false;
+        try {
+            if (provider != null && shardId != null) {
+                provider.register(shardId, statsTracker);
+                registered = true;
+            }
+        } catch (Throwable t) {
+            if (registered) {
+                try {
+                    provider.unregister(shardId);
+                } catch (Throwable rollbackErr) {
+                    logger.warn("Failed to unregister composite stats tracker during constructor rollback", rollbackErr);
+                }
+            }
+            throw t;
+        }
     }
 
     /**
@@ -231,6 +258,12 @@ public Exception getTragicException() {
      */
     @Override
     public RefreshResult refresh(RefreshInput refreshInput) throws IOException {
+        // recordTimeMillis owns the whole-refresh timing; incRefreshTotal counts every refresh.
+        statsTracker.incRefreshTotal();
+        return StatsRecorder.recordTimeMillis(() -> doRefresh(refreshInput), statsTracker::addRefreshTimeMillis);
+    }
+
+    private RefreshResult doRefresh(RefreshInput refreshInput) throws IOException {
         tryDeletePendingFiles();
 
         // All per-format engines refresh normally (primary passes through, secondary does addIndexes)
@@ -269,8 +302,14 @@ public RefreshResult refresh(RefreshInput refreshInput) throws IOException {
             if (onlyNew.size() > 1) {
                 try {
                     final long mergeStartNanos = System.nanoTime();
-                    MergeResult mergeResult = merger.merge(
-                        MergeInput.builder().segments(onlyNew).newWriterGeneration(refreshInput.nextAvailableGeneration()).build()
+                    // Counts merge-on-refresh attempts; a subset overlay of merge_total (also
+                    // incremented inside CompositeMerger.merge()).
+                    statsTracker.incRefreshMergeTotal();
+                    MergeResult mergeResult = StatsRecorder.recordTimeMillis(
+                        () -> merger.merge(
+                            MergeInput.builder().segments(onlyNew).newWriterGeneration(refreshInput.nextAvailableGeneration()).build()
+                        ),
+                        statsTracker::addRefreshMergeTimeMillis
                     );
 
                     if (mergeResult != null) {
@@ -316,6 +355,7 @@ public RefreshResult refresh(RefreshInput refreshInput) throws IOException {
                 } catch (Exception e) {
                     // Merge-on-refresh is best-effort. On failure, fall back to normal per-writer
                     // segments. Background merge will consolidate them later.
+                    statsTracker.incRefreshMergeFailures();
                     logger.warn("merge-on-refresh failed, falling back to per-writer segments", e);
                 }
             }
@@ -478,11 +518,20 @@ public CompositeDocumentInput newDocumentInput() {
      */
     @Override
     public void close() throws IOException {
+        CompositeStatsProvider provider = CompositeStatsProvider.getInstance();
+        if (provider != null && shardId != null) {
+            provider.unregister(shardId);
+        }
         IOUtils.closeWhileHandlingException(primaryEngine);
         secondaryEngines.forEach(IOUtils::closeWhileHandlingException);
         IOUtils.closeWhileHandlingException(committer);
     }
 
+    /** Returns this shard's composite stats tracker, used by the writer and merger to count. */
+    public CompositeShardStatsTracker statsTracker() {
+        return statsTracker;
+    }
+
     /**
      * Returns the primary delegate engine.
      *