[8.x] [Failure Store] Expose failure store lifecycle information via the GET data stream API (#126668) (#127255)

* [Failure Store] Expose failure store lifecycle information via the `GET` data stream API (#126668)

To retrieve the effective configuration you need to use the `GET` data
streams API, for example, if a data stream has empty data stream
options, it might still have failure store enabled from a cluster
setting. The failure store is managed by default with a lifecycle with
infinite (for now) retention, so the response will look like this:

```
GET _data_stream/*
{
  "data_streams": [
    {
      "name": "my-data-stream",
      "timestamp_field": {
        "name": "@timestamp"
      },
      .....
      "failure_store": {
        "enabled": true,
        "lifecycle": {
          "enabled": true
        },
        "rollover_on_write": false,
        "indices": [
           {
            "index_name": ".fs-my-data-stream-2099.03.08-000003",
            "index_uuid": "PA_JquKGSiKcAKBA8DJ5gw",
            "managed_by": "Data stream lifecycle"
          }
        ]
      }
    },...
]
```

In case there is a failure indexed managed by ILM the failure index info
will be displayed as follows.

```
      {
          "index_name": ".fs-my-data-stream-2099.03.08-000002",
          "index_uuid": "PA_JquKGSiKcAKBA8DJ5gw",
          "prefer_ilm": true,
          "ilm_policy": "my-lifecycle-policy",
          "managed_by": "Index Lifecycle Management"
        }
```

(cherry picked from commit db2992f)

# Conflicts:
#	modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsResponseTests.java
#	server/src/main/java/org/elasticsearch/action/datastreams/GetDataStreamAction.java

* Ignore correctly the failure store response

Author: gmarouli

docs/reference/data-streams/lifecycle/tutorial-migrate-data-stream-from-ilm-to-dsl.asciidoc

@@ -72,7 +72,7 @@ PUT _index_template/dsl-data-stream-template
 ----
 // TEST[continued]
 
-We'll now index a document targetting `dsl-data-stream` to create the data stream
+We'll now index a document targeting `dsl-data-stream` to create the data stream
 and we'll also manually rollover the data stream to have another generation index created:
 
 [source,console]
@@ -286,7 +286,7 @@ GET _data_stream/dsl-data-stream
 // TESTRESPONSE[s/"index_uuid": "xCEhwsp8Tey0-FLNFYVwSg"/"index_uuid": $body.data_streams.0.indices.0.index_uuid/]
 // TESTRESPONSE[s/"index_name": ".ds-dsl-data-stream-2023.10.19-000002"/"index_name": $body.data_streams.0.indices.1.index_name/]
 // TESTRESPONSE[s/"index_uuid": "PA_JquKGSiKcAKBA8DJ5gw"/"index_uuid": $body.data_streams.0.indices.1.index_uuid/]
-// TESTRESPONSE[s/"status": "GREEN"/"status": "YELLOW","failure_store":{"enabled": false, "indices": [], "rollover_on_write": true}/]
+// TESTRESPONSE[s/"status": "GREEN"/"status": "YELLOW","failure_store":$body.data_streams.0.failure_store/]
 
 <1> The existing backing index will continue to be managed by {ilm-init}
 <2> The existing backing index will continue to be managed by {ilm-init}
@@ -368,7 +368,7 @@ GET _data_stream/dsl-data-stream
 // TESTRESPONSE[s/"index_uuid": "PA_JquKGSiKcAKBA8DJ5gw"/"index_uuid": $body.data_streams.0.indices.1.index_uuid/]
 // TESTRESPONSE[s/"index_name": ".ds-dsl-data-stream-2023.10.19-000003"/"index_name": $body.data_streams.0.indices.2.index_name/]
 // TESTRESPONSE[s/"index_uuid": "PA_JquKGSiKcAKBA8abcd1"/"index_uuid": $body.data_streams.0.indices.2.index_uuid/]
-// TESTRESPONSE[s/"status": "GREEN"/"status": "YELLOW","failure_store":{"enabled": false, "indices": [], "rollover_on_write": true}/]
+// TESTRESPONSE[s/"status": "GREEN"/"status": "YELLOW","failure_store":$body.data_streams.0.failure_store/]
 
 <1> The backing indices that existed before rollover will continue to be managed by {ilm-init}
 <2> The backing indices that existed before rollover will continue to be managed by {ilm-init}
@@ -466,7 +466,7 @@ GET _data_stream/dsl-data-stream
 // TESTRESPONSE[s/"index_uuid": "PA_JquKGSiKcAKBA8DJ5gw"/"index_uuid": $body.data_streams.0.indices.1.index_uuid/]
 // TESTRESPONSE[s/"index_name": ".ds-dsl-data-stream-2023.10.19-000003"/"index_name": $body.data_streams.0.indices.2.index_name/]
 // TESTRESPONSE[s/"index_uuid": "PA_JquKGSiKcAKBA8abcd1"/"index_uuid": $body.data_streams.0.indices.2.index_uuid/]
-// TESTRESPONSE[s/"status": "GREEN"/"status": "YELLOW","failure_store":{"enabled": false, "indices": [], "rollover_on_write": true}/]
+// TESTRESPONSE[s/"status": "GREEN"/"status": "YELLOW","failure_store":$body.data_streams.0.failure_store/]
 <1> The write index is now managed by {ilm-init}
 <2> The `lifecycle` configured on the data stream is now disabled.
 <3> The next write index will be managed by {ilm-init}

modules/data-streams/src/main/java/org/elasticsearch/datastreams/rest/RestGetDataStreamsAction.java

@@ -44,6 +44,11 @@ public class RestGetDataStreamsAction extends BaseRestHandler {
             )
         )
     );
