forked from opensearch-project/data-prepper
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ENH: peer forwarding codec and model (opensearch-project#2256)
Signed-off-by: George Chen <[email protected]>
- Loading branch information
1 parent
a5e51a3
commit 5bc6a2c
Showing
17 changed files
with
523 additions
and
254 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
85 changes: 85 additions & 0 deletions
85
...c/main/java/org/opensearch/dataprepper/peerforwarder/codec/JacksonPeerForwarderCodec.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package org.opensearch.dataprepper.peerforwarder.codec; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.opensearch.dataprepper.model.event.DefaultEventMetadata; | ||
| import org.opensearch.dataprepper.model.event.Event; | ||
| import org.opensearch.dataprepper.model.event.JacksonEvent; | ||
| import org.opensearch.dataprepper.model.trace.JacksonSpan; | ||
| import org.opensearch.dataprepper.peerforwarder.model.PeerForwardingEvents; | ||
| import org.opensearch.dataprepper.peerforwarder.model.WireEvent; | ||
| import org.opensearch.dataprepper.peerforwarder.model.WireEvents; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| public class JacksonPeerForwarderCodec implements PeerForwarderCodec { | ||
| private static final String TRACE_EVENT_TYPE = "TRACE"; | ||
|
|
||
| private final ObjectMapper objectMapper; | ||
|
|
||
| public JacksonPeerForwarderCodec(final ObjectMapper objectMapper) { | ||
| this.objectMapper = objectMapper; | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] serialize(final PeerForwardingEvents peerForwardingEvents) throws IOException { | ||
| final WireEvents wireEvents = fromPeerForwardingEventsToWireEvents(peerForwardingEvents); | ||
| return objectMapper.writeValueAsBytes(wireEvents); | ||
| } | ||
|
|
||
| @Override | ||
| public PeerForwardingEvents deserialize(final byte[] bytes) throws IOException { | ||
| final WireEvents wireEvents = objectMapper.readValue(bytes, WireEvents.class); | ||
| return fromWireEventsToPeerForwardingEvents(wireEvents); | ||
| } | ||
|
|
||
| private WireEvents fromPeerForwardingEventsToWireEvents(final PeerForwardingEvents peerForwardingEvents) { | ||
| List<WireEvent> wireEventList = null; | ||
| if (peerForwardingEvents.getEvents() != null) { | ||
| wireEventList = peerForwardingEvents.getEvents().stream().map(event -> new WireEvent( | ||
| event.getMetadata().getEventType(), | ||
| event.getMetadata().getTimeReceived(), | ||
| event.getMetadata().getAttributes(), | ||
| event.toJsonString() | ||
| )).collect(Collectors.toList()); | ||
| } | ||
| return new WireEvents(wireEventList, | ||
| peerForwardingEvents.getDestinationPluginId(), peerForwardingEvents.getDestinationPipelineName()); | ||
| } | ||
|
|
||
| private PeerForwardingEvents fromWireEventsToPeerForwardingEvents(final WireEvents wireEvents) { | ||
| List<Event> eventList = null; | ||
| if (wireEvents.getEvents() != null) { | ||
| eventList = wireEvents.getEvents().stream().map(this::transformWireEvent).collect(Collectors.toList()); | ||
| } | ||
| return new PeerForwardingEvents( | ||
| eventList, wireEvents.getDestinationPluginId(), wireEvents.getDestinationPipelineName()); | ||
| } | ||
|
|
||
| private Event transformWireEvent(final WireEvent wireEvent) { | ||
| final DefaultEventMetadata eventMetadata = getEventMetadata(wireEvent); | ||
| Event event; | ||
|
|
||
| if (wireEvent.getEventType().equalsIgnoreCase(TRACE_EVENT_TYPE)) { | ||
| event = JacksonSpan.builder() | ||
| .withJsonData(wireEvent.getEventData()) | ||
| .withEventMetadata(eventMetadata) | ||
| .build(); | ||
| } else { | ||
| event = JacksonEvent.builder() | ||
| .withData(wireEvent.getEventData()) | ||
| .withEventMetadata(eventMetadata) | ||
| .build(); | ||
| } | ||
| return event; | ||
| } | ||
|
|
||
| private DefaultEventMetadata getEventMetadata(final WireEvent wireEvent) { | ||
| return DefaultEventMetadata.builder() | ||
| .withEventType(wireEvent.getEventType()) | ||
| .withTimeReceived(wireEvent.getEventTimeReceived()) | ||
| .withAttributes(wireEvent.getEventAttributes()) | ||
| .build(); | ||
| } | ||
| } |
30 changes: 30 additions & 0 deletions
30
.../src/main/java/org/opensearch/dataprepper/peerforwarder/codec/JavaPeerForwarderCodec.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package org.opensearch.dataprepper.peerforwarder.codec; | ||
|
|
||
| import org.opensearch.dataprepper.peerforwarder.model.PeerForwardingEvents; | ||
|
|
||
| import java.io.ByteArrayInputStream; | ||
| import java.io.ByteArrayOutputStream; | ||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.io.ObjectInputStream; | ||
| import java.io.ObjectOutputStream; | ||
|
|
||
| public class JavaPeerForwarderCodec implements PeerForwarderCodec { | ||
|
|
||
| @Override | ||
| public byte[] serialize(final PeerForwardingEvents events) throws IOException { | ||
| try (final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); | ||
| final ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream)) { | ||
| objectOutputStream.writeObject(events); | ||
| return byteArrayOutputStream.toByteArray(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public PeerForwardingEvents deserialize(final byte[] bytes) throws IOException, ClassNotFoundException { | ||
| try (final InputStream inputStream = new ByteArrayInputStream(bytes); | ||
| final ObjectInputStream objectInputStream = new ObjectInputStream(inputStream)) { | ||
| return (PeerForwardingEvents) objectInputStream.readObject(); | ||
| } | ||
| } | ||
| } |
9 changes: 9 additions & 0 deletions
9
...core/src/main/java/org/opensearch/dataprepper/peerforwarder/codec/PeerForwarderCodec.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| package org.opensearch.dataprepper.peerforwarder.codec; | ||
|
|
||
| import org.opensearch.dataprepper.peerforwarder.model.PeerForwardingEvents; | ||
|
|
||
| public interface PeerForwarderCodec { | ||
| byte[] serialize(PeerForwardingEvents peerForwardingEvents) throws Exception; | ||
|
|
||
| PeerForwardingEvents deserialize(byte[] bytes) throws Exception; | ||
| } |
44 changes: 44 additions & 0 deletions
44
...re/src/main/java/org/opensearch/dataprepper/peerforwarder/model/PeerForwardingEvents.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.dataprepper.peerforwarder.model; | ||
|
|
||
| import org.opensearch.dataprepper.model.event.Event; | ||
| import org.opensearch.dataprepper.peerforwarder.PeerForwarder; | ||
|
|
||
| import java.io.Serializable; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * A class to serialize list of {@link Event} and destination plugin ID used by {@link PeerForwarder} | ||
| * | ||
| * @since 2.0 | ||
| */ | ||
| public class PeerForwardingEvents implements Serializable { | ||
| private List<Event> events; | ||
| private String destinationPluginId; | ||
| private String destinationPipelineName; | ||
|
|
||
| public PeerForwardingEvents() { | ||
| } | ||
|
|
||
| public PeerForwardingEvents(final List<Event> events, final String destinationPluginId, final String destinationPipelineName) { | ||
| this.events = events; | ||
| this.destinationPluginId = destinationPluginId; | ||
| this.destinationPipelineName = destinationPipelineName; | ||
| } | ||
|
|
||
| public List<Event> getEvents() { | ||
| return events; | ||
| } | ||
|
|
||
| public String getDestinationPluginId() { | ||
| return destinationPluginId; | ||
| } | ||
|
|
||
| public String getDestinationPipelineName() { | ||
| return destinationPipelineName; | ||
| } | ||
| } |
Oops, something went wrong.