Json codec changes with specific json input codec config #5054

sb2k16 · 2024-10-11T22:50:16Z

Description

This PR is to add support for additional configuration entries in the JsonInputCodec so that CloudWatch logs from subscription filters could be decoded in Data Prepper. CloudWatch Subscription Filter

For the input message from CloudWatch

{
    "owner": "111111111111",
    "logGroup": "CloudTrail/logs",
    "logStream": "111111111111_CloudTrail/logs_us-east-1",
    "subscriptionFilters": [
        "Destination"
    ],
    "messageType": "DATA_MESSAGE",
    "logEvents": [
        {
            "id": "31953106606966983378809025079804211143289615424298221568",
            "timestamp": 1432826855000,
            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
        },
        {
            "id": "31953106606966983378809025079804211143289615424298221569",
            "timestamp": 1432826855000,
            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
        },
        {
            "id": "31953106606966983378809025079804211143289615424298221570",
            "timestamp": 1432826855000,
            "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}"
        }
    ]
}

and with the following configuration:

      codec:
        json:
          key_name: "logEvents"
          include_keys: ['owner', 'logGroup', 'logStream' ]
          include_keys_metadata: ['owner', 'logGroup', 'logStream' ]

the output event should be split into 3 separate events with additional data and metadata as specified in the config

{
  "id": "31953106606966983378809025079804211143289615424298221568",
  "timestamp": 1432826855000,
  "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}",
  "owner": "111111111111",
  "logGroup": "CloudTrail/logs",
  "logStream": "111111111111_CloudTrail/logs_us-east-1",
}

{
  "id": "31953106606966983378809025079804211143289615424298221569",
  "timestamp": 1432826855000,
  "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}",
  "owner": "111111111111",
  "logGroup": "CloudTrail/logs",
  "logStream": "111111111111_CloudTrail/logs_us-east-1",
}

{
  "id": "31953106606966983378809025079804211143289615424298221570",
  "timestamp": 1432826855000,
  "message": "{\"eventVersion\":\"1.03\",\"userIdentity\":{\"type\":\"Root\"}",
  "owner": "111111111111",
  "logGroup": "CloudTrail/logs",
  "logStream": "111111111111_CloudTrail/logs_us-east-1",
}

Issues Resolved

Resolves #5045

Check List

New functionality includes testing.
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Souvik Bose <souvbose@amazon.com>

dlvenable · 2024-10-11T23:20:23Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

