Skip to content

Commit af18b58

Browse files
Juan Pablo Diaz-VazJuan Pablo Diaz-Vaz
Juan Pablo Diaz-Vaz
authored and
Juan Pablo Diaz-Vaz
committed
SUMOK-26 SUMOK-27 CloudWatch logs decomposed and header added
1 parent 8b4c87d commit af18b58

4 files changed

Lines changed: 334 additions & 50 deletions

File tree

‎src/main/java/com/sumologic/client/CloudWatchLogsMessageModel.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package com.sumologic.client;
2+
3+
import java.util.ArrayList;
4+
import java.util.HashMap;
5+
import java.util.List;
6+
import java.util.Map;
7+
8+
import org.apache.commons.lang.builder.ToStringBuilder;
9+
10+
import com.sumologic.client.LogEvent;
11+
import com.fasterxml.jackson.annotation.JsonAnyGetter;
12+
import com.fasterxml.jackson.annotation.JsonAnySetter;
13+
import com.fasterxml.jackson.annotation.JsonIgnore;
14+
import com.fasterxml.jackson.annotation.JsonInclude;
15+
import com.fasterxml.jackson.annotation.JsonProperty;
16+
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
17+
18+
@JsonInclude(JsonInclude.Include.NON_NULL)
19+
@JsonPropertyOrder({
20+
"logEvents",
21+
"logGroup",
22+
"logStream",
23+
"messageType",
24+
"owner",
25+
"subscriptionFilters"
26+
})
27+
28+
public class CloudWatchLogsMessageModel {
29+
30+
@JsonProperty("logEvents")
31+
private List<LogEvent> logEvents = new ArrayList<LogEvent>();
32+
@JsonProperty("logGroup")
33+
private String logGroup;
34+
@JsonProperty("logStream")
35+
private String logStream;
36+
@JsonProperty("messageType")
37+
private String messageType;
38+
@JsonProperty("owner")
39+
private String owner;
40+
@JsonProperty("subscriptionFilters")
41+
private List<String> subscriptionFilters = new ArrayList<String>();
42+
@JsonIgnore
43+
private Map<String, Object> additionalProperties = new HashMap<String, Object>();
44+
45+
@JsonProperty("logEvents")
46+
public List<LogEvent> getLogEvents() {
47+
return logEvents;
48+
}
49+
50+
@JsonProperty("logEvents")
51+
public void setLogEvents(List<LogEvent> logEvents) {
52+
this.logEvents = logEvents;
53+
}
54+
55+
@JsonProperty("logGroup")
56+
public String getLogGroup() {
57+
return logGroup;
58+
}
59+
60+
@JsonProperty("logGroup")
61+
public void setLogGroup(String logGroup) {
62+
this.logGroup = logGroup;
63+
}
64+
65+
@JsonProperty("logStream")
66+
public String getLogStream() {
67+
return logStream;
68+
}
69+
70+
@JsonProperty("logStream")
71+
public void setLogStream(String logStream) {
72+
this.logStream = logStream;
73+
}
74+
75+
@JsonProperty("messageType")
76+
public String getMessageType() {
77+
return messageType;
78+
}
79+
80+
@JsonProperty("messageType")
81+
public void setMessageType(String messageType) {
82+
this.messageType = messageType;
83+
}
84+
85+
@JsonProperty("owner")
86+
public String getOwner() {
87+
return owner;
88+
}
89+
90+
@JsonProperty("owner")
91+
public void setOwner(String owner) {
92+
this.owner = owner;
93+
}
94+
95+
@JsonProperty("subscriptionFilters")
96+
public List<String> getSubscriptionFilters() {
97+
return subscriptionFilters;
98+
}
99+
100+
@JsonProperty("subscriptionFilters")
101+
public void setSubscriptionFilters(List<String> subscriptionFilters) {
102+
this.subscriptionFilters = subscriptionFilters;
103+
}
104+
105+
@JsonAnyGetter
106+
public Map<String, Object> getAdditionalProperties() {
107+
return this.additionalProperties;
108+
}
109+
110+
@JsonAnySetter
111+
public void setAdditionalProperty(String name, Object value) {
112+
this.additionalProperties.put(name, value);
113+
}
114+
115+
@Override
116+
public String toString() {
117+
return ToStringBuilder.reflectionToString(this);
118+
}
119+
}

