[Bug] During heavy indexing load it's possible for lazy rollover to trigger multiple rollovers (#109636) (#110031)

Let’s say we have `my-metrics`  data stream which is receiving a lot of
indexing requests. The following scenario can result in multiple
unnecessary rollovers:

1. We update the mapping and mark it to be lazy rolled over
2. We receive 5 bulk index requests that all contain a write request for this data stream.
3. Each of these requests are being picked up “at the same time”, they see that the data stream needs to be rolled over and they issue a lazy rollover request.
4. Currently, data stream my-metrics  has 5 tasks executing an unconditional rollover.
5. The data stream gets rolled over 5 times instead of one.

This scenario is captured in the `LazyRolloverDuringDisruptionIT`.

We have witnessed this also in the wild, where a data stream was rolled
over 281 times extra resulting in 281 empty indices.

This PR proposes:

- To create a new task queue with a more efficient executor that further batches/deduplicates the requests.
- We add two safe guards, the first to ensure we will not enqueue the rollover task if we see that a rollover has occurred already. The second safe guard is during task execution, if we see that the data stream does not have the `rolloverOnWrite` flag set to `true` we skip the rollover.
- When we skip the rollover we return the following response:

```
{
  "acknowledged": true,
  "shards_acknowledged": true,
  "old_index": ".ds-my-data-stream-2099.05.07-000002",
  "new_index": ".ds-my-data-stream-2099.05.07-000002",
  "rolled_over": false,
  "dry_run": false,
  "lazy": false,
}
```

Author: gmarouli

docs/changelog/109636.yaml

@@ -0,0 +1,5 @@
+pr: 109636
+summary: "Ensure a lazy rollover request will rollover the target data stream once."
+area: Data streams
+type: bug
+issues: []

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

@@ -0,0 +1,125 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.rollover.RolloverRequestBuilder;
+import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
+import org.elasticsearch.action.datastreams.CreateDataStreamAction;
+import org.elasticsearch.action.datastreams.GetDataStreamAction;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.disruption.IntermittentLongGCDisruption;
+import org.elasticsearch.test.disruption.SingleNodeDisruption;
+import org.elasticsearch.xcontent.XContentType;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, numClientNodes = 0)
+public class LazyRolloverDuringDisruptionIT extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(DataStreamsPlugin.class);
+    }
+
+    public void testRolloverIsExecutedOnce() throws ExecutionException, InterruptedException {
+        String masterNode = internalCluster().startMasterOnlyNode();
+        internalCluster().startDataOnlyNodes(3);
+        ensureStableCluster(4);
+
+        String dataStreamName = "my-data-stream";
+        createDataStream(dataStreamName);
+
+        // Mark it to lazy rollover
+        new RolloverRequestBuilder(client()).setRolloverTarget(dataStreamName).lazy(true).execute().get();
+
+        // Verify that the data stream is marked for rollover and that it has currently one index
+        DataStream dataStream = getDataStream(dataStreamName);
+        assertThat(dataStream.rolloverOnWrite(), equalTo(true));
+        assertThat(dataStream.getIndices().size(), equalTo(1));
+
+        // Introduce a disruption to the master node that should delay the rollover execution
+        SingleNodeDisruption masterNodeDisruption = new IntermittentLongGCDisruption(random(), masterNode, 100, 200, 30000, 60000);
+        internalCluster().setDisruptionScheme(masterNodeDisruption);
+        masterNodeDisruption.startDisrupting();
+
+        // Start indexing operations
+        int docs = randomIntBetween(5, 10);
+        CountDownLatch countDownLatch = new CountDownLatch(docs);
+        for (int i = 0; i < docs; i++) {
+            var indexRequest = new IndexRequest(dataStreamName).opType(DocWriteRequest.OpType.CREATE);
+            final String doc = "{ \"@timestamp\": \"2099-05-06T16:21:15.000Z\", \"message\": \"something cool happened\" }";
+            indexRequest.source(doc, XContentType.JSON);
+            client().index(indexRequest, new ActionListener<>() {
+                @Override
+                public void onResponse(DocWriteResponse docWriteResponse) {
+                    countDownLatch.countDown();
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    fail("Indexing request should have succeeded eventually, failed with " + e.getMessage());
+                }
+            });
+        }
+
+        // End the disruption so that all pending tasks will complete
+        masterNodeDisruption.stopDisrupting();
+
+        // Wait for all the indexing requests to be processed successfully
+        countDownLatch.await();
+
+        // Verify that the rollover has happened once
+        dataStream = getDataStream(dataStreamName);
+        assertThat(dataStream.rolloverOnWrite(), equalTo(false));
+        assertThat(dataStream.getIndices().size(), equalTo(2));
+    }
+
+    private DataStream getDataStream(String dataStreamName) {
+        return client().execute(GetDataStreamAction.INSTANCE, new GetDataStreamAction.Request(new String[] { dataStreamName }))
+            .actionGet()
+            .getDataStreams()
+            .get(0)
+            .getDataStream();
+    }
+
+    private void createDataStream(String dataStreamName) throws InterruptedException, ExecutionException {
+        final TransportPutComposableIndexTemplateAction.Request putComposableTemplateRequest =
+            new TransportPutComposableIndexTemplateAction.Request("my-template");
+        putComposableTemplateRequest.indexTemplate(
+            ComposableIndexTemplate.builder()
+                .indexPatterns(List.of(dataStreamName))
+                .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
+                .build()
+        );
+        final AcknowledgedResponse putComposableTemplateResponse = client().execute(
+            TransportPutComposableIndexTemplateAction.TYPE,
+            putComposableTemplateRequest
+        ).actionGet();
+        assertThat(putComposableTemplateResponse.isAcknowledged(), is(true));
+
+        final CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        final AcknowledgedResponse createDataStreamResponse = client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest)
+            .get();
+        assertThat(createDataStreamResponse.isAcknowledged(), is(true));
+    }
+}

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

