Avoid return negative value in CounterMetric (#71446)

A CounterMetric is used to track the number of completed and outstanding
items, for example, the number of executed refreshes, the currently used
memory by indexing, the current pending search requests. In all cases,
the current count of CounterMetric is always non-negative.

However, as this metric is implemented using a LongAdder, the returned
count is NOT an atomic snapshot; invocation in the absence of concurrent
updates returns an accurate result, but concurrent updates that occur
while the sum is being calculated might not be incorporated.

We can replace LongAdder with AtomicLong, but this commit chooses to
continue using LongAdder but returns 0 when the sum value is negative.

Relates #52411
Closes #70968

Author: dnhatn

server/src/main/java/org/elasticsearch/common/metrics/CounterMetric.java

@@ -8,29 +8,65 @@
 
 package org.elasticsearch.common.metrics;
 
+import org.elasticsearch.Assertions;
+
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 
-public class CounterMetric implements Metric {
 
+/**
+ * A {@link CounterMetric} is used to track the number of completed and outstanding items, for example, the number of executed refreshes,
+ * the currently used memory by indexing, the current pending search requests. In both cases, the current {@link CounterMetric#count} is
+ * always non-negative.
+ */
+public final class CounterMetric {
     private final LongAdder counter = new LongAdder();
+    private final AtomicLong assertingCounter = Assertions.ENABLED ? new AtomicLong() : null;
+
+    private boolean assertNonNegative(long n) {
+        assert n >= 0 : "CounterMetric value must always be non-negative; got: " + n;
+        return true;
+    }
 
     public void inc() {
         counter.increment();
+        assert assertNonNegative(assertingCounter.incrementAndGet());
     }
 
     public void inc(long n) {
         counter.add(n);
+        assert assertNonNegative(assertingCounter.addAndGet(n));
     }
 
     public void dec() {
         counter.decrement();
+        assert assertNonNegative(assertingCounter.decrementAndGet());
     }
 
     public void dec(long n) {
         counter.add(-n);
+        assert assertNonNegative(assertingCounter.addAndGet(-n));
     }
 
+    /**
+     * Returns the current count of this metric. The returned value is always non-negative.
+     * <p>
+     * As this metric is implemented using a {@link LongAdder}, the returned value is NOT an atomic snapshot;
+     * invocation in the absence of concurrent updates returns an accurate result, but concurrent updates that
+     * occur while the sum is being calculated might not be incorporated.
+     *
+     * @see LongAdder#sum()
+     */
     public long count() {
-        return counter.sum();
+        // The `counter.sum()` value is expected to always be non-negative. And if it's negative, then some concurrent updates
+        // aren't incorporated yet. In this case, we can immediately return 0L; but here we choose to retry several times
+        // to hopefully have a more accurate value than 0L.
+        for (int i = 0; i < 5; i++) {
+            final long count = counter.sum();
+            if (count >= 0L) {
+                return count;
+            }
+        }
+        return 0L;
     }
 }

server/src/main/java/org/elasticsearch/common/metrics/MeanMetric.java

@@ -10,7 +10,7 @@
 
 import java.util.concurrent.atomic.LongAdder;
 
-public class MeanMetric implements Metric {
+public final class MeanMetric {
 
     private final LongAdder counter = new LongAdder();
     private final LongAdder sum = new LongAdder();
@@ -20,13 +20,14 @@ public void inc(long n) {
         sum.add(n);
     }
 
-    public void dec(long n) {
-        counter.decrement();
-        sum.add(-n);
-    }
-
+    /**
+     * Returns the current count of this metric. This metric supports only {@link #inc(long)} that increases the counter
+     * whenever it's invoked; hence, the returned count is always non-negative.
+     */
     public long count() {
-        return counter.sum();
+        final long count = counter.sum();
+        assert count >= 0 : "Count of MeanMetric must always be non-negative; got " + count;
+        return count;
     }
 
     public long sum() {
@@ -40,9 +41,4 @@ public double mean() {
         }
         return 0.0;
     }
-
-    public void clear() {
-        counter.reset();
-        sum.reset();
-    }
 }

server/src/main/java/org/elasticsearch/common/metrics/Metric.java

@@ -71,10 +71,8 @@ public void onFailedQueryPhase(SearchContext searchContext) {
         computeStats(searchContext, statsHolder -> {
             if (searchContext.hasOnlySuggest()) {
                 statsHolder.suggestCurrent.dec();
-                assert statsHolder.suggestCurrent.count() >= 0;
             } else {
                 statsHolder.queryCurrent.dec();
-                assert statsHolder.queryCurrent.count() >= 0;
             }
         });
     }
@@ -85,11 +83,9 @@ public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
             if (searchContext.hasOnlySuggest()) {
                 statsHolder.suggestMetric.inc(tookInNanos);
                 statsHolder.suggestCurrent.dec();
-                assert statsHolder.suggestCurrent.count() >= 0;
             } else {
                 statsHolder.queryMetric.inc(tookInNanos);
                 statsHolder.queryCurrent.dec();
-                assert statsHolder.queryCurrent.count() >= 0;
             }
         });
     }
@@ -109,7 +105,6 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         computeStats(searchContext, statsHolder -> {
             statsHolder.fetchMetric.inc(tookInNanos);
             statsHolder.fetchCurrent.dec();
-            assert statsHolder.fetchCurrent.count() >= 0;
         });
     }
 

server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchStats.java

@@ -23,10 +23,10 @@ protected SnapshotLifecycleStats doParseInstance(XContentParser parser) throws I
 
     public static SnapshotLifecycleStats.SnapshotPolicyStats randomPolicyStats(String policyId) {
         return new SnapshotLifecycleStats.SnapshotPolicyStats(policyId,
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong());
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE));
     }
 
     public static SnapshotLifecycleStats randomLifecycleStats() {
@@ -37,10 +37,10 @@ public static SnapshotLifecycleStats randomLifecycleStats() {
             policyStats.put(policy, randomPolicyStats(policy));
         }
         return new SnapshotLifecycleStats(
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
-            randomBoolean() ? 0 : randomNonNegativeLong(),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
+            randomBoolean() ? 0 : randomIntBetween(0, Integer.MAX_VALUE),
             policyStats);
     }