+    private List<String> includeKeys;
+    private List<String> includeKeysMetadata;
+
+    public JsonDecoder(String keyName, List<String> includeKeys, List<String> includeKeysMetadata) {


Suggested change

public JsonDecoder(String keyName, List<String> includeKeys, List<String> includeKeysMetadata) {

public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<String> includeKeysMetadata) {

dlvenable · 2024-10-11T23:20:35Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Consumer;

 public class JsonDecoder implements ByteDecoder {
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final JsonFactory jsonFactory = new JsonFactory();
+    private String keyName;
+    private List<String> includeKeys;


Suggested change

private List<String> includeKeys;

private Collection<String> includeKeys;

dlvenable · 2024-10-11T23:20:43Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Consumer;

 public class JsonDecoder implements ByteDecoder {
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final JsonFactory jsonFactory = new JsonFactory();
+    private String keyName;
+    private List<String> includeKeys;
+    private List<String> includeKeysMetadata;


Suggested change

private List<String> includeKeysMetadata;

private Collection<String> includeKeysMetadata;

dlvenable · 2024-10-11T23:22:35Z

...or/src/test/java/org/opensearch/dataprepper/plugins/codec/json/JsonInputCodecConfigTest.java

+    @Test
+    public void testJsonInputCodecConfig() {
+        JsonInputCodecConfig jsonInputCodecConfig = createObjectUnderTest();
+        assertTrue(jsonInputCodecConfig.getKeyName().equals(JsonInputCodecConfig.DEFAULT_KEY_NAME));


Suggested change

assertTrue(jsonInputCodecConfig.getKeyName().equals(JsonInputCodecConfig.DEFAULT_KEY_NAME));

assertNulljsonInputCodecConfig.getKeyName());

Is this not the default behavior?

Thanks @dlvenable. This is the default behavior. I will make the change.

dlvenable · 2024-10-11T23:25:13Z

...rocessor/src/test/java/org/opensearch/dataprepper/plugins/codec/json/JsonInputCodecTest.java

@@ -215,6 +220,38 @@ void parse_with_InputStream_calls_Consumer_with_Event(final int numberOfObjects)
        }
    }

+    @ParameterizedTest
+    @ValueSource(ints = {1, 2, 10, 100})
+    void parse_with_InputStream_calls_Consumer_with_EventConfig(final int numberOfObjects) throws IOException {


I think ideally, each of these conditions is tested independently of each other. I see three things: 1. Using a key_name, 2. Keys as data, 3. Keys as metadata.

Thanks @dlvenable. I have added tests to cover the scenarios for 3 configs separately.

dlvenable · 2024-10-11T23:26:46Z

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/codec/JsonDecoderTest.java

+    @Nested
+    class JsonDecoderWithInputConfig {
+        private ObjectMapper objectMapper;
+        final List<String> include_keys = new ArrayList<>();


Rename to includeKeys.

Also, set this list in the @BeforeEach. Otherwise, tests may interfere with each other.

dlvenable · 2024-10-11T23:27:57Z

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/codec/JsonDecoderTest.java

+                receivedTime = ((Event)record.getData()).getEventHandle().getInternalOriginationTime();
+            });
+
+            records.forEach(record -> {


Assert the size of records first or this will miss an empty list.

dlvenable · 2024-10-11T23:28:19Z

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/codec/JsonDecoderTest.java

+                Map<String, Object> dataMap = record.getData().toMap();
+                Map<String, Object> metadataMap = record.getData().getMetadata().getAttributes();
+
+                for (String include_key: include_keys) {


Assert the size of include_keys first or you may miss an empty list. Also, rename to includeKeys.

dlvenable · 2024-10-11T23:28:57Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

+    private void parseRecordsArray(final JsonParser jsonParser,
+                                   final Instant timeReceived,
+                                   final Consumer<Record<Event>> eventConsumer,
+                                   Map<String, Object> includeKeysMap,


Suggested change

Map<String, Object> includeKeysMap,

final Map<String, Object> includeKeysMap,

dlvenable · 2024-10-11T23:29:05Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

+                                   final Instant timeReceived,
+                                   final Consumer<Record<Event>> eventConsumer,
+                                   Map<String, Object> includeKeysMap,
+                                   Map<String, Object> includeMetadataKeysMap


Suggested change

Map<String, Object> includeMetadataKeysMap

final Map<String, Object> includeMetadataKeysMap

Signed-off-by: Souvik Bose <souvbose@amazon.com>

kkondaka · 2024-10-15T02:10:09Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

            if (jsonParser.getCurrentToken() == JsonToken.START_ARRAY) {
-                parseRecordsArray(jsonParser, timeReceived, eventConsumer);
+                if (keyName != null && !nodeName.equals(keyName)) {


This only allows for first level of fields? Should we make it generic to allow nested fields, like

{ "key1" : "value1", "key2" : { "logEvents" : [ ] } }

* Json codec changes with specific json input codec config Signed-off-by: Souvik Bose <souvbose@amazon.com> * Modify the tests and address comments Signed-off-by: Souvik Bose <souvbose@amazon.com> --------- Signed-off-by: Souvik Bose <souvbose@amazon.com> Co-authored-by: Souvik Bose <souvbose@amazon.com> (cherry picked from commit 89bc937)

* Json codec changes with specific json input codec config Signed-off-by: Souvik Bose <souvbose@amazon.com> * Modify the tests and address comments Signed-off-by: Souvik Bose <souvbose@amazon.com> --------- Signed-off-by: Souvik Bose <souvbose@amazon.com> Co-authored-by: Souvik Bose <souvbose@amazon.com> (cherry picked from commit 89bc937) Co-authored-by: Souvik Bose <souvik04in@gmail.com>

…project#5054) * Json codec changes with specific json input codec config Signed-off-by: Souvik Bose <souvbose@amazon.com> * Modify the tests and address comments Signed-off-by: Souvik Bose <souvbose@amazon.com> --------- Signed-off-by: Souvik Bose <souvbose@amazon.com> Co-authored-by: Souvik Bose <souvbose@amazon.com>

oeyh · 2024-10-28T17:12:28Z

@sb2k16 Can we update the documentation website with this change?

Json codec changes with specific json input codec config

333b1dc

Signed-off-by: Souvik Bose <souvbose@amazon.com>

sb2k16 requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, KarstenSchnitter, dlvenable and oeyh as code owners October 11, 2024 22:50

dlvenable requested changes Oct 11, 2024

View reviewed changes

dlvenable added the backport 2.10 label Oct 11, 2024

Modify the tests and address comments

dc7f0f0

Signed-off-by: Souvik Bose <souvbose@amazon.com>

dlvenable approved these changes Oct 14, 2024

View reviewed changes

kkondaka reviewed Oct 15, 2024

View reviewed changes

kkondaka approved these changes Oct 15, 2024

View reviewed changes

kkondaka merged commit 89bc937 into opensearch-project:main Oct 15, 2024
46 of 47 checks passed

opensearch-trigger-bot bot mentioned this pull request Oct 15, 2024

[Backport 2.10] Json codec changes with specific json input codec config #5066

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Json codec changes with specific json input codec config #5054

Json codec changes with specific json input codec config #5054

sb2k16 commented Oct 11, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

sb2k16 Oct 12, 2024

dlvenable Oct 11, 2024

sb2k16 Oct 12, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

dlvenable Oct 11, 2024

kkondaka Oct 15, 2024

oeyh commented Oct 28, 2024

	public JsonDecoder(String keyName, List<String> includeKeys, List<String> includeKeysMetadata) {
	public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<String> includeKeysMetadata) {

	private List<String> includeKeys;
	private Collection<String> includeKeys;

	private List<String> includeKeysMetadata;
	private Collection<String> includeKeysMetadata;

	assertTrue(jsonInputCodecConfig.getKeyName().equals(JsonInputCodecConfig.DEFAULT_KEY_NAME));
	assertNulljsonInputCodecConfig.getKeyName());

	Map<String, Object> includeKeysMap,
	final Map<String, Object> includeKeysMap,

	Map<String, Object> includeMetadataKeysMap
	final Map<String, Object> includeMetadataKeysMap

Json codec changes with specific json input codec config #5054

Json codec changes with specific json input codec config #5054

Conversation

sb2k16 commented Oct 11, 2024

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oeyh commented Oct 28, 2024