Add check and handle negative SearchRequestStats

CHANGELOG.md

@@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add Setting to adjust the primary constraint weights ([#16471](https://github.com/opensearch-project/OpenSearch/pull/16471))
 - Switch from `buildSrc/version.properties` to Gradle version catalog (`gradle/libs.versions.toml`) to enable dependabot to perform automated upgrades on common libs ([#16284](https://github.com/opensearch-project/OpenSearch/pull/16284))
 - Add dynamic setting allowing size > 0 requests to be cached in the request cache ([#16483](https://github.com/opensearch-project/OpenSearch/pull/16483))
+- Add check and handle negative SearchRequestStats ([#16569](https://github.com/opensearch-project/OpenSearch/pull/16569))
 - Make IndexStoreListener a pluggable interface ([#16583](https://github.com/opensearch-project/OpenSearch/pull/16583))
 
 ### Dependencies

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("Coordinator request stats 'current' is negative: {}", current);
+                out.writeVLong(0);
+            } else {
+                out.writeVLong(current);
+            }
             out.writeVLong(total);
-            out.writeVLong(timeInMillis);
+            if (timeInMillis < 0) {
+                logger.warn("Coordinator request stats 'timeInMillis' is negative: {}", timeInMillis);
+                out.writeVLong(0);
+            } else {
+                out.writeVLong(timeInMillis);
+            }
         }
 
         PhaseStatsLongHolder() {
@@ -173,7 +185,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,6 +37,7 @@
 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;
@@ -52,6 +53,46 @@
 
 public class SearchStatsTests extends OpenSearchTestCase implements SearchRequestOperationsListenerSupport {
 
+    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.getSearchPhaseName()).thenReturn(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()) {
+            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));
+    }
+
     // https://github.com/elastic/elasticsearch/issues/7644
     public void testShardLevelSearchGroupStats() throws Exception {
         // let's create two dummy search stats with groups