Fix SearchStatsContributor to report per-shard fragment counts (opensearch-project#22291)

The analytics-engine SearchStatsContributor was reporting 1-per-query
(via queries.elapsed_ms.count), but Lucene's ShardSearchStats counts
1-per-shard via per-shard onPreQueryPhase / onQueryPhase callbacks.
Downstream consumers of _nodes/stats search counters were undercounting
analytics-engine traffic by the shard fan-out factor (e.g. ~50x on a
50-shard index, ~8x on 8 shards).

A secondary symptom: query_time_in_millis was reporting near-zero
because AnalyticsStatsCollector.recordExecution computes per-query
elapsed as (latestStageEnd - earliestStageStart) and skips recording
when either timestamp is 0, dropping most queries' latency entirely.

Fix: contribute from the fragments bucket (per-shard StageTasks)
instead of the per-query bucket. fragments.total is populated once
per shard execution with its own elapsed time, matching Lucene's
per-shard semantics for both count and time-in-millis.

Empirical confirmation on a 27-node OpenSearch 3.5 cluster:

50 PPL queries against a 50-shard index
  - Observed: cluster-wide query_total delta = 36
  - Expected (Lucene parity): 50 queries x 50 shards = 2500
  - Per-node breakdown: only 3 of 27 nodes incremented (coordinators)

100 PPL queries against an 8-shard index
  - Observed: cluster-wide query_total delta = 66
  - Expected (Lucene parity): 100 queries x 8 shards = 800

The ratio of undercount scales with shard count, exactly as
shard-fan-out-vs-1-per-query predicts.

Also splits SearchStatsContributorIT into three tests:
  - testQueryTotalScalesWithShardFanOut: 3-shard index, asserts
    delta >= N*shards/2 (catches the 1-per-query bug)
  - testQueryTimeIncrementsAfterAnalyticsQueries: asserts time
    delta > 0 (catches the all-or-nothing-window latency bug)
  - testQueryTotalIncrementsAtLeastOncePerQuery: asserts delta >= N
    (catches a regression where the contributor stops firing)

Signed-off-by: Michael Oviedo <mikeovi@amazon.com>

Author: OVI3D0

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java

@@ -260,11 +260,26 @@ static int schedulerPoolSize() {
 
     @Override
     public SearchStats contributeSearchStats() {
-        AnalyticsStats.LatencyStats elapsed = statsCollector.snapshot().queries().elapsedMs();
-        if (elapsed.count() == 0) {
+        // Contribute per-shard fragment task counts so the node-level search counters
+        // exposed via _nodes/stats match Lucene's per-shard accounting (one increment
+        // per shard query phase, not per user query).
+        //
+        // The queries.elapsed_ms bucket counts 1-per-query, which undercounts the
+        // rate by the shard fan-out factor and drops most queries' latency (the
+        // start/end window in AnalyticsStatsCollector#recordExecution is commonly
+        // 0 when stage timestamps aren't populated). The SHARD_FRAGMENT stage
+        // bucket counts 1-per-(query, node-hosting-shards), still missing the
+        // per-shard granularity Lucene reports. fragments.total walks each
+        // SHARD_FRAGMENT execution's per-shard StageTasks, giving per-shard
+        // counts matching Lucene's onPreQueryPhase semantics.
+        AnalyticsStats snapshot = statsCollector.snapshot();
+        AnalyticsStats.Fragments fragments = snapshot.fragments();
+        if (fragments == null || fragments.total() == 0) {
             return null;
         }
-        SearchStats.Stats stats = new SearchStats.Stats.Builder().queryCount(elapsed.count()).queryTimeInMillis(elapsed.sumMs()).build();
+        SearchStats.Stats stats = new SearchStats.Stats.Builder().queryCount(fragments.total())
+            .queryTimeInMillis(fragments.elapsedMs().sumMs())
+            .build();
         return new SearchStats(stats, 0, null);
     }
 

sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/SearchStatsContributorIT.java

@@ -25,42 +25,91 @@
 public class SearchStatsContributorIT extends AnalyticsRestTestCase {
 
     private static final Dataset DATASET = new Dataset("calcs", "calcs");
+    /**
+     * Multi-shard index so the test can distinguish the per-query contributor
+     * (queryCount delta ≈ N) from the per-shard Lucene-equivalent contributor
+     * (queryCount delta ≈ N × shards). Catches the regression class where the
+     * contributor counts 1-per-query and silently under-reports search rate
+     * by the shard fan-out factor to downstream consumers of node stats.
+     */
+    private static final int NUMBER_OF_SHARDS = 3;
     private static boolean dataProvisioned = false;
 
     private void ensureDataProvisioned() throws IOException {
         if (dataProvisioned == false) {
-            DatasetProvisioner.provision(client(), DATASET);
+            DatasetProvisioner.provision(client(), DATASET, NUMBER_OF_SHARDS);
             dataProvisioned = true;
         }
     }
 
-    public void testQueryTotalIncrementsAfterAnalyticsQueries() throws IOException {
+    /**
+     * Guards the 1-per-query bug: cluster-wide query_total must grow with the
+     * shard fan-out factor. A 1-per-query contributor would produce delta ≈ N
+     * (instead of N × shards), under-reporting search rate by the shard fan-out
+     * factor to downstream consumers of node stats.
+     */
+    public void testQueryTotalScalesWithShardFanOut() throws IOException {
         ensureDataProvisioned();
 
-        long baselineQueryTotal = getQueryTotalAcrossNodes();
-        long baselineQueryTimeMs = getQueryTimeAcrossNodes();
+        long before = getQueryTotalAcrossNodes();
+        fireQueries();
+        long after = getQueryTotalAcrossNodes();
 
-        // Fire a handful of analytics-engine PPL queries — varied shapes so the contributor
-        // sees both real elapsed time and a non-trivial count.
+        long delta = after - before;
+        long minExpected = (long) TOTAL_QUERIES * NUMBER_OF_SHARDS / 2;
+        assertTrue(
+            "search.query_total must grow with shard fan-out (fired " + TOTAL_QUERIES
+                + " on " + NUMBER_OF_SHARDS + "-shard index, expected ≥ " + minExpected
+                + ", observed delta=" + delta + ")",
+            delta >= minExpected
+        );
+    }
+
+    /**
+     * Guards the latency-aggregation bug: query_time_in_millis must move when
+     * real PPL work runs. A delta of 0 means the contributor's time source
+     * dropped every recording (e.g. an all-or-nothing window that collapsed).
+     */
+    public void testQueryTimeIncrementsAfterAnalyticsQueries() throws IOException {
+        ensureDataProvisioned();
+
+        long before = getQueryTimeAcrossNodes();
+        fireQueries();
+        long after = getQueryTimeAcrossNodes();
+
+        long delta = after - before;
+        assertTrue("search.query_time_in_millis must grow > 0 after PPL work, observed delta=" + delta, delta > 0);
+    }
+
+    /**
+     * Minimum-viable check: every fired query must contribute at least once to
+     * cluster-wide query_total. Catches regressions where the contributor stops
+     * firing entirely (e.g. returns null, or filters out all stages).
+     */
+    public void testQueryTotalIncrementsAtLeastOncePerQuery() throws IOException {
+        ensureDataProvisioned();
+
+        long before = getQueryTotalAcrossNodes();
+        fireQueries();
+        long after = getQueryTotalAcrossNodes();
+
+        long delta = after - before;
+        assertTrue(
+            "search.query_total must grow by at least one per query (fired " + TOTAL_QUERIES
+                + ", observed delta=" + delta + ")",
+            delta >= TOTAL_QUERIES
+        );
+    }
+
+    private static final int TOTAL_QUERIES = 15;
+
+    private void fireQueries() throws IOException {
+        // Varied shapes so the contributor sees both real elapsed time and a non-trivial count.
         for (int i = 0; i < 5; i++) {
             executePpl("source=" + DATASET.indexName + " | fields str0");
             executePpl("source=" + DATASET.indexName + " | where num0 > 0 | fields str0, num0");
             executePpl("source=" + DATASET.indexName + " | stats avg(num0)");
         }
-
-        long afterQueryTotal = getQueryTotalAcrossNodes();
-        long afterQueryTimeMs = getQueryTimeAcrossNodes();
-
-        long countDelta = afterQueryTotal - baselineQueryTotal;
-        assertTrue(
-            "search.query_total must increment by at least 1 after analytics queries, delta=" + countDelta,
-            countDelta >= 1
-        );
-        // query_time_in_millis is a sum so it should grow strictly monotonically with count;
-        // any single query that took 0ms is unlikely with non-trivial work, but the contract
-        // we care about is "the field is being populated", so >= 0 is enough.
-        long timeDelta = afterQueryTimeMs - baselineQueryTimeMs;
-        assertTrue("search.query_time_in_millis must not regress, delta=" + timeDelta, timeDelta >= 0);
     }
 
     private long getQueryTotalAcrossNodes() throws IOException {