diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/BaseStatsIT.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/BaseStatsIT.java index e64450a22599f..a83b02e9f4055 100644 --- a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/BaseStatsIT.java +++ b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/BaseStatsIT.java @@ -35,6 +35,14 @@ protected Map luceneIndexStats(String index, String... queryPara return StatsITHelpers.luceneIndexStats(getRestClient(), index, queryParams); } + protected Map compositeIndexStats(String index, String... queryParams) throws IOException { + return StatsITHelpers.compositeIndexStats(getRestClient(), index, queryParams); + } + + protected Map compositeNodeStats(String nodeIdOrEmpty, String... queryParams) throws IOException { + return StatsITHelpers.compositeNodeStats(getRestClient(), nodeIdOrEmpty, queryParams); + } + protected Map parquetNodeStats(String nodeIdOrEmpty, String... queryParams) throws IOException { return StatsITHelpers.parquetNodeStats(getRestClient(), nodeIdOrEmpty, queryParams); } diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeStatsEndpointIT.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/CompositeStatsEndpointIT.java new file mode 100644 index 0000000000000..5916365c9289a --- /dev/null +++ b/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 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 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 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 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 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); + } +} diff --git a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/StatsITHelpers.java b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/StatsITHelpers.java index a7146c1c90da5..b29af138a5e78 100644 --- a/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/StatsITHelpers.java +++ b/sandbox/plugins/composite-engine/src/internalClusterTest/java/org/opensearch/composite/StatsITHelpers.java @@ -41,6 +41,17 @@ static Map luceneIndexStats(RestClient rest, String index, Strin return fetchStats(rest, "/_plugins/lucene/" + index + "/_stats", queryParams); } + static Map compositeIndexStats(RestClient rest, String index, String... queryParams) throws IOException { + return fetchStats(rest, "/_plugins/composite/" + index + "/_stats", queryParams); + } + + static Map 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 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); diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormat.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormat.java index b474121550ef7..9b47cf58090c4 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormat.java +++ b/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 dataFormats; @@ -68,7 +71,7 @@ public DataFormat getPrimaryDataFormat() { @Override public String name() { - return "composite"; + return COMPOSITE_FORMAT_NAME; } @Override diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java index 19a38dcdaa454..d339da4abad8a 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDataFormatPlugin.java +++ b/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 createComponents( Supplier 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> getActions() { + return List.of( + new ActionPlugin.ActionHandler<>(CompositeStatsActionType.INSTANCE, CompositeStatsTransportAction.class), + new ActionPlugin.ActionHandler<>(CompositeNodeStatsActionType.INSTANCE, CompositeNodeStatsTransportAction.class) + ); + } + + @Override + public List getRestHandlers( + Settings settings, + RestController restController, + ClusterSettings clusterSettings, + IndexScopedSettings indexScopedSettings, + SettingsFilter settingsFilter, + IndexNameExpressionResolver indexNameExpressionResolver, + Supplier 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 diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java index 47ece1f7d36ea..5aba20d57ceb8 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java +++ b/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> 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. * diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java index 8999d645bce81..fe627f5c6271a 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java @@ -13,6 +13,7 @@ import org.apache.logging.log4j.message.ParameterizedMessage; import org.opensearch.common.annotation.ExperimentalApi; import org.opensearch.common.util.io.IOUtils; +import org.opensearch.composite.stats.CompositeShardStatsTracker; import org.opensearch.index.engine.dataformat.DataFormat; import org.opensearch.index.engine.dataformat.DocumentInput; import org.opensearch.index.engine.dataformat.FileInfos; @@ -55,6 +56,7 @@ class CompositeWriter implements Writer { private final Map>> secondaryWritersByFormat; private final long writerGeneration; private final FailureHandlerStrategy failureHandler; + private final CompositeShardStatsTracker statsTracker; private volatile boolean closed; private long mappingVersion; /** Successful addDoc count — every incoming rowId must equal this. */ @@ -103,6 +105,7 @@ class CompositeWriter implements Writer { } this.secondaryWritersByFormat = Collections.unmodifiableMap(secondaries); this.failureHandler = new FailureHandlerStrategy(); + this.statsTracker = engine.statsTracker(); } @Override @@ -116,11 +119,15 @@ public WriteResult addDoc(CompositeDocumentInput doc) throws IOException { throw new IllegalStateException("rowId [" + doc.getRowId() + "] does not match accepted row count [" + acceptedRows + "]"); } + // Count every write attempt so write_*_failures can be read as a rate. + statsTracker.incWriteTotal(); + // Roll back exactly the writers we've called addDoc on, in order. List>> touched = new ArrayList<>(); touched.add(primaryWriter); WriteResult primaryResult = primaryWriter.addDoc(doc.getPrimaryInput()); if (primaryResult instanceof WriteResult.Failure pf) { + statsTracker.incWritePrimaryFailures(); logger.warn( () -> new ParameterizedMessage("Failed to add document in primary format [{}], rolling back", primaryFormat.name()), pf.cause() @@ -136,6 +143,7 @@ public WriteResult addDoc(CompositeDocumentInput doc) throws IOException { touched.add(writer); WriteResult result = writer.addDoc(inputEntry.getValue()); if (result instanceof WriteResult.Failure sf) { + statsTracker.incWriteSecondaryFailures(); logger.warn( () -> new ParameterizedMessage("Failed to add document in secondary format [{}], rolling back", format.name()), sf.cause() @@ -212,6 +220,7 @@ public long mappingVersion() { @Override public void updateMappingVersion(long newVersion) { if (newVersion > this.mappingVersion) { + statsTracker.incMappingUpdateExecutedTotal(); this.mappingVersion = newVersion; primaryWriter.updateMappingVersion(newVersion); for (Writer w : secondaryWritersByFormat.values()) { diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/merge/CompositeMerger.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/merge/CompositeMerger.java index b32d50a1368f1..5c10c282ff0ce 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/merge/CompositeMerger.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/merge/CompositeMerger.java @@ -11,6 +11,7 @@ import org.opensearch.common.annotation.ExperimentalApi; import org.opensearch.composite.CompositeDataFormat; import org.opensearch.composite.CompositeIndexingExecutionEngine; +import org.opensearch.composite.stats.CompositeShardStatsTracker; import org.opensearch.index.engine.dataformat.DataFormat; import org.opensearch.index.engine.dataformat.IndexingExecutionEngine; import org.opensearch.index.engine.dataformat.MergeInput; @@ -18,6 +19,7 @@ import org.opensearch.index.engine.dataformat.Merger; import org.opensearch.index.engine.exec.Segment; import org.opensearch.index.engine.exec.WriterFileSet; +import org.opensearch.plugin.stats.StatsRecorder; import java.io.IOException; import java.util.ArrayList; @@ -40,18 +42,23 @@ public class CompositeMerger implements Merger { private final DataFormat primaryFormat; private final List secondaryFormats; private final CompositeMergeExecutor executor; + private final CompositeShardStatsTracker statsTracker; public CompositeMerger(CompositeIndexingExecutionEngine engine, CompositeDataFormat compositeDataFormat) { this.primaryFormat = compositeDataFormat.getPrimaryDataFormat(); this.secondaryFormats = resolveSecondaryFormats(compositeDataFormat, primaryFormat); this.executor = new CompositeMergeExecutor(buildMergerMap(engine)); + this.statsTracker = engine.statsTracker(); } @Override public MergeResult merge(MergeInput mergeInput) throws IOException { - Map> filesByFormat = extractFilesByFormat(mergeInput.segments()); - MergePlan plan = new MergePlan(mergeInput.newWriterGeneration(), primaryFormat, secondaryFormats, filesByFormat); - return executor.execute(plan); + // recordOutcome: time always, merge_total on success, merge_failures on throw. + return StatsRecorder.recordOutcome(() -> { + Map> filesByFormat = extractFilesByFormat(mergeInput.segments()); + MergePlan plan = new MergePlan(mergeInput.newWriterGeneration(), primaryFormat, secondaryFormats, filesByFormat); + return executor.execute(plan); + }, statsTracker::addMergeTimeMillis, statsTracker::incMergeTotal, statsTracker::incMergeFailures); } private Map> extractFilesByFormat(List segments) { diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeShardStats.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeShardStats.java new file mode 100644 index 0000000000000..3d520c752e5d3 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeShardStats.java @@ -0,0 +1,167 @@ +/* + * 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.stats; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.plugin.stats.DataFormatShardStats; + +import java.io.IOException; + +/** + * Immutable point-in-time snapshot of shard-level composite-engine statistics. + * Produced by {@link CompositeShardStatsTracker#stats()}. + * + * @opensearch.experimental + */ +@ExperimentalApi +public class CompositeShardStats implements DataFormatShardStats { + + // Refresh (engine-level, orchestrating all formats) + the merge-on-refresh breakdown. + private final long refreshTotal; + private final long refreshTimeMillis; + private final long refreshMergeTotal; + private final long refreshMergeTimeMillis; + private final long refreshMergeFailures; + + // Merge (standalone CompositeMerger path). + private final long mergeTotal; + private final long mergeTimeMillis; + private final long mergeFailures; + + // Write attempts + failures, split by primary vs secondary format. + private final long writeTotal; + private final long writePrimaryFailures; + private final long writeSecondaryFailures; + + // Dynamic mapping updates that were actually applied (newVersion > current), not no-op calls. + private final long mappingUpdateExecutedTotal; + + /** + * Returns an empty snapshot with all zero counters. Used by transport actions when a + * shard has no composite engine. + */ + public static CompositeShardStats empty() { + return new CompositeShardStats(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0); + } + + /** Constructs a snapshot with all values. */ + public CompositeShardStats( + long refreshTotal, + long refreshTimeMillis, + long refreshMergeTotal, + long refreshMergeTimeMillis, + long refreshMergeFailures, + long mergeTotal, + long mergeTimeMillis, + long mergeFailures, + long writeTotal, + long writePrimaryFailures, + long writeSecondaryFailures, + long mappingUpdateExecutedTotal + ) { + this.refreshTotal = refreshTotal; + this.refreshTimeMillis = refreshTimeMillis; + this.refreshMergeTotal = refreshMergeTotal; + this.refreshMergeTimeMillis = refreshMergeTimeMillis; + this.refreshMergeFailures = refreshMergeFailures; + this.mergeTotal = mergeTotal; + this.mergeTimeMillis = mergeTimeMillis; + this.mergeFailures = mergeFailures; + this.writeTotal = writeTotal; + this.writePrimaryFailures = writePrimaryFailures; + this.writeSecondaryFailures = writeSecondaryFailures; + this.mappingUpdateExecutedTotal = mappingUpdateExecutedTotal; + } + + public CompositeShardStats(StreamInput in) throws IOException { + this.refreshTotal = in.readVLong(); + this.refreshTimeMillis = in.readVLong(); + this.refreshMergeTotal = in.readVLong(); + this.refreshMergeTimeMillis = in.readVLong(); + this.refreshMergeFailures = in.readVLong(); + this.mergeTotal = in.readVLong(); + this.mergeTimeMillis = in.readVLong(); + this.mergeFailures = in.readVLong(); + this.writeTotal = in.readVLong(); + this.writePrimaryFailures = in.readVLong(); + this.writeSecondaryFailures = in.readVLong(); + this.mappingUpdateExecutedTotal = in.readVLong(); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeVLong(refreshTotal); + out.writeVLong(refreshTimeMillis); + out.writeVLong(refreshMergeTotal); + out.writeVLong(refreshMergeTimeMillis); + out.writeVLong(refreshMergeFailures); + out.writeVLong(mergeTotal); + out.writeVLong(mergeTimeMillis); + out.writeVLong(mergeFailures); + out.writeVLong(writeTotal); + out.writeVLong(writePrimaryFailures); + out.writeVLong(writeSecondaryFailures); + out.writeVLong(mappingUpdateExecutedTotal); + } + + @Override + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { + // Refresh — includes the merge-on-refresh breakdown. + builder.startObject("refresh"); + builder.field("refresh_total", refreshTotal); + builder.field("refresh_time_millis", refreshTimeMillis); + builder.field("refresh_merge_total", refreshMergeTotal); + builder.field("refresh_merge_time_millis", refreshMergeTimeMillis); + builder.field("refresh_merge_failures", refreshMergeFailures); + builder.endObject(); + + // Merge — standalone merger path. + builder.startObject("merge"); + builder.field("merge_total", mergeTotal); + builder.field("merge_time_millis", mergeTimeMillis); + builder.field("merge_failures", mergeFailures); + builder.endObject(); + + // Write attempts + failures by format role. + builder.startObject("write"); + builder.field("write_total", writeTotal); + builder.field("write_primary_failures", writePrimaryFailures); + builder.field("write_secondary_failures", writeSecondaryFailures); + builder.endObject(); + + // Dynamic mapping updates actually applied. + builder.startObject("mapping"); + builder.field("mapping_update_executed_total", mappingUpdateExecutedTotal); + builder.endObject(); + + return builder; + } + + /** Returns a new snapshot that is the element-wise sum of this and another. */ + @Override + public CompositeShardStats add(CompositeShardStats other) { + return new CompositeShardStats( + this.refreshTotal + other.refreshTotal, + this.refreshTimeMillis + other.refreshTimeMillis, + this.refreshMergeTotal + other.refreshMergeTotal, + this.refreshMergeTimeMillis + other.refreshMergeTimeMillis, + this.refreshMergeFailures + other.refreshMergeFailures, + this.mergeTotal + other.mergeTotal, + this.mergeTimeMillis + other.mergeTimeMillis, + this.mergeFailures + other.mergeFailures, + this.writeTotal + other.writeTotal, + this.writePrimaryFailures + other.writePrimaryFailures, + this.writeSecondaryFailures + other.writeSecondaryFailures, + this.mappingUpdateExecutedTotal + other.mappingUpdateExecutedTotal + ); + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeShardStatsTracker.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeShardStatsTracker.java new file mode 100644 index 0000000000000..d8bcd8a447d2d --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeShardStatsTracker.java @@ -0,0 +1,118 @@ +/* + * 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.stats; + +import org.opensearch.common.annotation.ExperimentalApi; + +import java.util.concurrent.atomic.LongAdder; + +/** + * Mutable, thread-safe shard-level statistics tracker for the composite engine. + * Uses {@link LongAdder} for high-throughput counters. Call {@link #stats()} for an + * immutable {@link CompositeShardStats} snapshot. + * + * @opensearch.experimental + */ +@ExperimentalApi +public class CompositeShardStatsTracker { + + // Refresh + merge-on-refresh breakdown. + private final LongAdder refreshTotal = new LongAdder(); + private final LongAdder refreshTimeMillis = new LongAdder(); + private final LongAdder refreshMergeTotal = new LongAdder(); + private final LongAdder refreshMergeTimeMillis = new LongAdder(); + private final LongAdder refreshMergeFailures = new LongAdder(); + + // Standalone merge. + private final LongAdder mergeTotal = new LongAdder(); + private final LongAdder mergeTimeMillis = new LongAdder(); + private final LongAdder mergeFailures = new LongAdder(); + + // Write attempts + failures by format role. + private final LongAdder writeTotal = new LongAdder(); + private final LongAdder writePrimaryFailures = new LongAdder(); + private final LongAdder writeSecondaryFailures = new LongAdder(); + + // Dynamic mapping updates actually applied. + private final LongAdder mappingUpdateExecutedTotal = new LongAdder(); + + /** Returns an immutable point-in-time snapshot of all tracked statistics. */ + public CompositeShardStats stats() { + return new CompositeShardStats( + refreshTotal.sum(), + refreshTimeMillis.sum(), + refreshMergeTotal.sum(), + refreshMergeTimeMillis.sum(), + refreshMergeFailures.sum(), + mergeTotal.sum(), + mergeTimeMillis.sum(), + mergeFailures.sum(), + writeTotal.sum(), + writePrimaryFailures.sum(), + writeSecondaryFailures.sum(), + mappingUpdateExecutedTotal.sum() + ); + } + + // --- Refresh --- + + public void incRefreshTotal() { + refreshTotal.increment(); + } + + public void addRefreshTimeMillis(long ms) { + refreshTimeMillis.add(ms); + } + + public void incRefreshMergeTotal() { + refreshMergeTotal.increment(); + } + + public void addRefreshMergeTimeMillis(long ms) { + refreshMergeTimeMillis.add(ms); + } + + public void incRefreshMergeFailures() { + refreshMergeFailures.increment(); + } + + // --- Merge --- + + public void incMergeTotal() { + mergeTotal.increment(); + } + + public void addMergeTimeMillis(long ms) { + mergeTimeMillis.add(ms); + } + + public void incMergeFailures() { + mergeFailures.increment(); + } + + // --- Write --- + + public void incWriteTotal() { + writeTotal.increment(); + } + + public void incWritePrimaryFailures() { + writePrimaryFailures.increment(); + } + + public void incWriteSecondaryFailures() { + writeSecondaryFailures.increment(); + } + + // --- Mapping --- + + public void incMappingUpdateExecutedTotal() { + mappingUpdateExecutedTotal.increment(); + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeStatsProvider.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeStatsProvider.java new file mode 100644 index 0000000000000..2f34992549663 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/CompositeStatsProvider.java @@ -0,0 +1,100 @@ +/* + * 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.stats; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.composite.CompositeDataFormat; +import org.opensearch.core.common.io.stream.Writeable; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.plugin.stats.DataFormatStatsProvider; +import org.opensearch.plugin.stats.DataFormatStatsProviderRegistry; + +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Composite-engine implementation of {@link DataFormatStatsProvider}. + * + *

Maintains a per-shard registry of {@link CompositeShardStatsTracker} instances. + * {@code CompositeIndexingExecutionEngine} self-registers on construction and unregisters + * on close. The plugin's REST + transport classes read stats from this provider. + * + *

Singleton pattern: a static {@code INSTANCE} is set during plugin construction so the + * engine/writer/merger layers can reach the tracker without DI plumbing. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeStatsProvider implements DataFormatStatsProvider { + + public static final String FORMAT_NAME = CompositeDataFormat.COMPOSITE_FORMAT_NAME; + + private static volatile CompositeStatsProvider INSTANCE; + + private final Map trackers = new ConcurrentHashMap<>(); + + public CompositeStatsProvider() { + // First instance wins. Subsequent constructions are no-ops on the singleton slot. + if (INSTANCE == null) { + INSTANCE = this; + } + DataFormatStatsProviderRegistry.INSTANCE.register(this); + } + + /** Returns the singleton, or {@code null} if the plugin has not been constructed yet. */ + public static CompositeStatsProvider getInstance() { + return INSTANCE; + } + + /** Registers a tracker for a shard. Called from the composite engine's constructor. */ + public void register(ShardId shardId, CompositeShardStatsTracker tracker) { + trackers.put(shardId, tracker); + } + + /** Unregisters a tracker. Called from the engine's {@code close()}. */ + public void unregister(ShardId shardId) { + trackers.remove(shardId); + } + + /** Returns the tracker for a shard, or {@code null} if none — used by writer/merger to count. */ + public CompositeShardStatsTracker getTracker(ShardId shardId) { + return trackers.get(shardId); + } + + // --- DataFormatStatsProvider --- + + @Override + public String formatName() { + return FORMAT_NAME; + } + + @Override + public Optional shardStats(ShardId shardId) { + CompositeShardStatsTracker tracker = trackers.get(shardId); + return tracker == null ? Optional.empty() : Optional.of(tracker.stats()); + } + + @Override + public Optional aggregateNodeStats() { + if (trackers.isEmpty()) { + return Optional.empty(); + } + CompositeShardStats agg = CompositeShardStats.empty(); + for (CompositeShardStatsTracker t : trackers.values()) { + agg = agg.add(t.stats()); + } + return Optional.of(agg); + } + + @Override + public Writeable.Reader shardStatsReader() { + return CompositeShardStats::new; + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/package-info.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/package-info.java new file mode 100644 index 0000000000000..9fc811468697e --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/package-info.java @@ -0,0 +1,14 @@ +/* + * 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. + */ + +/** + * Statistics collection for the composite engine plugin. + * + * @opensearch.experimental + */ +package org.opensearch.composite.stats; diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsActionType.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsActionType.java new file mode 100644 index 0000000000000..b4aa957ea1b34 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsActionType.java @@ -0,0 +1,30 @@ +/* + * 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.stats.transport; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.composite.stats.CompositeShardStats; +import org.opensearch.composite.stats.CompositeStatsProvider; +import org.opensearch.plugin.stats.transport.FormatNodeStatsActionType; + +/** + * Action type for the composite per-node stats endpoint + * ({@code GET /_plugins/composite/_nodes/[{nodeId}/]_stats}). + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeNodeStatsActionType extends FormatNodeStatsActionType { + + public static final CompositeNodeStatsActionType INSTANCE = new CompositeNodeStatsActionType(); + + private CompositeNodeStatsActionType() { + super(CompositeStatsProvider.FORMAT_NAME, CompositeShardStats::new); + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsRestAction.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsRestAction.java new file mode 100644 index 0000000000000..bc1b395c8de93 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsRestAction.java @@ -0,0 +1,34 @@ +/* + * 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.stats.transport; + +import org.opensearch.action.ActionType; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.composite.stats.CompositeStatsProvider; +import org.opensearch.plugin.stats.transport.BaseFormatNodeStatsRestAction; +import org.opensearch.plugin.stats.transport.FormatNodeStatsResponse; + +/** + * REST handler for {@code GET /_plugins/composite/_nodes/[{nodeId}/]_stats}. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeNodeStatsRestAction extends BaseFormatNodeStatsRestAction { + + @Override + protected String formatName() { + return CompositeStatsProvider.FORMAT_NAME; + } + + @Override + protected ActionType> actionType() { + return CompositeNodeStatsActionType.INSTANCE; + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsTransportAction.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsTransportAction.java new file mode 100644 index 0000000000000..6016cbe508804 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeNodeStatsTransportAction.java @@ -0,0 +1,46 @@ +/* + * 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.stats.transport; + +import org.opensearch.action.support.ActionFilters; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.common.inject.Inject; +import org.opensearch.composite.stats.CompositeShardStats; +import org.opensearch.composite.stats.CompositeStatsProvider; +import org.opensearch.plugin.stats.transport.BaseTransportFormatNodeStatsAction; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; + +/** + * Per-node stats transport action for the composite engine. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeNodeStatsTransportAction extends BaseTransportFormatNodeStatsAction { + + @Inject + public CompositeNodeStatsTransportAction( + ThreadPool threadPool, + ClusterService clusterService, + TransportService transportService, + ActionFilters actionFilters + ) { + super( + CompositeNodeStatsActionType.INSTANCE.name(), + CompositeStatsProvider.FORMAT_NAME, + CompositeShardStats::new, + threadPool, + clusterService, + transportService, + actionFilters + ); + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsActionType.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsActionType.java new file mode 100644 index 0000000000000..ea3dd7b99df0a --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsActionType.java @@ -0,0 +1,30 @@ +/* + * 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.stats.transport; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.composite.stats.CompositeShardStats; +import org.opensearch.composite.stats.CompositeStatsProvider; +import org.opensearch.plugin.stats.transport.FormatStatsActionType; + +/** + * Action type for the composite per-index stats endpoint + * ({@code GET /_plugins/composite/{index}/_stats}). + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeStatsActionType extends FormatStatsActionType { + + public static final CompositeStatsActionType INSTANCE = new CompositeStatsActionType(); + + private CompositeStatsActionType() { + super(CompositeStatsProvider.FORMAT_NAME, CompositeShardStats::new); + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsRestAction.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsRestAction.java new file mode 100644 index 0000000000000..a9bd710d70baa --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsRestAction.java @@ -0,0 +1,34 @@ +/* + * 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.stats.transport; + +import org.opensearch.action.ActionType; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.composite.stats.CompositeStatsProvider; +import org.opensearch.plugin.stats.transport.BaseFormatStatsRestAction; +import org.opensearch.plugin.stats.transport.FormatStatsResponse; + +/** + * REST handler for {@code GET /_plugins/composite/{index}/_stats}. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeStatsRestAction extends BaseFormatStatsRestAction { + + @Override + protected String formatName() { + return CompositeStatsProvider.FORMAT_NAME; + } + + @Override + protected ActionType> actionType() { + return CompositeStatsActionType.INSTANCE; + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsTransportAction.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsTransportAction.java new file mode 100644 index 0000000000000..aee0871c3c339 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/CompositeStatsTransportAction.java @@ -0,0 +1,49 @@ +/* + * 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.stats.transport; + +import org.opensearch.action.support.ActionFilters; +import org.opensearch.cluster.metadata.IndexNameExpressionResolver; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.common.inject.Inject; +import org.opensearch.composite.stats.CompositeShardStats; +import org.opensearch.composite.stats.CompositeStatsProvider; +import org.opensearch.plugin.stats.transport.BaseTransportFormatStatsAction; +import org.opensearch.transport.TransportService; + +/** + * Per-index stats transport action for the composite engine. + * + *

All broadcast/aggregation logic lives in {@link BaseTransportFormatStatsAction}; this + * class supplies the action name, format name, and the {@link CompositeShardStats} reader. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class CompositeStatsTransportAction extends BaseTransportFormatStatsAction { + + @Inject + public CompositeStatsTransportAction( + ClusterService clusterService, + TransportService transportService, + ActionFilters actionFilters, + IndexNameExpressionResolver indexNameExpressionResolver + ) { + super( + CompositeStatsActionType.INSTANCE.name(), + CompositeStatsProvider.FORMAT_NAME, + CompositeShardStats::new, + clusterService, + transportService, + actionFilters, + indexNameExpressionResolver + ); + } +} diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/package-info.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/package-info.java new file mode 100644 index 0000000000000..5ae87b4bdec61 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/stats/transport/package-info.java @@ -0,0 +1,22 @@ +/* + * 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. + */ + +/** + * Transport and REST actions for the composite per-format statistics endpoints. + * + *

Wires {@link org.opensearch.composite.stats.CompositeStatsProvider} into the + * {@code /_plugins/composite/{index}/_stats} and {@code /_plugins/composite/_nodes/_stats} + * REST + transport surfaces by extending the abstract bases in + * {@link org.opensearch.plugin.stats.transport}. + * + * @opensearch.experimental + */ +@ExperimentalApi +package org.opensearch.composite.stats.transport; + +import org.opensearch.common.annotation.ExperimentalApi; diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/merge/CompositeMergerTests.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/merge/CompositeMergerTests.java index 7ea86d22eaae1..dfa984277a39f 100644 --- a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/merge/CompositeMergerTests.java +++ b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/merge/CompositeMergerTests.java @@ -14,6 +14,7 @@ import org.opensearch.common.settings.Settings; import org.opensearch.composite.CompositeDataFormat; import org.opensearch.composite.CompositeIndexingExecutionEngine; +import org.opensearch.composite.stats.CompositeShardStatsTracker; import org.opensearch.core.index.Index; import org.opensearch.core.index.shard.ShardId; import org.opensearch.index.IndexSettings; @@ -79,6 +80,7 @@ public void setUp() throws Exception { IndexingExecutionEngine secondaryEngine = mockEngine(secondaryFormat, secondaryMerger); compositeEngine = mock(CompositeIndexingExecutionEngine.class); + when(compositeEngine.statsTracker()).thenReturn(new CompositeShardStatsTracker()); doReturn(primaryEngine).when(compositeEngine).getPrimaryDelegate(); doReturn(Set.of(secondaryEngine)).when(compositeEngine).getSecondaryDelegates(); when(compositeEngine.getNextWriterGeneration()).thenReturn(99L); @@ -118,6 +120,7 @@ public void testDoMergeSuccessWithPrimaryAndSecondary() throws IOException { public void testDoMergePrimaryOnlyNoSecondaries() throws IOException { CompositeIndexingExecutionEngine engineNoSecondary = mock(CompositeIndexingExecutionEngine.class); + when(engineNoSecondary.statsTracker()).thenReturn(new CompositeShardStatsTracker()); IndexingExecutionEngine primaryEngine = mockEngine(primaryFormat, primaryMerger); doReturn(primaryEngine).when(engineNoSecondary).getPrimaryDelegate(); doReturn(Set.of()).when(engineNoSecondary).getSecondaryDelegates(); @@ -193,6 +196,7 @@ public void testDoMergeMultipleSecondariesFailsFastOnFirstError() throws IOExcep Merger secondaryMerger2 = mock(Merger.class); CompositeIndexingExecutionEngine multiEngine = mock(CompositeIndexingExecutionEngine.class); + when(multiEngine.statsTracker()).thenReturn(new CompositeShardStatsTracker()); IndexingExecutionEngine primaryEngine = mockEngine(primaryFormat, primaryMerger); doReturn(primaryEngine).when(multiEngine).getPrimaryDelegate(); doReturn(Set.of(mockEngine(secondaryFormat, secondaryMerger), mockEngine(secondaryFormat2, secondaryMerger2))).when(multiEngine) @@ -354,6 +358,7 @@ public void testDoMergeSkipsSecondaryThatEqualsPrimary() throws IOException { IndexingExecutionEngine duplicateEngine = mockEngine(primaryFormat, primaryMerger); CompositeIndexingExecutionEngine dupEngine = mock(CompositeIndexingExecutionEngine.class); + when(dupEngine.statsTracker()).thenReturn(new CompositeShardStatsTracker()); doReturn(primaryEngine).when(dupEngine).getPrimaryDelegate(); doReturn(Set.of(duplicateEngine)).when(dupEngine).getSecondaryDelegates(); when(dupEngine.getNextWriterGeneration()).thenReturn(99L); diff --git a/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/stats/CompositeShardStatsTests.java b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/stats/CompositeShardStatsTests.java new file mode 100644 index 0000000000000..d4eda10758ad9 --- /dev/null +++ b/sandbox/plugins/composite-engine/src/test/java/org/opensearch/composite/stats/CompositeShardStatsTests.java @@ -0,0 +1,112 @@ +/* + * 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.stats; + +import org.opensearch.common.io.stream.BytesStreamOutput; +import org.opensearch.common.xcontent.XContentFactory; +import org.opensearch.common.xcontent.XContentHelper; +import org.opensearch.common.xcontent.json.JsonXContent; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.xcontent.ToXContent; +import org.opensearch.test.OpenSearchTestCase; + +import java.util.Map; + +/** + * Unit tests for {@link CompositeShardStats} and {@link CompositeShardStatsTracker}: + * counter increments, snapshot fidelity, aggregation, and wire/XContent round-trips. + */ +public class CompositeShardStatsTests extends OpenSearchTestCase { + + public void testTrackerCountersFlowIntoSnapshot() { + CompositeShardStatsTracker tracker = new CompositeShardStatsTracker(); + tracker.incRefreshTotal(); + tracker.addRefreshTimeMillis(10); + tracker.incRefreshMergeTotal(); + tracker.addRefreshMergeTimeMillis(4); + tracker.incRefreshMergeFailures(); + tracker.incMergeTotal(); + tracker.addMergeTimeMillis(7); + tracker.incMergeFailures(); + tracker.incWriteTotal(); + tracker.incWritePrimaryFailures(); + tracker.incWriteSecondaryFailures(); + tracker.incMappingUpdateExecutedTotal(); + + CompositeShardStats s = tracker.stats(); + Map json = toMap(s); + + assertEquals(1L, get(json, "refresh.refresh_total")); + assertEquals(10L, get(json, "refresh.refresh_time_millis")); + assertEquals(1L, get(json, "refresh.refresh_merge_total")); + assertEquals(4L, get(json, "refresh.refresh_merge_time_millis")); + assertEquals(1L, get(json, "refresh.refresh_merge_failures")); + assertEquals(1L, get(json, "merge.merge_total")); + assertEquals(7L, get(json, "merge.merge_time_millis")); + assertEquals(1L, get(json, "merge.merge_failures")); + assertEquals(1L, get(json, "write.write_total")); + assertEquals(1L, get(json, "write.write_primary_failures")); + assertEquals(1L, get(json, "write.write_secondary_failures")); + assertEquals(1L, get(json, "mapping.mapping_update_executed_total")); + } + + public void testEmptyIsAllZero() { + Map json = toMap(CompositeShardStats.empty()); + assertEquals(0L, get(json, "refresh.refresh_total")); + assertEquals(0L, get(json, "merge.merge_failures")); + assertEquals(0L, get(json, "write.write_primary_failures")); + assertEquals(0L, get(json, "mapping.mapping_update_executed_total")); + } + + public void testAddSumsAllCounters() { + CompositeShardStats a = new CompositeShardStats(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12); + CompositeShardStats b = new CompositeShardStats(10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120); + Map json = toMap(a.add(b)); + + assertEquals(11L, get(json, "refresh.refresh_total")); + assertEquals(33L, get(json, "refresh.refresh_merge_total")); + assertEquals(55L, get(json, "refresh.refresh_merge_failures")); + assertEquals(66L, get(json, "merge.merge_total")); + assertEquals(88L, get(json, "merge.merge_failures")); + assertEquals(99L, get(json, "write.write_total")); + assertEquals(110L, get(json, "write.write_primary_failures")); + assertEquals(132L, get(json, "mapping.mapping_update_executed_total")); + } + + public void testStreamRoundTrip() throws Exception { + CompositeShardStats original = new CompositeShardStats(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12); + BytesStreamOutput out = new BytesStreamOutput(); + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + CompositeShardStats restored = new CompositeShardStats(in); + assertEquals(toMap(original), toMap(restored)); + } + } + + private static Map toMap(CompositeShardStats stats) { + try { + var builder = XContentFactory.jsonBuilder().startObject(); + stats.toXContent(builder, ToXContent.EMPTY_PARAMS); + builder.endObject(); + return XContentHelper.convertToMap(JsonXContent.jsonXContent, builder.toString(), true); + } catch (Exception e) { + throw new AssertionError(e); + } + } + + @SuppressWarnings("unchecked") + private static long get(Map json, String dotted) { + String[] parts = dotted.split("\\."); + Object cur = json; + for (String p : parts) { + cur = ((Map) cur).get(p); + } + return ((Number) cur).longValue(); + } +} diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java index 5831078a12bd2..272100284a5e2 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/ParquetDataFormatPlugin.java @@ -94,6 +94,8 @@ public class ParquetDataFormatPlugin extends Plugin implements DataFormatPlugin, /** Thread pool name for background native Parquet writes during VSR rotation. */ public static final String PARQUET_THREAD_POOL_NAME = "parquet_native_write"; + + public static final int PARQUET_THREAD_POOL_QUEUE_SIZE = 10_000; private static final StoreStrategy storeStrategy = new ParquetStoreStrategy(); public static final ParquetDataFormat PARQUET_DATA_FORMAT = new ParquetDataFormat(); /** Initialized to EMPTY to avoid NPE if indexingEngine() is called before createComponents(). */ @@ -121,6 +123,11 @@ public Collection createComponents( ) { this.settings = clusterService.getSettings(); this.threadPool = threadPool; + // Hand the node thread pool to the stats provider so per-node stats can read the live + // parquet_native_write pool (queue depth / active / rejected). + if (ParquetStatsProvider.getInstance() != null) { + ParquetStatsProvider.getInstance().setThreadPool(threadPool); + } this.nativeAllocator = pluginComponentRegistry.getComponent(ArrowNativeAllocator.class).orElse(null); // Initialize native write/merge memory pools @@ -233,7 +240,7 @@ public List> getExecutorBuilders(Settings settings) { settings, PARQUET_THREAD_POOL_NAME, OpenSearchExecutors.allocatedProcessors(settings), - -1, + PARQUET_THREAD_POOL_QUEUE_SIZE, "thread_pool." + PARQUET_THREAD_POOL_NAME ) ); diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetIngestPoolStats.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetIngestPoolStats.java new file mode 100644 index 0000000000000..febb61684892b --- /dev/null +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetIngestPoolStats.java @@ -0,0 +1,92 @@ +/* + * 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.parquet.stats; + +import org.opensearch.common.annotation.ExperimentalApi; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.common.io.stream.Writeable; +import org.opensearch.core.xcontent.ToXContentFragment; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.threadpool.ThreadPoolStats; + +import java.io.IOException; + +/** + * Live, node-level snapshot of the {@code parquet_native_write} thread pool that backs Parquet + * background ingestion writes. Surfaces queue saturation and rejection pressure. + * + *

Node-level only: rendered under the {@code native_ingest_pool} block of the per-node stats + * aggregate, never on a per-shard payload. + * + * @opensearch.experimental + */ +@ExperimentalApi +public final class ParquetIngestPoolStats implements Writeable, ToXContentFragment { + + private final int threads; + private final int queueDepth; + private final int active; + private final long rejected; + private final long completed; + + public ParquetIngestPoolStats(int threads, int queueDepth, int active, long rejected, long completed) { + this.threads = threads; + this.queueDepth = queueDepth; + this.active = active; + this.rejected = rejected; + this.completed = completed; + } + + public ParquetIngestPoolStats(StreamInput in) throws IOException { + this.threads = in.readVInt(); + this.queueDepth = in.readVInt(); + this.active = in.readVInt(); + this.rejected = in.readVLong(); + this.completed = in.readVLong(); + } + + /** + * Builds a snapshot from a {@link ThreadPoolStats.Stats} reading of the {@code parquet_native_write} + * pool, or {@code null} if {@code stats} is {@code null} (pool not found). + */ + public static ParquetIngestPoolStats from(ThreadPoolStats.Stats stats) { + if (stats == null) { + return null; + } + return new ParquetIngestPoolStats( + stats.getThreads(), + stats.getQueue(), + stats.getActive(), + stats.getRejected(), + stats.getCompleted() + ); + } + + @Override + public void writeTo(StreamOutput out) throws IOException { + out.writeVInt(threads); + out.writeVInt(queueDepth); + out.writeVInt(active); + out.writeVLong(rejected); + out.writeVLong(completed); + } + + @Override + public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { + builder.startObject("native_ingest_pool"); + builder.field("threads", threads); + builder.field("queue_depth", queueDepth); + builder.field("active", active); + builder.field("rejected", rejected); + builder.field("completed", completed); + builder.endObject(); + return builder; + } +} diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetShardStats.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetShardStats.java index dc0ca757ec6d7..2e643340d0d97 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetShardStats.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/stats/ParquetShardStats.java @@ -40,6 +40,8 @@ public class ParquetShardStats implements DataFormatShardStats trackers = new ConcurrentHashMap<>(); + // Node-level thread pool backing parquet ingestion writes. Set post-construction (the pool + // isn't available when the provider is built in createGuiceModules). May be null in tests. + private volatile org.opensearch.threadpool.ThreadPool threadPool; + public ParquetStatsProvider() { // First instance wins. Subsequent constructions are no-ops on the singleton slot. if (INSTANCE == null) { @@ -61,6 +65,11 @@ public static ParquetStatsProvider getInstance() { return INSTANCE; } + /** Sets the node thread pool used to read live {@code parquet_native_write} pool stats. */ + public void setThreadPool(org.opensearch.threadpool.ThreadPool threadPool) { + this.threadPool = threadPool; + } + /** Registers a tracker for a shard. Called from {@code ParquetIndexingEngine}'s constructor. */ public void register(ShardId shardId, ParquetShardStatsTracker tracker) { trackers.put(shardId, tracker); @@ -102,19 +111,38 @@ public Optional aggregateNodeStats() { for (ParquetShardStatsTracker t : trackers.values()) { agg = agg.add(t.stats()); } - // Attach native runtime metrics — only at node level. If the FFM/JNI bridge is - // unavailable (e.g., test infra without lib loaded) we still return the per-shard - // aggregate; missing runtime metrics shouldn't break the whole stats request. - // Catching Exception (not Throwable) so we still propagate JVM-fatal Errors. + // Node-level decoration: the live parquet_native_write pool snapshot + the Rust runtime + // metrics. Both are best-effort — a null pool or an unavailable native bridge simply + // leaves that block off rather than failing the whole stats request. + ParquetIngestPoolStats ingestPool = collectIngestPoolStats(); + ParquetNativeRuntimeStats runtime = null; + try { + runtime = ParquetNativeRuntimeStats.fromArray(org.opensearch.parquet.bridge.RustBridge.collectRuntimeMetrics()); + } catch (Exception e) { + logger.warn("Failed to collect native runtime metrics; node stats will omit the native_runtime block", e); + } + return Optional.of(agg.withNodeStats(runtime, ingestPool)); + } + + /** + * Reads the live {@code parquet_native_write} pool stats from the node thread pool, or returns + * {@code null} if the thread pool is unavailable or the pool is not present. + */ + private ParquetIngestPoolStats collectIngestPoolStats() { + org.opensearch.threadpool.ThreadPool tp = threadPool; + if (tp == null) { + return null; + } try { - ParquetNativeRuntimeStats runtime = ParquetNativeRuntimeStats.fromArray( - org.opensearch.parquet.bridge.RustBridge.collectRuntimeMetrics() - ); - return Optional.of(agg.withNativeRuntime(runtime)); + for (org.opensearch.threadpool.ThreadPoolStats.Stats s : tp.stats()) { + if (org.opensearch.parquet.ParquetDataFormatPlugin.PARQUET_THREAD_POOL_NAME.equals(s.getName())) { + return ParquetIngestPoolStats.from(s); + } + } } catch (Exception e) { - logger.warn("Failed to collect native runtime metrics; returning per-shard aggregate without runtime block", e); - return Optional.of(agg); + logger.warn("Failed to collect parquet ingest pool stats", e); } + return null; } @Override diff --git a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java index 9e1c994cb1c71..ea65e1ad798b5 100644 --- a/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java +++ b/sandbox/plugins/parquet-data-format/src/main/java/org/opensearch/parquet/vsr/VSRManager.java @@ -14,6 +14,7 @@ import org.apache.arrow.vector.types.pojo.Schema; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; import org.opensearch.index.IndexSettings; import org.opensearch.index.engine.dataformat.DocumentInput; import org.opensearch.index.engine.dataformat.RowIdMapping; @@ -320,7 +321,13 @@ public void maybeRotateActiveVSR() throws IOException { vsrPool.completeVSR(frozenVSR); vsrPool.unsetFrozenVSR(); }; - pendingWrite = threadPool.executor(vsrRotationThread).submit(writeTask); + try { + pendingWrite = threadPool.executor(vsrRotationThread).submit(writeTask); + } catch (OpenSearchRejectedExecutionException e) { + // Pool saturated — count the rejection and re-throw (surfaces as HTTP 429). + stats.incNativeWriteRejections(); + throw e; + } } ManagedVSR newVSR = vsrPool.getActiveVSR(); if (newVSR == null) { diff --git a/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/stats/ParquetIngestPoolStatsTests.java b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/stats/ParquetIngestPoolStatsTests.java new file mode 100644 index 0000000000000..e0422436a7add --- /dev/null +++ b/sandbox/plugins/parquet-data-format/src/test/java/org/opensearch/parquet/stats/ParquetIngestPoolStatsTests.java @@ -0,0 +1,83 @@ +/* + * 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.parquet.stats; + +import org.opensearch.common.io.stream.BytesStreamOutput; +import org.opensearch.common.xcontent.XContentFactory; +import org.opensearch.common.xcontent.XContentHelper; +import org.opensearch.common.xcontent.json.JsonXContent; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.xcontent.ToXContent; +import org.opensearch.test.OpenSearchTestCase; +import org.opensearch.threadpool.ThreadPoolStats; + +import java.util.Map; + +/** Unit tests for {@link ParquetIngestPoolStats}: XContent shape, stream round-trip, factory. */ +public class ParquetIngestPoolStatsTests extends OpenSearchTestCase { + + public void testToXContentShape() { + ParquetIngestPoolStats stats = new ParquetIngestPoolStats(4, 12, 3, 7L, 100L); + Map json = toMap(stats); + @SuppressWarnings("unchecked") + Map pool = (Map) json.get("native_ingest_pool"); + assertNotNull("native_ingest_pool block must be present", pool); + assertEquals(4, ((Number) pool.get("threads")).intValue()); + assertEquals(12, ((Number) pool.get("queue_depth")).intValue()); + assertEquals(3, ((Number) pool.get("active")).intValue()); + assertEquals(7L, ((Number) pool.get("rejected")).longValue()); + assertEquals(100L, ((Number) pool.get("completed")).longValue()); + } + + public void testStreamRoundTrip() throws Exception { + ParquetIngestPoolStats original = new ParquetIngestPoolStats(2, 5, 1, 9L, 42L); + BytesStreamOutput out = new BytesStreamOutput(); + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + ParquetIngestPoolStats restored = new ParquetIngestPoolStats(in); + assertEquals(toMap(original), toMap(restored)); + } + } + + public void testFromThreadPoolStats() { + ThreadPoolStats.Stats s = new ThreadPoolStats.Stats.Builder().name("parquet_native_write") + .threads(4) + .queue(12) + .active(3) + .rejected(7L) + .largest(20) + .completed(100L) + .build(); + ParquetIngestPoolStats stats = ParquetIngestPoolStats.from(s); + assertNotNull(stats); + Map pool = poolBlock(stats); + assertEquals(12, ((Number) pool.get("queue_depth")).intValue()); + assertEquals(7L, ((Number) pool.get("rejected")).longValue()); + } + + public void testFromNullReturnsNull() { + assertNull(ParquetIngestPoolStats.from(null)); + } + + private static Map toMap(ParquetIngestPoolStats stats) { + try { + var builder = XContentFactory.jsonBuilder().startObject(); + stats.toXContent(builder, ToXContent.EMPTY_PARAMS); + builder.endObject(); + return XContentHelper.convertToMap(JsonXContent.jsonXContent, builder.toString(), true); + } catch (Exception e) { + throw new AssertionError(e); + } + } + + @SuppressWarnings("unchecked") + private static Map poolBlock(ParquetIngestPoolStats stats) { + return (Map) toMap(stats).get("native_ingest_pool"); + } +}