Introduce new setting for default failure store

gmarouli · gmarouli · commit 88e49f0290c7 · 2025-04-30T18:42:03.000+03:00
diff --git a/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamGlobalRetentionIT.java b/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamGlobalRetentionIT.java
@@ -147,7 +147,12 @@ public void testDataStreamRetention() throws Exception {
     @SuppressWarnings("unchecked")
     public void testDefaultRetention() throws Exception {
         // Set default global retention
-        updateClusterSettings(Settings.builder().put("data_streams.lifecycle.retention.default", "10s").build());
+        updateClusterSettings(
+            Settings.builder()
+                .put("data_streams.lifecycle.retention.default", "10s")
+                .put("data_streams.lifecycle.retention.failures_default", "10s")
+                .build()
+        );
 
         // Verify that the effective retention matches the default retention
         {
@@ -163,7 +168,7 @@ public void testDefaultRetention() throws Exception {
             assertThat(lifecycle.get("data_retention"), nullValue());
             Map<String, Object> failuresLifecycle = ((Map<String, Map<String, Object>>) dataStream.get("failure_store")).get("lifecycle");
             assertThat(failuresLifecycle.get("effective_retention"), is("10s"));
-            assertThat(failuresLifecycle.get("retention_determined_by"), is("default_global_retention"));
+            assertThat(failuresLifecycle.get("retention_determined_by"), is("default_failures_retention"));
             assertThat(failuresLifecycle.get("data_retention"), nullValue());
         }
 
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamGlobalRetention.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamGlobalRetention.java
@@ -23,6 +23,10 @@
 public record DataStreamGlobalRetention(@Nullable TimeValue defaultRetention, @Nullable TimeValue maxRetention) implements Writeable {
 
     public static final TimeValue MIN_RETENTION_VALUE = TimeValue.timeValueSeconds(10);
+    public static final DataStreamGlobalRetention DEFAULT_FAILURES_CONFIG = new DataStreamGlobalRetention(
+        DataStreamGlobalRetentionSettings.FAILURES_DEFAULT_RETENTION,
+        null
+    );
 
     /**
      * @param defaultRetention the default retention or null if it's undefined
@@ -46,6 +50,21 @@ public DataStreamGlobalRetention(TimeValue defaultRetention, TimeValue maxRetent
         this.maxRetention = maxRetention;
     }
 
+    /**
+     * Helper method that creates a global retention object or returns null in case both retentions are null
+     */
+    @Nullable
+    public static DataStreamGlobalRetention create(@Nullable TimeValue defaultRetention, @Nullable TimeValue maxRetention) {
+        if (defaultRetention == null && maxRetention == null) {
+            return null;
+        }
+        if (maxRetention == null
+            && DataStreamGlobalRetentionSettings.FAILURES_DEFAULT_RETENTION.getMillis() == defaultRetention.getMillis()) {
+            return DEFAULT_FAILURES_CONFIG;
+        }
+        return new DataStreamGlobalRetention(defaultRetention, maxRetention);
+    }
+
     private boolean validateRetentionValue(@Nullable TimeValue retention) {
         return retention == null || retention.getMillis() >= MIN_RETENTION_VALUE.getMillis();
     }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamGlobalRetentionSettings.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamGlobalRetentionSettings.java
@@ -13,6 +13,7 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 
@@ -24,8 +25,10 @@
  * This class holds the data stream global retention settings. It defines, validates and monitors the settings.
  * <p>
  * The global retention settings apply to non-system data streams that are managed by the data stream lifecycle. They consist of:
- * - The default retention which applies to data streams that do not have a retention defined.
- * - The max retention which applies to all data streams that do not have retention or their retention has exceeded this value.
+ * - The default retention which applies to the backing indices of data streams that do not have a retention defined.
+ * - The max retention which applies to backing and failure indices of data streams that do not have retention or their
+ * retention has exceeded this value.
+ * - The failures default retention which applied to the failure indices of data streams that do not have retention defined.
  */
 public class DataStreamGlobalRetentionSettings {
 
@@ -82,27 +85,66 @@ public Iterator<Setting<?>> settings() {
         Setting.Property.Dynamic
     );
 
+    static final TimeValue FAILURES_DEFAULT_RETENTION = TimeValue.timeValueDays(30);
+    public static final Setting<TimeValue> FAILURE_STORE_DEFAULT_RETENTION_SETTING = Setting.timeSetting(
+        "data_streams.lifecycle.retention.failures_default",
+        FAILURES_DEFAULT_RETENTION,
+        new Setting.Validator<>() {
+            @Override
+            public void validate(TimeValue value) {}
+
+            @Override
+            public void validate(final TimeValue settingValue, final Map<Setting<?>, Object> settings) {
+                TimeValue defaultRetention = getSettingValueOrNull(settingValue);
+                // Currently, we do not validate the default for the failure store against the max because
+                // we start with a default value that might conflict the max retention.
+                validateIsolatedRetentionValue(defaultRetention, FAILURE_STORE_DEFAULT_RETENTION_SETTING.getKey());
+            }
+        },
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
     @Nullable
     private volatile TimeValue defaultRetention;
     @Nullable
     private volatile TimeValue maxRetention;
+    @Nullable
+    private volatile TimeValue failuresDefaultRetention;
+    /** We cache the global retention objects, volatile is sufficient we only "write" this values in the settings appliers which
+     * are executed by {@link org.elasticsearch.common.settings.AbstractScopedSettings#applySettings(Settings)} which is synchronised.
+     */
+    @Nullable
+    private volatile DataStreamGlobalRetention dataGlobalRetention;
+    @Nullable
+    private volatile DataStreamGlobalRetention failuresGlobalRetention;
 
     private DataStreamGlobalRetentionSettings() {
 
     }
 
+    /**
+     * @return the max retention that applies to all data stream data
+     */
     @Nullable
     public TimeValue getMaxRetention() {
         return maxRetention;
     }
 
+    /**
+     * @return the default retention that applies either to the data component
+     */
     @Nullable
     public TimeValue getDefaultRetention() {
         return defaultRetention;
     }
 
-    public boolean areDefined() {
-        return getDefaultRetention() != null || getMaxRetention() != null;
+    /**
+     * @return the default retention that applies either to the data or the failures component
+     */
+    @Nullable
+    public TimeValue getDefaultRetention(boolean failureStore) {
+        return failureStore ? failuresDefaultRetention : defaultRetention;
     }
 
     /**
@@ -113,17 +155,33 @@ public static DataStreamGlobalRetentionSettings create(ClusterSettings clusterSe
         DataStreamGlobalRetentionSettings dataStreamGlobalRetentionSettings = new DataStreamGlobalRetentionSettings();
         clusterSettings.initializeAndWatch(DATA_STREAMS_DEFAULT_RETENTION_SETTING, dataStreamGlobalRetentionSettings::setDefaultRetention);
         clusterSettings.initializeAndWatch(DATA_STREAMS_MAX_RETENTION_SETTING, dataStreamGlobalRetentionSettings::setMaxRetention);
+        clusterSettings.initializeAndWatch(
+            FAILURE_STORE_DEFAULT_RETENTION_SETTING,
+            dataStreamGlobalRetentionSettings::setFailuresDefaultRetention
+        );
         return dataStreamGlobalRetentionSettings;
     }
 
     private void setMaxRetention(TimeValue maxRetention) {
         this.maxRetention = getSettingValueOrNull(maxRetention);
-        logger.info("Updated max factory retention to [{}]", this.maxRetention == null ? null : maxRetention.getStringRep());
+        this.dataGlobalRetention = createDataStreamGlobalRetention(false);
+        this.failuresGlobalRetention = createDataStreamGlobalRetention(true);
+        logger.info("Updated global max retention to [{}]", this.maxRetention == null ? null : maxRetention.getStringRep());
     }
 
     private void setDefaultRetention(TimeValue defaultRetention) {
         this.defaultRetention = getSettingValueOrNull(defaultRetention);
-        logger.info("Updated default factory retention to [{}]", this.defaultRetention == null ? null : defaultRetention.getStringRep());
+        this.dataGlobalRetention = createDataStreamGlobalRetention(false);
+        logger.info("Updated global default retention to [{}]", this.defaultRetention == null ? null : defaultRetention.getStringRep());
+    }
+
+    private void setFailuresDefaultRetention(TimeValue failuresDefaultRetention) {
+        this.failuresDefaultRetention = getSettingValueOrNull(failuresDefaultRetention);
+        this.failuresGlobalRetention = createDataStreamGlobalRetention(true);
+        logger.info(
+            "Updated failures default retention to [{}]",
+            this.failuresDefaultRetention == null ? null : failuresDefaultRetention.getStringRep()
+        );
     }
 
     private static void validateIsolatedRetentionValue(@Nullable TimeValue retention, String settingName) {
@@ -150,12 +208,36 @@ private static void validateGlobalRetentionConfiguration(@Nullable TimeValue def
         }
     }
 
+    /**
+     * @return the global retention of backing indices
+     */
     @Nullable
     public DataStreamGlobalRetention get() {
-        if (areDefined() == false) {
+        return get(false);
+    }
+
+    /**
+     * Returns the global retention that applies to the data or failures of a data stream
+     * @param failureStore, true if we are retrieving the global retention that applies to failure store, false otherwise.
+     */
+    @Nullable
+    public DataStreamGlobalRetention get(boolean failureStore) {
+        return failureStore ? failuresGlobalRetention : dataGlobalRetention;
+    }
+
+    @Nullable
+    private DataStreamGlobalRetention createDataStreamGlobalRetention(boolean failureStore) {
+        if (areDefined(failureStore) == false) {
             return null;
         }
-        return new DataStreamGlobalRetention(getDefaultRetention(), getMaxRetention());
+        TimeValue defaultRetention = getDefaultRetention(failureStore);
+        TimeValue maxRetention = getMaxRetention();
+        // We ensure that we create valid DataStreamGlobalRetention where default is less or equal to max.
+        // If it's not we set it to null.
+        if (defaultRetention != null && maxRetention != null && defaultRetention.getMillis() > maxRetention.getMillis()) {
+            return new DataStreamGlobalRetention(null, getMaxRetention());
+        }
+        return new DataStreamGlobalRetention(defaultRetention, maxRetention);
     }
 
     /**
@@ -169,4 +251,8 @@ public DataStreamGlobalRetention get() {
     private static TimeValue getSettingValueOrNull(TimeValue value) {
         return value == null || value.equals(TimeValue.MINUS_ONE) ? null : value;
     }
+
+    private boolean areDefined(boolean failureStore) {
+        return getDefaultRetention(failureStore) != null || getMaxRetention() != null;
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamLifecycle.java b/server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamLifecycle.java
@@ -233,7 +233,10 @@ public Tuple<TimeValue, RetentionSource> getEffectiveDataRetentionWithSource(
         }
         if (dataRetention() == null) {
             return globalRetention.defaultRetention() != null
-                ? Tuple.tuple(globalRetention.defaultRetention(), RetentionSource.DEFAULT_GLOBAL_RETENTION)
+                ? Tuple.tuple(
+                    globalRetention.defaultRetention(),
+                    targetsFailureStore() ? RetentionSource.DEFAULT_FAILURES_RETENTION : RetentionSource.DEFAULT_GLOBAL_RETENTION
+                )
                 : Tuple.tuple(globalRetention.maxRetention(), RetentionSource.MAX_GLOBAL_RETENTION);
         }
         if (globalRetention.maxRetention() != null && globalRetention.maxRetention().getMillis() < dataRetention().getMillis()) {
@@ -510,7 +513,8 @@ public static ToXContent.Params addEffectiveRetentionParams(ToXContent.Params pa
     public enum RetentionSource {
         DATA_STREAM_CONFIGURATION,
         DEFAULT_GLOBAL_RETENTION,
-        MAX_GLOBAL_RETENTION;
+        MAX_GLOBAL_RETENTION,
+        DEFAULT_FAILURES_RETENTION;
 
         public String displayName() {
             return this.toString().toLowerCase(Locale.ROOT);
diff --git a/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java b/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java
@@ -631,6 +631,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         TransportService.ENABLE_STACK_OVERFLOW_AVOIDANCE,
         DataStreamGlobalRetentionSettings.DATA_STREAMS_DEFAULT_RETENTION_SETTING,
         DataStreamGlobalRetentionSettings.DATA_STREAMS_MAX_RETENTION_SETTING,
+        DataStreamGlobalRetentionSettings.FAILURE_STORE_DEFAULT_RETENTION_SETTING,
         ShardsAvailabilityHealthIndicatorService.REPLICA_UNASSIGNED_BUFFER_TIME,
         DataStreamFailureStoreSettings.DATA_STREAM_FAILURE_STORED_ENABLED_SETTING,
         IndexingStatsSettings.RECENT_WRITE_LOAD_HALF_LIFE_SETTING,
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamGlobalRetentionSettingsTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamGlobalRetentionSettingsTests.java
@@ -27,6 +27,8 @@ public void testDefaults() {
 
         assertThat(globalRetentionSettings.getDefaultRetention(), nullValue());
         assertThat(globalRetentionSettings.getMaxRetention(), nullValue());
+        assertThat(globalRetentionSettings.get(false), nullValue());
+        assertThat(globalRetentionSettings.get(true), equalTo(DataStreamGlobalRetention.create(TimeValue.timeValueDays(30), null)));
     }
 
     public void testMonitorsDefaultRetention() {
@@ -43,7 +45,8 @@ public void testMonitorsDefaultRetention() {
             .build();
         clusterSettings.applySettings(newSettings);
 
-        assertThat(newDefaultRetention, equalTo(globalRetentionSettings.getDefaultRetention()));
+        assertThat(globalRetentionSettings.getDefaultRetention(), equalTo(newDefaultRetention));
+        assertThat(globalRetentionSettings.get(false), equalTo(DataStreamGlobalRetention.create(newDefaultRetention, null)));
 
         // Test invalid update
         Settings newInvalidSettings = Settings.builder()
@@ -57,20 +60,33 @@ public void testMonitorsDefaultRetention() {
             exception.getCause().getMessage(),
             containsString("Setting 'data_streams.lifecycle.retention.default' should be greater than")
         );
+        assertThat(globalRetentionSettings.get(false), equalTo(DataStreamGlobalRetention.create(newDefaultRetention, null)));
     }
 
     public void testMonitorsMaxRetention() {
         ClusterSettings clusterSettings = ClusterSettings.createBuiltInClusterSettings();
         DataStreamGlobalRetentionSettings globalRetentionSettings = DataStreamGlobalRetentionSettings.create(clusterSettings);
 
         // Test valid update
-        TimeValue newMaxRetention = TimeValue.timeValueDays(randomIntBetween(10, 30));
+        TimeValue newMaxRetention = TimeValue.timeValueDays(randomIntBetween(10, 29));
         Settings newSettings = Settings.builder()
             .put(DataStreamGlobalRetentionSettings.DATA_STREAMS_MAX_RETENTION_SETTING.getKey(), newMaxRetention.toHumanReadableString(0))
             .build();
         clusterSettings.applySettings(newSettings);
 
-        assertThat(newMaxRetention, equalTo(globalRetentionSettings.getMaxRetention()));
+        assertThat(globalRetentionSettings.getMaxRetention(), equalTo(newMaxRetention));
+        assertThat(globalRetentionSettings.get(false), equalTo(DataStreamGlobalRetention.create(null, newMaxRetention)));
+        assertThat(globalRetentionSettings.get(true), equalTo(DataStreamGlobalRetention.create(null, newMaxRetention)));
+
+        newMaxRetention = TimeValue.timeValueDays(100);
+        newSettings = Settings.builder()
+            .put(DataStreamGlobalRetentionSettings.DATA_STREAMS_MAX_RETENTION_SETTING.getKey(), newMaxRetention.toHumanReadableString(0))
+            .build();
+        clusterSettings.applySettings(newSettings);
+        assertThat(
+            globalRetentionSettings.get(true),
+            equalTo(DataStreamGlobalRetention.create(TimeValue.timeValueDays(30), newMaxRetention))
+        );
 
         // Test invalid update
         Settings newInvalidSettings = Settings.builder()
@@ -84,11 +100,57 @@ public void testMonitorsMaxRetention() {
             exception.getCause().getMessage(),
             containsString("Setting 'data_streams.lifecycle.retention.max' should be greater than")
         );
+        assertThat(globalRetentionSettings.get(false), equalTo(DataStreamGlobalRetention.create(null, newMaxRetention)));
+    }
+
+    public void testMonitorsDefaultFailuresRetention() {
+        ClusterSettings clusterSettings = ClusterSettings.createBuiltInClusterSettings();
+        DataStreamGlobalRetentionSettings globalRetentionSettings = DataStreamGlobalRetentionSettings.create(clusterSettings);
+
+        // Test valid update
+        TimeValue newDefaultRetention = TimeValue.timeValueDays(randomIntBetween(1, 10));
+        Settings newSettings = Settings.builder()
+            .put(
+                DataStreamGlobalRetentionSettings.FAILURE_STORE_DEFAULT_RETENTION_SETTING.getKey(),
+                newDefaultRetention.toHumanReadableString(0)
+            )
+            .build();
+        clusterSettings.applySettings(newSettings);
+
+        assertThat(globalRetentionSettings.getDefaultRetention(true), equalTo(newDefaultRetention));
+        assertThat(globalRetentionSettings.get(true), equalTo(DataStreamGlobalRetention.create(newDefaultRetention, null)));
+
+        // Test update default failures retention to infinite retention
+        newDefaultRetention = TimeValue.MINUS_ONE;
+        newSettings = Settings.builder()
+            .put(
+                DataStreamGlobalRetentionSettings.FAILURE_STORE_DEFAULT_RETENTION_SETTING.getKey(),
+                newDefaultRetention.toHumanReadableString(0)
+            )
+            .build();
+        clusterSettings.applySettings(newSettings);
+
+        assertThat(globalRetentionSettings.getDefaultRetention(true), nullValue());
+        assertThat(globalRetentionSettings.get(true), nullValue());
+
+        // Test invalid update
+        Settings newInvalidSettings = Settings.builder()
+            .put(DataStreamGlobalRetentionSettings.FAILURE_STORE_DEFAULT_RETENTION_SETTING.getKey(), TimeValue.ZERO)
+            .build();
+        IllegalArgumentException exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> clusterSettings.applySettings(newInvalidSettings)
+        );
+        assertThat(
+            exception.getCause().getMessage(),
+            containsString("Setting 'data_streams.lifecycle.retention.failures_default' should be greater than")
+        );
+        assertThat(globalRetentionSettings.get(true), nullValue());
     }
 
     public void testCombinationValidation() {
         ClusterSettings clusterSettings = ClusterSettings.createBuiltInClusterSettings();
-        DataStreamGlobalRetentionSettings.create(clusterSettings);
+        DataStreamGlobalRetentionSettings dataStreamGlobalRetentionSettings = DataStreamGlobalRetentionSettings.create(clusterSettings);
 
         // Test invalid update
         Settings newInvalidSettings = Settings.builder()
@@ -105,5 +167,17 @@ public void testCombinationValidation() {
                 "Setting [data_streams.lifecycle.retention.default=90d] cannot be greater than [data_streams.lifecycle.retention.max=30d]"
             )
         );
+
+        // Test valid update even if the failures default is greater than max.
+        Settings newValidSettings = Settings.builder()
+            .put(DataStreamGlobalRetentionSettings.FAILURE_STORE_DEFAULT_RETENTION_SETTING.getKey(), TimeValue.timeValueDays(90))
+            .put(DataStreamGlobalRetentionSettings.DATA_STREAMS_MAX_RETENTION_SETTING.getKey(), TimeValue.timeValueDays(30))
+            .build();
+        clusterSettings.applySettings(newValidSettings);
+        assertThat(dataStreamGlobalRetentionSettings.getDefaultRetention(true), equalTo(TimeValue.timeValueDays(90)));
+        assertThat(
+            dataStreamGlobalRetentionSettings.getDefaultRetention(true),
+            equalTo(DataStreamGlobalRetention.create(null, TimeValue.timeValueDays(30)))
+        );
     }
 }
