Skip to content

Commit 71939cf

Browse files
mcplusamcplusa
committed
Merge pull request #2 from jpdiazvaz/master
Separated CloudWatch logs into separate JSON lines
2 parents c5b389c + af18b58 commit 71939cf

7 files changed

Lines changed: 395 additions & 21 deletions

File tree

‎build.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
<get src= "http://central.maven.org/maven2/org/apache/lucene/lucene-core/4.8.1/lucene-core-4.8.1.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3434
<get src= "http://central.maven.org/maven2/commons-codec/commons-codec/1.10/commons-codec-1.10.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3535
<get src= "http://central.maven.org/maven2/com/ning/async-http-client/1.9.30/async-http-client-1.9.30.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
36+
<get src= "http://central.maven.org/maven2/com/google/code/gson/gson/2.3.1/gson-2.3.1.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3637

3738
<get src= "https://hamcrest.googlecode.com/files/hamcrest-core-1.3.jar" dest="${external.dir}/lib" usetimestamp="true" verbose="true"/>
3839

‎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: 81 additions & 11 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,18 +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;
30+
import com.google.gson.Gson;
31+
2032
/**
21-
* 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
2234
* usable for insertions to Sumologic.
2335
*/
24-
public class CloudWatchMessageModelSumologicTransformer implements
25-
SumologicTransformer<SimpleKinesisMessageModel> {
36+
public class CloudWatchMessageModelSumologicTransformer
37+
implements SumologicTransformer<CloudWatchLogsMessageModel> {
2638

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

2943
/**
3044
* Creates a new KinesisMessageModelSumologicTransformer.
@@ -34,24 +48,70 @@ public CloudWatchMessageModelSumologicTransformer() {
3448
}
3549

3650
@Override
37-
public String fromClass(SimpleKinesisMessageModel message) {
38-
return message.toString();
39-
}
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()));
4068

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+
4184
@Override
42-
public SimpleKinesisMessageModel toClass(Record record) throws IOException {
85+
public CloudWatchLogsMessageModel toClass(Record record) {
4386
byte[] decodedRecord = record.getData().array();
4487
String stringifiedRecord = decompressGzip(decodedRecord);
4588

4689
if (stringifiedRecord == null) {
47-
LOG.error("Unable to decompress the record: "+new String(record.getData().array()));
48-
LOG.error("Not attempting to transform into a Message Model");
90+
LOG.error("Unable to decompress the record: "+new String(record.getData().array())
91+
+"\nNot attempting to transform into a Message Model");
4992
return null;
5093
}
94+
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());
100+
}
101+
record.setData(bufferedData);
51102

52-
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+
53113
}
54-
114+
55115
public static String decompressGzip(byte[] compressedData) {
56116
try {
57117
GZIPInputStream gis = new GZIPInputStream(new ByteArrayInputStream(compressedData));
@@ -83,4 +143,14 @@ public static String byteBufferToString(ByteBuffer buffer){
83143
return data;
84144
}
85145

146+
private static final Gson gson = new Gson();
147+
public static boolean verifyJSON(String json) {
148+
try {
149+
gson.fromJson(json, Object.class);
150+
return true;
151+
} catch(com.google.gson.JsonSyntaxException ex) {
152+
return false;
153+
}
154+
}
155+
86156
}

‎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+
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ public boolean sendToSumologic(String data) throws IOException{
4848
BoundRequestBuilder builder = null;
4949
builder = this.clientPreparePost(url);
5050

51-
byte[] compressedData = compressGzip(data);
51+
byte[] compressedData = SumologicSender.compressGzip(data);
5252

5353
builder.setHeader("Content-Encoding", "gzip");
5454
builder.setBody(compressedData);
@@ -82,7 +82,7 @@ public boolean sendToSumologic(String data) throws IOException{
8282
}
8383
}
8484

85-
public byte[] compressGzip(String data) {
85+
public static byte[] compressGzip(String data) {
8686
if (data == null || data.length() == 0) {
8787
return null;
8888
}

0 commit comments

Comments
 (0)