Skip to content

Commit 41773e6

Browse files
kkondakakkondaka
authored
Add support for pipeline DLQ to opensearch sink (#6137)
* Add support for pipeline DLQ to opensearch sink Signed-off-by: Kondaka <krishkdk@amazon.com> * Added _failure_metadata field to event before sending pipeline DLQ Signed-off-by: Kondaka <krishkdk@amazon.com> * Fix failing tests Signed-off-by: Kondaka <krishkdk@amazon.com> * Fix failing tests Signed-off-by: Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Kondaka <krishkdk@amazon.com> * Modified to reuse event Signed-off-by: Kondaka <krishkdk@amazon.com> * Fix checkstyle errors Signed-off-by: Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Kondaka <krishkdk@amazon.com>
1 parent d252490 commit 41773e6

11 files changed

Lines changed: 301 additions & 30 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -274,6 +274,8 @@ default void putIfAbsent(final String key, final Class clazz, final Object value
274274

275275
JsonStringBuilder jsonBuilder();
276276

277+
EventFailureMetadata updateFailureMetadata();
278+
277279
abstract class JsonStringBuilder {
278280
private String tagsKey;
279281

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package org.opensearch.dataprepper.model.event;
7+
8+
public interface EventFailureMetadata {
9+
EventFailureMetadata with(String key, Object value);
10+
}
11+

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,14 @@
5858
* @since 1.2
5959
*/
6060
public class JacksonEvent implements Event {
61+
class DefaultEventFailureMetadata implements EventFailureMetadata {
62+
static final String FAILURE_METADATA = "_failure_metadata";
63+
64+
public DefaultEventFailureMetadata with(String key, Object value) {
65+
put(FAILURE_METADATA+"/"+key, value);
66+
return this;
67+
}
68+
}
6169

6270
private static final Logger LOG = LoggerFactory.getLogger(JacksonEvent.class);
6371

@@ -204,6 +212,10 @@ public void put(String key, final Object value, final boolean replaceInvalidChar
204212
put(key, value);
205213
}
206214

215+
public EventFailureMetadata updateFailureMetadata() {
216+
return new DefaultEventFailureMetadata();
217+
}
218+
207219
/**
208220
* Adds or updates the key with a given value in the Event.
209221
*

‎data-prepper-api/src/main/java/org/opensearch/dataprepper/model/failures/DlqObject.java‎

Lines changed: 50 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import com.fasterxml.jackson.annotation.JsonIgnore;
1010
import org.apache.commons.lang3.StringUtils;
1111
import org.opensearch.dataprepper.model.event.EventHandle;
12+
import org.opensearch.dataprepper.model.event.Event;
1213
import org.opensearch.dataprepper.model.configuration.PluginSetting;
1314

1415
import java.time.Instant;
@@ -20,14 +21,18 @@
2021

2122
import static com.google.common.base.Preconditions.checkArgument;
2223
import static com.google.common.base.Preconditions.checkNotNull;
24+
import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
25+
26+
import org.slf4j.Logger;
27+
import org.slf4j.LoggerFactory;
2328

2429
/**
2530
* A model representing DLQ objects in Data Prepper
2631
*
2732
* @since 2.2
2833
*/
2934
public class DlqObject {
30-
35+
private static final Logger LOG = LoggerFactory.getLogger(DlqObject.class);
3136
private static final String ISO8601_FORMAT_STRING = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
3237
private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern(ISO8601_FORMAT_STRING)
3338
.withZone(ZoneId.systemDefault());;
@@ -42,14 +47,14 @@ public class DlqObject {
4247

4348
private final String timestamp;
4449

50+
// DLQ object is created either with EventHandles or event.
4551
@JsonIgnore
46-
private final EventHandle eventHandle;
47-
52+
private final Event event;
4853
@JsonIgnore
4954
private final List<EventHandle> eventHandles;
5055

5156
private DlqObject(final String pluginId, final String pluginName, final String pipelineName,
52-
final String timestamp, final Object failedData, final List<EventHandle> eventHandles) {
57+
final String timestamp, final Object failedData, final List<EventHandle> eventHandles, final Event event) {
5358

5459
checkNotNull(pluginId, "pluginId cannot be null");
5560
checkArgument(!pluginId.isEmpty(), "pluginId cannot be an empty string");
@@ -58,13 +63,14 @@ private DlqObject(final String pluginId, final String pluginName, final String p
5863
checkNotNull(pipelineName, "pipelineName cannot be null");
5964
checkArgument(!pipelineName.isEmpty(), "pipelineName cannot be an empty string");
6065
checkNotNull(failedData, "failedData cannot be null");
66+
checkArgument((eventHandles == null || event == null), "Only one of eventhandles and event should be non-null");
6167

6268
this.pluginId = pluginId;
6369
this.pluginName = pluginName;
6470
this.pipelineName = pipelineName;
6571
this.failedData = failedData;
6672
this.eventHandles = eventHandles;
67-
this.eventHandle = null;
73+
this.event = event;
6874

6975
this.timestamp = StringUtils.isEmpty(timestamp) ? FORMATTER.format(Instant.now()) : timestamp;
7076
}
@@ -89,17 +95,31 @@ public String getTimestamp() {
8995
return timestamp;
9096
}
9197

98+
public Event getEvent() {
99+
return event;
100+
}
101+
92102
public List<EventHandle> getEventHandles() {
93103
return eventHandles;
94104
}
95105

96106
public void releaseEventHandle(boolean result) {
97-
if (eventHandles != null && eventHandles.size() == 1) {
107+
if (event != null) {
108+
// This should not happen. DLQ objects with event should be sent to DLQ pipeline and should not be released.
109+
LOG.warn(NOISY, "Attempted to release DLQObject with event");
110+
} else if (eventHandles != null && eventHandles.size() == 1) {
98111
eventHandles.get(0).release(result);
99112
}
100113
}
101114

102115
public void releaseEventHandles(boolean result) {
116+
if (eventHandles == null) {
117+
if (event != null) {
118+
// This should not happen. DLQ objects with event should be sent to DLQ pipeline and should not be released.
119+
LOG.warn(NOISY, "Attempted to release DLQObject with event");
120+
}
121+
return;
122+
}
103123
for (final EventHandle eventHandle: eventHandles) {
104124
eventHandle.release(result);
105125
}
@@ -111,11 +131,12 @@ public boolean equals(final Object o) {
111131
if (o == null || getClass() != o.getClass()) return false;
112132
final DlqObject that = (DlqObject) o;
113133
return Objects.equals(failedData, that.getFailedData())
114-
&& Objects.equals(pluginId, that.pluginId)
115-
&& Objects.equals(pluginName, that.pluginName)
116-
&& Objects.equals(pipelineName, that.pipelineName)
117-
&& Objects.equals(eventHandles, that.eventHandles)
118-
&& Objects.equals(timestamp, that.getTimestamp());
134+
&& Objects.equals(pluginId, that.pluginId)
135+
&& Objects.equals(pluginName, that.pluginName)
136+
&& Objects.equals(pipelineName, that.pipelineName)
137+
&& Objects.equals(eventHandles, that.eventHandles)
138+
&& Objects.equals(event, that.event)
139+
&& Objects.equals(timestamp, that.getTimestamp());
119140
}
120141

121142
@Override
@@ -137,6 +158,18 @@ public String toString() {
137158
public static DlqObject createDlqObject(PluginSetting pluginSetting, List<EventHandle> eventHandles, Object failedData) {
138159
return DlqObject.builder()
139160
.withEventHandles(eventHandles)
161+
.withEvent(null)
162+
.withFailedData(failedData)
163+
.withPluginName(pluginSetting.getName())
164+
.withPipelineName(pluginSetting.getPipelineName())
165+
.withPluginId(pluginSetting.getName())
166+
.build();
167+
}
168+
169+
public static DlqObject createDlqObject(PluginSetting pluginSetting, Event event, Object failedData) {
170+
return DlqObject.builder()
171+
.withEventHandles(null)
172+
.withEvent(event)
140173
.withFailedData(failedData)
141174
.withPluginName(pluginSetting.getName())
142175
.withPipelineName(pluginSetting.getPipelineName())
@@ -155,6 +188,7 @@ public static class Builder {
155188
private String pipelineName;
156189
private Object failedData;
157190
private List<EventHandle> eventHandles;
191+
private Event event;
158192

159193
private String timestamp;
160194

@@ -188,6 +222,10 @@ public Builder withEventHandles(final List<EventHandle> eventHandles) {
188222
return this;
189223
}
190224

225+
public Builder withEvent(final Event event) {
226+
this.event = event;
227+
return this;
228+
}
191229
public Builder withEventHandle(final EventHandle eventHandle) {
192230
this.eventHandles = new ArrayList<>();
193231
this.eventHandles.add(eventHandle);
@@ -200,7 +238,7 @@ public Builder withTimestamp(final Instant instant) {
200238
}
201239

202240
public DlqObject build() {
203-
return new DlqObject(this.pluginId, this.pluginName, this.pipelineName, this.timestamp, this.failedData, this.eventHandles);
241+
return new DlqObject(this.pluginId, this.pluginName, this.pipelineName, this.timestamp, this.failedData, this.eventHandles, this.event);
204242
}
205243

206244
}

‎data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/JacksonEventTest.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,29 @@ void testPutAndGet_withArrays_eventKey() {
186186
assertThat(resultValue, equalTo(newValue));
187187
}
188188

189+
@Test
190+
void testUpdateFailureMetadata() {
191+
Object failureMetadata = event.updateFailureMetadata();
192+
assertThat(failureMetadata, instanceOf(JacksonEvent.DefaultEventFailureMetadata.class));
193+
assertThat(event.get(JacksonEvent.DefaultEventFailureMetadata.FAILURE_METADATA, Map.class), is(nullValue()));
194+
((EventFailureMetadata)failureMetadata).with("key", "value");
195+
assertThat(event.get(JacksonEvent.DefaultEventFailureMetadata.FAILURE_METADATA, Map.class), is(notNullValue()));
196+
}
197+
198+
@Test
199+
public void testDefaultEventFailureMetadata() {
200+
String eventType = UUID.randomUUID().toString();
201+
202+
Event event = JacksonEvent.builder()
203+
.withEventType(eventType)
204+
.build();
205+
206+
EventFailureMetadata eventFailureMetadata = event.updateFailureMetadata();
207+
eventFailureMetadata.with("key1", "value1").with("key2", 2);
208+
assertThat(event.get(JacksonEvent.DefaultEventFailureMetadata.FAILURE_METADATA+"/key1", String.class), equalTo("value1"));
209+
assertThat(event.get(JacksonEvent.DefaultEventFailureMetadata.FAILURE_METADATA+"/key2", Integer.class), equalTo(2));
210+
}
211+
189212
@Test
190213
void testPutAndGet_withArrays_out_of_bounds_on_end_of_list_creates_new_element() {
191214

0 commit comments

Comments
 (0)