[Failure store] Introduce dedicated failure store lifecycle configuration (elastic#127314)

The failure store is a set of data stream indices that are used to store certain type of ingestion failures. Until this moment they were sharing the configuration of the backing indices. We understand that the two data sets have different lifecycle needs.

We believe that typically the failures will need to be retained much less than the data. Considering this we believe the lifecycle needs of the failures also more limited and they fit better the simplicity of the data stream lifecycle feature.

This allows the user to only set the desired retention and we will perform the rollover and other maintenance tasks without the user having to think about them. Furthermore, having only one lifecycle management feature allows us to ensure that these data is managed by default.

This PR introduces the following:

Configuration

We extend the failure store configuration to allow lifecycle configuration too, this configuration reflects the user's configuration only as shown below:

PUT _data_stream/*/options
{
  "failure_store": {
     "lifecycle": {
       "data_retention": "5d"
     }
  }
}

GET _data_stream/*/options

{
  "data_streams": [
    {
      "name": "my-ds",
      "options": {
        "failure_store": {
          "lifecycle": {
            "data_retention": "5d"
          }
        }
      }
    }
  ]
}
To retrieve the effective configuration you need to use the GET data streams API, see elastic#126668

Functionality

The data stream lifecycle (DLM) will manage the failure indices regardless if the failure store is enabled or not. This will ensure that if the failure store gets disabled we will not have stagnant data.
The data stream options APIs reflect only the user's configuration.
The GET data stream API should be used to check the current state of the effective failure store configuration.
Telemetry
We extend the data stream failure store telemetry to also include the lifecycle telemetry.

{
  "data_streams": {
     "available": true,
     "enabled": true,
     "data_streams": 10,
     "indices_count": 50,
     "failure_store": {
       "explicitly_enabled_count": 1,
       "effectively_enabled_count": 15,
       "failure_indices_count": 30
       "lifecycle": {
         "explicitly_enabled_count": 5,
         "effectively_enabled_count": 20,
         "data_retention": {
           "configured_data_streams": 5,
           "minimum_millis": X,
           "maximum_millis": Y,
           "average_millis": Z,
          },
          "effective_retention": {
            "retained_data_streams": 20,
            "minimum_millis": X,
            "maximum_millis": Y,
            "average_millis": Z
          },
         "global_retention": {
           "max": {
             "defined": false
           },
           "default": {
             "defined": true,  <------ this is the default value applicable for the failure store
             "millis": X
           }
        }
      }
   }
}
Implementation details

We ensure that partially reset failure store will create valid failure store configuration.
We ensure that when a node communicates with a note with a previous version it will ensure it will not send an invalid failure store configuration enabled: null.

(cherry picked from commit 03d7781)

# Conflicts:
#	modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java
#	modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java
#	modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java
#	server/src/main/java/org/elasticsearch/TransportVersions.java
#	server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDataStreamsService.java
#	server/src/test/java/org/elasticsearch/cluster/metadata/DataStreamTests.java

Author: gmarouli

docs/changelog/127314.yaml

@@ -0,0 +1,5 @@
+pr: 127314
+summary: "[Failure store] Introduce dedicated failure store lifecycle configuration"
+area: Data streams
+type: enhancement
+issues: []

modules/data-streams/build.gradle

@@ -78,4 +78,5 @@ tasks.named("yamlRestTestV7CompatTransform").configure({ task ->
   task.skipTest("data_stream/210_rollover_failure_store/Lazily roll over a data stream's failure store after an ingest failure", "Rolling over a data stream using target_failure_store is no longer supported.")
   task.skipTest("data_stream/210_rollover_failure_store/Lazily roll over a data stream's failure store after a shard failure", "Rolling over a data stream using target_failure_store is no longer supported.")
   task.skipTest("data_stream/210_rollover_failure_store/Roll over a data stream's failure store without conditions", "Rolling over a data stream using target_failure_store is no longer supported.")
+  task.skipTest("data_stream/240_failure_store_info/Get failure store info from explicitly enabled failure store and disabled lifecycle", "failure store lifecycle is not using anymore the data stream lifecycle")
 })

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java

@@ -28,19 +28,20 @@
 import org.elasticsearch.action.datastreams.DeleteDataStreamAction;
 import org.elasticsearch.action.datastreams.GetDataStreamAction;
 import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
+import org.elasticsearch.action.datastreams.PutDataStreamOptionsAction;
 import org.elasticsearch.action.datastreams.lifecycle.ErrorEntry;
 import org.elasticsearch.action.datastreams.lifecycle.ExplainDataStreamLifecycleAction;
 import org.elasticsearch.action.datastreams.lifecycle.ExplainIndexDataStreamLifecycle;
 import org.elasticsearch.action.datastreams.lifecycle.PutDataStreamLifecycleAction;
-import org.elasticsearch.action.downsample.DownsampleConfig;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.cluster.coordination.StableMasterHealthIndicatorService;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamAction;
+import org.elasticsearch.cluster.metadata.DataStreamFailureStore;
 import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
-import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
+import org.elasticsearch.cluster.metadata.DataStreamOptions;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -70,7 +71,6 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.SystemIndexPlugin;
 import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.xcontent.ToXContent;
@@ -1030,17 +1030,8 @@ public void testReenableDataStreamLifecycle() throws Exception {
     }
 
     public void testLifecycleAppliedToFailureStore() throws Exception {
-        // We configure a lifecycle with downsampling to ensure it doesn't fail
-        DataStreamLifecycle.Template lifecycle = DataStreamLifecycle.dataLifecycleBuilder()
+        DataStreamLifecycle.Template lifecycle = DataStreamLifecycle.failuresLifecycleBuilder()
             .dataRetention(TimeValue.timeValueSeconds(20))
-            .downsampling(
-                List.of(
-                    new DataStreamLifecycle.DownsamplingRound(
-                        TimeValue.timeValueMillis(10),
-                        new DownsampleConfig(new DateHistogramInterval("10m"))
-                    )
-                )
-            )
             .buildTemplate();
 
         putComposableIndexTemplate("id1", """
@@ -1054,7 +1045,7 @@ public void testLifecycleAppliedToFailureStore() throws Exception {
                    "type": "boolean"
                  }
              }
-            }""", List.of("metrics-fs*"), Settings.builder().put("index.number_of_replicas", 0).build(), null, lifecycle, true);
+            }""", List.of("metrics-fs*"), Settings.builder().put("index.number_of_replicas", 0).build(), null, null, lifecycle, true);
 
         String dataStreamName = "metrics-fs";
         CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(
@@ -1109,7 +1100,7 @@ public void testLifecycleAppliedToFailureStore() throws Exception {
             );
         });
 
-        updateLifecycle(dataStreamName, TimeValue.timeValueSeconds(1));
+        updateFailureStoreConfiguration(dataStreamName, true, TimeValue.timeValueSeconds(1));
 
         // And finally apply retention
         assertBusy(() -> {
@@ -1202,7 +1193,20 @@ static void putComposableIndexTemplate(
         List<String> patterns,
         @Nullable Settings settings,
         @Nullable Map<String, Object> metadata,
-        @Nullable DataStreamLifecycle.Template lifecycle,
+        @Nullable DataStreamLifecycle.Template dataLifecycle,
+        boolean withFailureStore
+    ) throws IOException {
+        putComposableIndexTemplate(id, mappings, patterns, settings, metadata, dataLifecycle, null, withFailureStore);
+    }
+
+    static void putComposableIndexTemplate(
+        String id,
+        @Nullable String mappings,
+        List<String> patterns,
+        @Nullable Settings settings,
+        @Nullable Map<String, Object> metadata,
+        @Nullable DataStreamLifecycle.Template dataLifecycle,
+        @Nullable DataStreamLifecycle.Template failuresLifecycle,
         boolean withFailureStore
     ) throws IOException {
         TransportPutComposableIndexTemplateAction.Request request = new TransportPutComposableIndexTemplateAction.Request(id);
@@ -1213,8 +1217,10 @@ static void putComposableIndexTemplate(
                     Template.builder()
                         .settings(settings)
                         .mappings(mappings == null ? null : CompressedXContent.fromJSON(mappings))
-                        .lifecycle(lifecycle)
-                        .dataStreamOptions(DataStreamTestHelper.createDataStreamOptionsTemplate(withFailureStore))
+                        .lifecycle(dataLifecycle)
+                        .dataStreamOptions(
+                            new DataStreamOptions.Template(new DataStreamFailureStore.Template(withFailureStore, failuresLifecycle))
+                        )
                 )
                 .metadata(metadata)
                 .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate())
@@ -1233,6 +1239,16 @@ static void updateLifecycle(String dataStreamName, TimeValue dataRetention) {
         assertAcked(client().execute(PutDataStreamLifecycleAction.INSTANCE, putDataLifecycleRequest));
     }
 
+    static void updateFailureStoreConfiguration(String dataStreamName, boolean enabled, TimeValue retention) {
+        PutDataStreamOptionsAction.Request putDataOptionsRequest = new PutDataStreamOptionsAction.Request(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            new String[] { dataStreamName },
+            new DataStreamFailureStore(enabled, DataStreamLifecycle.failuresLifecycleBuilder().dataRetention(retention).build())
+        );
+        assertAcked(client().execute(PutDataStreamOptionsAction.INSTANCE, putDataOptionsRequest));
+    }
+
     /*
      * This test plugin adds `.system-test` as a known system data stream. The data stream is not created by this plugin. But if it is
      * created, it will be a system data stream.

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/ExplainDataStreamLifecycleIT.java

@@ -264,7 +264,7 @@ public void testExplainFailuresLifecycle() throws Exception {
             List.of("metrics-foo*"),
             null,
             null,
-            DataStreamLifecycle.Template.DATA_DEFAULT,
+            null,
             new DataStreamOptions.Template(DataStreamFailureStore.builder().enabled(true).buildTemplate())
         );
         String dataStreamName = "metrics-foo";
@@ -354,26 +354,30 @@ public void testExplainFailuresLifecycle() throws Exception {
             ).actionGet();
             assertThat(response.getIndices().size(), is(1));
             for (ExplainIndexDataStreamLifecycle explainIndex : response.getIndices()) {
-                assertThat(explainIndex.isManagedByLifecycle(), is(true));
-                assertThat(explainIndex.getIndexCreationDate(), notNullValue());
-                assertThat(explainIndex.getLifecycle(), notNullValue());
-                assertThat(explainIndex.getLifecycle().dataRetention(), nullValue());
-
-                if (internalCluster().numDataNodes() > 1) {
-                    // If the number of nodes is 1 then the cluster will be yellow so forcemerge will report an error if it has run
-                    assertThat(explainIndex.getError(), nullValue());
-                }
-
-                if (explainIndex.getIndex().equals(firstGenerationIndex)) {
-                    // first generation index was rolled over
-                    assertThat(explainIndex.getRolloverDate(), notNullValue());
-                    assertThat(explainIndex.getTimeSinceRollover(System::currentTimeMillis), notNullValue());
-                    assertThat(explainIndex.getGenerationTime(System::currentTimeMillis), notNullValue());
+                if (explainIndex.getIndex().startsWith(DataStream.BACKING_INDEX_PREFIX)) {
+                    assertThat(explainIndex.isManagedByLifecycle(), is(false));
                 } else {
-                    // the write index has not been rolled over yet
-                    assertThat(explainIndex.getRolloverDate(), nullValue());
-                    assertThat(explainIndex.getTimeSinceRollover(System::currentTimeMillis), nullValue());
-                    assertThat(explainIndex.getGenerationTime(System::currentTimeMillis), nullValue());
+                    assertThat(explainIndex.isManagedByLifecycle(), is(true));
+                    assertThat(explainIndex.getIndexCreationDate(), notNullValue());
+                    assertThat(explainIndex.getLifecycle(), notNullValue());
+                    assertThat(explainIndex.getLifecycle().dataRetention(), nullValue());
+
+                    if (internalCluster().numDataNodes() > 1) {
+                        // If the number of nodes is 1 then the cluster will be yellow so forcemerge will report an error if it has run
+                        assertThat(explainIndex.getError(), nullValue());
+                    }
+
+                    if (explainIndex.getIndex().equals(firstGenerationIndex)) {
+                        // first generation index was rolled over
+                        assertThat(explainIndex.getRolloverDate(), notNullValue());
+                        assertThat(explainIndex.getTimeSinceRollover(System::currentTimeMillis), notNullValue());
+                        assertThat(explainIndex.getGenerationTime(System::currentTimeMillis), notNullValue());
+                    } else {
+                        // the write index has not been rolled over yet
+                        assertThat(explainIndex.getRolloverDate(), nullValue());
+                        assertThat(explainIndex.getTimeSinceRollover(System::currentTimeMillis), nullValue());
+                        assertThat(explainIndex.getGenerationTime(System::currentTimeMillis), nullValue());
+                    }
                 }
             }
         }

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamGlobalRetentionIT.java

@@ -42,16 +42,35 @@ public void setup() throws IOException {
               "index_patterns": ["my-data-stream*"],
               "data_stream": {},
               "template": {
-                "lifecycle": {}
+                "settings": {
+                  "number_of_replicas": 0
+                },
+                "mappings": {
+                  "properties": {
+                    "count": {
+                     "type": "long"
+                    }
+                  }
+                },
+                "lifecycle": {},
+                "data_stream_options": {
+                  "failure_store": {
+                    "enabled": true
+                  }
+                }
               }
             }
             """);
         assertOK(client().performRequest(putComposableIndexTemplateRequest));
 
-        // Create a data streams with one doc
+        // Index one doc, this will trigger a rollover
         Request createDocRequest = new Request("POST", "/my-data-stream/_doc?refresh=true");
         createDocRequest.setJsonEntity("{ \"@timestamp\": \"2022-12-12\"}");
         assertOK(client().performRequest(createDocRequest));
+        // Index one doc that will fail, this will create the failure store
+        createDocRequest = new Request("POST", "/my-data-stream/_doc?refresh=true");
+        createDocRequest.setJsonEntity("{ \"@timestamp\": \"2022-12-12\", \"count\": \"not-a-number\"}");
+        assertOK(client().performRequest(createDocRequest));
     }
 
     @After
