Type mappings transformation

DocumentsFromSnapshotMigration/build.gradle

@@ -30,7 +30,7 @@ dependencies {
     implementation project(":RFS")
     implementation project(":transformation")
     implementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonMessageTransformerLoaders')
-    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:openSearch23PlusTargetTransformerProvider')
+    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonTypeMappingsSanitizationTransformerProvider')
 
     implementation group: 'org.apache.logging.log4j', name: 'log4j-api'
     implementation group: 'org.apache.logging.log4j', name: 'log4j-core'

RFS/build.gradle

@@ -62,7 +62,7 @@ dependencies {
     testImplementation group: 'org.mockito', name: 'mockito-junit-jupiter'
 
     testImplementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonMessageTransformerLoaders')
-    testRuntimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:openSearch23PlusTargetTransformerProvider')
+    testRuntimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonTypeMappingsSanitizationTransformerProvider')
 
     testRuntimeOnly group: 'org.junit.jupiter', name: 'junit-jupiter-engine'
 

TrafficCapture/dockerSolution/src/main/docker/docker-compose.yml

@@ -78,8 +78,8 @@ services:
         condition: service_started
       opensearchtarget:
         condition: service_started
-    command: /bin/sh -c "/runJavaWithClasspath.sh org.opensearch.migrations.replay.TrafficReplayer --speedup-factor 2 https://opensearchtarget:9200 --auth-header-value Basic\\ YWRtaW46bXlTdHJvbmdQYXNzd29yZDEyMyE= --insecure --kafka-traffic-brokers kafka:9092 --kafka-traffic-topic logging-traffic-topic --kafka-traffic-group-id logging-group-default --otelCollectorEndpoint http://otel-collector:4317" #--transformer-config-base64 W3sgIkpzb25Kb2x0VHJhbnNmb3JtZXJQcm92aWRlciI6ClsKICB7CiAgICAic2NyaXB0IjogewogICAgICAib3BlcmF0aW9uIjogInNoaWZ0IiwKICAgICAgInNwZWMiOiB7CiAgICAgICAgIm1ldGhvZCI6ICJtZXRob2QiLAogICAgICAgICJVUkkiOiAiVVJJIiwKICAgICAgICAiaGVhZGVycyI6ICJoZWFkZXJzIiwKICAgICAgICAicGF5bG9hZCI6IHsKICAgICAgICAgICJpbmxpbmVkSnNvbkJvZHkiOiB7CiAgICAgICAgICAgICJ0b3AiOiB7CiAgICAgICAgICAgICAgInRhZ1RvRXhjaXNlIjogewogICAgICAgICAgICAgICAgIioiOiAicGF5bG9hZC5pbmxpbmVkSnNvbkJvZHkudG9wLiYiIAogICAgICAgICAgICAgIH0sCiAgICAgICAgICAgICAgIioiOiAicGF5bG9hZC5pbmxpbmVkSnNvbkJvZHkudG9wLiYiCiAgICAgICAgICAgIH0sCiAgICAgICAgICAiKiI6ICJwYXlsb2FkLmlubGluZWRKc29uQm9keS4mIgogICAgICAgICAgfQogICAgICAgIH0KICAgICAgfQogICAgfQogIH0sIAogewogICAic2NyaXB0IjogewogICAgICJvcGVyYXRpb24iOiAibW9kaWZ5LW92ZXJ3cml0ZS1iZXRhIiwKICAgICAic3BlYyI6IHsKICAgICAgICJVUkkiOiAiPXNwbGl0KCcvZXh0cmFUaGluZ1RvUmVtb3ZlJyxAKDEsJikpIgogICAgIH0KICB9CiB9LAogewogICAic2NyaXB0IjogewogICAgICJvcGVyYXRpb24iOiAibW9kaWZ5LW92ZXJ3cml0ZS1iZXRhIiwKICAgICAic3BlYyI6IHsKICAgICAgICJVUkkiOiAiPWpvaW4oJycsQCgxLCYpKSIKICAgICB9CiAgfQogfQpdCn1dCg=="
-
+#   command: /bin/sh -c "/runJavaWithClasspath.sh org.opensearch.migrations.replay.TrafficReplayer --speedup-factor 2 https://opensearchtarget:9200 --auth-header-value Basic\\ YWRtaW46bXlTdHJvbmdQYXNzd29yZDEyMyE= --insecure --kafka-traffic-brokers kafka:9092 --kafka-traffic-topic logging-traffic-topic --kafka-traffic-group-id logging-group-default --otelCollectorEndpoint http://otel-collector:4317 --transformer-config "
+    command: /bin/sh -c "/runJavaWithClasspath.sh org.opensearch.migrations.replay.TrafficReplayer --speedup-factor 2 https://opensearchtarget:9200 --auth-header-value Basic\\ YWRtaW46bXlTdHJvbmdQYXNzd29yZDEyMyE= --insecure --kafka-traffic-brokers kafka:9092 --kafka-traffic-topic logging-traffic-topic --kafka-traffic-group-id logging-group-default --otelCollectorEndpoint http://otel-collector:4317 --transformer-config '[{\"TypeMappingSanitizationTransformerProvider\":\"\"}]'"
   opensearchtarget:
     image: 'opensearchproject/opensearch:2.15.0'
     environment:

TrafficCapture/trafficReplayer/README.md

@@ -145,15 +145,15 @@ transform to add GZIP encoding and another to apply a new header would be config
 ```
 
 To run only one transformer without any configuration, the `--transformer-config` argument can simply 
-be set to the name of the transformer (e.g. 'JsonTransformerForOpenSearch23PlusTargetTransformerProvider', 
+be set to the name of the transformer (e.g. 'TypeMappingSanitizationTransformerProvider', 
 without quotes or any json surrounding it).
 
 The user can also specify a file to read the transformations from using the `--transformer-config-file`.  Users can
 also pass the script as an argument via `--transformer-config-base64`.  Each of the `transformer-config` options
 is mutually exclusive.
 
 Some simple transformations are included to change headers to add compression or to force an HTTP message payload to 
-be chunked.  Another transformer, [JsonTypeMappingTransformer.java](../../transformation/transformationPlugins/jsonMessageTransformers/openSearch23PlusTargetTransformerProvider/src/main/java/org/opensearch/migrations/transform/JsonTypeMappingTransformer.java),
+be chunked.  Another transformer, [TypeMappingSanitizationTransformer.java](../../transformation/transformationPlugins/jsonMessageTransformers/jsonTypeMappingsSanitizationTransformer/src/main/java/org/opensearch/migrations/transform/TypeMappingsSanitizationTransformer.java),
 is a work-in-progress to excise type mapping references from URIs and message payloads since versions of OpenSource
 greater than 2.3 do not support them.
 

TrafficCapture/trafficReplayer/build.gradle

@@ -23,7 +23,7 @@ dependencies {
     implementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonMessageTransformerInterface')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJMESPathMessageTransformerProvider')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJoltMessageTransformerProvider')
-    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:openSearch23PlusTargetTransformerProvider')
+    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonTypeMappingsSanitizationTransformerProvider')
 
     implementation group: 'org.jcommander', name: 'jcommander'
     implementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind'
@@ -59,7 +59,7 @@ dependencies {
     testImplementation testFixtures(project(path: ':coreUtilities'))
     testImplementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJMESPathMessageTransformerProvider')
     testImplementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJoltMessageTransformerProvider')
-    testImplementation project(':transformation:transformationPlugins:jsonMessageTransformers:openSearch23PlusTargetTransformerProvider')
+    testImplementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonTypeMappingsSanitizationTransformerProvider')
 
     testImplementation group: 'org.apache.httpcomponents.client5', name: 'httpclient5'
     testImplementation group: 'org.junit.jupiter', name:'junit-jupiter-api'

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/datahandlers/PayloadAccessFaultingMap.java

@@ -32,6 +32,7 @@ public class PayloadAccessFaultingMap extends AbstractMap<String, Object> {
     @Getter
     @Setter
     private boolean disableThrowingPayloadNotLoaded;
+    private boolean payloadWasAccessed;
 
     public PayloadAccessFaultingMap(StrictCaseInsensitiveHttpHeadersMap headers) {
         underlyingMap = new TreeMap<>();
@@ -51,19 +52,19 @@ public Iterator<Map.Entry<String, Object>> iterator() {
                     return new Iterator<>() {
                         @Override
                         public boolean hasNext() {
-                            throw PayloadNotLoadedException.getInstance();
+                            throw makeFault();
                         }
 
                         @Override
                         public Map.Entry<String, Object> next() {
-                            throw PayloadNotLoadedException.getInstance();
+                            throw makeFault();
                         }
                     };
                 }
 
                 @Override
                 public int size() {
-                    throw PayloadNotLoadedException.getInstance();
+                    throw makeFault();
                 }
             };
         } else {
@@ -80,8 +81,21 @@ public Object put(String key, Object value) {
     public Object get(Object key) {
         var value = super.get(key);
         if (value == null && !disableThrowingPayloadNotLoaded) {
-            throw PayloadNotLoadedException.getInstance();
+            throw makeFault();
         }
         return value;
     }
+
+    public boolean missingPaylaodWasAccessed() {
+        return payloadWasAccessed;
+    }
+
+    public void resetMissingPaylaodWasAccessed() {
+        payloadWasAccessed = false;
+    }
+
+    private PayloadNotLoadedException makeFault() throws PayloadNotLoadedException {
+        payloadWasAccessed = true;
+        return PayloadNotLoadedException.getInstance();
+    }
 }

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/datahandlers/http/NettyDecodedHttpRequestPreliminaryTransformHandler.java

@@ -7,7 +7,6 @@
 import java.util.Optional;
 
 import org.opensearch.migrations.replay.datahandlers.PayloadAccessFaultingMap;
-import org.opensearch.migrations.replay.datahandlers.PayloadNotLoadedException;
 import org.opensearch.migrations.replay.tracing.IReplayContexts;
 import org.opensearch.migrations.transform.IAuthTransformer;
 import org.opensearch.migrations.transform.IJsonTransformer;
@@ -67,25 +66,34 @@ public void channelRead(@NonNull ChannelHandlerContext ctx, @NonNull Object msg)
             IAuthTransformer authTransformer = requestPipelineOrchestrator.authTransfomerFactory.getAuthTransformer(
                 httpJsonMessage
             );
+            HttpJsonRequestWithFaultingPayload transformedMessage = null;
             try {
+                transformedMessage = transform(transformer, httpJsonMessage);
+            } catch (Exception e) {
+                var payload = (PayloadAccessFaultingMap) httpJsonMessage.payload();
+                if (payload.missingPaylaodWasAccessed()) {
+                    payload.resetMissingPaylaodWasAccessed();
+                    log.atDebug().setMessage("The transforms for this message require payload manipulation, "
+                        + "all content handlers are being loaded.").log();
+                    // make a fresh message and its headers
+                    requestPipelineOrchestrator.addJsonParsingHandlers(
+                        ctx,
+                        transformer,
+                        getAuthTransformerAsStreamingTransformer(authTransformer)
+                    );
+                    ctx.fireChannelRead(handleAuthHeaders(httpJsonMessage, authTransformer));
+                } else{
+                    throw new TransformationException(e);
+                }
+            }
+
+            if (transformedMessage != null) {
                 handlePayloadNeutralTransformationOrThrow(
                     ctx,
                     originalHttpJsonMessage,
-                    transform(transformer, httpJsonMessage),
+                    transformedMessage,
                     authTransformer
                 );
-            } catch (PayloadNotLoadedException pnle) {
-                log.debug(
-                    "The transforms for this message require payload manipulation, "
-                        + "all content handlers are being loaded."
-                );
-                // make a fresh message and its headers
-                requestPipelineOrchestrator.addJsonParsingHandlers(
-                    ctx,
-                    transformer,
-                    getAuthTransformerAsStreamingTransformer(authTransformer)
-                );
-                ctx.fireChannelRead(handleAuthHeaders(httpJsonMessage, authTransformer));
             }
         } else if (msg instanceof HttpContent) {
             ctx.fireChannelRead(msg);
@@ -131,43 +139,35 @@ private void handlePayloadNeutralTransformationOrThrow(
 
         var pipeline = ctx.pipeline();
         if (streamingAuthTransformer != null) {
-            log.info(
-                diagnosticLabel
-                    + "An Authorization Transformation is required for this message.  "
-                    + "The headers and payload will be parsed and reformatted."
-            );
+            log.atInfo().setMessage("{} An Authorization Transformation is required for this message.  "
+                    + "The headers and payload will be parsed and reformatted.")
+                .addArgument(diagnosticLabel).log();
             requestPipelineOrchestrator.addContentRepackingHandlers(ctx, streamingAuthTransformer);
             ctx.fireChannelRead(httpJsonMessage);
         } else if (headerFieldsAreIdentical(originalRequest, httpJsonMessage)) {
-            log.info(
-                diagnosticLabel
-                    + "Transformation isn't necessary.  "
-                    + "Resetting the processing pipeline to let the caller send the original network bytes as-is."
-            );
+            log.atInfo().setMessage("{} Transformation isn't necessary.  "
+                    + "Resetting the processing pipeline to let the caller send the original network bytes as-is.")
+                .addArgument(diagnosticLabel)
+                .log();
             RequestPipelineOrchestrator.removeAllHandlers(pipeline);
-
         } else if (headerFieldIsIdentical("content-encoding", originalRequest, httpJsonMessage)
             && headerFieldIsIdentical("transfer-encoding", originalRequest, httpJsonMessage)) {
-                log.info(
-                    diagnosticLabel
-                        + "There were changes to the headers that require the message to be reformatted "
-                        + "but the payload doesn't need to be transformed."
-                );
-                // By adding the baseline handlers and removing this and previous handlers in reverse order,
-                // we will cause the upstream handlers to flush their in-progress accumulated ByteBufs downstream
-                // to be processed accordingly
-                requestPipelineOrchestrator.addBaselineHandlers(pipeline);
-                ctx.fireChannelRead(httpJsonMessage);
-                RequestPipelineOrchestrator.removeThisAndPreviousHandlers(pipeline, this);
-            } else {
-                log.info(
-                    diagnosticLabel
-                        + "New headers have been specified that require the payload stream to be "
-                        + "reformatted.  Setting up the processing pipeline to parse and reformat the request payload."
-                );
-                requestPipelineOrchestrator.addContentRepackingHandlers(ctx, streamingAuthTransformer);
-                ctx.fireChannelRead(httpJsonMessage);
-            }
+            log.atInfo().setMessage("{} There were changes to the headers that require the message to be reformatted "
+                    + "but the payload doesn't need to be transformed.")
+                .addArgument(diagnosticLabel).log();
+            // By adding the baseline handlers and removing this and previous handlers in reverse order,
+            // we will cause the upstream handlers to flush their in-progress accumulated ByteBufs downstream
+            // to be processed accordingly
+            requestPipelineOrchestrator.addBaselineHandlers(pipeline);
+            ctx.fireChannelRead(httpJsonMessage);
+            RequestPipelineOrchestrator.removeThisAndPreviousHandlers(pipeline, this);
+        } else {
+            log.atInfo().setMessage("{} New headers have been specified that require the payload stream to be "
+                    + "reformatted.  Setting up the processing pipeline to parse and reformat the request payload.")
+                .addArgument(diagnosticLabel).log();
+            requestPipelineOrchestrator.addContentRepackingHandlers(ctx, streamingAuthTransformer);
+            ctx.fireChannelRead(httpJsonMessage);
+        }
     }
 
     private static HttpJsonRequestWithFaultingPayload handleAuthHeaders(

TrafficCapture/trafficReplayer/src/test/java/org/opensearch/migrations/replay/PayloadNotFoundTest.java

@@ -0,0 +1,37 @@
+package org.opensearch.migrations.replay;
+
+
+import org.opensearch.migrations.replay.datahandlers.PayloadAccessFaultingMap;
+import org.opensearch.migrations.replay.datahandlers.http.HttpJsonRequestWithFaultingPayload;
+import org.opensearch.migrations.replay.datahandlers.http.ListKeyAdaptingCaseInsensitiveHeadersMap;
+import org.opensearch.migrations.replay.datahandlers.http.StrictCaseInsensitiveHttpHeadersMap;
+import org.opensearch.migrations.transform.TransformationLoader;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PayloadNotFoundTest {
+    @Test
+    public void testTransformsPropagateExceptionProperly() throws JsonProcessingException {
+        final HttpJsonRequestWithFaultingPayload FAULTING_MAP = new HttpJsonRequestWithFaultingPayload();
+        FAULTING_MAP.setMethod("PUT");
+        FAULTING_MAP.setPath("/_bulk");
+        FAULTING_MAP.setHeaders(new ListKeyAdaptingCaseInsensitiveHeadersMap(new StrictCaseInsensitiveHttpHeadersMap()));
+        FAULTING_MAP.headers().put("Content-Type", "application/json");
+        FAULTING_MAP.setPayloadFaultMap(new PayloadAccessFaultingMap(FAULTING_MAP.headers().asStrictMap()));
+        final String EXPECTED = "{\n"
+            + "  \"method\": \"PUT\",\n"
+            + "  \"URI\": \"/_bulk\",\n"
+            + "  \"headers\": {\n"
+            + "      \"Content-Type\": \"application/json\"\n"
+            + "  }\n"
+            + "}\n";
+
+        var transformer = new TransformationLoader().getTransformerFactoryLoader("newhost", null,
+            "[{\"TypeMappingSanitizationTransformerProvider\":\"\"}]");
+        var e = Assertions.assertThrows(Exception.class,
+            () -> transformer.transformJson(FAULTING_MAP));
+        Assertions.assertTrue(((PayloadAccessFaultingMap)FAULTING_MAP.payload()).missingPaylaodWasAccessed());
+    }
+}

TrafficCapture/trafficReplayer/src/test/java/org/opensearch/migrations/replay/datahandlers/http/HttpJsonTransformingConsumerTest.java

@@ -331,7 +331,7 @@ public void testMalformedPayload_andThrowingTransformation_IsPassedThrough() thr
             new TransformationLoader().getTransformerFactoryLoader(
                 HOST_NAME,
                 null,
-                "[{\"JsonTransformerForOpenSearch23PlusTargetTransformerProvider\":\"\"}]"
+                "[{\"TypeMappingSanitizationTransformerProvider\":\"\"}]"
             ),
             null,
             testPacketCapture,

commonDependencyVersionConstraints/build.gradle

@@ -55,6 +55,8 @@ dependencies {
 
     def jackson = '2.16.2'
     api group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: jackson
+    api group: 'com.fasterxml.jackson.core', name: 'jackson-dataformat-yaml', version: jackson
+    api group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: jackson
     api group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-smile', version: jackson
 
     def jupiter = '5.10.2'

testHelperFixtures/src/testFixtures/java/org/opensearch/migrations/testutils/JsonNormalizer.java

@@ -0,0 +1,23 @@
+package org.opensearch.migrations.testutils;
+
+import java.util.SortedMap;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import lombok.SneakyThrows;
+
+public class JsonNormalizer {
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
+        .configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS, true)
+        .configure(SerializationFeature.INDENT_OUTPUT, true);
+
+    @SneakyThrows
+    public static String fromString(String input) {
+        return OBJECT_MAPPER.writeValueAsString(OBJECT_MAPPER.readValue(input, SortedMap.class));
+    }
+
+    @SneakyThrows
+    public static String fromObject(Object obj) {
+        return fromString(OBJECT_MAPPER.writeValueAsString(obj));
+    }
+}