Added the router to the Data Prepper pipeline model. #1666

dlvenable · 2022-08-15T13:17:41Z

Description

Adds router to the pipeline model to support conditional routing.

Some additional code and tests had to change to support the new model design.

Issues Resolved

Resolves #1665

Check List

New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
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: David Venable <[email protected]>

cmanning09 · 2022-08-15T15:37:31Z

data-prepper-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineModel.java

        this.processors = processors;
+        this.router = router != null ? router : new ArrayList<>();


processors are optional and we do not perform this check. Are we building these objects in a different way?

I'd probably want to make processors return an empty list too.

Right now, this class has two constructors. One used by Jackson and the other used by other code in Data Prepper. I'd like to replace the second constructor with a builder pattern. It would be too much for this PR though. In that scenario, we could always use empty arrays instead of null to make development easier.

I created #1669 to track these two changes. I propose there that processors should return an empty list.

cmanning09 · 2022-08-15T15:41:30Z

...-prepper-api/src/test/java/com/amazon/dataprepper/model/configuration/PipelineModelTest.java

+        assertThat(pipelineModel.getRouter(), notNullValue());
+        assertThat(pipelineModel.getRouter(), notNullValue());


Duplicate assertions

cmanning09 · 2022-08-15T15:44:38Z

data-prepper-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineModel.java

