diff --git a/CHANGELOG.md b/CHANGELOG.md index 047431f465a39..2216043f0b749 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Add support for enabling pluggable data formats, starting with phase-1 of decoupling shard from engine, and introducing basic abstractions ([#20675](https://github.com/opensearch-project/OpenSearch/pull/20675)) - Add warmup phase to wait for lag to catch up in pull-based ingestion before serving ([#20526](https://github.com/opensearch-project/OpenSearch/pull/20526)) +- Make warmup settings dynamic for pull-based ingestion ([#20931](https://github.com/opensearch-project/OpenSearch/pull/20931)) ### Changed - Make telemetry `Tags` immutable ([#20788](https://github.com/opensearch-project/OpenSearch/pull/20788)) - Move Randomness from server to libs/common ([#20570](https://github.com/opensearch-project/OpenSearch/pull/20570)) diff --git a/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java b/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java index eb528115bbc02..2573b1b1e031d 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java @@ -1041,7 +1041,7 @@ public Iterator> settings() { TimeValue.timeValueMillis(-1), TimeValue.timeValueMillis(-1), Property.IndexScope, - Property.Final + Property.Dynamic ); /** @@ -1054,7 +1054,7 @@ public Iterator> settings() { 100L, 0L, Property.IndexScope, - Property.Final + Property.Dynamic ); /** diff --git a/server/src/main/java/org/opensearch/index/IndexSettings.java b/server/src/main/java/org/opensearch/index/IndexSettings.java index 2dc33b3017572..ccddaf57ee8c1 100644 --- a/server/src/main/java/org/opensearch/index/IndexSettings.java +++ b/server/src/main/java/org/opensearch/index/IndexSettings.java @@ -1070,6 +1070,16 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) { */ private volatile boolean isStarTreeIndexEnabled; + /** + * The warmup timeout for pull-based ingestion. + */ + private volatile TimeValue warmupTimeout; + + /** + * The warmup lag threshold for pull-based ingestion. + */ + private volatile long warmupLagThreshold; + /** * Returns the default search fields for this index. */ @@ -1239,6 +1249,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti setDocIdFuzzySetFalsePositiveProbability(scopedSettings.get(INDEX_DOC_ID_FUZZY_SET_FALSE_POSITIVE_PROBABILITY_SETTING)); isCompositeIndex = scopedSettings.get(StarTreeIndexSettings.IS_COMPOSITE_INDEX_SETTING); isStarTreeIndexEnabled = scopedSettings.get(StarTreeIndexSettings.STAR_TREE_SEARCH_ENABLED_SETTING); + this.warmupTimeout = IndexMetadata.INGESTION_SOURCE_WARMUP_TIMEOUT_SETTING.get(nodeSettings); + this.warmupLagThreshold = IndexMetadata.INGESTION_SOURCE_WARMUP_LAG_THRESHOLD_SETTING.get(nodeSettings); scopedSettings.addSettingsUpdateConsumer( TieredMergePolicyProvider.INDEX_COMPOUND_FORMAT_SETTING, tieredMergePolicyProvider::setNoCFSRatio @@ -1381,6 +1393,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti this::setRemoteStoreTranslogRepository ); scopedSettings.addSettingsUpdateConsumer(StarTreeIndexSettings.STAR_TREE_SEARCH_ENABLED_SETTING, this::setStarTreeIndexEnabled); + scopedSettings.addSettingsUpdateConsumer(IndexMetadata.INGESTION_SOURCE_WARMUP_TIMEOUT_SETTING, this::setWarmupTimeout); + scopedSettings.addSettingsUpdateConsumer(IndexMetadata.INGESTION_SOURCE_WARMUP_LAG_THRESHOLD_SETTING, this::setWarmupLagThreshold); } private void setSearchIdleAfter(TimeValue searchIdleAfter) { @@ -2002,6 +2016,22 @@ public boolean getStarTreeIndexEnabled() { return isStarTreeIndexEnabled; } + private void setWarmupTimeout(TimeValue warmupTimeout) { + this.warmupTimeout = warmupTimeout; + } + + private void setWarmupLagThreshold(long warmupLagThreshold) { + this.warmupLagThreshold = warmupLagThreshold; + } + + public TimeValue getWarmupTimeout() { + return warmupTimeout; + } + + public long getWarmupLagThreshold() { + return warmupLagThreshold; + } + /** * Returns the merge policy that should be used for this index. * diff --git a/server/src/main/java/org/opensearch/index/engine/IngestionEngine.java b/server/src/main/java/org/opensearch/index/engine/IngestionEngine.java index ebcbd776135cd..909ec4820b0f4 100644 --- a/server/src/main/java/org/opensearch/index/engine/IngestionEngine.java +++ b/server/src/main/java/org/opensearch/index/engine/IngestionEngine.java @@ -486,6 +486,15 @@ public void close() throws IOException { super.close(); } + /** + * Updates warmup configuration dynamically. + */ + public void updateWarmupConfig(IngestionSource.WarmupConfig newConfig) { + if (streamPoller != null) { + streamPoller.updateWarmupConfig(newConfig); + } + } + public DocumentMapperForType getDocumentMapperForType() { return documentMapperForType; } diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java index 1c155c897acba..f8338bc89c21e 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -68,6 +68,7 @@ import org.opensearch.action.support.replication.ReplicationResponse; import org.opensearch.cluster.metadata.DataStream; import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.cluster.metadata.IngestionSource; import org.opensearch.cluster.metadata.MappingMetadata; import org.opensearch.cluster.node.DiscoveryNode; import org.opensearch.cluster.node.DiscoveryNodes; @@ -3338,6 +3339,19 @@ public void onSettingsChanged() { indexSettings.getSoftDeleteRetentionOperations() ); } + + // Update warmup config if this is an ingestion engine + Indexer indexer = getIndexerOrNull(); + if (indexer instanceof EngineBackedIndexer) { + Engine engine = ((EngineBackedIndexer) indexer).getEngine(); + if (engine instanceof IngestionEngine) { + IngestionSource.WarmupConfig newConfig = new IngestionSource.WarmupConfig( + indexSettings.getWarmupTimeout(), + indexSettings.getWarmupLagThreshold() + ); + ((IngestionEngine) engine).updateWarmupConfig(newConfig); + } + } } private void turnOffTranslogRetention() { diff --git a/server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java b/server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java index 4dc51e0539d83..93d92e1f14368 100644 --- a/server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java +++ b/server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java @@ -63,10 +63,10 @@ public class DefaultStreamPoller implements StreamPoller { private volatile long lastPointerBasedLagUpdateTime = 0; // Warmup configuration and state - private final IngestionSource.WarmupConfig warmupConfig; + private volatile IngestionSource.WarmupConfig warmupConfig; private volatile boolean warmupComplete = false; private volatile long warmupStartTime = 0; - private final CountDownLatch warmupLatch = new CountDownLatch(1); + private volatile CountDownLatch warmupLatch = new CountDownLatch(1); @Nullable private IngestionShardConsumer consumer; @@ -430,6 +430,31 @@ public boolean awaitWarmupComplete(long timeoutMs) throws InterruptedException { return completed; } + /** + * Updates the warmup configuration dynamically. + * Called when index settings are changed at runtime. + */ + @Override + public void updateWarmupConfig(IngestionSource.WarmupConfig newConfig) { + IngestionSource.WarmupConfig oldConfig = this.warmupConfig; + this.warmupConfig = newConfig; + + // If warmup was enabled and is now disabled, mark as complete + if (oldConfig.isEnabled() && !newConfig.isEnabled() && !warmupComplete) { + warmupComplete = true; + warmupLatch.countDown(); + logger.info("Warmup disabled for index {} shard {} via dynamic settings update", indexName, shardId); + } + + logger.info( + "Warmup config updated for index {} shard {}: timeout={}, lagThreshold={}", + indexName, + shardId, + newConfig.timeout(), + newConfig.lagThreshold() + ); + } + /** * Check if warmup conditions are met and mark warmup as complete if so. * diff --git a/server/src/main/java/org/opensearch/indices/pollingingest/StreamPoller.java b/server/src/main/java/org/opensearch/indices/pollingingest/StreamPoller.java index 1d9249cb3f57c..d0617098660a3 100644 --- a/server/src/main/java/org/opensearch/indices/pollingingest/StreamPoller.java +++ b/server/src/main/java/org/opensearch/indices/pollingingest/StreamPoller.java @@ -83,6 +83,12 @@ public interface StreamPoller extends Closeable, ClusterStateListener { */ void requestConsumerReinitialization(IngestionSource updatedIngestionSource); + /** + * Updates the warmup configuration dynamically. + * Called when index settings are changed at runtime. + */ + void updateWarmupConfig(IngestionSource.WarmupConfig config); + /** * @return true if the warmup phase is complete and the shard is ready to serve */ diff --git a/server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java b/server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java index 7979c2c86dec1..20433445cf6e0 100644 --- a/server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java +++ b/server/src/test/java/org/opensearch/indices/pollingingest/DefaultStreamPollerTests.java @@ -976,4 +976,142 @@ public void testWarmupAwaitReturnsImmediatelyWhenAlreadyComplete() throws Interr warmupPoller.close(); } + + // ==================== Dynamic Warmup Config Update Tests ==================== + + public void testUpdateWarmupConfigDisableWhileInProgress() throws InterruptedException { + // Create a poller with warmup enabled + IngestionSource.WarmupConfig enabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 100L); + DefaultStreamPoller warmupPoller = new DefaultStreamPoller( + new FakeIngestionSource.FakeIngestionShardPointer(0), + fakeConsumerFactory, + "", + 0, + partitionedBlockingQueueContainer, + StreamPoller.ResetState.NONE, + "", + errorStrategy, + StreamPoller.State.NONE, + 1000, + 1000, + 10000, + indexSettings, + new DefaultIngestionMessageMapper(), + enabledConfig + ); + + // Warmup should not be complete yet + assertFalse(warmupPoller.isWarmupComplete()); + + // Dynamically disable warmup (timeout=-1) + IngestionSource.WarmupConfig disabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMillis(-1), 100L); + warmupPoller.updateWarmupConfig(disabledConfig); + + // Warmup should now be complete since we disabled it + assertTrue(warmupPoller.isWarmupComplete()); + + warmupPoller.close(); + } + + public void testUpdateWarmupConfigThresholdWhileInProgress() { + // Create a poller with warmup enabled and a high threshold + IngestionSource.WarmupConfig initialConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 100L); + DefaultStreamPoller warmupPoller = new DefaultStreamPoller( + new FakeIngestionSource.FakeIngestionShardPointer(0), + fakeConsumerFactory, + "", + 0, + partitionedBlockingQueueContainer, + StreamPoller.ResetState.NONE, + "", + errorStrategy, + StreamPoller.State.NONE, + 1000, + 1000, + 10000, + indexSettings, + new DefaultIngestionMessageMapper(), + initialConfig + ); + + // Warmup should not be complete yet + assertFalse(warmupPoller.isWarmupComplete()); + + // Update threshold to a different value while warmup is in progress + IngestionSource.WarmupConfig updatedConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 50L); + warmupPoller.updateWarmupConfig(updatedConfig); + + // Warmup should still not be complete (we just changed threshold, not disabled it) + assertFalse(warmupPoller.isWarmupComplete()); + + warmupPoller.close(); + } + + public void testUpdateWarmupConfigDoesNotReEnableAfterCompletion() { + // Create a poller with warmup disabled (warmup immediately complete) + IngestionSource.WarmupConfig disabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMillis(-1), 100L); + DefaultStreamPoller warmupPoller = new DefaultStreamPoller( + new FakeIngestionSource.FakeIngestionShardPointer(0), + fakeConsumerFactory, + "", + 0, + partitionedBlockingQueueContainer, + StreamPoller.ResetState.NONE, + "", + errorStrategy, + StreamPoller.State.NONE, + 1000, + 1000, + 10000, + indexSettings, + new DefaultIngestionMessageMapper(), + disabledConfig + ); + + // Warmup should be complete since it was disabled + assertTrue(warmupPoller.isWarmupComplete()); + + // Dynamically enable warmup - should NOT re-trigger since shard is already serving + IngestionSource.WarmupConfig enabledConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(5), 50L); + warmupPoller.updateWarmupConfig(enabledConfig); + + // Warmup should still be complete (not re-triggered) + assertTrue(warmupPoller.isWarmupComplete()); + + warmupPoller.close(); + } + + public void testUpdateWarmupConfigTimeoutWhileInProgress() { + // Create a poller with warmup enabled with a short timeout + IngestionSource.WarmupConfig initialConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueSeconds(30), 100L); + DefaultStreamPoller warmupPoller = new DefaultStreamPoller( + new FakeIngestionSource.FakeIngestionShardPointer(0), + fakeConsumerFactory, + "", + 0, + partitionedBlockingQueueContainer, + StreamPoller.ResetState.NONE, + "", + errorStrategy, + StreamPoller.State.NONE, + 1000, + 1000, + 10000, + indexSettings, + new DefaultIngestionMessageMapper(), + initialConfig + ); + + // Warmup should not be complete yet + assertFalse(warmupPoller.isWarmupComplete()); + + // Update timeout to a longer value while warmup is in progress + IngestionSource.WarmupConfig updatedConfig = new IngestionSource.WarmupConfig(TimeValue.timeValueMinutes(10), 100L); + warmupPoller.updateWarmupConfig(updatedConfig); + + // Warmup should still not be complete (we just changed timeout) + assertFalse(warmupPoller.isWarmupComplete()); + + warmupPoller.close(); + } }