diff --git a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java index b59e67749782b..ef7f4ed324e7d 100644 --- a/server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java +++ b/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, diff --git a/server/src/main/java/org/opensearch/common/settings/Setting.java b/server/src/main/java/org/opensearch/common/settings/Setting.java index d31ab400223ea..8e9b55d9f5d40 100644 --- a/server/src/main/java/org/opensearch/common/settings/Setting.java +++ b/server/src/main/java/org/opensearch/common/settings/Setting.java @@ -2109,6 +2109,16 @@ public static Setting byteSizeSetting(String key, Setting(key, fallbackSetting, new ByteSizeValueParser(key), properties); } + public static Setting byteSizeSetting( + String key, + Setting fallbackSetting, + ByteSizeValue minValue, + ByteSizeValue maxValue, + Property... properties + ) { + return new Setting<>(key, fallbackSetting, new ByteSizeValueParser(minValue, maxValue, key), properties); + } + public static Setting byteSizeSetting(String key, Function defaultValue, Property... properties) { return new Setting<>(key, defaultValue, new ByteSizeValueParser(key), properties); } diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java index 4b9ca27890e97..72f49f1c8324d 100644 --- a/server/src/main/java/org/opensearch/index/IndexSettings.java +++ b/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 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 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 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. */ diff --git a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java index 997161ffb60c2..db4bc20410150 100644 --- a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java +++ b/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 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 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 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(); diff --git a/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java b/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java index a7a4406e37abd..30e2158ae4d0d 100644 --- a/server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java +++ b/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); diff --git a/server/src/test/java/org/opensearch/index/IndexSettingsTests.java b/server/src/test/java/org/opensearch/index/IndexSettingsTests.java index 5fb949cdbbaa2..ffab2d8ae956b 100644 --- a/server/src/test/java/org/opensearch/index/IndexSettingsTests.java +++ b/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(); diff --git a/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java b/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java index 6cefab864c960..e4fa8bf92f28f 100644 --- a/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java +++ b/server/src/test/java/org/opensearch/index/engine/InternalEngineTests.java @@ -6812,6 +6812,66 @@ public void testShouldPeriodicallyFlushAfterMerge() throws Exception { assertThat(engine.shouldPeriodicallyFlush(), equalTo(false)); } + public void testShouldPeriodicallyFlushOnUncommittedSegmentBytes() throws Exception { + assertThat("Empty engine does not need flushing", engine.shouldPeriodicallyFlush(), equalTo(false)); + ParsedDocument doc = testParsedDocument("0", null, testDocumentWithTextField(), SOURCE, null); + engine.index(indexForDoc(doc)); + engine.refresh("test"); + assertThat("Nothing published yet", engine.shouldPeriodicallyFlush(), equalTo(false)); + + // simulate the remote segment upload path publishing the post-refresh local file sizes + final Map localSegmentsSizeMap = new HashMap<>(); + try (GatedCloseable snapshot = engine.getSegmentInfosSnapshot()) { + for (String file : snapshot.get().files(false)) { + localSegmentsSizeMap.put(file, engine.store.directory().fileLength(file)); + } + } + engine.updateUncommittedSegmentBytes(localSegmentsSizeMap); + assertThat( + "Uncommitted bytes below the default threshold inherited from the translog flush threshold", + engine.shouldPeriodicallyFlush(), + equalTo(false) + ); + + final IndexSettings indexSettings = engine.config().getIndexSettings(); + updateIndexSettings( + indexSettings, + Settings.builder().put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + ); + assertThat("Uncommitted bytes breach the lowered threshold", engine.shouldPeriodicallyFlush(), equalTo(true)); + + updateIndexSettings( + indexSettings, + Settings.builder() + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING.getKey(), false) + ); + assertThat("Disabling the condition takes effect immediately", engine.shouldPeriodicallyFlush(), equalTo(false)); + + updateIndexSettings( + indexSettings, + Settings.builder() + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_ENABLED_SETTING.getKey(), true) + ); + assertThat("Re-enabling restores the condition", engine.shouldPeriodicallyFlush(), equalTo(true)); + + engine.flush(); + assertThat("Stale commit generation stamp is ignored after flush", engine.shouldPeriodicallyFlush(), equalTo(false)); + + // republishing against the new commit computes zero uncommitted bytes, hence no flush loop + engine.updateUncommittedSegmentBytes(localSegmentsSizeMap); + assertThat("All published files are committed now", engine.shouldPeriodicallyFlush(), equalTo(false)); + } + + private static void updateIndexSettings(IndexSettings indexSettings, Settings.Builder settingsBuilder) { + indexSettings.updateIndexMetadata( + IndexMetadata.builder(indexSettings.getIndexMetadata()) + .settings(Settings.builder().put(indexSettings.getSettings()).put(settingsBuilder.build())) + .build() + ); + } + public void testStressShouldPeriodicallyFlush() throws Exception { final long flushThreshold = randomLongBetween(120, 5000); final long generationThreshold = randomLongBetween(1000, 5000); diff --git a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java index d34db204a112f..5f747bc074d76 100644 --- a/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java +++ b/test/framework/src/main/java/org/opensearch/remotestore/RemoteStoreCoreTestCase.java @@ -16,6 +16,7 @@ import org.opensearch.action.admin.indices.flush.FlushRequest; import org.opensearch.action.admin.indices.recovery.RecoveryResponse; import org.opensearch.action.admin.indices.settings.put.UpdateSettingsRequest; +import org.opensearch.action.bulk.BulkRequestBuilder; import org.opensearch.action.index.IndexResponse; import org.opensearch.action.search.SearchPhaseExecutionException; import org.opensearch.cluster.health.ClusterHealthStatus; @@ -27,7 +28,9 @@ import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.concurrent.BufferedAsyncIOProcessor; +import org.opensearch.index.IndexService; import org.opensearch.index.IndexSettings; +import org.opensearch.index.seqno.SequenceNumbers; import org.opensearch.index.shard.IndexShard; import org.opensearch.index.shard.IndexShardClosedException; import org.opensearch.index.translog.Translog; @@ -41,6 +44,7 @@ import org.opensearch.repositories.blobstore.BlobStoreRepository; import org.opensearch.snapshots.SnapshotInfo; import org.opensearch.snapshots.SnapshotState; +import org.opensearch.test.InternalSettingsPlugin; import org.opensearch.test.InternalTestCluster; import org.opensearch.test.OpenSearchIntegTestCase; import org.opensearch.test.transport.MockTransportService; @@ -59,6 +63,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -75,7 +80,9 @@ import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount; import static org.hamcrest.Matchers.comparesEqualTo; import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.lessThan; import static org.hamcrest.Matchers.oneOf; @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) @@ -85,7 +92,8 @@ public class RemoteStoreCoreTestCase extends RemoteStoreBaseIntegTestCase { @Override protected Collection> nodePlugins() { - return Stream.concat(super.nodePlugins().stream(), Stream.of(MockTransportService.TestPlugin.class)).collect(Collectors.toList()); + return Stream.concat(super.nodePlugins().stream(), Stream.of(MockTransportService.TestPlugin.class, InternalSettingsPlugin.class)) + .collect(Collectors.toList()); } @Override @@ -138,6 +146,128 @@ private void testPeerRecovery(int numberOfIterations, boolean invokeFlush) throw ); } + public void testPeriodicFlushOnUncommittedSegmentBytes() throws Exception { + String dataNode = internalCluster().startNodes(1).get(0); + // lower the uncommitted segment bytes threshold so that any refreshed but uncommitted segment breaches it; + // the condition itself is enabled by default + createIndex( + INDEX_NAME, + Settings.builder() + .put(remoteStoreIndexSettings(0)) + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "1b") + .build() + ); + ensureGreen(INDEX_NAME); + IndexShard indexShard = getIndexShard(dataNode, INDEX_NAME); + assertEquals(0, indexShard.flushStats().getPeriodic()); + // each iteration drives one refresh (and hence one remote segments sync publishing the uncommitted bytes) + // followed by a write operation which polls the periodic flush condition and triggers the async flush + assertBusy(() -> { + indexSingleDoc(INDEX_NAME); + refresh(INDEX_NAME); + indexSingleDoc(INDEX_NAME); + assertThat(indexShard.flushStats().getPeriodic(), greaterThan(0L)); + }, 30, TimeUnit.SECONDS); + } + + /** + * End-to-end reproduction of the scenario in issue #22802: an update-heavy workload with regular refreshes on a + * remote-store shard. Every successful segments upload advances minSeqNoToKeep which suppresses the translog + * based flush condition, so without the uncommitted-segment-bytes condition no periodic flush ever happens, the + * safe commit never advances, and the soft-deleted documents produced by the updates stay protected from merge + * reclamation indefinitely. This test asserts the full recovery chain: periodic flushes fire, the safe commit + * advances, and a merge that does NOT flush (flush=false) is able to reclaim the tombstones. + */ + public void testPeriodicFlushReclaimsSoftDeletesUnderUpdateHeavyWorkload() throws Exception { + final int docCount = 20; + final int rounds = 8; + String dataNode = internalCluster().startNodes(1).get(0); + createIndex( + INDEX_NAME, + Settings.builder() + .put(remoteStoreIndexSettings(0)) + // refreshes are driven manually by the workload below + .put("index.refresh_interval", "-1") + // do not make writes wait on the translog upload batching window + .put(IndexSettings.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING.getKey(), "0ms") + // let the primary's own peer-recovery retention lease follow the global checkpoint at test speed; + // otherwise the renewal gate (half the 12h lease period) keeps retention pinned at the initial + // checkpoint for the whole test regardless of commits + .put(IndexSettings.INDEX_SOFT_DELETES_RETENTION_LEASE_PERIOD_SETTING.getKey(), "0ms") + .put(IndexService.RETENTION_LEASE_SYNC_INTERVAL_SETTING.getKey(), "1s") + // pin the (suppressed anyway) translog condition high so any observed periodic flush is attributable + // to the uncommitted-segment-bytes condition only + .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), "1gb") + .put(IndexSettings.INDEX_REMOTE_STORE_FLUSH_ON_UNCOMMITTED_SEGMENTS_THRESHOLD_SIZE_SETTING.getKey(), "5kb") + // any segment with at least one reclaimable delete qualifies for the expunge-deletes merge below, + // independent of the randomized merge policy defaults + .put("index.merge.policy.expunge_deletes_allowed", "0.0") + // the random index template may set retention.operations up to 1000 which would retain the whole + // workload below the global checkpoint no matter how far the safe commit advances + .put(IndexSettings.INDEX_SOFT_DELETES_RETENTION_OPERATIONS_SETTING.getKey(), 0) + .build() + ); + ensureGreen(INDEX_NAME); + IndexShard indexShard = getIndexShard(dataNode, INDEX_NAME); + assertEquals(0, indexShard.flushStats().getPeriodic()); + + // overwrite the same document ids round after round; every round turns the previous round's documents into + // soft-deleted tombstones, and the refresh drives a remote segments upload + for (int round = 0; round < rounds; round++) { + BulkRequestBuilder bulk = client().prepareBulk(); + for (int i = 0; i < docCount; i++) { + bulk.add( + client().prepareIndex(INDEX_NAME) + .setId(String.valueOf(i)) + .setSource("field", "value-" + round + "-" + randomAlphaOfLength(100)) + ); + } + assertFalse(bulk.get().hasFailures()); + refresh(INDEX_NAME); + } + + // the periodic flush condition is polled on write operations and the counter is republished on each + // successful upload, so nudge with write+refresh cycles until the flushes have carried the safe commit past + // the whole workload (the flush is async, so a single trip is not enough on its own) + final long maxSeqNoAfterWorkload = indexShard.seqNoStats().getMaxSeqNo(); + assertBusy(() -> { + client().prepareIndex(INDEX_NAME).setId("0").setSource("field", randomAlphaOfLength(100)).get(); + refresh(INDEX_NAME); + assertThat(indexShard.flushStats().getPeriodic(), greaterThan(0L)); + long committedCheckpoint = Long.parseLong(indexShard.commitStats().getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)); + assertThat(committedCheckpoint, greaterThanOrEqualTo(maxSeqNoAfterWorkload)); + }, 30, TimeUnit.SECONDS); + + // a merge WITHOUT a flush can now reclaim tombstones below the advanced safe commit. Without the fix the + // soft-deletes floor never moves off the initial commit, nothing is reclaimable, and the count can never + // drop below the full tombstone pile -- so ANY reclamation proves the chain end to end. Each retry keeps + // writing so the retention lease and the async flush pipeline keep advancing. + final long tombstonesCreated = (long) (rounds - 1) * docCount; + final AtomicInteger nudge = new AtomicInteger(); + assertBusy(() -> { + client().prepareIndex(INDEX_NAME).setId("nudge-" + nudge.incrementAndGet()).setSource("field", "x").get(); + refresh(INDEX_NAME); + assertEquals( + 0, + client().admin().indices().prepareForceMerge(INDEX_NAME).setOnlyExpungeDeletes(true).setFlush(false).get().getFailedShards() + ); + refresh(INDEX_NAME); + long deletedDocs = client().admin().indices().prepareStats(INDEX_NAME).get().getPrimaries().getDocs().getDeleted(); + assertThat( + "deleted=" + + deletedDocs + + " commitCkpt=" + + indexShard.commitStats().getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY) + + " gcp=" + + indexShard.getLastSyncedGlobalCheckpoint() + + " leases=" + + indexShard.getRetentionLeases(), + deletedDocs, + lessThan(tombstonesCreated) + ); + }, 30, TimeUnit.SECONDS); + } + public void testRemoteStoreIndexCreationAndDeletionWithReferencedStore() throws InterruptedException, ExecutionException { String dataNode = internalCluster().startNodes(1).get(0); createIndex(INDEX_NAME, remoteStoreIndexSettings(0));