Restore size-based periodic flush on remote-store shards

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -250,6 +250,8 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 // Settings for remote translog
                 IndexSettings.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING,
                 IndexSettings.INDEX_REMOTE_TRANSLOG_KEEP_EXTRA_GEN_SETTING,
+                IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING,
+                IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING,
 
                 // Settings for remote store enablement
                 IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING,

server/src/main/java/org/opensearch/common/settings/Setting.java

@@ -2109,6 +2109,16 @@ public static Setting<ByteSizeValue> byteSizeSetting(String key, Setting<ByteSiz
         return new Setting<>(key, fallbackSetting, new ByteSizeValueParser(key), properties);
     }
 
+    public static Setting<ByteSizeValue> byteSizeSetting(
+        String key,
+        Setting<ByteSizeValue> fallbackSetting,
+        ByteSizeValue minValue,
+        ByteSizeValue maxValue,
+        Property... properties
+    ) {
+        return new Setting<>(key, fallbackSetting, new ByteSizeValueParser(minValue, maxValue, key), properties);
+    }
+
     public static Setting<ByteSizeValue> byteSizeSetting(String key, Function<Settings, String> defaultValue, Property... properties) {
         return new Setting<>(key, defaultValue, new ByteSizeValueParser(key), properties);
     }

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -881,6 +881,38 @@ public static IndexMergePolicy fromString(String text) {
         Property.IndexScope
     );
 
