Handle negative search request nodes stats

Signed-off-by: David Zane <[email protected]>

Author: dzane17

CHANGELOG.md

@@ -76,6 +76,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix ingestion state xcontent serialization in IndexMetadata and fail fast on mapping errors([#19320](https://github.com/opensearch-project/OpenSearch/pull/19320))
 - Fix updated keyword field params leading to stale responses from request cache ([#19385](https://github.com/opensearch-project/OpenSearch/pull/19385))
 - Implement SslHandler retrieval logic for transport-reactor-netty4 plugin ([#19458](https://github.com/opensearch-project/OpenSearch/pull/19458))
+- Handle negative search request nodes stats ([#19340](https://github.com/opensearch-project/OpenSearch/pull/19340))
 
 ### Dependencies
 - Bump `com.gradleup.shadow:shadow-gradle-plugin` from 8.3.5 to 8.3.9 ([#19400](https://github.com/opensearch-project/OpenSearch/pull/19400))

server/src/main/java/org/opensearch/action/search/SearchRequestStats.java

@@ -8,6 +8,8 @@
 
 package org.opensearch.action.search;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.metrics.CounterMetric;
@@ -28,6 +30,7 @@
 public final class SearchRequestStats extends SearchRequestOperationsListener {
     Map<SearchPhaseName, StatsHolder> phaseStatsMap = new EnumMap<>(SearchPhaseName.class);
     StatsHolder tookStatsHolder;
+    private static final Logger logger = LogManager.getLogger(SearchRequestStats.class);
 
     public static final String SEARCH_REQUEST_STATS_ENABLED_KEY = "search.request_stats_enabled";
     public static final Setting<Boolean> SEARCH_REQUEST_STATS_ENABLED = Setting.boolSetting(
@@ -80,6 +83,9 @@ protected void onPhaseStart(SearchPhaseContext context) {
     protected void onPhaseEnd(SearchPhaseContext context, SearchRequestContext searchRequestContext) {
         context.getCurrentPhase().getSearchPhaseNameOptional().ifPresent(name -> {
             StatsHolder phaseStats = phaseStatsMap.get(name);
+            if (phaseStats.current.count() == 0) {
+                logger.warn("SearchRequestStats 'current' went negative for phase '{}'", name);
+            }
             phaseStats.current.dec();
             phaseStats.total.inc();
             phaseStats.timing.inc(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - context.getCurrentPhase().getStartTimeInNanos()));

server/src/main/java/org/opensearch/index/search/stats/SearchStats.java

@@ -32,6 +32,8 @@
 
 package org.opensearch.index.search.stats;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.Version;
 import org.opensearch.action.search.SearchPhaseName;
 import org.opensearch.action.search.SearchRequestStats;
@@ -67,7 +69,7 @@ public class SearchStats implements Writeable, ToXContentFragment {
      */
     @PublicApi(since = "1.0.0")
     public static class PhaseStatsLongHolder implements Writeable {
-
+        private static final Logger logger = LogManager.getLogger(PhaseStatsLongHolder.class);
         long current;
         long total;
         long timeInMillis;
@@ -86,9 +88,19 @@ public long getTimeInMillis() {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            out.writeVLong(current);
+            if (current < 0) {
+                logger.warn("'current' is negative [{}]", current);
+                out.writeVLong(0);
+            } else {
+                out.writeVLong(current);
+            }
             out.writeVLong(total);
-            out.writeVLong(timeInMillis);
+            if (timeInMillis < 0) {
+                logger.warn("'timeInMillis' is negative: [{}]", timeInMillis);
+                out.writeVLong(0);
+            } else {
+                out.writeVLong(timeInMillis);
+            }
         }
 
         PhaseStatsLongHolder() {
@@ -177,7 +189,7 @@ public RequestStatsLongHolder getRequestStatsLongHolder() {
             return requestStatsLongHolder;
         }
 
-        private Stats() {
+        Stats() {
             // for internal use, initializes all counts to 0
         }
 

server/src/test/java/org/opensearch/index/search/stats/SearchStatsTests.java

@@ -37,10 +37,12 @@
 import org.opensearch.action.search.SearchPhaseName;
 import org.opensearch.action.search.SearchRequestOperationsListenerSupport;
 import org.opensearch.action.search.SearchRequestStats;
+import org.opensearch.common.io.stream.BytesStreamOutput;
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.search.stats.SearchStats.Stats;
 import org.opensearch.test.OpenSearchTestCase;
+import org.junit.Assert;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -159,4 +161,45 @@ private static void assertStats(Stats stats, long equalTo) {
         // avg_concurrency is not summed up across stats
         assertEquals(1, stats.getConcurrentAvgSliceCount(), 0);
     }
+
+    public void testNegativeRequestStats() throws Exception {
+        SearchStats searchStats = new SearchStats(new Stats(), 0, new HashMap<>());
+
+        long paramValue = randomIntBetween(2, 50);
+
+        // Testing for request stats
+        ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        SearchRequestStats testRequestStats = new SearchRequestStats(clusterSettings);
+        SearchPhaseContext ctx = mock(SearchPhaseContext.class);
+        for (SearchPhaseName searchPhaseName : SearchPhaseName.values()) {
+            SearchPhase mockSearchPhase = mock(SearchPhase.class);
+            when(ctx.getCurrentPhase()).thenReturn(mockSearchPhase);
+            when(mockSearchPhase.getStartTimeInNanos()).thenReturn(System.nanoTime() - TimeUnit.SECONDS.toNanos(paramValue));
+            when(mockSearchPhase.getSearchPhaseNameOptional()).thenReturn(Optional.ofNullable(searchPhaseName));
+            for (int iterator = 0; iterator < paramValue; iterator++) {
+                onPhaseStart(testRequestStats, ctx);
+                onPhaseEnd(testRequestStats, ctx);
+                onPhaseEnd(testRequestStats, ctx); // call onPhaseEnd() twice to make 'current' negative
+            }
+        }
+        searchStats.setSearchRequestStats(testRequestStats);
+        for (SearchPhaseName searchPhaseName : SearchPhaseName.values()) {
+            Assert.assertNotNull(searchStats.getTotal().getRequestStatsLongHolder());
+            assertEquals(
+                -1 * paramValue,    // current is negative, equals -1 * paramValue (num loop iterations)
+                searchStats.getTotal().getRequestStatsLongHolder().getRequestStatsHolder().get(searchPhaseName.getName()).current
+            );
+            assertEquals(
+                2 * paramValue,
+                searchStats.getTotal().getRequestStatsLongHolder().getRequestStatsHolder().get(searchPhaseName.getName()).total
+            );
+            assertThat(
+                searchStats.getTotal().getRequestStatsLongHolder().getRequestStatsHolder().get(searchPhaseName.getName()).timeInMillis,
+                greaterThanOrEqualTo(paramValue)
+            );
+        }
+
+        // Ensure writeTo() does not throw error with negative 'current'
+        searchStats.writeTo(new BytesStreamOutput(10));
+    }
 }