Support for Event Json input and output codecs

[kkondaka]

Description

Support for event_json input and output codecs

Issues Resolved

Resolves #4404

Check List

• [X ] New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• [X ] 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.

[graytaylor0]

You could probably have Jackson handle these conversions by annotating the DefaultEventMetadata properties and deserializing the metadata to the DefaultEventMetadata class

[dlvenable]

Is the time an Instant in production use cases?

[graytaylor0]

We should make this configurable and default to not passing anything here so that the new time the Event reaches the source is applied to the Event.

[kkondaka]

event time received should not be optional, I think. Tags, attributes and externalorigination time are optional.

[graytaylor0]

The default behavior when using this codec should be to create a new time received time, which is the default behavior when an Event is created.

[graytaylor0]

There's still some missing test coverage in data-prepper-api

[dinujoh]

Do we need this ?

[dlvenable]

The Java convention for packages with spaces is to use underscore. So you could rename to:

package org.opensearch.dataprepper.plugins.codec.event_json;

[dlvenable]

There is no guarantee that this consumer is even called.

A simple change is to update the consumer to put the records into a collection. Then assert the size. Then assert the format.

List<Record> records = new LinkedList<>();
inputCodec.parse(new ByteArrayInputStream(outputStream.toByteArray()), records::add);

assertThat(records.size(), equalTo(1));
for(Record record : records) {
  Event e = record.getData();
  // All the same from here.
}

[dlvenable]

It appears that you are starting an object. But, there is no array. So won't it look like this?

{{"data":..., "metadata"...},{"data":..., "metadata"...},{"data":..., "metadata"...}}

[kkondaka]

Good point. Modified to generate an array.

[dlvenable]

Similar comment as below about the assertions here.

[dlvenable]

We should have a test that operates against multiple events.

[dlvenable]

I like that this test is asserting symmetry between the codecs.

We should also have test suites for the individual codecs with known values to ensure that the format is consistent. That is, somebody could make a change which causes the input codec to no longer work with the same values. The current tests would pass, but we would have broken backward compatibility which could pose problems when using this to write to S3.

[dlvenable]

Do we need this?

[kkondaka]

Just keeping it for future use where we may want to control what fields should be included in the output.

[dlvenable]

Make these different tests for isolation:

@ParameterizedTest
@ValueSource(strings = { "[]", "[{}]", "{}" })
...

[dlvenable]

We should do something to assert this.

Probably:

Consumer<Record<Event>> consumer = mock(Consumer.class);
inputCodec.parse(inputStream, consumer);
verifyNoInteractions(consumer);

[dlvenable]

We should keep the writeStartObject.

This could allow us to also include metadata such as the version.

e.g.

{
  "metadata" : {"version" : "2.8"},
  "events" : [
    {"data":...,"metadata"},
    {"data":...,"metadata"},
    {"data":...,"metadata"},
  ]
}

See this code:

data-prepper/data-prepper-plugins/parse-json-processor/src/main/java/org/opensearch/dataprepper/plugins/codec/json/JsonOutputCodec.java

Lines 55 to 57 in f9a3a60

	generator.writeStartObject();
	generator.writeFieldName(config.getKeyName());
	generator.writeStartArray();

Though, I don't think we need to make the events key configurable.

[kkondaka]

Good point. But the following should be enough.

{
  "version" : "2.8",
  "events" : [
    {"data":...,"metadata"},
    {"data":...,"metadata"},
    {"data":...,"metadata"},
  ]
}

[graytaylor0]

I think you want to run this part if override_time_received is false

[kkondaka]

You mean, the overrideTimeReceived flag "true" means it will override with the value from the data and "false" means it will take Instant.now()?

[graytaylor0]

Let's move this to setup and then add a test for when the override_time_received is true

[kkondaka]

Modified as per the suggestion

[dlvenable]

Do we need strict matching here? One thought is that we could be strict on the major version and ignore the minor version.

[dlvenable]

It may be good to move this check into a small method to help make the code more readable.

[dlvenable]

The DataPrepperVersion class provides a lot of this.

final String versionStr = (String)json.get(EventJsonDefines.VERSION);
final DataPrepperVersion definedVersion = DataPrepperVersion.parse(versionStr);
if(definedVersion.getMajorVersion() != DataPrepperVersion.getCurrentVersion().getMajorVersion()) {
  LOG.error("Version mismatch! Current version {} Received data version {}", DataPrepperVersion.getCurrentVersion(), definedVersion);
            return false;
}

[dlvenable]

Thanks for adding this and making these improvements to the feature!