Add remote refresh segment pressure service, settings and tracker (op…

…ensearch-project#7227) Signed-off-by: Ashish Singh <ssashish@amazon.com>
linuxpi · May 2, 2023 · 18390f9 · 18390f9
1 parent 44ebf81
commit 18390f9
Show file tree

Hide file tree

Showing 10 changed files with 1,914 additions and 8 deletions.
diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java
@@ -39,6 +39,7 @@
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.IndexingPressure;
+import org.opensearch.index.remote.RemoteRefreshSegmentPressureSettings;
 import org.opensearch.index.SegmentReplicationPressureService;
 import org.opensearch.index.ShardIndexingPressureMemoryManager;
 import org.opensearch.index.ShardIndexingPressureSettings;
@@ -638,7 +639,17 @@ public void apply(Settings value, Settings current, Settings previous) {
                 SegmentReplicationPressureService.MAX_ALLOWED_STALE_SHARDS,
 
                 // Settings related to Searchable Snapshots
-                Node.NODE_SEARCH_CACHE_SIZE_SETTING
+                Node.NODE_SEARCH_CACHE_SIZE_SETTING,
+
+                // Settings related to Remote Refresh Segment Pressure
+                RemoteRefreshSegmentPressureSettings.REMOTE_REFRESH_SEGMENT_PRESSURE_ENABLED,
+                RemoteRefreshSegmentPressureSettings.MIN_SEQ_NO_LAG_LIMIT,
+                RemoteRefreshSegmentPressureSettings.BYTES_LAG_VARIANCE_FACTOR,
+                RemoteRefreshSegmentPressureSettings.UPLOAD_TIME_LAG_VARIANCE_FACTOR,
+                RemoteRefreshSegmentPressureSettings.MIN_CONSECUTIVE_FAILURES_LIMIT,
+                RemoteRefreshSegmentPressureSettings.UPLOAD_BYTES_MOVING_AVERAGE_WINDOW_SIZE,
+                RemoteRefreshSegmentPressureSettings.UPLOAD_BYTES_PER_SEC_MOVING_AVERAGE_WINDOW_SIZE,
+                RemoteRefreshSegmentPressureSettings.UPLOAD_TIME_MOVING_AVERAGE_WINDOW_SIZE
             )
         )
     );

diff --git a/server/src/main/java/org/opensearch/common/util/MovingAverage.java b/server/src/main/java/org/opensearch/common/util/MovingAverage.java
@@ -17,19 +17,47 @@ public class MovingAverage {
     private final int windowSize;
     private final long[] observations;
 
-    private long count = 0;
-    private long sum = 0;
-    private double average = 0;
+    private volatile long count = 0;
+    private volatile long sum = 0;
+    private volatile double average = 0;
 
     public MovingAverage(int windowSize) {
-        if (windowSize <= 0) {
-            throw new IllegalArgumentException("window size must be greater than zero");
-        }
-
+        checkWindowSize(windowSize);
         this.windowSize = windowSize;
         this.observations = new long[windowSize];
     }
 
+    /**
+     * Used for changing the window size of {@code MovingAverage}.
+     *
+     * @param newWindowSize new window size.
+     * @return copy of original object with updated size.
+     */
+    public MovingAverage copyWithSize(int newWindowSize) {
+        MovingAverage copy = new MovingAverage(newWindowSize);
+        // Start is inclusive, but end is exclusive
+        long start, end = count;
+        if (isReady() == false) {
+            start = 0;
+        } else {
+            start = end - windowSize;
+        }
+        // If the newWindow Size is smaller than the elements eligible to be copied over, then we adjust the start value
+        if (end - start > newWindowSize) {
+            start = end - newWindowSize;
+        }
+        for (int i = (int) start; i < end; i++) {
+            copy.record(observations[i % observations.length]);
+        }
+        return copy;
+    }
+
+    private void checkWindowSize(int size) {
+        if (size <= 0) {
+            throw new IllegalArgumentException("window size must be greater than zero");
+        }
+    }
+
     /**
      * Records a new observation and evicts the n-th last observation.
      */