Cluster state and recovery constructs for in-place shard split

Signed-off-by: vikasvb90 <vikasvb@amazon.com>

Author: vikasvb90

server/src/main/java/org/opensearch/action/admin/indices/split/InPlaceShardSplitClusterStateUpdateRequest.java

@@ -0,0 +1,49 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.admin.indices.split;
+
+import org.opensearch.cluster.ack.ClusterStateUpdateRequest;
+import org.opensearch.common.annotation.ExperimentalApi;
+
+/**
+ * Cluster state update request for in-place shard split.
+ *
+ * @opensearch.experimental
+ */
+@ExperimentalApi
+public class InPlaceShardSplitClusterStateUpdateRequest extends ClusterStateUpdateRequest<InPlaceShardSplitClusterStateUpdateRequest> {
+
+    private final String index;
+    private final int shardId;
+    private final int splitInto;
+    private final String cause;
+
+    public InPlaceShardSplitClusterStateUpdateRequest(String cause, String index, int shardId, int splitInto) {
+        this.index = index;
+        this.shardId = shardId;
+        this.splitInto = splitInto;
+        this.cause = cause;
+    }
+
+    public String getIndex() {
+        return index;
+    }
+
+    public int getSplitInto() {
+        return splitInto;
+    }
+
+    public String cause() {
+        return cause;
+    }
+
+    public int getShardId() {
+        return shardId;
+    }
+}

server/src/main/java/org/opensearch/cluster/metadata/MetadataInPlaceShardSplitService.java

@@ -0,0 +1,131 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.metadata;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.opensearch.action.admin.indices.split.InPlaceShardSplitClusterStateUpdateRequest;
+import org.opensearch.cluster.AckedClusterStateUpdateTask;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.ack.ClusterStateUpdateResponse;
+import org.opensearch.cluster.routing.RoutingTable;
+import org.opensearch.cluster.routing.allocation.AllocationService;
+import org.opensearch.cluster.service.ClusterManagerTask;
+import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.Priority;
+import org.opensearch.common.annotation.ExperimentalApi;
+import org.opensearch.core.action.ActionListener;
+
+import java.util.function.BiFunction;
+
+/**
+ * Service responsible for applying in-place shard split requests to cluster state.
+ *
+ * @opensearch.experimental
+ */
+@ExperimentalApi
+public class MetadataInPlaceShardSplitService {
+    private static final Logger logger = LogManager.getLogger(MetadataInPlaceShardSplitService.class);
+
+    private final ClusterService clusterService;
+    private final AllocationService allocationService;
+    private final ClusterManagerTaskThrottler.ThrottlingKey splitShardTaskKey;
+
+    public MetadataInPlaceShardSplitService(final ClusterService clusterService, final AllocationService allocationService) {
+        this.clusterService = clusterService;
+        this.allocationService = allocationService;
+        this.splitShardTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTask.IN_PLACE_SHARD_SPLIT, true);
+    }
+
+    /**
+     * Submits a cluster state update task to split a shard in-place.
+     */
+    public void split(final InPlaceShardSplitClusterStateUpdateRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {
+        clusterService.submitStateUpdateTask(
+            "in-place-split-shard [" + request.getShardId() + "] of index [" + request.getIndex() + "], cause [" + request.cause() + "]",
+            new AckedClusterStateUpdateTask<>(Priority.URGENT, request, listener) {
+                @Override
+                protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
+                    return new ClusterStateUpdateResponse(acknowledged);
+                }
+
+                @Override
+                public ClusterManagerTaskThrottler.ThrottlingKey getClusterManagerThrottlingKey() {
+                    return splitShardTaskKey;
+                }
+
+                @Override
+                public ClusterState execute(ClusterState currentState) {
+                    return applyShardSplitRequest(currentState, request, allocationService::reroute);
+                }
+
+                @Override
+                public void onFailure(String source, Exception e) {
+                    logger.trace(
+                        () -> new ParameterizedMessage(
+                            "[{}] of index [{}] failed to split online",
+                            request.getShardId(),
+                            request.getIndex()
+                        ),
+                        e
+                    );
+                    super.onFailure(source, e);
+                }
+            }
+        );
+    }
+
+    /**
+     * Applies a shard split request to the given cluster state. Updates the split metadata
+     * in the index metadata and triggers a reroute so that child shards get allocated.
+     */
+    static ClusterState applyShardSplitRequest(
+        ClusterState currentState,
+        InPlaceShardSplitClusterStateUpdateRequest request,
+        BiFunction<ClusterState, String, ClusterState> rerouteRoutingTable
+    ) {
+        IndexMetadata curIndexMetadata = currentState.metadata().index(request.getIndex());
+        if (curIndexMetadata == null) {
+            throw new IllegalArgumentException("Index [" + request.getIndex() + "] not found");
+        }
+
+        if (curIndexMetadata.getNumberOfVirtualShards() != -1) {
+            throw new IllegalArgumentException(
+                "In-place shard split is not supported on index [" + request.getIndex() + "] with virtual shards enabled"
+            );
+        }
+
+        int shardId = request.getShardId();
+        SplitShardsMetadata splitShardsMetadata = curIndexMetadata.getSplitShardsMetadata();
+
+        if (splitShardsMetadata.getInProgressSplitShardIds().contains(shardId)) {
+            throw new IllegalArgumentException("Splitting of shard [" + shardId + "] is already in progress");
+        }
+
+        if (splitShardsMetadata.isSplitParent(shardId)) {
+            throw new IllegalArgumentException("Shard [" + shardId + "] has already been split.");
+        }
+
+        RoutingTable.Builder routingTableBuilder = RoutingTable.builder(currentState.routingTable());
+        Metadata.Builder metadataBuilder = Metadata.builder(currentState.metadata());
+        IndexMetadata.Builder indexMetadataBuilder = IndexMetadata.builder(curIndexMetadata);
+
+        SplitShardsMetadata.Builder splitMetadataBuilder = new SplitShardsMetadata.Builder(splitShardsMetadata);
+        splitMetadataBuilder.splitShard(shardId, request.getSplitInto());
+        indexMetadataBuilder.splitShardsMetadata(splitMetadataBuilder.build());
+
+        RoutingTable routingTable = routingTableBuilder.build();
+        metadataBuilder.put(indexMetadataBuilder);
+
+        ClusterState updatedState = ClusterState.builder(currentState).metadata(metadataBuilder).routingTable(routingTable).build();
+        return rerouteRoutingTable.apply(updatedState, "shard [" + shardId + "] of index [" + request.getIndex() + "] split");
+    }
+}