Type mappings transformation

RFS/build.gradle

@@ -21,7 +21,9 @@ dependencies {
 
     implementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonMessageTransformerInterface')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJMESPathMessageTransformerProvider')
+    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJinjavaTransformerProvider')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJoltMessageTransformerProvider')
+    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonTypeMappingsSanitizationTransformerProvider')
 
     implementation group: 'org.jcommander', name: 'jcommander'
     implementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind'

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 "
-    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\":\"\"}]'"
+#   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  "
+    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\":{\"sourceProperties\":{\"version\":{\"major\":7,\"minor\":10}}}}]'"
   opensearchtarget:
     image: 'opensearchproject/opensearch:2.15.0'
     environment:

TrafficCapture/trafficReplayer/build.gradle

@@ -22,6 +22,7 @@ dependencies {
     implementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonMessageTransformerLoaders')
     implementation project(':transformation:transformationPlugins:jsonMessageTransformers:jsonMessageTransformerInterface')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJMESPathMessageTransformerProvider')
+    runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJinjavaTransformerProvider')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonJoltMessageTransformerProvider')
     runtimeOnly project(':transformation:transformationPlugins:jsonMessageTransformers:jsonTypeMappingsSanitizationTransformerProvider')
 

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/ParsedHttpMessagesAsDicts.java