‎src/main/java/com/sumologic/client/CloudWatchMessageModelSumologicTransformer.java‎

Lines changed: 65 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,11 @@
22

33
import java.io.IOException;
44

5+
import com.amazonaws.services.kinesis.connectors.BasicJsonTransformer;
56
import com.amazonaws.services.kinesis.model.Record;
7+
import com.amazonaws.util.json.JSONArray;
8+
import com.amazonaws.util.json.JSONException;
9+
import com.amazonaws.util.json.JSONObject;
610
import com.sumologic.client.SimpleKinesisMessageModel;
711
import com.sumologic.client.implementations.SumologicTransformer;
812

@@ -13,20 +17,28 @@
1317
import java.io.BufferedReader;
1418
import java.io.InputStreamReader;
1519
import java.nio.ByteBuffer;
20+
import java.nio.CharBuffer;
21+
import java.nio.charset.CharacterCodingException;
1622
import java.nio.charset.Charset;
1723
import java.nio.charset.CharsetDecoder;
24+
import java.nio.charset.CharsetEncoder;
25+
import java.util.List;
1826
import java.util.zip.GZIPInputStream;
1927

28+
import com.fasterxml.jackson.databind.JsonMappingException;
29+
import com.fasterxml.jackson.databind.ObjectMapper;
2030
import com.google.gson.Gson;
2131

