Add ShardIndexingPressure framework level construct and Stats (opense…

…arch-project#1015) * Add ShardIndexingPressure framework level construct and related Stats artefacts. * Test and code refactoring for shard indexing pressure. * Moved the average calculation logic to common memory manager util. * Add wrapper for releasable in ShardIndexingPressure operations. Signed-off-by: Saurabh Singh <sisurab@amazon.com>
getsaurabh02 · Oct 6, 2021 · b222f3f · b222f3f
1 parent d4a48be
commit b222f3f
Show file tree

Hide file tree

Showing 8 changed files with 2,618 additions and 1 deletion.
diff --git a/server/src/main/java/org/opensearch/action/admin/indices/stats/CommonStatsFlags.java b/server/src/main/java/org/opensearch/action/admin/indices/stats/CommonStatsFlags.java
@@ -36,6 +36,7 @@
 import org.opensearch.common.io.stream.StreamInput;
 import org.opensearch.common.io.stream.StreamOutput;
 import org.opensearch.common.io.stream.Writeable;
+import org.opensearch.index.ShardIndexingPressureSettings;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -53,6 +54,8 @@ public class CommonStatsFlags implements Writeable, Cloneable {
     private String[] completionDataFields = null;
     private boolean includeSegmentFileSizes = false;
     private boolean includeUnloadedSegments = false;
+    private boolean includeAllShardIndexingPressureTrackers = false;
+    private boolean includeOnlyTopIndexingPressureMetrics = false;
 
     /**
      * @param flags flags to set. If no flags are supplied, default flags will be set.
@@ -80,6 +83,10 @@ public CommonStatsFlags(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(LegacyESVersion.V_7_2_0)) {
             includeUnloadedSegments = in.readBoolean();
         }
+        if (ShardIndexingPressureSettings.isShardIndexingPressureAttributeEnabled()) {
+            includeAllShardIndexingPressureTrackers = in.readBoolean();
+            includeOnlyTopIndexingPressureMetrics = in.readBoolean();
+        }
     }
 
     @Override
@@ -98,6 +105,10 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(LegacyESVersion.V_7_2_0)) {
             out.writeBoolean(includeUnloadedSegments);
         }
+        if (ShardIndexingPressureSettings.isShardIndexingPressureAttributeEnabled()) {
+            out.writeBoolean(includeAllShardIndexingPressureTrackers);
+            out.writeBoolean(includeOnlyTopIndexingPressureMetrics);
+        }
     }
 
     /**
@@ -111,6 +122,8 @@ public CommonStatsFlags all() {
         completionDataFields = null;
         includeSegmentFileSizes = false;
         includeUnloadedSegments = false;
+        includeAllShardIndexingPressureTrackers = false;
+        includeOnlyTopIndexingPressureMetrics = false;
         return this;
     }
 
@@ -125,6 +138,8 @@ public CommonStatsFlags clear() {
         completionDataFields = null;
         includeSegmentFileSizes = false;
         includeUnloadedSegments = false;
+        includeAllShardIndexingPressureTrackers = false;
+        includeOnlyTopIndexingPressureMetrics = false;
         return this;
     }
 
@@ -198,10 +213,28 @@ public CommonStatsFlags includeUnloadedSegments(boolean includeUnloadedSegments)
         return this;
     }
 
+    public CommonStatsFlags includeAllShardIndexingPressureTrackers(boolean includeAllShardPressureTrackers) {
+        this.includeAllShardIndexingPressureTrackers = includeAllShardPressureTrackers;
+        return this;
+    }
+
+    public CommonStatsFlags includeOnlyTopIndexingPressureMetrics(boolean includeOnlyTopIndexingPressureMetrics) {
+        this.includeOnlyTopIndexingPressureMetrics = includeOnlyTopIndexingPressureMetrics;
+        return this;
+    }
+
     public boolean includeUnloadedSegments() {
         return this.includeUnloadedSegments;
     }
 
+    public boolean includeAllShardIndexingPressureTrackers() {
+        return this.includeAllShardIndexingPressureTrackers;
+    }
+
+    public boolean includeOnlyTopIndexingPressureMetrics() {
+        return this.includeOnlyTopIndexingPressureMetrics;
+    }
+
     public boolean includeSegmentFileSizes() {
         return this.includeSegmentFileSizes;
     }

diff --git a/server/src/main/java/org/opensearch/index/ShardIndexingPressure.java b/server/src/main/java/org/opensearch/index/ShardIndexingPressure.java
diff --git a/server/src/main/java/org/opensearch/index/ShardIndexingPressureMemoryManager.java b/server/src/main/java/org/opensearch/index/ShardIndexingPressureMemoryManager.java
@@ -282,6 +282,26 @@ ShardIndexingPressureTracker getShardIndexingPressureTracker(ShardId shardId) {
         return shardIndexingPressureStore.getShardIndexingPressureTracker(shardId);
     }
 
+    Map<ShardId, ShardIndexingPressureTracker> getShardIndexingPressureHotStore() {
+        return shardIndexingPressureStore.getShardIndexingPressureHotStore();
+    }
+
+    Map<ShardId, ShardIndexingPressureTracker> getShardIndexingPressureColdStore() {
+        return shardIndexingPressureStore.getShardIndexingPressureColdStore();
+    }
+
+    void tryTrackerCleanupFromHotStore(ShardIndexingPressureTracker tracker, BooleanSupplier condition) {
+        shardIndexingPressureStore.tryTrackerCleanupFromHotStore(tracker, condition);
+    }
+
+    double calculateMovingAverage(long currentAverage, double frontValue, double currentValue, int count) {
+        if(count > 0) {
+            return ((Double.longBitsToDouble(currentAverage) * count) + currentValue - frontValue) / count;
+        } else {
+            return currentValue;
+        }
+    }
+
     long getTotalNodeLimitsBreachedRejections() {
         return totalNodeLimitsBreachedRejections.get();
     }
@@ -417,7 +437,7 @@ private boolean evaluateThroughputDegradationLimitsBreached(PerformanceTracker p
      */
     private boolean evaluateLastSuccessfulRequestDurationLimitsBreached(PerformanceTracker performanceTracker, long requestStartTime) {
         return (performanceTracker.getLastSuccessfulRequestTimestamp() > 0) &&
-            (requestStartTime - performanceTracker.getLastSuccessfulRequestTimestamp()) > this.successfulRequestElapsedTimeout.millis() &&
+            (requestStartTime - performanceTracker.getLastSuccessfulRequestTimestamp()) > this.successfulRequestElapsedTimeout.nanos() &&
             performanceTracker.getTotalOutstandingRequests() > this.maxOutstandingRequests;
     }
 

diff --git a/server/src/main/java/org/opensearch/index/ShardIndexingPressureTracker.java b/server/src/main/java/org/opensearch/index/ShardIndexingPressureTracker.java
@@ -139,6 +139,10 @@ public long incrementTotalBytes(long bytes) {
         public long getRequestCount() {
             return requestCount.get();
         }
+
+        public long incrementRequestCount() {
+            return requestCount.incrementAndGet();
+        }
     }
 
     /**
@@ -161,6 +165,10 @@ public long getTotalRejections() {
             return totalRejections.get();
         }
 
+        public long incrementTotalRejections() {
+            return totalRejections.incrementAndGet();
+        }
+
         public long getNodeLimitsBreachedRejections() {
             return nodeLimitsBreachedRejections.get();
         }
@@ -232,6 +240,10 @@ public long incrementTotalOutstandingRequests() {
             return totalOutstandingRequests.incrementAndGet();
         }
 
+        public void resetTotalOutstandingRequests() {
+            totalOutstandingRequests.set(0L);
+        }
+
         public long getThroughputMovingAverage() {
             return throughputMovingAverage.get();
         }
@@ -275,5 +287,9 @@ public long incrementCurrentCombinedCoordinatingAndPrimaryBytes(long bytes) {
         public long getTotalCombinedCoordinatingAndPrimaryBytes() {
             return totalCombinedCoordinatingAndPrimaryBytes.get();
         }
+
+        public long incrementTotalCombinedCoordinatingAndPrimaryBytes(long bytes) {
+            return totalCombinedCoordinatingAndPrimaryBytes.addAndGet(bytes);
+        }
     }
 }