@@ -42,6 +42,10 @@ public class ParsedHttpMessagesAsDicts {
     public static final String STATUS_CODE_KEY = "Status-Code";
     public static final String RESPONSE_TIME_MS_KEY = "response_time_ms";
     public static final String EXCEPTION_KEY_STRING = "Exception";
+    public static final String REQUEST_URI_KEY = "Request-URI";
+    public static final String METHOD_KEY = "Method";
+    public static final String HTTP_VERSION_KEY = "HTTP-Version";
+    public static final String PAYLOAD_KEY = "payload";
 
     public final Optional<Map<String, Object>> sourceRequestOp;
     public final Optional<Map<String, Object>> sourceResponseOp;
@@ -183,15 +187,15 @@ private static Map<String, Object> convertRequest(
                 var message = (HttpJsonRequestWithFaultingPayload) messageHolder.get();
                 if (message != null) {
                     var map = new LinkedHashMap<>(message.headers());
-                    map.put("Request-URI", message.path());
-                    map.put("Method", message.method());
-                    map.put("HTTP-Version", message.protocol());
+                    map.put(REQUEST_URI_KEY, message.path());
+                    map.put(METHOD_KEY, message.method());
+                    map.put(HTTP_VERSION_KEY, message.protocol());
                     context.setMethod(message.method());
                     context.setEndpoint(message.path());
                     context.setHttpVersion(message.protocol());
                     encodeBinaryPayloadIfExists(message);
                     if (!message.payload().isEmpty()) {
-                        map.put("payload", message.payload());
+                        map.put(PAYLOAD_KEY, message.payload());
                     }
                     return map;
                 } else {
@@ -223,14 +227,14 @@ private static Map<String, Object> convertResponse(
                 var message = (HttpJsonResponseWithFaultingPayload) messageHolder.get();
                 if (message != null) {
                     var map = new LinkedHashMap<>(message.headers());
-                    map.put("HTTP-Version", message.protocol());
+                    map.put(HTTP_VERSION_KEY, message.protocol());
                     map.put(STATUS_CODE_KEY, Integer.parseInt(message.code()));
                     map.put("Reason-Phrase", message.reason());
                     map.put(RESPONSE_TIME_MS_KEY, latency.toMillis());
                     context.setHttpVersion(message.protocol());
                     encodeBinaryPayloadIfExists(message);
                     if (!message.payload().isEmpty()) {
-                        map.put("payload", message.payload());
+                        map.put(PAYLOAD_KEY, message.payload());
                     }
                     return map;
                 } else {

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/ResultsToLogsConsumer.java

@@ -161,6 +161,7 @@ public static String getTransactionSummaryStringPreamble() {
             .add("SOURCE_STATUS_CODE/TARGET_STATUS_CODE...")
             .add("SOURCE_RESPONSE_SIZE_BYTES/TARGET_RESPONSE_SIZE_BYTES...")
             .add("SOURCE_LATENCY_MS/TARGET_LATENCY_MS...")
+            .add("URI...")
             .toString();
     }
 
@@ -218,6 +219,11 @@ public static String toTransactionSummaryString(
                     transformStreamToString(parsed.targetResponseList.stream(),
                         r -> "" + r.get(ParsedHttpMessagesAsDicts.RESPONSE_TIME_MS_KEY))
             )
+            // uri
+            .add(
+                parsed.sourceRequestOp
+                    .map(r -> (String) r.get(ParsedHttpMessagesAsDicts.REQUEST_URI_KEY))
+                    .orElse(MISSING_STR))
             .toString();
     }
 

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

@@ -116,10 +116,10 @@ public Object getNextTopLevelObject() throws IOException {
                     pushCompletedValue(parser.getText());
                     break;
                 case VALUE_NUMBER_INT:
-                    pushCompletedValue(parser.getIntValue());
+                    pushCompletedValue(parser.getLongValue());
                     break;
                 case VALUE_NUMBER_FLOAT:
-                    pushCompletedValue(parser.getFloatValue());
+                    pushCompletedValue(parser.getDoubleValue());
                     break;
                 case NOT_AVAILABLE:
                     // pipeline stall - need more data

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

@@ -35,6 +35,7 @@ public class PayloadAccessFaultingMap extends AbstractMap<String, Object> {
     private boolean payloadWasAccessed;
 
     public PayloadAccessFaultingMap(StrictCaseInsensitiveHttpHeadersMap headers) {
+        disableThrowingPayloadNotLoaded = true;
         underlyingMap = new TreeMap<>();
         isJson = Optional.ofNullable(headers.get("content-type"))
             .map(list -> list.stream().anyMatch(s -> s.startsWith("application/json")))
@@ -86,11 +87,11 @@ public Object get(Object key) {
         return value;
     }
 
-    public boolean missingPaylaodWasAccessed() {
+    public boolean missingPayloadWasAccessed() {
         return payloadWasAccessed;
     }
 
-    public void resetMissingPaylaodWasAccessed() {
+    public void resetMissingPayloadWasAccessed() {
         payloadWasAccessed = false;
     }
 

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

@@ -67,12 +67,14 @@ public void channelRead(@NonNull ChannelHandlerContext ctx, @NonNull Object msg)
                 httpJsonMessage
             );
             HttpJsonRequestWithFaultingPayload transformedMessage = null;
+            final var payloadMap = (PayloadAccessFaultingMap) httpJsonMessage.payload();
             try {
+                payloadMap.setDisableThrowingPayloadNotLoaded(false);
                 transformedMessage = transform(transformer, httpJsonMessage);
             } catch (Exception e) {
                 var payload = (PayloadAccessFaultingMap) httpJsonMessage.payload();
-                if (payload.missingPaylaodWasAccessed()) {
-                    payload.resetMissingPaylaodWasAccessed();
+                if (payload.missingPayloadWasAccessed()) {
+                    payload.resetMissingPayloadWasAccessed();
                     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
@@ -85,6 +87,8 @@ public void channelRead(@NonNull ChannelHandlerContext ctx, @NonNull Object msg)
                 } else{
                     throw new TransformationException(e);
                 }
+            } finally {
+                payloadMap.setDisableThrowingPayloadNotLoaded(true);
             }
 
             if (transformedMessage != null) {
@@ -139,32 +143,40 @@ private void handlePayloadNeutralTransformationOrThrow(
 
         var pipeline = ctx.pipeline();
         if (streamingAuthTransformer != null) {
-            log.atInfo().setMessage("{} An Authorization Transformation is required for this message.  "
-                    + "The headers and payload will be parsed and reformatted.")
-                .addArgument(diagnosticLabel).log();
+            log.info(
+                diagnosticLabel
+                    + "An Authorization Transformation is required for this message.  "
+                    + "The headers and payload will be parsed and reformatted."
+            );
             requestPipelineOrchestrator.addContentRepackingHandlers(ctx, streamingAuthTransformer);
             ctx.fireChannelRead(httpJsonMessage);
         } else if (headerFieldsAreIdentical(originalRequest, httpJsonMessage)) {
-            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();
+            log.info(
+                diagnosticLabel
+                    + "Transformation isn't necessary.  "
+                    + "Resetting the processing pipeline to let the caller send the original network bytes as-is."
+            );
             RequestPipelineOrchestrator.removeAllHandlers(pipeline);
+
         } else if (headerFieldIsIdentical("content-encoding", originalRequest, httpJsonMessage)
             && headerFieldIsIdentical("transfer-encoding", originalRequest, 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();
+            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.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();
+            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);
         }

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

@@ -8,6 +8,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Predicate;
 
 import org.opensearch.migrations.replay.datahandlers.JsonAccumulator;
 import org.opensearch.migrations.replay.tracing.IReplayContexts;
@@ -24,6 +25,7 @@
 import io.netty.util.ReferenceCountUtil;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.slf4j.event.Level;
 
 /**
  * This accumulates HttpContent messages through a JsonAccumulator and eventually fires off a
@@ -93,7 +95,10 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                     }
                 }
             } catch (JacksonException e) {
-                log.atInfo().setCause(e).setMessage("Error parsing json body.  " +
+                log.atLevel(hasRequestContentTypeMatching(capturedHttpJsonMessage,
+                        // a JacksonException for non-json data doesn't need to be surfaced to a user
+                        v -> v.startsWith("application/json")) ?  Level.INFO : Level.TRACE)
+                    .setCause(e).setMessage("Error parsing json body.  " +
                     "Will pass all payload bytes directly as a ByteBuf within the payload map").log();
                 jsonWasInvalid = true;
                 parsedJsonObjects.clear();
@@ -123,7 +128,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 
                     var leftoverBody = accumulatedBody.slice(jsonBodyByteLength,
                         accumulatedBody.readableBytes() - jsonBodyByteLength);
-                    if (jsonBodyByteLength == 0 && isRequestContentTypeNotText(capturedHttpJsonMessage)) {
+                    if (jsonBodyByteLength == 0 &&
+                        hasRequestContentTypeMatching(capturedHttpJsonMessage, v -> !v.startsWith("text/")))
+                    {
                         context.onPayloadSetBinary();
                         capturedHttpJsonMessage.payload()
                             .put(JsonKeysForHttpMessage.INLINED_BINARY_BODY_DOCUMENT_KEY,
@@ -157,12 +164,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
         }
     }
 
-    private boolean isRequestContentTypeNotText(HttpJsonMessageWithFaultingPayload message) {
+    private boolean hasRequestContentTypeMatching(HttpJsonMessageWithFaultingPayload message,
+                                                  Predicate<String> contentTypeFilter) {
         // ContentType not text if specified and has a value with / and that value does not start with text/
         return Optional.ofNullable(capturedHttpJsonMessage.headers().insensitiveGet(HttpHeaderNames.CONTENT_TYPE.toString()))
             .map(s -> s.stream()
                 .filter(v -> v.contains("/"))
-                .filter(v -> !v.startsWith("text/"))
+                .filter(contentTypeFilter)
                 .count() > 1
             )
             .orElse(false);

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

@@ -19,7 +19,9 @@ public void testTransformsPropagateExceptionProperly() throws JsonProcessingExce
         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()));
+        var payloadMap = new PayloadAccessFaultingMap(FAULTING_MAP.headers().asStrictMap());
+        FAULTING_MAP.setPayloadFaultMap(payloadMap);
+        payloadMap.setDisableThrowingPayloadNotLoaded(false);
         final String EXPECTED = "{\n"
             + "  \"method\": \"PUT\",\n"
             + "  \"URI\": \"/_bulk\",\n"
@@ -32,6 +34,6 @@ public void testTransformsPropagateExceptionProperly() throws JsonProcessingExce
             "[{\"TypeMappingSanitizationTransformerProvider\":\"\"}]");
         var e = Assertions.assertThrows(Exception.class,
             () -> transformer.transformJson(FAULTING_MAP));
-        Assertions.assertTrue(((PayloadAccessFaultingMap)FAULTING_MAP.payload()).missingPaylaodWasAccessed());
+        Assertions.assertTrue(((PayloadAccessFaultingMap)FAULTING_MAP.payload()).missingPayloadWasAccessed());
     }
 }

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

@@ -25,6 +25,7 @@
 import org.opensearch.migrations.transform.TransformationLoader;
 import org.opensearch.migrations.utils.TrackedFuture;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCountUtil;
@@ -331,8 +332,11 @@ public void testMalformedPayload_andThrowingTransformation_IsPassedThrough() thr
             new TransformationLoader().getTransformerFactoryLoader(
                 HOST_NAME,
                 null,
-                "[{\"TypeMappingSanitizationTransformerProvider\":\"\"}]"
-            ),
+                new ObjectMapper().writeValueAsString(List.of(
+                    Map.of("JsonJinjavaTransformerProvider", Map.of(
+                        "template", "{%- throw \"intentional exception\" -%}"
+                    ))
+                ))),
             null,
             testPacketCapture,
             rootContext.getTestConnectionRequestContext(0)
@@ -362,10 +366,7 @@ public void testMalformedPayload_andThrowingTransformation_IsPassedThrough() thr
         );
         var outputAndResult = finalizationFuture.get();
         Assertions.assertInstanceOf(TransformationException.class,
-            TrackedFuture.unwindPossibleCompletionException(outputAndResult.transformationStatus.getException()),
-            "It's acceptable for now that the OpenSearch upgrade transformation can't handle non-json " +
-                "content.  If that Transform wants to handle this on its own, we'll need to use another transform " +
-                "configuration so that it throws and we can do this test.");
+            TrackedFuture.unwindPossibleCompletionException(outputAndResult.transformationStatus.getException()));
         var combinedOutputBuf = outputAndResult.transformedOutput.getResponseAsByteBuf();
         Assertions.assertTrue(combinedOutputBuf.readableBytes() == 0);
         combinedOutputBuf.release();