@@ -64,6 +65,9 @@ else if (preppers != null) {
    @JsonInclude(JsonInclude.Include.NON_NULL)
    private final PluginModel buffer;

+    @JsonProperty("router")
+    private List<PipelineConditionalRoute> router;


Is a router a list of PipelineConditionalRoutes?

From a users perspective, am I creating routes or a router when I put a route in my pipeline? Should the JsonProperty be routes?

The RFC calls for:

router: - myRouteA: '/some/conditional' - myRouteA: '/some/conditional'

I tried to use Jackson to create a Router class as an abstraction of the whole router concept. It could then have routes (which are not exposed in the YAML model, only the Java model) But, I hit quite a few limitations with Jackson that prevented it from working the way I would have liked.

There is a bit of discussion in that RFC, but overall the concept of a router seemed most appropriate. We could rename it to routes. We could also consider making a model like the following:

router: routes: - myRouteA: '/some/conditional' - myRouteB: '/some/conditional'

This last approach actually lines up with what I was trying to do with the Java model. I think that extra nesting is not valuable to pipeline authors.

cmanning09 · 2022-08-15T15:45:46Z

...r-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRoute.java

+ */
+@JsonSerialize(using = PipelineConditionalRoute.PipelineConditionalRouteSerializer.class)
+@JsonDeserialize(using = PipelineConditionalRoute.PipelineConditionalRouteDeserializer.class)
+public class PipelineConditionalRoute {


PipelineCondiationalRoute seems overly verbose. Are we expecting to have routes which are not part of the pipeliens or conditional based? Can we simplify the naming here?

If we are or do plan on extending (to support non-conditional routes). We should define an interface and have this class be the first implementation.

The java docs does not mention anything about being a conditional route either.

I'm not expecting non-conditional routes. But, at some point in the code we will have routes applied to events. I was hoping to distinguish between these routes here and then routes which are added to events at runtime.

I'll try to think about a better name here.

I renamed it to ConditionalRoute. I think this may help clear it up from a "route" which has already been applied to an event at runtime.

cmanning09 · 2022-08-15T15:59:15Z

...-prepper-api/src/test/java/com/amazon/dataprepper/model/configuration/PipelineModelTest.java

        final List<PluginModel> expectedPreppersPluginModel = validPreppersPluginModel();
        final PipelineModel pipelineModel = new PipelineModel(
                validSourcePluginModel(),
                null,
                expectedPreppersPluginModel,
                null,
+                null,


I am curious as to why you elected not to include a router here? This test should be eliminated when prepper is deprecated. I want to make sure we are capturing this test long term if it's required.

We plan to remove prepper as you noted, so I'm not too worried about this test. But, we do need to cover the scenario for now (or test coverage is too low).

graytaylor0 · 2022-08-15T16:15:01Z

data-prepper-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineModel.java

@@ -64,6 +65,9 @@ else if (preppers != null) {
    @JsonInclude(JsonInclude.Include.NON_NULL)
    private final PluginModel buffer;

+    @JsonProperty("router")


Should this have the @JsonInclude(JsonInclude.Include.NON_NULL annotation? It is an optional configuration similar to buffer and processor

I updated this to use NON_EMPTY. This will not serialize an empty list. This makes null and empty list semantically the same and allows Java clients to not have to worry about null.

I create #1669 to make some related changes. In that, I will change processor to also have the same behavior. This should make it easier to code against these Java models (no null lists).

dinujoh · 2022-08-15T17:59:54Z

...r-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRoute.java

+
+        @Override
+        public PipelineConditionalRoute deserialize(final JsonParser parser, final DeserializationContext context) throws IOException, JacksonException {
+            final JsonNode node = parser.getCodec().readTree(parser);


Should the input to readTree be some serialized data ?

dinujoh · 2022-08-15T18:00:16Z

...r-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRoute.java

+        }
+
+        @Override
+        public PipelineConditionalRoute deserialize(final JsonParser parser, final DeserializationContext context) throws IOException, JacksonException {


how is context used here ?

dinujoh · 2022-08-15T18:01:44Z

...r-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRoute.java

+@JsonDeserialize(using = PipelineConditionalRoute.PipelineConditionalRouteDeserializer.class)
+public class PipelineConditionalRoute {
+    private final String name;
+    private final String condition;


Should we think about an interface for condition ? This will help us do some validation over condition.

I thought about this, but I really think it should be a larger change that covers any Data Prepper condition (or even expression). It thus affects the data-prepper-expression package and would be a fairly significant change.

dinujoh · 2022-08-15T18:03:52Z

...r-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRoute.java

+        }
+
+        @Override
+        public void serialize(final PipelineConditionalRoute value, final JsonGenerator gen, final SerializerProvider provider) throws IOException {


Do we plan to use the serializer provider ?

No, but this is an overridden method from a Jackson interface.

dinujoh · 2022-08-15T18:04:31Z

...r-api/src/main/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRoute.java

+            final JsonNode node = parser.getCodec().readTree(parser);
+
+            final Iterator<Map.Entry<String, JsonNode>> fields = node.fields();
+            final Map.Entry<String, JsonNode> onlyField = fields.next();


Do we need to do validation for single value entry ?

dinujoh · 2022-08-15T18:10:51Z

...i/src/test/java/com/amazon/dataprepper/model/configuration/PipelineConditionalRouteTest.java

+        final StringBuilder stringBuilder = new StringBuilder();
+        try (final Reader reader = new BufferedReader(new InputStreamReader(inputStream, Charset.forName(StandardCharsets.UTF_8.name())))) {
+            int counter = 0;
+            while ((counter = reader.read()) != -1) {
+                stringBuilder.append((char) counter);
+            }
+        }
+        return stringBuilder.toString();
+    }


Java 9:

new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);

…alize router as non-empty, negative test cases. Signed-off-by: David Venable <[email protected]>

…ect#1666) Added the router to the Data Prepper pipeline model. Signed-off-by: David Venable <[email protected]>

Added the router to the Data Prepper pipeline model.

26c5603

Signed-off-by: David Venable <[email protected]>

dlvenable requested a review from a team as a code owner August 15, 2022 13:17

cmanning09 reviewed Aug 15, 2022

View reviewed changes

graytaylor0 reviewed Aug 15, 2022

View reviewed changes

dinujoh reviewed Aug 15, 2022

View reviewed changes

Responded to PR feedback with: Updated name to ConditionalRoute, seri…

5059326

…alize router as non-empty, negative test cases. Signed-off-by: David Venable <[email protected]>

dinujoh approved these changes Aug 18, 2022

View reviewed changes

graytaylor0 approved these changes Aug 18, 2022

View reviewed changes

dlvenable merged commit fabbb17 into opensearch-project:main Aug 18, 2022

engechas pushed a commit to engechas/data-prepper that referenced this pull request Sep 12, 2022

Added the router to the Data Prepper pipeline model. (opensearch-proj…

d0ebc3c

…ect#1666) Added the router to the Data Prepper pipeline model. Signed-off-by: David Venable <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added the router to the Data Prepper pipeline model. #1666

Added the router to the Data Prepper pipeline model. #1666

dlvenable commented Aug 15, 2022

cmanning09 Aug 15, 2022

dlvenable Aug 15, 2022

dlvenable Aug 16, 2022

cmanning09 Aug 15, 2022

cmanning09 Aug 15, 2022

dlvenable Aug 15, 2022 •

edited

Loading

cmanning09 Aug 15, 2022

dlvenable Aug 15, 2022

dlvenable Aug 18, 2022

cmanning09 Aug 15, 2022

dlvenable Aug 15, 2022

graytaylor0 Aug 15, 2022

dlvenable Aug 16, 2022

dinujoh Aug 15, 2022

dinujoh Aug 15, 2022

dinujoh Aug 15, 2022

dlvenable Aug 15, 2022

dinujoh Aug 15, 2022

dlvenable Aug 15, 2022

dinujoh Aug 15, 2022

dinujoh Aug 15, 2022

		this.processors = processors;
		this.router = router != null ? router : new ArrayList<>();

		assertThat(pipelineModel.getRouter(), notNullValue());
		assertThat(pipelineModel.getRouter(), notNullValue());

Added the router to the Data Prepper pipeline model. #1666

Added the router to the Data Prepper pipeline model. #1666

Conversation

dlvenable commented Aug 15, 2022

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

dlvenable Aug 15, 2022 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable Aug 15, 2022 •

edited

Loading