Add MultiTypeMappingTransformation and test

DocumentsFromSnapshotMigration/src/test/java/org/opensearch/migrations/bulkload/EndToEndTest.java

@@ -85,23 +85,23 @@ private void migrationDocumentsWithClusters(
             // So we explicitly set it.
             String body = String.format(
                 "{" +
-                "  \"settings\": {" +
-                "    \"index\": {" +
-                "      \"number_of_shards\": %d," +
-                "      \"number_of_replicas\": 0" +
-                "    }" +
-                "  }" +
-                "}",
+                    "  \"settings\": {" +
+                    "    \"index\": {" +
+                    "      \"number_of_shards\": %d," +
+                    "      \"number_of_replicas\": 0" +
+                    "    }" +
+                    "  }" +
+                    "}",
                 numberOfShards
             );
             sourceClusterOperations.createIndex(indexName, body);
             targetClusterOperations.createIndex(indexName, body);
 
 
             sourceClusterOperations.createDocument(indexName, "222", "{\"author\":\"Tobias Funke\"}");
-            sourceClusterOperations.createDocument(indexName, "223", "{\"author\":\"Tobias Funke\", \"category\": \"cooking\"}", "1");
-            sourceClusterOperations.createDocument(indexName, "224", "{\"author\":\"Tobias Funke\", \"category\": \"cooking\"}", "1");
-            sourceClusterOperations.createDocument(indexName, "225", "{\"author\":\"Tobias Funke\", \"category\": \"tech\"}", "2");
+            sourceClusterOperations.createDocument(indexName, "223", "{\"author\":\"Tobias Funke\", \"category\": \"cooking\"}", "1", null);
+            sourceClusterOperations.createDocument(indexName, "224", "{\"author\":\"Tobias Funke\", \"category\": \"cooking\"}", "1", null);
+            sourceClusterOperations.createDocument(indexName, "225", "{\"author\":\"Tobias Funke\", \"category\": \"tech\"}", "2", null);
 
             // === ACTION: Take a snapshot ===
             var snapshotName = "my_snap";

DocumentsFromSnapshotMigration/src/test/java/org/opensearch/migrations/bulkload/SourceTestBase.java

@@ -54,10 +54,10 @@ public class SourceTestBase {
     public static final long TOLERABLE_CLIENT_SERVER_CLOCK_DIFFERENCE_SECONDS = 3600;
 
     protected static Object[] makeParamsForBase(SearchClusterContainer.ContainerVersion baseSourceImage) {
-        return new Object[] {
+        return new Object[]{
             baseSourceImage,
             GENERATOR_BASE_IMAGE,
-            new String[] { "/root/runTestBenchmarks.sh", "--endpoint", "http://" + SOURCE_SERVER_ALIAS + ":9200/" } };
+            new String[]{"/root/runTestBenchmarks.sh", "--endpoint", "http://" + SOURCE_SERVER_ALIAS + ":9200/"}};
     }
 
     @AllArgsConstructor
@@ -102,7 +102,7 @@ public static int migrateDocumentsSequentially(
         Version version,
         boolean compressionEnabled
     ) {
-        for (int runNumber = 1;; ++runNumber) {
+        for (int runNumber = 1; ; ++runNumber) {
             try {
                 var workResult = migrateDocumentsWithOneWorker(
                     sourceRepo,
@@ -146,7 +146,8 @@ public Flux<RfsLuceneDocument> readDocuments(int startSegmentIndex, int startDoc
         }
     }
 
-    static class LeasePastError extends Error {}
+    static class LeasePastError extends Error {
+    }
 
     @SneakyThrows
     public static DocumentsRunner.CompletionStatus migrateDocumentsWithOneWorker(

MetadataMigration/build.gradle

@@ -27,6 +27,7 @@ dependencies {
     testImplementation group: 'org.mockito', name: 'mockito-junit-jupiter'
     testImplementation group: 'org.hamcrest', name: 'hamcrest'
     testImplementation group: 'org.testcontainers', name: 'testcontainers'
+    testImplementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind'
 
     testRuntimeOnly group: 'org.junit.jupiter', name: 'junit-jupiter-engine'
 }

MetadataMigration/docs/DESIGN.md

@@ -142,17 +142,17 @@ Migration Candidates:
 
 Transformations:
    Index:
-      ERROR - IndexMappingTypeRemoval is Unsupported on Index `logs-181998` "Multiple mapping types are not supported""
+      ERROR - IndexMappingTypeRemoval is Unsupported on Index `logs-181998` "No multi type resolution behavior declared, specify --multi-type-behavior to process""
    Index Template:
-      ERROR - IndexMappingTypeRemoval is Unsupported on Index Template `daily_logs` "Multiple mapping types are not supported"
+      ERROR - IndexMappingTypeRemoval is Unsupported on Index Template `daily_logs` "No multi type resolution behavior declared, specify --multi-type-behavior to process"
    DEBUG - 6 transformations did not apply, add --`full` to see all results
 
 Result:
    2 migration issues detected
 
 Issues:
-   IndexMappingTypeRemoval is Unsupported on Index `logs-181998` "Multiple mapping types are not supported""
-   IndexMappingTypeRemoval is Unsupported on Index Template `daily_logs` "Multiple mapping types are not supported"
+   IndexMappingTypeRemoval is Unsupported on Index `logs-181998` "No multi type resolution behavior declared, specify --multi-type-behavior to process""
+   IndexMappingTypeRemoval is Unsupported on Index Template `daily_logs` "No multi type resolution behavior declared, specify --multi-type-behavior to process"
 ```
 ### Exclude incompatible rolling logs indices
 

MetadataMigration/src/main/java/org/opensearch/migrations/MigrateOrEvaluateArgs.java

@@ -4,8 +4,11 @@
 
 import org.opensearch.migrations.bulkload.common.http.ConnectionContext;
 import org.opensearch.migrations.bulkload.models.DataFilterArgs;
+import org.opensearch.migrations.bulkload.transformers.MetadataTransformerParams;
 import org.opensearch.migrations.transform.TransformerParams;
+import org.opensearch.migrations.transformation.rules.IndexMappingTypeRemoval;
 
+import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParametersDelegate;
 import lombok.Getter;
@@ -56,10 +59,20 @@ public class MigrateOrEvaluateArgs {
     public Version sourceVersion = null;
 
     @ParametersDelegate
-    public TransformerParams metadataTransformationParams = new MetadataTransformerParams();
+    public MetadataTransformationParams metadataTransformationParams = new MetadataTransformationParams();
+
+    @ParametersDelegate
+    public TransformerParams metadataCustomTransformationParams = new MetadataCustomTransformationParams();
+
+    @Getter
+    public static class MetadataTransformationParams implements MetadataTransformerParams {
+        @Parameter(names = {"--multi-type-behavior"}, description = "Define behavior for resolving multi type mappings.")
+        public IndexMappingTypeRemoval.MultiTypeResolutionBehavior multiTypeResolutionBehavior = IndexMappingTypeRemoval.MultiTypeResolutionBehavior.NONE;
+    }
 
     @Getter
-    public static class MetadataTransformerParams implements TransformerParams {
+    public static class MetadataCustomTransformationParams implements TransformerParams {
+
         public String getTransformerConfigParameterArgPrefix() {
             return "";
         }
@@ -89,4 +102,11 @@ public String getTransformerConfigParameterArgPrefix() {
                 description = "Path to the JSON configuration file of metadata transformers.")
         private String transformerConfigFile;
     }
+
+    static class MultiTypeResolutionBehaviorConverter implements IStringConverter<IndexMappingTypeRemoval.MultiTypeResolutionBehavior> {
+        @Override
+        public IndexMappingTypeRemoval.MultiTypeResolutionBehavior convert(String value) {
+            return IndexMappingTypeRemoval.MultiTypeResolutionBehavior.valueOf(value.toUpperCase());
+        }
+    }
 }

MetadataMigration/src/main/java/org/opensearch/migrations/commands/MigratorEvaluatorBase.java

@@ -55,7 +55,7 @@ protected Clusters createClusters() {
     }
 
     protected Transformer getCustomTransformer() {
-        var transformerConfig = TransformerConfigUtils.getTransformerConfig(arguments.metadataTransformationParams);
+        var transformerConfig = TransformerConfigUtils.getTransformerConfig(arguments.metadataCustomTransformationParams);
         if (transformerConfig != null) {
             log.atInfo().setMessage("Metadata Transformations config string: {}")
                     .addArgument(transformerConfig).log();
@@ -72,7 +72,8 @@ protected Transformer selectTransformer(Clusters clusters) {
         var versionTransformer = TransformFunctions.getTransformer(
             clusters.getSource().getVersion(),
             clusters.getTarget().getVersion(),
-            arguments.minNumberOfReplicas
+            arguments.minNumberOfReplicas,
+            arguments.metadataTransformationParams
         );
         var customTransformer = getCustomTransformer();
         var compositeTransformer = new CompositeTransformer(customTransformer, versionTransformer);

MetadataMigration/src/test/java/org/opensearch/migrations/BaseMigrationTest.java

@@ -0,0 +1,144 @@
+package org.opensearch.migrations;
+
+import java.io.File;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import org.opensearch.migrations.bulkload.common.FileSystemSnapshotCreator;
+import org.opensearch.migrations.bulkload.common.OpenSearchClient;
+import org.opensearch.migrations.bulkload.common.http.ConnectionContextTestParams;
+import org.opensearch.migrations.bulkload.framework.SearchClusterContainer;
+import org.opensearch.migrations.bulkload.http.ClusterOperations;
+import org.opensearch.migrations.bulkload.worker.SnapshotRunner;
+import org.opensearch.migrations.commands.MigrationItemResult;
+import org.opensearch.migrations.metadata.tracing.MetadataMigrationTestContext;
+import org.opensearch.migrations.snapshot.creation.tracing.SnapshotTestContext;
+
+import lombok.Getter;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Base test class providing shared setup and utility methods for migration tests.
+ */
+@Slf4j
+abstract class BaseMigrationTest {
+
+    @TempDir
+    protected File localDirectory;
+
+    @Getter
+    protected SearchClusterContainer sourceCluster;
+    @Getter
+    protected SearchClusterContainer targetCluster;
+
+    protected ClusterOperations sourceOperations;
+    protected ClusterOperations targetOperations;
+
+    /**
+     * Starts the source and target clusters.
+     */
+    protected void startClusters() {
+        CompletableFuture.allOf(
+                CompletableFuture.runAsync(sourceCluster::start),
+                CompletableFuture.runAsync(targetCluster::start)
+        ).join();
+
+        sourceOperations = new ClusterOperations(sourceCluster.getUrl());
+        targetOperations = new ClusterOperations(targetCluster.getUrl());
+    }
+
+    /**
+     * Sets up a snapshot repository and takes a snapshot of the source cluster.
+     *
+     * @param snapshotName Name of the snapshot.
+     * @return The name of the created snapshot.
+     */
+    @SneakyThrows
+    protected String createSnapshot(String snapshotName) {
+        var snapshotContext = SnapshotTestContext.factory().noOtelTracking();
+        var sourceClient = new OpenSearchClient(ConnectionContextTestParams.builder()
+                .host(sourceCluster.getUrl())
+                .insecure(true)
+                .build()
+                .toConnectionContext());
+        var snapshotCreator = new org.opensearch.migrations.bulkload.common.FileSystemSnapshotCreator(
+                snapshotName,
+                sourceClient,
+                SearchClusterContainer.CLUSTER_SNAPSHOT_DIR,
+                List.of(),
+                snapshotContext.createSnapshotCreateContext()
+        );
+        org.opensearch.migrations.bulkload.worker.SnapshotRunner.runAndWaitForCompletion(snapshotCreator);
+        sourceCluster.copySnapshotData(localDirectory.toString());
+        return snapshotName;
+    }
+
+    /**
+     * Prepares migration arguments for snapshot-based migrations.
+     *
+     * @param snapshotName Name of the snapshot.
+     * @return Prepared migration arguments.
+     */
+    protected MigrateOrEvaluateArgs prepareSnapshotMigrationArgs(String snapshotName) {
+        var arguments = new MigrateOrEvaluateArgs();
+        arguments.fileSystemRepoPath = localDirectory.getAbsolutePath();
+        arguments.snapshotName = snapshotName;
+        arguments.sourceVersion = sourceCluster.getContainerVersion().getVersion();
+        arguments.targetArgs.host = targetCluster.getUrl();
+        return arguments;
+    }
+
+    /**
+     * Executes the migration command (either migrate or evaluate).
+     *
+     * @param arguments Migration arguments.
+     * @param command   The migration command to execute.
+     * @return The result of the migration.
+     */
+    protected MigrationItemResult executeMigration(MigrateOrEvaluateArgs arguments, MetadataCommands command) {
+        var metadataContext = MetadataMigrationTestContext.factory().noOtelTracking();
+        var metadata = new MetadataMigration();
+
+        if (MetadataCommands.MIGRATE.equals(command)) {
+            return metadata.migrate(arguments).execute(metadataContext);
+        } else {
+            return metadata.evaluate(arguments).execute(metadataContext);
+        }
+    }
+
+    /**
+     * Creates an OpenSearch client for the given cluster.
+     *
+     * @param cluster The cluster container.
+     * @return An OpenSearch client.
+     */
+    protected OpenSearchClient createClient(SearchClusterContainer cluster) {
+        return new OpenSearchClient(ConnectionContextTestParams.builder()
+                .host(cluster.getUrl())
+                .insecure(true)
+                .build()
+                .toConnectionContext());
+    }
+
+    protected SnapshotTestContext createSnapshotContext() {
+        return SnapshotTestContext.factory().noOtelTracking();
+    }
+
+    protected FileSystemSnapshotCreator createSnapshotCreator(String snapshotName, OpenSearchClient client, SnapshotTestContext context) {
+        return new FileSystemSnapshotCreator(
+                snapshotName,
+                client,
+                SearchClusterContainer.CLUSTER_SNAPSHOT_DIR,
+                List.of(),
+                context.createSnapshotCreateContext()
+        );
+    }
+
+    @SneakyThrows
+    protected void runSnapshotAndCopyData(FileSystemSnapshotCreator snapshotCreator, SearchClusterContainer cluster) {
+        SnapshotRunner.runAndWaitForCompletion(snapshotCreator);
+        cluster.copySnapshotData(localDirectory.toString());
+    }
+}