Add remote refresh segment pressure service, settings and tracker (#7227

)

Signed-off-by: Ashish Singh <ssashish@amazon.com>

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
  )
  )
  );

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.
  */