+    public static final String FAILURES_LIFECYCLE_API_CAPABILITY = "failure_store.lifecycle";
+    private static final Set<String> CAPABILITIES = Set.of(
+        DataStreamLifecycle.EFFECTIVE_RETENTION_REST_API_CAPABILITY,
+        FAILURES_LIFECYCLE_API_CAPABILITY
+    );
 
     @Override
     public String getName() {
@@ -74,7 +79,7 @@ public boolean allowSystemIndexAccessByDefault() {
 
     @Override
     public Set<String> supportedCapabilities() {
-        return Set.of(DataStreamLifecycle.EFFECTIVE_RETENTION_REST_API_CAPABILITY);
+        return CAPABILITIES;
     }
 
     @Override

modules/data-streams/src/test/java/org/elasticsearch/datastreams/action/GetDataStreamsResponseTests.java

@@ -38,6 +38,7 @@
 import static org.elasticsearch.cluster.metadata.DataStream.getDefaultBackingIndexName;
 import static org.elasticsearch.cluster.metadata.DataStream.getDefaultFailureStoreName;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 
 public class GetDataStreamsResponseTests extends AbstractWireSerializingTestCase<Response> {
@@ -163,21 +164,16 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
                     is(ManagedBy.LIFECYCLE.displayValue)
                 );
 
-                if (DataStream.isFailureStoreFeatureFlagEnabled()) {
-                    var failureStore = (Map<String, Object>) dataStreamMap.get(DataStream.FAILURE_STORE_FIELD.getPreferredName());
-                    List<Object> failureStoresRepresentation = (List<Object>) failureStore.get(DataStream.INDICES_FIELD.getPreferredName());
-                    Map<String, Object> failureStoreRepresentation = (Map<String, Object>) failureStoresRepresentation.get(0);
-                    assertThat(failureStoreRepresentation.get("index_name"), is(failureStoreIndex.getName()));
-                    assertThat(failureStoreRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
-                    assertThat(
-                        failureStoreRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()),
-                        is(nullValue())
-                    );
-                    assertThat(
-                        failureStoreRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
-                        is(ManagedBy.LIFECYCLE.displayValue)
-                    );
-                }
+                var failureStore = (Map<String, Object>) dataStreamMap.get(DataStream.FAILURE_STORE_FIELD.getPreferredName());
+                List<Object> failureIndices = (List<Object>) failureStore.get(DataStream.INDICES_FIELD.getPreferredName());
+                Map<String, Object> failureIndexRepresentation = (Map<String, Object>) failureIndices.get(0);
+                assertThat(failureIndexRepresentation.get("index_name"), is(failureStoreIndex.getName()));
+                assertThat(failureIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), nullValue());
+                assertThat(failureIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(nullValue()));
+                assertThat(
+                    failureIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.LIFECYCLE.displayValue)
+                );
             }
         }
 
@@ -256,21 +252,75 @@ public void testResponseIlmAndDataStreamLifecycleRepresentation() throws Excepti
                     is(ManagedBy.UNMANAGED.displayValue)
                 );
 
