Implement lazy rollover for failure stores (#108108)

Allow marking failure stores for lazy rollover.

Author: nielsbauman

modules/data-streams/build.gradle

@@ -14,7 +14,7 @@ esplugin {
 restResources {
   restApi {
     include 'bulk', 'count', 'search', '_common', 'indices', 'index', 'cluster', 'rank_eval', 'reindex', 'update_by_query', 'delete_by_query',
-      'eql', 'data_stream', 'ingest', 'cat'
+      'eql', 'data_stream', 'ingest', 'cat', 'capabilities'
   }
 }
 

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/200_rollover_failure_store.yml

@@ -3,7 +3,7 @@ setup:
   - requires:
       cluster_features: ["gte_v8.15.0"]
       reason: "data stream failure stores REST structure changed in 8.15+"
-      test_runner_features: allowed_warnings
+      test_runner_features: [allowed_warnings, contains, capabilities]
 
   - do:
       allowed_warnings:
@@ -27,13 +27,24 @@ setup:
         name: data-stream-for-rollover
 
 ---
+teardown:
+  - do:
+      indices.delete_data_stream:
+        name: data-stream-for-lazy-rollover
+        ignore: 404
+
+  - do:
+      ingest.delete_pipeline:
+        id: failing_pipeline
+        ignore: 404
+---
 "Roll over a data stream's failure store without conditions":
-  # rollover data stream to create new backing index
   - do:
       indices.rollover:
         alias: "data-stream-for-rollover"
         target_failure_store: true
 
+  - match: { acknowledged: true }
   - match: { old_index: "/\\.fs-data-stream-for-rollover-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000001/" }
   - match: { new_index: "/\\.fs-data-stream-for-rollover-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000002/" }
   - match: { rolled_over: true }
@@ -54,7 +65,6 @@ setup:
 
 ---
 "Roll over a data stream's failure store with conditions":
-  # index first document and wait for refresh
   - do:
       index:
         index: data-stream-for-rollover
@@ -63,7 +73,6 @@ setup:
           '@timestamp': '2020-12-12'
           count: 'invalid value'
 
-  # rollover data stream to create new backing index
   - do:
       indices.rollover:
         alias: "data-stream-for-rollover"
@@ -72,6 +81,7 @@ setup:
           conditions:
             max_docs: 1
 
+  - match: { acknowledged: true }
   - match: { old_index: "/\\.fs-data-stream-for-rollover-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000001/" }
   - match: { new_index: "/\\.fs-data-stream-for-rollover-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000002/" }
   - match: { rolled_over: true }
@@ -92,7 +102,6 @@ setup:
 
 ---
 "Don't roll over a data stream's failure store when conditions aren't met":
-  # rollover data stream to create new backing index
   - do:
       indices.rollover:
         alias: "data-stream-for-rollover"
@@ -101,14 +110,219 @@ setup:
           conditions:
             max_docs: 1
 
+  - match: { acknowledged: false }
+  - match: { rolled_over: false }
+  - match: { dry_run: false }
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - match: { data_streams.0.name: data-stream-for-rollover }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 1 }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - length: { data_streams.0.failure_store.indices: 1 }
+  - match: { data_streams.0.failure_store.indices.0.index_name: '/\.fs-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+
+---
+"Lazily roll over a data stream's failure store after a shard failure":
+  - requires:
+      reason: "data stream failure store lazy rollover only supported in 8.15+"
+      test_runner_features: [allowed_warnings, capabilities]
+      capabilities:
+        - method: POST
+          path: /{index}/_rollover
+          capabilities: [lazy-rollover-failure-store]
+
+  # Mark the failure store for lazy rollover
+  - do:
+      indices.rollover:
+        alias: "data-stream-for-rollover"
+        target_failure_store: true
+        lazy: true
+
+  - match: { acknowledged: true }
+  - match: { rolled_over: false }
+  - match: { dry_run: false }
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - match: { data_streams.0.name: data-stream-for-rollover }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 1 }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - length: { data_streams.0.failure_store.indices: 1 }
+  - match: { data_streams.0.failure_store.indices.0.index_name: '/\.fs-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+
+  - do:
+      index:
+        index: data-stream-for-rollover
+        refresh: true
+        body:
+          '@timestamp': '2020-12-12'
+          count: 'invalid value'
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - match: { data_streams.0.name: data-stream-for-rollover }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  # Both backing and failure indices use the same generation field.
+  - match: { data_streams.0.generation: 2 }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - length: { data_streams.0.failure_store.indices: 2 }
+  - match: { data_streams.0.failure_store.indices.0.index_name: '/\.fs-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - match: { data_streams.0.failure_store.indices.1.index_name: '/\.fs-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000002/' }
+
+  - do:
+      search:
+        index: .fs-data-stream-for-rollover-*
+  - length: { hits.hits: 1  }
+  - match: { hits.hits.0._index: "/\\.fs-data-stream-for-rollover-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000002/" }
+  - exists: hits.hits.0._source.@timestamp
+  - not_exists: hits.hits.0._source.count
+  - match: { hits.hits.0._source.document.index: 'data-stream-for-rollover' }
+  - match: { hits.hits.0._source.document.source.@timestamp: '2020-12-12' }
+  - match: { hits.hits.0._source.document.source.count: 'invalid value' }
+  - match: { hits.hits.0._source.error.type: 'document_parsing_exception' }
+
+---
+"Lazily roll over a data stream's failure store after an ingest failure":
+  - requires:
+      reason: "data stream failure store lazy rollover only supported in 8.15+"
+      test_runner_features: [allowed_warnings, capabilities]
+      capabilities:
+        - method: POST
+          path: /{index}/_rollover
+          capabilities: [lazy-rollover-failure-store]
+
+  - do:
+      ingest.put_pipeline:
+        id: "failing_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "fail" : {
+                  "message" : "error_message"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      allowed_warnings:
+        - "index template [generic_logs_template] has index patterns [logs-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [generic_logs_template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template
+        body:
+          index_patterns: [data-*]
+          data_stream:
+            failure_store: true
+          template:
+            settings:
+              index:
+                default_pipeline: "failing_pipeline"
+
+  - do:
+      indices.create_data_stream:
+        name: data-stream-for-lazy-rollover
+
+  # Mark the failure store for lazy rollover
+  - do:
+      indices.rollover:
+        alias: data-stream-for-lazy-rollover
+        target_failure_store: true
+        lazy: true
+
+  - match: { acknowledged: true }
   - match: { rolled_over: false }
   - match: { dry_run: false }
 
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - match: { data_streams.0.name: data-stream-for-lazy-rollover }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 1 }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-lazy-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - length: { data_streams.0.failure_store.indices: 1 }
+  - match: { data_streams.0.failure_store.indices.0.index_name: '/\.fs-data-stream-for-lazy-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+
+  - do:
+      index:
+        index: data-stream-for-lazy-rollover
+        refresh: true
+        body:
+          '@timestamp': '2020-12-12'
+          count: 1
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - match: { data_streams.0.name: data-stream-for-lazy-rollover }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  # Both backing and failure indices use the same generation field.
+  - match: { data_streams.0.generation: 2 }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-lazy-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - length: { data_streams.0.failure_store.indices: 2 }
+  - match: { data_streams.0.failure_store.indices.0.index_name: '/\.fs-data-stream-for-lazy-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - match: { data_streams.0.failure_store.indices.1.index_name: '/\.fs-data-stream-for-lazy-rollover-(\d{4}\.\d{2}\.\d{2}-)?000002/' }
+
+---
+"A failure store marked for lazy rollover should only be rolled over when there is a failure":
+  - requires:
+      reason: "data stream failure store lazy rollover only supported in 8.15+"
+      test_runner_features: [allowed_warnings, capabilities]
+      capabilities:
+        - method: POST
+          path: /{index}/_rollover
+          capabilities: [lazy-rollover-failure-store]
+
+  # Mark the failure store for lazy rollover
+  - do:
+      indices.rollover:
+        alias: "data-stream-for-rollover"
+        target_failure_store: true
+        lazy: true
+
+  - match: { acknowledged: true }
+  - match: { rolled_over: false }
+  - match: { dry_run: false }
+
+  - do:
+      indices.get_data_stream:
+        name: "*"
+  - match: { data_streams.0.name: data-stream-for-rollover }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 1 }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - length: { data_streams.0.failure_store.indices: 1 }
+  - match: { data_streams.0.failure_store.indices.0.index_name: '/\.fs-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+
+  - do:
+      index:
+        index: data-stream-for-rollover
+        refresh: true
+        body:
+          '@timestamp': '2020-12-12'
+          count: 3
+
   - do:
       indices.get_data_stream:
         name: "*"
   - match: { data_streams.0.name: data-stream-for-rollover }
   - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  # Both backing and failure indices use the same generation field.
   - match: { data_streams.0.generation: 1 }
   - length: { data_streams.0.indices: 1 }
   - match: { data_streams.0.indices.0.index_name: '/\.ds-data-stream-for-rollover-(\d{4}\.\d{2}\.\d{2}-)?000001/' }

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/LazyRolloverAction.java

@@ -99,7 +99,7 @@ protected void masterOperation(
                 rolloverRequest.getRolloverTarget(),
                 rolloverRequest.getNewIndexName(),
                 rolloverRequest.getCreateIndexRequest(),
-                false
+                rolloverRequest.indicesOptions().failureStoreOptions().includeFailureIndices()
             );
             final String trialSourceIndexName = trialRolloverNames.sourceName();
             final String trialRolloverIndexName = trialRolloverNames.rolloverName();
@@ -121,13 +121,9 @@ protected void masterOperation(
             String source = "lazy_rollover source [" + trialRolloverIndexName + "] to target [" + trialRolloverIndexName + "]";
             // We create a new rollover request to ensure that it doesn't contain any other parameters apart from the data stream name
             // This will provide a more resilient user experience
-            RolloverTask rolloverTask = new RolloverTask(
-                new RolloverRequest(rolloverRequest.getRolloverTarget(), null),
-                null,
-                trialRolloverResponse,
-                null,
-                listener
-            );
+            var newRolloverRequest = new RolloverRequest(rolloverRequest.getRolloverTarget(), null);
+            newRolloverRequest.setIndicesOptions(rolloverRequest.indicesOptions());
+            RolloverTask rolloverTask = new RolloverTask(newRolloverRequest, null, trialRolloverResponse, null, listener);
             submitRolloverTask(rolloverRequest, source, rolloverTask);
         }
     }

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverRequest.java

@@ -153,10 +153,6 @@ public ActionRequestValidationException validate() {
             );
         }
 
-        if (failureStoreOptions.includeFailureIndices() && lazy) {
-            validationException = addValidationError("lazily rolling over a failure store is currently not supported", validationException);
-        }
-
         return validationException;
     }