[RFC] Internal Model Proposal

[cmanning09]

What kind of business use case are you trying to solve? What are your requirements?

Existing preppers consume and emit serialized JSON strings. This wastes CPU cycles when chaining preppers due to excessive de/serialization. Users of the DP have encountered runtime exceptions due to conflicting data requirements of their preppers. Model definitions are duplicated throughout prepper plugins. (e.g. otel-trace-group-prepper TraceGroup and otel-trace-raw-prepper)

Requirements:

1. Extendable - the model should scale beyond the existing trace analytics support.
2. Type Safety - the model should be prescriptive enough to enable type safety checks throughout the pipelines in the future.
3. Eliminate need to duplicate code between plugins
4. Allow preppers to operate on internal data in a generic way
5. Remove excessing serialization

What is the problem? What is preventing you from meeting the requirements?

Currently, data flows through Data Prepper as a Collection<Records>. Records are generic types that allow for any type to flow through. Trace events have been been defined as a Collection of Records as type String. The strings are serialized representations of JSON objects conforming to the OTEL Spec.

What are you proposing? What do you suggest we do to solve the problem or improve the existing situation?

We will deprecate Records and define explicit object models for Traces and Logs. Traces and Logs will implement a new interface called Event. Events will be the new data type flowing through Data Prepper.

Source plugins will be responsible for translating the external requests into Events. Sink plugins will be responsible for transforming the Events into the correct output schema. Preppers will only accept Events as inputs and outputs or subtypes. This will effectively create internal boundaries for our model between sources and sinks.

Event

Events will be managed through public putting, deleting and fetching methods. An additional method for generating a JSON String is included to support the sinks.

/**
* Add or update the key with a the given value in the Event
*/
void put(String key, Object value);

/**
* Retrieves the given key from the Event
*/
<T> T get(String key, Class<T> type);

/**
* Deletes the given key from the Event
*/
void delete(String key);

/**
* Generates a Json representation of the Event
*/
JsonNode toJsonNode();

/**
* Get Metadata
*/
Metadata getMetadata();

EventMetadata

Will be a class with a slight refactor of the RecordMetadata Class. Currently, RecordMetadata maintains a map of attributes and has one required recordType attribute. However, the recordType has been historically ignored. The new model will have required attributes recategorized as POJO fields. The eventType will help preserve the type (i.e log, span), for casting and type validation. The EventMetadata class will still maintain a mapping for custom metadata defined in attributes.

public class EventMetadata {

private String eventType;
private long timeReceived;
private Map<String, Object> attributes;

Span

Span will be a new model to support the traces. It will implement the Event interface and maintain the same attributes as the current RawSpan Object. This will ensure backwards compatibility with our existing preppers.

Phased Approach

This design includes breaking changes and will be broken into two phases. One phase allows us to build support for the new model and onboard log ingestion and trace analytics. The second phase will deprecate the old model and will be a part of the 2.0 release

What are your assumptions or prerequisites?

The design and changes to the pipelines to enforce type safety are out of scope and should addressed in a separate review. However, the output of this design should not hinder but enable type safety enforcement.

This aligns with the proposal for Log Ingestion RFC

What are remaining open questions?

• Which library should we use to support the underlying interfaces? (JsonPath or Jackson) JsonPath is a library for reading and updating JSON documents. It natively supports the dot notation. Jackson is fast JSON library for parsing json objects and supports Json Pointers for managing objects. Both libraries will work. Jackson will be ideal to reduce dependencies.

[cmanning09]

I built a small POC locally to demonstrate using Jackson over JsonPath. This will reduce our dependencies while supporting the functionality we need. Here is the example code:

public class Events {

    private JsonNode jsonNode;
    private ObjectMapper mapper;

    // Constructors ...

    public <T> T get(String key, Class<T> type) {
        JsonNode node = jsonNode.at(key);
        if (node == null) {
            return null;
        }
        try {
            return mapper.treeToValue(node, type);
        } catch (JsonProcessingException e) {
            return null;
        }
    }

    public void put(String key, Object value) {

        LinkedList<String> keys = new LinkedList<>(Arrays.asList(key.split("\\.")));
        putKeys(jsonNode, keys, value);
    }

    private void putKeys(JsonNode node, LinkedList<String> keys, Object value) {
        if (keys.size() == 1) {
            JsonNode valueNode = mapper.valueToTree(value);
            ((ObjectNode) node).set(keys.get(0), valueNode);
        } else {
            JsonNode newNode = node.get(keys.get(0));
            if (newNode == null) {
                newNode = mapper.createObjectNode();
                ((ObjectNode) node).set(keys.get(0), newNode);
            }
            keys.remove(0);
            putKeys(newNode,  keys, value);
        }
    }

I don't foresee any issues following this appraoch at this time.

[cmanning09]

Steps for implementation:

• Build Interfaces
• Build implementations for Events
• Build Log and Trace implementations
• Extend Prepper, Sink, etc interfaces to support new model
• Begin plugin implementation with new model...

[cmanning09]

Task Status:

• Build Interfaces New Internal Model Interfaces #405
• Build implementations for Events Implement metadata & event model #434
• Build Log and Trace implementations Implement Log & Trace Model #436
• Integration Log Plugins with new model Log Ingestion Plugin Event Model Integration #538
• Integrat Trace Plugins with new model
• Deprecate Record and remove it from DP

[cmanning09]

Re-openning to have a migration discussion.

I have begun the process of extending the prepper, sink, source and buffer interfaces. This is proving to be more invasive than I originally anticipated. While extending the interfaces is easy, swapping which interface functionality to leverage internally is proving to be more invasive and requiring a significant amount of work. Supporting both model flows will require an architecture / design which supports both model flows.

At this moment I can think of a few proposals to still meet our objectives:

1. Continue on the path of interface extension. Introduce a new configuration option (TBD on the name) for sources, buffer and the ProcessWorker to toggle which method to call which can toggle which interface methods and models need to be leveraged to read and write from a buffer. The configuration option will control the type of buffer as well. Default value will be Record. We would need enforcement for each plugin in the pipeline or provide a mechanism to provide this globally (However plugins are instantiated through their own separate configs). This toggle will be deprecated in 2.0. Additionally, each interface and abstract implementation is highly coupled to the type Records in its class interface signature and extending the interface is turning into a larger code smell. Ultimately continuing down this path will introduce more tech debt, more work and poorly typed interfaces which use a different model.

2. Avoiding extension of interfaces DP, I would like to propose we leverage the existing interfaces and implementations and current architecture. Instead of using Strings as the type for Records the plugins will be updated to support Events as the type of Records. This simplifies the implementation and leverages the existing flexibility of the original design. It eliminates the need to support two separate model flows. It aligns with our existing documentation, metrics, etc while still eliminating the original bottleneck of excessive serialization. Downsides are we have a construct of Record. There is some overlap in functionality between these two models that will need to be addressed.

I would like to propose option 2. Ultimately, I believe this is the cleanest and most elegant extension of our existing code base that achieves our original requirements.

I would greatly appreciate feedback from others (@dlvenable, @graytaylor0, @chenqi0805).

[cmanning09]

Moving forward with option 2 to leverage Record<Event> for 1.2 Log ingestion Plugins. Trace plugins will be updated to support Record<Event> in 1.3. Records will still be deprecated in 2.0.