Merge remote-tracking branch 'origin/try1-limit-byte-sizes' into try1-limit-byte-sizes

Author: prdoyle

docs/changelog/119542.yaml

@@ -0,0 +1,5 @@
+pr: 119542
+summary: Wait while index is blocked
+area: Transform
+type: enhancement
+issues: []

server/src/test/java/org/elasticsearch/index/mapper/blockloader/KeywordFieldBlockLoaderTests.java

@@ -13,10 +13,12 @@
 import org.elasticsearch.index.mapper.BlockLoaderTestCase;
 import org.elasticsearch.logsdb.datageneration.FieldType;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.stream.Collectors;
+import java.util.function.Function;
+import java.util.stream.Stream;
 
 public class KeywordFieldBlockLoaderTests extends BlockLoaderTestCase {
     public KeywordFieldBlockLoaderTests() {
@@ -29,25 +31,30 @@ protected Object expected(Map<String, Object> fieldMapping, Object value, boolea
         if (value == null) {
             return null;
         }
+
+        var ignoreAbove = fieldMapping.get("ignore_above") == null
+            ? Integer.MAX_VALUE
+            : ((Number) fieldMapping.get("ignore_above")).intValue();
+
         if (value instanceof String s) {
-            return convert(s);
+            return convert(s, ignoreAbove);
         }
 
-        var nonNullStream = ((List<String>) value).stream().filter(Objects::nonNull);
+        Function<Stream<String>, Stream<BytesRef>> convertValues = s -> s.map(v -> convert(v, ignoreAbove)).filter(Objects::nonNull);
 
         if ((boolean) fieldMapping.getOrDefault("doc_values", false)) {
             // Sorted and no duplicates
-            return maybeFoldList(nonNullStream.collect(Collectors.toSet()).stream().sorted().map(this::convert).toList());
-        }
 
-        if ((boolean) fieldMapping.getOrDefault("store", false)) {
-            return maybeFoldList(nonNullStream.map(this::convert).toList());
+            var values = new HashSet<>((List<String>) value);
+            var resultList = convertValues.compose(s -> values.stream().filter(Objects::nonNull).sorted())
+                .andThen(Stream::toList)
+                .apply(values.stream());
+            return maybeFoldList(resultList);
         }
 
-        // Using source (either stored or synthetic).
-        // Original order is preserved and values longer than ignore_above are returned.
-        // TODO actual ignore_above support in data generation
-        return maybeFoldList(nonNullStream.map(this::convert).toList());
+        // store: "true" and source
+        var resultList = convertValues.andThen(Stream::toList).apply(((List<String>) value).stream());
+        return maybeFoldList(resultList);
     }
 
     private Object maybeFoldList(List<?> list) {
@@ -62,7 +69,11 @@ private Object maybeFoldList(List<?> list) {
         return list;
     }
 
-    private BytesRef convert(String value) {
-        return new BytesRef(value);
+    private BytesRef convert(String value, int ignoreAbove) {
+        if (value == null) {
+            return null;
+        }
+
+        return value.length() <= ignoreAbove ? new BytesRef(value) : null;
     }
 }

test/framework/src/main/java/org/elasticsearch/logsdb/datageneration/datasource/DefaultMappingParametersHandler.java

@@ -66,7 +66,7 @@ private Supplier<Map<String, Object>> keywordMapping(
             }
 
             if (ESTestCase.randomDouble() <= 0.2) {
-                injected.put("ignore_above", ESTestCase.randomIntBetween(1, 10000));
+                injected.put("ignore_above", ESTestCase.randomIntBetween(1, 100));
             }
 
             return injected;

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/MigratePlugin.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.settings.SettingsModule;
@@ -58,6 +59,7 @@
 import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.migrate.action.ReindexDataStreamAction.REINDEX_DATA_STREAM_FEATURE_FLAG;
+import static org.elasticsearch.xpack.migrate.task.ReindexDataStreamPersistentTaskExecutor.MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING;
 
 public class MigratePlugin extends Plugin implements ActionPlugin, PersistentTaskPlugin {
 
@@ -153,4 +155,11 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(
             return List.of();
         }
     }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        List<Setting<?>> pluginSettings = new ArrayList<>();
+        pluginSettings.add(MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING);
+        return pluginSettings;
+    }
 }

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamPersistentTaskExecutor.java

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.xpack.migrate.task;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.rollover.RolloverAction;
@@ -19,6 +21,7 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamAction;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
@@ -39,6 +42,19 @@
 import static org.elasticsearch.xpack.core.deprecation.DeprecatedIndexPredicate.getReindexRequiredPredicate;
 
 public class ReindexDataStreamPersistentTaskExecutor extends PersistentTasksExecutor<ReindexDataStreamTaskParams> {
+    /*
+     * This setting controls how many indices we reindex concurrently for a single data stream. This is not an overall limit -- if five
+     * data streams are being reindexed, then each of them can have this many indices being reindexed at once. This setting is dynamic,
+     * but changing it does not have an impact if the task is already running (unless the task is restarted or moves to another node).
+     */
+    public static final Setting<Integer> MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING = Setting.intSetting(
+        "migrate.max_concurrent_indices_reindexed_per_data_stream",
+        1,
+        1,
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+    private static final Logger logger = LogManager.getLogger(ReindexDataStreamPersistentTaskExecutor.class);
     private static final TimeValue TASK_KEEP_ALIVE_TIME = TimeValue.timeValueDays(1);
     private final Client client;
     private final ClusterService clusterService;
@@ -165,8 +181,9 @@ private void reindexIndices(
         CountDownActionListener listener = new CountDownActionListener(indicesToBeReindexed.size() + 1, ActionListener.wrap(response1 -> {
             completeSuccessfulPersistentTask(reindexDataStreamTask, updatedState);
         }, exception -> { completeFailedPersistentTask(reindexDataStreamTask, updatedState, exception); }));
+        final int maxConcurrentIndices = clusterService.getClusterSettings().get(MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING);
         List<Index> indicesRemaining = Collections.synchronizedList(new ArrayList<>(indicesToBeReindexed));
-        final int maxConcurrentIndices = 1;
+        logger.debug("Reindexing {} indices, with up to {} handled concurrently", indicesRemaining.size(), maxConcurrentIndices);
         for (int i = 0; i < maxConcurrentIndices; i++) {
             maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, reindexClient, sourceDataStream, listener, parentTaskId);
         }

x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.xpack.core.transform.transforms.TimeRetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TimeSyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
+import org.elasticsearch.xpack.core.transform.transforms.TransformStats;
 import org.elasticsearch.xpack.core.transform.transforms.pivot.PivotConfig;
 import org.elasticsearch.xpack.core.transform.transforms.pivot.SingleGroupSource;
 import org.elasticsearch.xpack.core.transform.transforms.pivot.TermsGroupSource;
@@ -152,7 +153,7 @@ public void testBasicTransformStats() throws Exception {
     public void testContinuousTransformCrud() throws Exception {
         var transformId = "transform-continuous-crud";
         var indexName = "continuous-crud-reviews";
-        createContinuousTransform(indexName, transformId);
+        createContinuousTransform(indexName, transformId, "reviews-by-user-business-day");
         var transformStats = getBasicTransformStats(transformId);
         assertThat(transformStats.get("state"), equalTo("started"));
 
@@ -181,7 +182,7 @@ public void testContinuousTransformCrud() throws Exception {
         deleteTransform(transformId);
     }
 
-    private void createContinuousTransform(String indexName, String transformId) throws Exception {
+    private void createContinuousTransform(String indexName, String transformId, String destinationIndex) throws Exception {
         createReviewsIndex(indexName, 100, NUM_USERS, TransformIT::getUserIdForRow, TransformIT::getDateStringForRow);
 
         var groups = Map.of(
@@ -197,8 +198,9 @@ private void createContinuousTransform(String indexName, String transformId) thr
             .addAggregator(AggregationBuilders.avg("review_score").field("stars"))
             .addAggregator(AggregationBuilders.max("timestamp").field("timestamp"));
 
-        var config = createTransformConfigBuilder(transformId, "reviews-by-user-business-day", QueryConfig.matchAll(), indexName)
-            .setPivotConfig(createPivotConfig(groups, aggs))
+        var config = createTransformConfigBuilder(transformId, destinationIndex, QueryConfig.matchAll(), indexName).setPivotConfig(
+            createPivotConfig(groups, aggs)
+        )
             .setSyncConfig(new TimeSyncConfig("timestamp", TimeValue.timeValueSeconds(1)))
             .setSettings(new SettingsConfig.Builder().setAlignCheckpoints(false).build())
             .build();
@@ -216,7 +218,7 @@ private void createContinuousTransform(String indexName, String transformId) thr
     @SuppressWarnings("unchecked")
     public void testBasicContinuousTransformStats() throws Exception {
         var transformId = "transform-continuous-basic-stats";
-        createContinuousTransform("continuous-basic-stats-reviews", transformId);
+        createContinuousTransform("continuous-basic-stats-reviews", transformId, "reviews-by-user-business-day");
         var transformStats = getBasicTransformStats(transformId);
 
         assertEquals("started", XContentMapValues.extractValue("state", transformStats));
@@ -230,6 +232,40 @@ public void testBasicContinuousTransformStats() throws Exception {
         deleteTransform(transformId);
     }
 
+    public void testDestinationIndexBlocked() throws Exception {
+        var transformId = "transform-continuous-blocked-destination";
+        var sourceIndexName = "source-reviews";
+        var destIndexName = "destination-reviews";
+
+        // create transform & indices, wait until 1st checkpoint is finished
+        createContinuousTransform(sourceIndexName, transformId, destIndexName);
+
+        // block destination index
+        Request request = new Request("PUT", destIndexName + "/_block/write");
+        assertAcknowledged(adminClient().performRequest(request));
+
+        // index more docs so the checkpoint tries to run, wait until transform stops
+        assertBusy(() -> {
+            indexDoc(42, sourceIndexName);
+            assertEquals(TransformStats.State.WAITING.value(), getTransformState(transformId));
+        }, 30, TimeUnit.SECONDS);
+
+        // unblock index
+        request = new Request("PUT", destIndexName + "/_settings");
+        request.setJsonEntity("""
+                { "blocks.write": false }
+            """);
+        assertAcknowledged(adminClient().performRequest(request));
+
+        assertBusy(() -> {
+            indexDoc(42, sourceIndexName);
+            assertEquals(TransformStats.State.STARTED.value(), getTransformState(transformId));
+        }, 30, TimeUnit.SECONDS);
+
+        stopTransform(transformId);
+        deleteTransform(transformId);
+    }
+
     public void testTransformLifecycleInALoop() throws Exception {
         String transformId = "lifecycle-in-a-loop";
         String indexName = transformId + "-src";
@@ -652,4 +688,17 @@ private void indexMoreDocs(long timestamp, long userId, String index) throws Exc
         bulkBuilder.append("\r\n");
         doBulk(bulkBuilder.toString(), true);
     }
+
+    private void indexDoc(long userId, String index) throws Exception {
+        StringBuilder bulkBuilder = new StringBuilder();
+        bulkBuilder.append(format("""
+            {"create":{"_index":"%s"}}
+            """, index));
+        String source = format("""
+            {"user_id":"user_%s","count":%s,"business_id":"business_%s","stars":%s,"timestamp":%s}
+            """, userId, 1, 2, 5, Instant.now().toEpochMilli());
+        bulkBuilder.append(source);
+        bulkBuilder.append("\r\n");
+        doBulk(bulkBuilder.toString(), true);
+    }
 }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java

@@ -270,6 +270,9 @@ static TransformStats deriveStats(TransformTask transformTask, @Nullable Transfo
             && derivedState.equals(TransformStats.State.FAILED) == false) {
             derivedState = TransformStats.State.STOPPING;
             reason = Strings.isNullOrEmpty(reason) ? "transform is set to stop at the next checkpoint" : reason;
+        } else if (derivedState.equals(TransformStats.State.STARTED) && transformTask.getContext().isWaitingForIndexToUnblock()) {
+            derivedState = TransformStats.State.WAITING;
+            reason = Strings.isNullOrEmpty(reason) ? "transform is paused while destination index is blocked" : reason;
         }
         return new TransformStats(
             transformTask.getTransformId(),

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportScheduleNowTransformAction.java

@@ -128,6 +128,10 @@ protected void taskOperation(
         TransformTask transformTask,
         ActionListener<Response> listener
     ) {
+        if (transformTask.getContext().isWaitingForIndexToUnblock()) {
+            logger.debug("[{}] Destination index is blocked. User requested a retry.", transformTask.getTransformId());
+            transformTask.getContext().setIsWaitingForIndexToUnblock(false);
+        }
         transformScheduler.scheduleNow(request.getId());
         listener.onResponse(Response.TRUE);
     }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/ClientTransformIndexer.java

@@ -25,8 +25,10 @@
 import org.elasticsearch.action.search.TransportClosePointInTimeAction;
 import org.elasticsearch.action.search.TransportOpenPointInTimeAction;
 import org.elasticsearch.action.search.TransportSearchAction;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.client.internal.ParentTaskAssigningClient;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -443,9 +445,36 @@ public boolean maybeTriggerAsyncJob(long now) {
             logger.debug("[{}] schedule was triggered but the Transform is upgrading. Ignoring trigger.", getJobId());
             return false;
         }
+        if (context.isWaitingForIndexToUnblock()) {
+            if (destinationIndexHasWriteBlock()) {
+                logger.debug("[{}] schedule was triggered but the destination index has a write block. Ignoring trigger.", getJobId());
+                return false;
+            }
+            logger.debug("[{}] destination index is no longer blocked.", getJobId());
+            context.setIsWaitingForIndexToUnblock(false);
+        }
+
         return super.maybeTriggerAsyncJob(now);
     }
 
+    private boolean destinationIndexHasWriteBlock() {
+        var clusterState = clusterService.state();
+        if (clusterState == null) {
+            // if we can't determine if the index is blocked, we assume it isn't, even though the bulk request may fail again
+            return false;
+        }
+
+        var destinationIndexName = transformConfig.getDestination().getIndex();
+        var destinationIndex = indexNameExpressionResolver.concreteWriteIndex(
+            clusterState,
+            IndicesOptions.lenientExpandOpen(),
+            destinationIndexName,
+            true,
+            false
+        );
+        return destinationIndex != null && clusterState.blocks().indexBlocked(ClusterBlockLevel.WRITE, destinationIndex.getName());
+    }
+
     @Override
     protected void onStop() {
         closePointInTime();

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformContext.java

@@ -49,6 +49,14 @@ public interface Listener {
     private volatile AuthorizationState authState;
     private volatile int pageSize = 0;
 
+    /**
+     * If the destination index is blocked (e.g. during a reindex), the Transform will fail to write to it.
+     * {@link TransformFailureHandler} will silence the error so the Transform automatically retries.
+     * Every time the Transform runs, it will check if the index is unblocked and reset this to false.
+     * Users can override this via the `_schedule_now` API.
+     */
+    private volatile boolean isWaitingForIndexToUnblock = false;
+
     // the checkpoint of this transform, storing the checkpoint until data indexing from source to dest is _complete_
     // Note: Each indexer run creates a new future checkpoint which becomes the current checkpoint only after the indexer run finished
     private final AtomicLong currentCheckpoint;
@@ -183,6 +191,14 @@ public void setShouldRecreateDestinationIndex(boolean shouldRecreateDestinationI
         this.shouldRecreateDestinationIndex = shouldRecreateDestinationIndex;
     }
 
+    public boolean isWaitingForIndexToUnblock() {
+        return isWaitingForIndexToUnblock;
+    }
+
+    public void setIsWaitingForIndexToUnblock(boolean isWaitingForIndexToUnblock) {
+        this.isWaitingForIndexToUnblock = isWaitingForIndexToUnblock;
+    }
+
     public AuthorizationState getAuthState() {
         return authState;
     }