Add include_keys and exclude_keys options to sink #2989

daixba · 2023-07-07T07:43:54Z

Description

Add include_keys and exclude_keys options to sink

Issues Resolved

Resolve #2975

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.

dlvenable · 2023-07-07T16:45:04Z

@daixba , Thank you for this submission. I'd like to discuss some in the original issue about finding a way to solve this with multiple pipelines. That is largely what pipeline connectors are to help solve, and I'm concerned that this approach will grow such that we have two sets of processors - the normal processors and sink processors.

graytaylor0

Overall implementation looks really good. Just had a couple questions. Also the build is failing due to some missing code test coverage in data-prepper-api

graytaylor0 · 2023-07-12T22:06:27Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/SinkModel.java

+                return new ArrayList<>();
+            }
+            List<String> result = keys.stream()
+                    .map(k -> k.startsWith("/") ? k : "/" + k)


Do we have to require the json pointer at the start? I think just some_key and /some_key should both work

This is one-time work during the start to unify all the json paths for check. Customers can put both in the yaml.

graytaylor0 · 2023-07-12T22:12:17Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

+                            break;
+                        } else if (key.startsWith(keyPath)) {
+                            found = true;
+                            valueList.add("\"" + entry.getKey() + "\":" + searchAndFilter(entry.getValue(), keyPath, filterKeys, filterAction));


Is there any reason to prefer the recursive or non-recursive stack based implementation of DFS here?

Non-recursive requires an extra use of stack in memory but it's not a problem. I think both ways have the same Os, it's just recursive is more easier to read and implement, especially for array nodes. Let me know if any concerns here.

The depth of these events will probably not be deep enough to make stack overflow an issue. I think this is acceptable.

dlvenable · 2023-07-19T16:23:41Z

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

     * @throws IOException throws IOException when invalid input is received or not able to create wrapping
     */
-    void start(OutputStream outputStream) throws IOException;
+    void start(OutputStream outputStream, SinkContext sinkContext) throws IOException;


It may make sense to pass a different object here rather than SinkContext. For one, the sink really shouldn't need to know anything about routes. For another, this could vary independently of the SinkContext.

Perhaps:

interface OutputCodecOptions { List<String> getIncludeKeys(); ...

Then you can even create an internal adapter to adapt from SinkContext to OutputCodecOptions. Or just do a mapping.

Thanks David.

As the OutputCodec are loaded via plugin factory and may have its own configuration options (via pluginConfiguration), if we name it something like OutputCodecOptions, it seems to me it's some configuration options belongs to the OutputCodec, in such case, the usage will be

- sink - s3 codec: xxxx: include_keys: [...]

For in our case, the usage as below.

- sink - s3 codec: xxxx: include_keys: [...] ...

So I am not sure if OutputCodecOptions is the right way. Also, another thing I am not sure is whether or not we are going to throw more info to the SinkContext. So to me, Context seems to be more suitable than Options here.

Thoughts?

dlvenable · 2023-07-19T16:24:27Z

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

-        Map<String, Object> eventData = objectMapper.readValue(eventJsonString, new TypeReference<>() {
-        });
-        return JacksonLog.builder().withData(eventData).build();
+    default String buildJsonString(Event event, String tagsTargetKey, List<String> includeKeys, List<String> excludeKeys) throws JsonProcessingException {


I don't think we should have this method here. Only the JSON codec cares about building a JSON string. So each codec should handle this independently.

dlvenable · 2023-07-19T16:44:28Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/SinkModel.java

            super(pluginSettings);
            this.routes = routes != null ? routes : new ArrayList<>();
+            this.includeKeys = includeKeys != null ? preprocessingKeys(includeKeys) : new ArrayList<>();


nit:

Instead of new ArrayList<>() you could provide Collections.emptyList().

dlvenable · 2023-07-19T16:47:38Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/SinkModel.java

        }

-        private SinkInternalJsonModel(final List<String> routes, final String tagsTargetKey, final Map<String, Object> pluginSettings) {
+        private SinkInternalJsonModel(final List<String> routes, final String tagsTargetKey, final List<String> includeKeys, final List<String> excludeKeys, final Map<String, Object> pluginSettings) {


Can the customer provide both include and exclude? It seems that it should one or the other.

It makes no sense to have both. Although right now, we do allow customers to add both options, but in the code, if include_keys is provided, exclude_keys will be ignored. We can make this clear in the doc.

@daixba , Can we have a validation on the code as well? It seems right now the user can add both and the results may not be what they expect.

Sure, will add validation in the next PR.

dlvenable · 2023-07-19T16:50:06Z

.../main/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedOutputCodec.java

This class has been renamed. Please see #3030. You should probably rebase.

dlvenable · 2023-07-19T16:51:08Z

...main/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedOutputConfig.java

-    public List<String> getExcludeKeys() {
-        return excludeKeys;
-    }
+//    private static final List<String> DEFAULT_EXCLUDE_KEYS = new ArrayList<>();


Let's not leave commented code. Just delete it.

Signed-off-by: Aiden Dai <[email protected]>

dlvenable · 2023-07-28T14:36:18Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/SinkModel.java

        }

-        private SinkInternalJsonModel(final List<String> routes, final String tagsTargetKey, final Map<String, Object> pluginSettings) {
+        private SinkInternalJsonModel(final List<String> routes, final String tagsTargetKey, final List<String> includeKeys, final List<String> excludeKeys, final Map<String, Object> pluginSettings) {


@daixba , Can we have a validation on the code as well? It seems right now the user can add both and the results may not be what they expect.

dlvenable · 2023-07-28T14:37:16Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/SinkModel.java

@@ -46,6 +48,15 @@ public Collection<String> getRoutes() {
        return this.<SinkInternalJsonModel>getInternalJsonModel().routes;
    }

+    public List<String> getIncludeKeys() {


This can be a Collection<String> since the order doesn't matter to users. It should help add clarity when authors create new sinks.

public Collection<String> getIncludeKeys()

As matter of fact, the keys are sorted in the code, we do care about order, this is to improve performance in the jsonbuilder.

dlvenable · 2023-07-28T14:37:36Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/SinkModel.java

+        return this.<SinkInternalJsonModel>getInternalJsonModel().includeKeys;
+    }
+
+    public List<String> getExcludeKeys() {


Same as the comment above:

public Collection<String> getExcludeKeys()

Same as above.

dlvenable · 2023-07-28T14:38:08Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/Event.java

@@ -135,9 +143,15 @@ public interface Event extends Serializable {

    JsonStringBuilder jsonBuilder();

-    public abstract class JsonStringBuilder {
+    abstract class JsonStringBuilder {


Thanks for making this improvement - inner classes on interfaces are automatically public, so no keyword is necessary.

dlvenable · 2023-07-28T14:40:54Z

.../src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/DocumentBuilder.java

+                .toJsonString();
+
+        if (document == null || !document.startsWith("{")) {
+            return String.format("{\"data\": %s}", document);


You didn't introduce this, but String.format is slow. String concatenation would be faster.

Feel free to update that, but since you didn't introduce it, no need to make the change to accept the PR.

dlvenable · 2023-07-28T14:41:45Z

...src/test/java/org/opensearch/dataprepper/model/configuration/PipelinesDataFlowModelTest.java

@@ -50,7 +50,7 @@ void testSerializing_PipelinesDataFlowModel_empty_Plugins_with_nonEmpty_delay_an

        final PluginModel source = new PluginModel("testSource", (Map<String, Object>) null);
        final List<PluginModel> processors = Collections.singletonList(new PluginModel("testProcessor", (Map<String, Object>) null));
-        final List<SinkModel> sinks = Collections.singletonList(new SinkModel("testSink", Collections.emptyList(), null, null));
+        final List<SinkModel> sinks = Collections.singletonList(new SinkModel("testSink", Collections.emptyList(), null, Collections.emptyList(), Collections.emptyList(), null));


You should be able to update these tests to Collections.emptyCollection() after making the change I suggested about the collection.

As above, it's a list.

dlvenable · 2023-07-28T14:43:34Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

+                            break;
+                        } else if (key.startsWith(keyPath)) {
+                            found = true;
+                            valueList.add("\"" + entry.getKey() + "\":" + searchAndFilter(entry.getValue(), keyPath, filterKeys, filterAction));


The depth of these events will probably not be deep enough to make stack overflow an issue. I think this is acceptable.

dlvenable · 2023-07-28T14:46:53Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

+                            found = true;
+                            // To keep the order.
+                            if (filterAction == RETAIN_ALL) {
+                                valueList.add("\"" + entry.getKey() + "\":" + entry.getValue().toString());


Are we sure that entry.gevValue().toString() provides the right value here? What if the string has double quotes in it? Are they being escaped? Any other special characters?

Yes, entry.getValue().toString() is the right value here. All the special characters are already taken care of by this. The output could be a number (e.g. 10), a character with double quotes (e.g. "ye\"llow"), or even the full json tree (e.g. ["msg 1","msg 2","msg 3"], {"name":"hello","age":"world"} etc.)

dlvenable

Improvements can come in follow-on PR.

daixba requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, cmanning09, asifsmohammed, dlvenable and oeyh as code owners July 7, 2023 07:43

graytaylor0 reviewed Jul 12, 2023

View reviewed changes

daixba force-pushed the sink-with-keys branch from 61919d6 to 6d472b5 Compare July 17, 2023 06:11

dlvenable requested changes Jul 19, 2023

View reviewed changes

daixba force-pushed the sink-with-keys branch from e5e2b35 to cd9645b Compare July 21, 2023 01:45

daixba force-pushed the sink-with-keys branch from cd9645b to 4a7d847 Compare July 28, 2023 02:30

Add include_keys and exclude_keys to sink

994edd2

Signed-off-by: Aiden Dai <[email protected]>

daixba force-pushed the sink-with-keys branch from 4a7d847 to 994edd2 Compare July 28, 2023 03:01

dlvenable requested changes Jul 28, 2023

View reviewed changes

dlvenable approved these changes Jul 28, 2023

View reviewed changes

kkondaka approved these changes Jul 28, 2023

View reviewed changes

kkondaka merged commit b519a82 into opensearch-project:main Jul 28, 2023
44 checks passed

This was referenced Jul 31, 2023

PR feedback improvements from #2989 #3080

Closed

Add include_keys and exclude_keys to S3 sink. #3102

Closed

dlvenable mentioned this pull request Aug 14, 2023

Add include_keys and exclude_keys options under the sink #2975

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add include_keys and exclude_keys options to sink #2989

Add include_keys and exclude_keys options to sink #2989

daixba commented Jul 7, 2023

dlvenable commented Jul 7, 2023

graytaylor0 left a comment •

edited

Loading

graytaylor0 Jul 12, 2023

daixba Jul 13, 2023

graytaylor0 Jul 12, 2023

daixba Jul 13, 2023

dlvenable Jul 28, 2023

dlvenable Jul 19, 2023

daixba Jul 20, 2023

dlvenable Jul 19, 2023

dlvenable Jul 19, 2023

dlvenable Jul 19, 2023

daixba Jul 20, 2023

dlvenable Jul 28, 2023

daixba Aug 1, 2023

dlvenable Jul 19, 2023

dlvenable Jul 19, 2023

dlvenable Jul 28, 2023

dlvenable Jul 28, 2023

daixba Aug 1, 2023 •

edited

Loading

dlvenable Jul 28, 2023

daixba Aug 1, 2023

dlvenable Jul 28, 2023

dlvenable Jul 28, 2023

dlvenable Jul 28, 2023

daixba Aug 1, 2023

dlvenable Jul 28, 2023

dlvenable Jul 28, 2023

daixba Aug 1, 2023 •

edited

Loading

dlvenable left a comment

Add include_keys and exclude_keys options to sink #2989

Add include_keys and exclude_keys options to sink #2989

Conversation

daixba commented Jul 7, 2023

Description

Issues Resolved

Check List

dlvenable commented Jul 7, 2023

graytaylor0 left a comment • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

daixba Aug 1, 2023 • 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

daixba Aug 1, 2023 • edited Loading

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

graytaylor0 left a comment •

edited

Loading

daixba Aug 1, 2023 •

edited

Loading

daixba Aug 1, 2023 •

edited

Loading