-                if (DataStream.isFailureStoreFeatureFlagEnabled()) {
-                    var failureStore = (Map<String, Object>) dataStreamMap.get(DataStream.FAILURE_STORE_FIELD.getPreferredName());
-                    List<Object> failureStoresRepresentation = (List<Object>) failureStore.get(DataStream.INDICES_FIELD.getPreferredName());
-                    Map<String, Object> failureStoreRepresentation = (Map<String, Object>) failureStoresRepresentation.get(0);
-                    assertThat(failureStoreRepresentation.get("index_name"), is(failureStoreIndex.getName()));
-                    assertThat(failureStoreRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), is(false));
-                    assertThat(
-                        failureStoreRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()),
-                        is(nullValue())
-                    );
-                    assertThat(
-                        failureStoreRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
-                        is(ManagedBy.UNMANAGED.displayValue)
-                    );
-                }
+                var failureStore = (Map<String, Object>) dataStreamMap.get(DataStream.FAILURE_STORE_FIELD.getPreferredName());
+                List<Object> failureStoresRepresentation = (List<Object>) failureStore.get(DataStream.INDICES_FIELD.getPreferredName());
+                Map<String, Object> failureStoreRepresentation = (Map<String, Object>) failureStoresRepresentation.get(0);
+                assertThat(failureStoreRepresentation.get("index_name"), is(failureStoreIndex.getName()));
+                assertThat(failureStoreRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), nullValue());
+                assertThat(failureStoreRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(nullValue()));
+                assertThat(
+                    failureStoreRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.UNMANAGED.displayValue)
+                );
+            }
+        }
+
+        {
+            // one failure index that have ILM policy
+            DataStream logs = DataStream.builder("logs", indices)
+                .setGeneration(3)
+                .setAllowCustomRouting(true)
+                .setIndexMode(IndexMode.STANDARD)
+                .setLifecycle(DataStreamLifecycle.DEFAULT_DATA_LIFECYCLE)
+                .setDataStreamOptions(DataStreamOptions.FAILURE_STORE_ENABLED)
+                .setFailureIndices(DataStream.DataStreamIndices.failureIndicesBuilder(failureStores).build())
+                .build();
+
+            String ilmPolicyName = "rollover-30days";
+            Map<Index, Response.IndexProperties> indexSettingsValues = Map.of(
+                firstGenerationIndex,
+                new Response.IndexProperties(true, ilmPolicyName, ManagedBy.ILM),
+                secondGenerationIndex,
+                new Response.IndexProperties(false, ilmPolicyName, ManagedBy.LIFECYCLE),
+                writeIndex,
+                new Response.IndexProperties(true, null, ManagedBy.LIFECYCLE),
+                failureStoreIndex,
+                new Response.IndexProperties(randomBoolean(), ilmPolicyName, ManagedBy.LIFECYCLE)
+            );
+
+            Response.DataStreamInfo dataStreamInfo = new Response.DataStreamInfo(
+                logs,
+                true,
+                ClusterHealthStatus.GREEN,
+                "index-template",
+                null,
+                null,
+                indexSettingsValues,
+                false,
+                null
+            );
+            Response response = new Response(List.of(dataStreamInfo));
+            XContentBuilder contentBuilder = XContentFactory.jsonBuilder();
+            response.toXContent(contentBuilder, ToXContent.EMPTY_PARAMS);
+
+            BytesReference bytes = BytesReference.bytes(contentBuilder);
+            try (XContentParser parser = createParser(JsonXContent.jsonXContent, bytes)) {
+                Map<String, Object> map = parser.map();
+                List<Object> dataStreams = (List<Object>) map.get(Response.DATA_STREAMS_FIELD.getPreferredName());
+                assertThat(dataStreams.size(), is(1));
+                Map<String, Object> dataStreamMap = (Map<String, Object>) dataStreams.get(0);
+                assertThat(dataStreamMap.get(DataStream.NAME_FIELD.getPreferredName()), is(dataStreamName));
+
+                var failureStore = (Map<String, Object>) dataStreamMap.get(DataStream.FAILURE_STORE_FIELD.getPreferredName());
+                List<Object> failureIndices = (List<Object>) failureStore.get(DataStream.INDICES_FIELD.getPreferredName());
+                Map<String, Object> failureIndexRepresentation = (Map<String, Object>) failureIndices.get(0);
+                assertThat(failureIndexRepresentation.get("index_name"), is(failureStoreIndex.getName()));
+                assertThat(failureIndexRepresentation.get(Response.DataStreamInfo.PREFER_ILM.getPreferredName()), notNullValue());
+                assertThat(failureIndexRepresentation.get(Response.DataStreamInfo.ILM_POLICY_FIELD.getPreferredName()), is(ilmPolicyName));
+                assertThat(
+                    failureIndexRepresentation.get(Response.DataStreamInfo.MANAGED_BY.getPreferredName()),
+                    is(ManagedBy.LIFECYCLE.displayValue)
+                );
             }
         }
     }