2232
/**
23-
* A custom transfomer for {@link SimpleKinesisMessageModel} records in JSON format. The output is in a format
33+
* A custom transfomer for {@link CloudWatchLogsMessageModel} records in JSON format. The output is in a format
2434
* usable for insertions to Sumologic.
2535
*/
26-
public class CloudWatchMessageModelSumologicTransformer implements
27-
SumologicTransformer<SimpleKinesisMessageModel> {
36+
public class CloudWatchMessageModelSumologicTransformer
37+
implements SumologicTransformer<CloudWatchLogsMessageModel> {
2838

2939
private static final Log LOG = LogFactory.getLog(CloudWatchMessageModelSumologicTransformer.class);
40+
41+
private static CharsetEncoder encoder = Charset.forName("UTF-8").newEncoder();
3042

3143
/**
3244
* Creates a new KinesisMessageModelSumologicTransformer.
@@ -36,12 +48,41 @@ public CloudWatchMessageModelSumologicTransformer() {
3648
}
3749

3850
@Override
39-
public String fromClass(SimpleKinesisMessageModel message) {
40-
return message.toString();
41-
}
51+
public String fromClass(CloudWatchLogsMessageModel message) {
52+
String jsonMessage = "";
53+
JSONObject outputObject;
54+
55+
List<LogEvent> logEvents = message.getLogEvents();
56+
int logEventsSize = logEvents.size();
57+
for (int i=0;i<logEventsSize;i++) {
58+
LogEvent log = logEvents.get(i);
59+
60+
outputObject = new JSONObject();
61+
try {
62+
// Header
63+
outputObject.put("logGroup", message.getLogGroup());
64+
outputObject.put("logStream", message.getLogStream());
65+
outputObject.put("messageType", message.getMessageType());
66+
outputObject.put("owner", message.getOwner());
67+
outputObject.put("subscriptionFilters", new JSONArray(message.getSubscriptionFilters()));
4268

69+
// Body
70+
outputObject.put("id", log.getId());
71+
outputObject.put("message", log.getMessage());
72+
outputObject.put("timestamp", log.getTimestamp());
73+
} catch (JSONException e) {
74+
LOG.error("Unable to convert message into JSON String: "+e.getMessage());
75+
}
76+
jsonMessage += outputObject.toString();
77+
if (i < logEventsSize - 1) {
78+
jsonMessage += '\n';
79+
}
80+
}
81+
return jsonMessage;
82+
}
83+
4384
@Override
44-
public SimpleKinesisMessageModel toClass(Record record) throws IOException {
85+
public CloudWatchLogsMessageModel toClass(Record record) {
4586
byte[] decodedRecord = record.getData().array();
4687
String stringifiedRecord = decompressGzip(decodedRecord);
4788

@@ -51,15 +92,26 @@ public SimpleKinesisMessageModel toClass(Record record) throws IOException {
5192
return null;
5293
}
5394

54-
if (!verifyJSON(stringifiedRecord)) {
55-
LOG.error("The record is not a valid JSON: "+stringifiedRecord
56-
+"\nNot attempting to transform into a Message Model");
57-
return null;
95+
ByteBuffer bufferedData = null;
96+
try {
97+
bufferedData = encoder.encode(CharBuffer.wrap(stringifiedRecord));
98+
} catch (CharacterCodingException e) {
99+
LOG.error("Unable to set the decompressed Record for serializing "+e.getMessage());
58100
}
101+
record.setData(bufferedData);
59102

60-
return new SimpleKinesisMessageModel(stringifiedRecord);
103+
try {
104+
return new ObjectMapper().readValue(
105+
record.getData().array(),
106+
CloudWatchLogsMessageModel.class);
107+
} catch (IOException e) {
108+
LOG.error("Unable to convert the Record into a POJO: "+stringifiedRecord
109+
+"\nerror: "+e.getMessage());
110+
}
111+
return null;
112+
61113
}
62-
114+
63115
public static String decompressGzip(byte[] compressedData) {
64116
try {
65117
GZIPInputStream gis = new GZIPInputStream(new ByteArrayInputStream(compressedData));

‎src/main/java/com/sumologic/client/LogEvent.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
package com.sumologic.client;
2+
3+
import java.util.HashMap;
4+
import java.util.Map;
5+
6+
import com.fasterxml.jackson.annotation.JsonAnyGetter;
7+
import com.fasterxml.jackson.annotation.JsonAnySetter;
8+
import com.fasterxml.jackson.annotation.JsonIgnore;
9+
import com.fasterxml.jackson.annotation.JsonInclude;
10+
import com.fasterxml.jackson.annotation.JsonProperty;
11+
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
12+
13+
@JsonInclude(JsonInclude.Include.NON_NULL)
14+
@JsonPropertyOrder({
15+
"id",
16+
"message",
17+
"timestamp"
18+
})
19+
20+
public class LogEvent {
21+
22+
@JsonProperty("id")
23+
private String id;
24+
@JsonProperty("message")
25+
private String message;
26+
@JsonProperty("timestamp")
27+
private Long timestamp;
28+
@JsonIgnore
29+
private Map<String, Object> additionalProperties = new HashMap<String, Object>();
30+
31+
@JsonProperty("id")
32+
public String getId() {
33+
return id;
34+
}
35+
36+
@JsonProperty("id")
37+
public void setId(String id) {
38+
this.id = id;
39+
}
40+
41+
@JsonProperty("message")
42+
public String getMessage() {
43+
return message;
44+
}
45+
46+
@JsonProperty("message")
47+
public void setMessage(String message) {
48+
this.message = message;
49+
}
50+
51+
@JsonProperty("timestamp")
52+
public Long getTimestamp() {
53+
return timestamp;
54+
}
55+
56+
@JsonProperty("timestamp")
57+
public void setTimestamp(Long timestamp) {
58+
this.timestamp = timestamp;
59+
}
60+
61+
@JsonAnyGetter
62+
public Map<String, Object> getAdditionalProperties() {
63+
return this.additionalProperties;
64+
}
65+
66+
@JsonAnySetter
67+
public void setAdditionalProperty(String name, Object value) {
68+
this.additionalProperties.put(name, value);
69+
}
70+
71+
}

0 commit comments

Comments
 (0)