+    /**
+     * Controls whether a remote-store shard periodically flushes based on the total size of segment bytes that are
+     * refreshed and uploaded to the remote store but not yet referenced by the last commit point. On remote-store
+     * shards the translog based flush threshold ({@code index.translog.flush_threshold_size}) is ineffective because
+     * uploaded translog generations are trimmed continuously; this condition restores an equivalent size-based flush
+     * signal driven by uncommitted segment bytes.
+     */
+    public static final Setting<Boolean> INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING = Setting.boolSetting(
+        "index.remote_store.flush_on_uncommitted_segments.enabled",
+        true,
+        Property.Dynamic,
+        Property.IndexScope
+    );
+
+    /**
+     * The minimum total size of segment bytes not yet referenced by the last commit point which triggers a flush on a
+     * remote-store shard. Only takes effect when
+     * {@link #INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING} is enabled. Defaults to the current
+     * value of {@code index.translog.flush_threshold_size} of the index.
+     */
+    public static final Setting<ByteSizeValue> INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING = Setting
+        .byteSizeSetting(
+            "index.remote_store.flush_on_uncommitted_segments.threshold_size",
+            INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING,
+            // a zero or negative threshold would flush on every successful segments sync; disablement has its own
+            // explicit setting above
+            new ByteSizeValue(1, ByteSizeUnit.BYTES),
+            new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES),
+            Property.Dynamic,
+            Property.IndexScope
+        );
+
     public static final Setting<Long> INDEX_CONTEXT_CREATED_VERSION = Setting.longSetting(
         "index.context.created_version",
         0,
@@ -943,6 +975,8 @@ public static IndexMergePolicy fromString(String text) {
     // For warm index we would partially store files in local.
     private final boolean isWarmIndex;
     private volatile TimeValue remoteTranslogUploadBufferInterval;
+    private volatile boolean flushOnUncommittedSegmentsEnabled;
+    private volatile ByteSizeValue flushOnUncommittedSegmentsThresholdSize;
     private volatile String remoteStoreTranslogRepository;
     private volatile String remoteStoreRepository;
     private volatile String remoteStoreSegmentPathPrefix;
@@ -1173,6 +1207,10 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
 
         remoteStoreTranslogRepository = settings.get(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY);
         remoteTranslogUploadBufferInterval = INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING.get(settings);
+        flushOnUncommittedSegmentsEnabled = scopedSettings.get(INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING);
+        flushOnUncommittedSegmentsThresholdSize = scopedSettings.get(
+            INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING
+        );
         remoteStoreRepository = settings.get(IndexMetadata.SETTING_REMOTE_SEGMENT_STORE_REPOSITORY);
         this.remoteTranslogKeepExtraGen = INDEX_REMOTE_TRANSLOG_KEEP_EXTRA_GEN_SETTING.get(settings);
         String rawPrefix = IndexMetadata.INDEX_REMOTE_STORE_SEGMENT_PATH_PREFIX.get(settings);
@@ -1389,6 +1427,14 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             this::setRemoteTranslogUploadBufferInterval
         );
         scopedSettings.addSettingsUpdateConsumer(INDEX_REMOTE_TRANSLOG_KEEP_EXTRA_GEN_SETTING, this::setRemoteTranslogKeepExtraGen);
+        scopedSettings.addSettingsUpdateConsumer(
+            INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING,
+            this::setFlushOnUncommittedSegmentsEnabled
+        );
+        scopedSettings.addSettingsUpdateConsumer(
+            INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING,
+            this::setFlushOnUncommittedSegmentsThresholdSize
+        );
         this.autoForcemergeEnabled = scopedSettings.get(INDEX_AUTO_FORCE_MERGES_ENABLED);
         scopedSettings.addSettingsUpdateConsumer(INDEX_AUTO_FORCE_MERGES_ENABLED, this::setAutoForcemergeEnabled);
         scopedSettings.addSettingsUpdateConsumer(INDEX_DOC_ID_FUZZY_SET_ENABLED_SETTING, this::setEnableFuzzySetForDocId);
@@ -1774,6 +1820,31 @@ public int getRemoteTranslogExtraKeep() {
         return remoteTranslogKeepExtraGen;
     }
 
+    /**
+     * Returns whether the periodic flush condition based on uncommitted segment bytes is enabled for this
+     * (remote-store) index.
+     */
+    public boolean isFlushOnUncommittedSegmentsEnabled() {
+        return flushOnUncommittedSegmentsEnabled;
+    }
+
+    /**
+     * Returns the threshold size of segment bytes not yet referenced by the last commit point when to forcefully
+     * flush a remote-store shard. Only takes effect when {@link #isFlushOnUncommittedSegmentsEnabled()}. Defaults to
+     * the current value of {@code index.translog.flush_threshold_size} of this index.
+     */
+    public ByteSizeValue getFlushOnUncommittedSegmentsThresholdSize() {
+        return flushOnUncommittedSegmentsThresholdSize;
+    }
+
+    private void setFlushOnUncommittedSegmentsEnabled(boolean flushOnUncommittedSegmentsEnabled) {
+        this.flushOnUncommittedSegmentsEnabled = flushOnUncommittedSegmentsEnabled;
+    }
+
+    private void setFlushOnUncommittedSegmentsThresholdSize(ByteSizeValue flushOnUncommittedSegmentsThresholdSize) {
+        this.flushOnUncommittedSegmentsThresholdSize = flushOnUncommittedSegmentsThresholdSize;
+    }
+
     /**
      * Returns true iff the remote translog buffer interval setting exists or in other words is explicitly set.
      */

server/src/main/java/org/opensearch/index/engine/InternalEngine.java

@@ -37,6 +37,7 @@
 import org.apache.lucene.document.NumericDocValuesField;
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.IndexCommit;
+import org.apache.lucene.index.IndexFileNames;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.index.IndexWriterConfig;
 import org.apache.lucene.index.LeafReaderContext;
@@ -118,6 +119,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -161,6 +163,24 @@ public class InternalEngine extends Engine {
     protected final AtomicBoolean shouldPeriodicallyFlushAfterBigMerge = new AtomicBoolean(false);
     protected final NumericDocValuesField softDeletesField = Lucene.newSoftDeletesField();
 
+    /**
+     * Size of segment bytes not yet referenced by the last commit point on a remote-store shard, published by the
+     * remote segment upload path after every successful segments sync (see RemoteStoreRefreshListener). Stamped with
+     * the commit generation it was computed against: a value whose generation does not match the current last commit
+     * is stale (e.g. right after a flush) and is ignored until the next sync republishes.
+     */
+    private volatile UncommittedSegmentBytes uncommittedSegmentBytes;
+
+    private static final class UncommittedSegmentBytes {
+        private final long bytes;
+        private final long committedInfosGeneration;
+
+        private UncommittedSegmentBytes(long bytes, long committedInfosGeneration) {
+            this.bytes = bytes;
+            this.committedInfosGeneration = committedInfosGeneration;
+        }
+    }
+
     // A uid (in the form of BytesRef) to the version map
     // we use the hashed variant since we iterate over it and check removal and additions on existing keys
     protected final LiveVersionMap versionMap = new LiveVersionMap();
@@ -1496,6 +1516,9 @@ public boolean shouldPeriodicallyFlush() {
         if (shouldPeriodicallyFlushAfterBigMerge.get()) {
             return true;
         }
+        if (shouldFlushOnUncommittedSegmentBytes()) {
+            return true;
+        }
         final long localCheckpointOfLastCommit = Long.parseLong(
             lastCommittedSegmentInfos.userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)
         );
@@ -1505,6 +1528,68 @@ public boolean shouldPeriodicallyFlush() {
         );
     }
 
+    /**
+     * Updates the uncommitted segment bytes accounting for this (remote-store) shard. Invoked by the remote segment
+     * upload path after a successful segments sync, passing the post-refresh local segment file sizes. The uncommitted
+     * bytes are computed as the total size of local segment files that are not referenced by the last commit point,
+     * which includes newly written segments as well as updated per-segment files (e.g. live docs and doc-values
+     * updates) of already committed segments. The value is stamped with the generation of the commit point it was
+     * computed against, so a commit implicitly invalidates it.
+     *
+     * @param localSegmentsSizeMap post-refresh local segment file names mapped to their sizes in bytes
+     */
+    public void updateUncommittedSegmentBytes(Map<String, Long> localSegmentsSizeMap) {
+        // the file set and the generation stamp are both taken from this single snapshot, so the published value is
+        // always internally consistent. If a flush lands between this snapshot and the publish below, the stamp no
+        // longer matches the new last commit and shouldFlushOnUncommittedSegmentBytes() rejects the value -- the
+        // race can only suppress a flush trigger until the next successful segments sync republishes, never cause
+        // a spurious flush.
+        final SegmentInfos committedInfos = this.lastCommittedSegmentInfos;
+        if (committedInfos == null) {
+            return;
+        }
+        final Set<String> committedFiles;
+        try {
+            committedFiles = new HashSet<>(committedInfos.files(false));
+        } catch (IOException e) {
+            // best-effort accounting, a failed computation must never affect the segment upload path
+            logger.debug("failed to compute uncommitted segment bytes", e);
+            return;
+        }
+        long bytes = 0;
+        for (Map.Entry<String, Long> file : localSegmentsSizeMap.entrySet()) {
+            if (committedFiles.contains(file.getKey()) == false && file.getKey().startsWith(IndexFileNames.SEGMENTS) == false) {
+                bytes += file.getValue();
+            }
+        }
+        uncommittedSegmentBytes = new UncommittedSegmentBytes(bytes, committedInfos.getGeneration());
+    }
+
+    /**
+     * Checks whether the uncommitted segment bytes published by the remote segment upload path breach
+     * {@link IndexSettings#INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING}. Only ever
+     * effective on remote-store shards since the accounting is only published there, and only when
+     * {@link IndexSettings#INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING} is enabled; the stamped
+     * commit generation must match the current last commit so that stale values (e.g. right after a flush, before the
+     * next successful segments sync) can never re-trigger a flush.
+     */
+    private boolean shouldFlushOnUncommittedSegmentBytes() {
+        final UncommittedSegmentBytes current = this.uncommittedSegmentBytes;
+        if (current == null) {
+            return false;
+        }
+        final IndexSettings indexSettings = config().getIndexSettings();
+        if (indexSettings.isFlushOnUncommittedSegmentsEnabled() == false) {
+            return false;
+        }
+        // snapshot the volatile once so the null check and the generation comparison observe the same commit point
+        final SegmentInfos committedInfos = this.lastCommittedSegmentInfos;
+        return committedInfos != null
+            && current.bytes > 0
+            && current.bytes >= indexSettings.getFlushOnUncommittedSegmentsThresholdSize().getBytes()
+            && current.committedInfosGeneration == committedInfos.getGeneration();
+    }
+
     @Override
     public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
         ensureOpen();

server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java

@@ -436,6 +436,14 @@ private void onSuccessfulSegmentsSync(
         resetBackOffDelayIterator();
         // Set the minimum sequence number for keeping translog
         indexShard.getIndexer().translogManager().setMinSeqNoToKeep(lastRefreshedCheckpoint + 1);
+        // The above trimming makes the translog-size based periodic flush condition ineffective on remote-store
+        // shards, hence publish the size of segment bytes not yet referenced by the last commit point so that the
+        // engine can flush once they breach the configured threshold.
+        if (indexShard.indexSettings().isFlushOnUncommittedSegmentsEnabled()
+            && indexShard.getIndexer() instanceof EngineBackedIndexer engineBacked
+            && engineBacked.getEngine() instanceof InternalEngine internalEngine) {
+            internalEngine.updateUncommittedSegmentBytes(localFileSizeMap);
+        }
         // Publishing the new checkpoint which is used for remote store + segrep indexes
         checkpointPublisher.publish(indexShard, checkpoint);
         logger.debug("onSuccessfulSegmentsSync lastRefreshedCheckpoint={} checkpoint={}", lastRefreshedCheckpoint, checkpoint);

server/src/test/java/org/opensearch/index/IndexSettingsTests.java

@@ -42,6 +42,7 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.settings.SettingsException;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.unit.ByteSizeUnit;
 import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.replication.common.ReplicationType;
@@ -593,6 +594,55 @@ public void testTranslogFlushSizeThreshold() {
         assertEquals(actualNewTranslogFlushThresholdSize, settings.getFlushThresholdSize());
     }
 
+    public void testFlushOnUncommittedSegmentsThresholdSize() {
+        // defaults to the index's translog flush threshold via setting fallback
+        IndexMetadata metadata = newIndexMeta(
+            "index",
+            Settings.builder()
+                .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT)
+                .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "128mb")
+                .build()
+        );
+        IndexSettings settings = new IndexSettings(metadata, Settings.EMPTY);
+        assertEquals(new ByteSizeValue(128, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize());
+        // a dynamic update of only the fallback setting is reflected
+        settings.updateIndexMetadata(
+            newIndexMeta(
+                "index",
+                Settings.builder().put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "256mb").build()
+            )
+        );
+        assertEquals(new ByteSizeValue(256, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize());
+        // an explicit value wins over the fallback
+        settings.updateIndexMetadata(
+            newIndexMeta(
+                "index",
+                Settings.builder()
+                    .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "256mb")
+                    .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "64mb")
+                    .build()
+            )
+        );
+        assertEquals(new ByteSizeValue(64, ByteSizeUnit.MB), settings.getFlushOnUncommittedSegmentsThresholdSize());
+        // zero and negative values are rejected: they would flush on every successful segments sync
+        for (String invalid : new String[] { "0b", "-1" }) {
+            IllegalArgumentException e = expectThrows(
+                IllegalArgumentException.class,
+                () -> new IndexSettings(
+                    newIndexMeta(
+                        "index",
+                        Settings.builder()
+                            .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT)
+                            .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), invalid)
+                            .build()
+                    ),
+                    Settings.EMPTY
+                )
+            );
+            assertTrue(e.getMessage(), e.getMessage().contains("failed to parse value [" + invalid + "]"));
+        }
+    }
+
     public void testTranslogGenerationSizeThreshold() {
         final ByteSizeValue size = new ByteSizeValue(Math.abs(randomInt()));
         final String key = IndexSettings.INDEX_TRANSLOG_GENERATION_THRESHOLD_SIZE_SETTING.getKey();