@@ -64,7 +83,7 @@ public void cleanUp() throws IOException {
 
     @SuppressWarnings("unchecked")
     public void testDataStreamRetention() throws Exception {
-        // Set global retention and add retention to the data stream
+        // Set global retention and add retention to the data stream & failure store
         {
             updateClusterSettings(
                 Settings.builder()
@@ -78,6 +97,18 @@ public void testDataStreamRetention() throws Exception {
                   "data_retention": "10s"
                 }""");
             assertAcknowledged(client().performRequest(request));
+
+            request = new Request("PUT", "_data_stream/my-data-stream/_options");
+            request.setJsonEntity("""
+                {
+                  "failure_store": {
+                    "enabled": true,
+                    "lifecycle": {
+                      "data_retention": "10s"
+                    }
+                  }
+                }""");
+            assertAcknowledged(client().performRequest(request));
         }
 
         // Verify that the effective retention matches the default retention
@@ -92,6 +123,10 @@ public void testDataStreamRetention() throws Exception {
             assertThat(lifecycle.get("effective_retention"), is("10s"));
             assertThat(lifecycle.get("retention_determined_by"), is("data_stream_configuration"));
             assertThat(lifecycle.get("data_retention"), is("10s"));
+            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("data_stream_configuration"));
+            assertThat(failuresLifecycle.get("data_retention"), is("10s"));
         }
 
         // Verify that the first generation index was removed
@@ -101,8 +136,11 @@ public void testDataStreamRetention() throws Exception {
             assertThat(dataStream.get("name"), is("my-data-stream"));
             List<Object> backingIndices = (List<Object>) dataStream.get("indices");
             assertThat(backingIndices.size(), is(1));
+            List<Object> failureIndices = (List<Object>) ((Map<String, Object>) dataStream.get("failure_store")).get("indices");
+            assertThat(failureIndices.size(), is(1));
             // 2 backing indices created + 1 for the deleted index
-            assertThat(dataStream.get("generation"), is(3));
+            // 2 failure indices created + 1 for the deleted failure index
+            assertThat(dataStream.get("generation"), is(6));
         }, 20, TimeUnit.SECONDS);
     }
 
@@ -123,6 +161,10 @@ public void testDefaultRetention() throws Exception {
             assertThat(lifecycle.get("effective_retention"), is("10s"));
             assertThat(lifecycle.get("retention_determined_by"), is("default_global_retention"));
             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("data_retention"), nullValue());
         }
 
         // Verify that the first generation index was removed
@@ -132,8 +174,11 @@ public void testDefaultRetention() throws Exception {
             assertThat(dataStream.get("name"), is("my-data-stream"));
             List<Object> backingIndices = (List<Object>) dataStream.get("indices");
             assertThat(backingIndices.size(), is(1));
+            List<Object> failureIndices = (List<Object>) ((Map<String, Object>) dataStream.get("failure_store")).get("indices");
+            assertThat(failureIndices.size(), is(1));
             // 2 backing indices created + 1 for the deleted index
-            assertThat(dataStream.get("generation"), is(3));
+            // 2 failure indices created + 1 for the deleted failure index
+            assertThat(dataStream.get("generation"), is(6));
         }, 20, TimeUnit.SECONDS);
     }
 
@@ -157,6 +202,24 @@ public void testMaxRetention() throws Exception {
                     containsString("The retention provided [30d] is exceeding the max allowed data retention of this project [10s]")
                 );
             }
+            try {
+                Request request = new Request("PUT", "_data_stream/my-data-stream/_options");
+                request.setJsonEntity("""
+                    {
+                      "failure_store": {
+                        "lifecycle": {
+                          "data_retention": "30d"
+                        }
+                      }
+                    }""");
+                assertAcknowledged(client().performRequest(request));
+                fail("Should have returned a warning about data retention exceeding the max retention");
+            } catch (WarningFailureException warningFailureException) {
+                assertThat(
+                    warningFailureException.getMessage(),
+                    containsString("The retention provided [30d] is exceeding the max allowed data retention of this project [10s]")
+                );
+            }
         }
 
         // Verify that the effective retention matches the max retention
@@ -175,6 +238,14 @@ public void testMaxRetention() throws Exception {
             } else {
                 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("max_global_retention"));
+            if (withDataStreamLevelRetention) {
+                assertThat(failuresLifecycle.get("data_retention"), is("30d"));
+            } else {
+                assertThat(failuresLifecycle.get("data_retention"), nullValue());
+            }
         }
 
         // Verify that the first generation index was removed
@@ -184,8 +255,11 @@ public void testMaxRetention() throws Exception {
             assertThat(dataStream.get("name"), is("my-data-stream"));
             List<Object> backingIndices = (List<Object>) dataStream.get("indices");
             assertThat(backingIndices.size(), is(1));
+            List<Object> failureIndices = (List<Object>) ((Map<String, Object>) dataStream.get("failure_store")).get("indices");
+            assertThat(failureIndices.size(), is(1));
             // 2 backing indices created + 1 for the deleted index
-            assertThat(dataStream.get("generation"), is(3));
+            // 2 failure indices created + 1 for the deleted failure index
+            assertThat(dataStream.get("generation"), is(6));
         }, 20, TimeUnit.SECONDS);
     }
 }