@@ -7,33 +7,50 @@
  */
 package org.elasticsearch.action.admin.indices.rollover;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.datastreams.autosharding.DataStreamAutoShardingService;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.ActiveShardsObserver;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateTaskExecutor;
+import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.MetadataDataStreamsService;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
+import org.elasticsearch.cluster.routing.allocation.allocator.AllocationActionMultiListener;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
+import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 
 /**
  * API that lazily rolls over a data stream that has the flag {@link DataStream#rolloverOnWrite()} enabled. These requests always
  * originate from requests that write into the data stream.
  */
 public final class LazyRolloverAction extends ActionType<RolloverResponse> {
 
+    private static final Logger logger = LogManager.getLogger(LazyRolloverAction.class);
+
     public static final NodeFeature DATA_STREAM_LAZY_ROLLOVER = new NodeFeature("data_stream.rollover.lazy");
 
     public static final LazyRolloverAction INSTANCE = new LazyRolloverAction();
@@ -50,6 +67,8 @@ public String name() {
 
     public static final class TransportLazyRolloverAction extends TransportRolloverAction {
 
+        private final MasterServiceTaskQueue<LazyRolloverTask> lazyRolloverTaskQueue;
+
         @Inject
         public TransportLazyRolloverAction(
             TransportService transportService,
@@ -76,6 +95,11 @@ public TransportLazyRolloverAction(
                 metadataDataStreamsService,
                 dataStreamAutoShardingService
             );
+            this.lazyRolloverTaskQueue = clusterService.createTaskQueue(
+                "lazy-rollover",
+                Priority.NORMAL,
+                new LazyRolloverExecutor(clusterService, allocationService, rolloverService, threadPool)
+            );
         }
 
         @Override
@@ -93,6 +117,12 @@ protected void masterOperation(
                 : "The auto rollover action does not expect any other parameters in the request apart from the data stream name";
 
             Metadata metadata = clusterState.metadata();
+            DataStream dataStream = metadata.dataStreams().get(rolloverRequest.getRolloverTarget());
+            // Skip submitting the task if we detect that the lazy rollover has been already executed.
+            if (dataStream.rolloverOnWrite() == false) {
+                listener.onResponse(noopLazyRolloverResponse(dataStream));
+                return;
+            }
             // We evaluate the names of the source index as well as what our newly created index would be.
             final MetadataRolloverService.NameResolution trialRolloverNames = MetadataRolloverService.resolveRolloverNames(
                 clusterState,
@@ -107,28 +137,164 @@ protected void masterOperation(
 
             assert metadata.dataStreams().containsKey(rolloverRequest.getRolloverTarget()) : "Auto-rollover applies only to data streams";
 
-            final RolloverResponse trialRolloverResponse = new RolloverResponse(
-                trialSourceIndexName,
-                trialRolloverIndexName,
-                Map.of(),
-                false,
-                false,
-                false,
-                false,
-                false
-            );
-
             String source = "lazy_rollover source [" + trialSourceIndexName + "] 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),
+            var newRolloverRequest = new RolloverRequest(rolloverRequest.getRolloverTarget(), null);
+            newRolloverRequest.setIndicesOptions(rolloverRequest.indicesOptions());
+            LazyRolloverTask rolloverTask = new LazyRolloverTask(newRolloverRequest, listener);
+            lazyRolloverTaskQueue.submitTask(source, rolloverTask, rolloverRequest.masterNodeTimeout());
+        }
+    }
+
+    /**
+     * A lazy rollover task holds the rollover request and the listener.
+     */
+    record LazyRolloverTask(RolloverRequest rolloverRequest, ActionListener<RolloverResponse> listener)
+        implements
+            ClusterStateTaskListener {
+
+        @Override
+        public void onFailure(Exception e) {
+            listener.onFailure(e);
+        }
+    }
+
+    /**
+     * Performs a lazy rollover when required and notifies the listener. Due to the nature of the lazy rollover we are able
+     * to perform certain optimisations like identifying duplicate requests and executing them once. This is an optimisation
+     * that can work since we do not take into consideration any stats or auto-sharding conditions here.
+     */
+    record LazyRolloverExecutor(
+        ClusterService clusterService,
+        AllocationService allocationService,
+        MetadataRolloverService rolloverService,
+        ThreadPool threadPool
+    ) implements ClusterStateTaskExecutor<LazyRolloverTask> {
+
+        @Override
+        public ClusterState execute(BatchExecutionContext<LazyRolloverTask> batchExecutionContext) {
+            final var listener = new AllocationActionMultiListener<RolloverResponse>(threadPool.getThreadContext());
+            final var results = new ArrayList<MetadataRolloverService.RolloverResult>(batchExecutionContext.taskContexts().size());
+            var state = batchExecutionContext.initialState();
+            Map<RolloverRequest, List<TaskContext<LazyRolloverTask>>> groupedRequests = new HashMap<>();
+            for (final var taskContext : batchExecutionContext.taskContexts()) {
+                groupedRequests.computeIfAbsent(taskContext.getTask().rolloverRequest(), ignored -> new ArrayList<>()).add(taskContext);
+            }
+            for (final var entry : groupedRequests.entrySet()) {
+                List<TaskContext<LazyRolloverTask>> rolloverTaskContexts = entry.getValue();
+                try {
+                    RolloverRequest rolloverRequest = entry.getKey();
+                    state = executeTask(state, rolloverRequest, results, rolloverTaskContexts, listener);
+                } catch (Exception e) {
+                    rolloverTaskContexts.forEach(taskContext -> taskContext.onFailure(e));
+                } finally {
+                    rolloverTaskContexts.forEach(taskContext -> taskContext.captureResponseHeaders().close());
+                }
+            }
+
+            if (state != batchExecutionContext.initialState()) {
+                var reason = new StringBuilder();
+                Strings.collectionToDelimitedStringWithLimit(
+                    (Iterable<String>) () -> Iterators.map(results.iterator(), t -> t.sourceIndexName() + "->" + t.rolloverIndexName()),
+                    ",",
+                    "lazy bulk rollover [",
+                    "]",
+                    1024,
+                    reason
+                );
+                try (var ignored = batchExecutionContext.dropHeadersContext()) {
+                    state = allocationService.reroute(state, reason.toString(), listener.reroute());
+                }
+            } else {
+                listener.noRerouteNeeded();
+            }
+            return state;
+        }
+
+        public ClusterState executeTask(
+            ClusterState currentState,
+            RolloverRequest rolloverRequest,
+            List<MetadataRolloverService.RolloverResult> results,
+            List<TaskContext<LazyRolloverTask>> rolloverTaskContexts,
+            AllocationActionMultiListener<RolloverResponse> allocationActionMultiListener
+        ) throws Exception {
+
+            // If the data stream has been rolled over since it was marked for lazy rollover, this operation is a noop
+            final DataStream dataStream = currentState.metadata().dataStreams().get(rolloverRequest.getRolloverTarget());
+            assert dataStream != null;
+
+            if (dataStream.rolloverOnWrite() == false) {
+                var noopResponse = noopLazyRolloverResponse(dataStream);
+                notifyAllListeners(rolloverTaskContexts, context -> context.getTask().listener.onResponse(noopResponse));
+                return currentState;
+            }
+
+            // Perform the actual rollover
+            final var rolloverResult = rolloverService.rolloverClusterState(
+                currentState,
+                rolloverRequest.getRolloverTarget(),
+                rolloverRequest.getNewIndexName(),
+                rolloverRequest.getCreateIndexRequest(),
+                List.of(),
+                Instant.now(),
+                false,
+                false,
                 null,
-                trialRolloverResponse,
                 null,
-                listener
+                rolloverRequest.targetsFailureStore()
             );
-            submitRolloverTask(rolloverRequest, source, rolloverTask);
+            results.add(rolloverResult);
+            logger.trace("lazy rollover result [{}]", rolloverResult);
+
+            final var rolloverIndexName = rolloverResult.rolloverIndexName();
+            final var sourceIndexName = rolloverResult.sourceIndexName();
+
+            final var waitForActiveShardsTimeout = rolloverRequest.masterNodeTimeout().millis() < 0
+                ? null
+                : rolloverRequest.masterNodeTimeout();
+
+            notifyAllListeners(rolloverTaskContexts, context -> {
+                // Now assuming we have a new state and the name of the rolled over index, we need to wait for the configured number of
+                // active shards, as well as return the names of the indices that were rolled/created
+                ActiveShardsObserver.waitForActiveShards(
+                    clusterService,
+                    new String[] { rolloverIndexName },
+                    rolloverRequest.getCreateIndexRequest().waitForActiveShards(),
+                    waitForActiveShardsTimeout,
+                    allocationActionMultiListener.delay(context.getTask().listener())
+                        .map(
+                            isShardsAcknowledged -> new RolloverResponse(
+                                // Note that we use the actual rollover result for these, because even though we're single threaded,
+                                // it's possible for the rollover names generated before the actual rollover to be different due to
+                                // things like date resolution
+                                sourceIndexName,
+                                rolloverIndexName,
+                                Map.of(),
+                                false,
+                                true,
+                                true,
+                                isShardsAcknowledged,
+                                false
+                            )
+                        )
+                );
+            });
+
+            // Return the new rollover cluster state, which includes the changes that create the new index
+            return rolloverResult.clusterState();
         }
     }
+
+    private static void notifyAllListeners(
+        List<ClusterStateTaskExecutor.TaskContext<LazyRolloverTask>> taskContexts,
+        Consumer<ClusterStateTaskExecutor.TaskContext<LazyRolloverTask>> onPublicationSuccess
+    ) {
+        taskContexts.forEach(context -> context.success(() -> onPublicationSuccess.accept(context)));
+    }
+
+    private static RolloverResponse noopLazyRolloverResponse(DataStream dataStream) {
+        String latestWriteIndex = dataStream.getWriteIndex().getName();
+        return new RolloverResponse(latestWriteIndex, latestWriteIndex, Map.of(), false, false, true, true, false);
+    }
 }